bambuddy/backend/app/api/routes/metrics.py

"""Prometheus metrics endpoint for external monitoring."""

from fastapi import APIRouter, Depends, Header, HTTPException, Response
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.database import get_db
from backend.app.models.archive import PrintArchive
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.models.settings import Settings
from backend.app.services.printer_manager import printer_manager, supports_chamber_temp

router = APIRouter(tags=["metrics"])


async def get_prometheus_settings(db: AsyncSession) -> tuple[bool, str]:
    """Get Prometheus settings from database."""
    result = await db.execute(select(Settings).where(Settings.key.in_(["prometheus_enabled", "prometheus_token"])))
    settings_dict = {s.key: s.value for s in result.scalars().all()}

    enabled = settings_dict.get("prometheus_enabled", "false").lower() == "true"
    token = settings_dict.get("prometheus_token", "")
    return enabled, token


def format_labels(**labels: str) -> str:
    """Format label key-value pairs for Prometheus."""
    if not labels:
        return ""
    pairs = [f'{k}="{v}"' for k, v in labels.items() if v is not None]
    return "{" + ",".join(pairs) + "}"


def state_to_numeric(state: str) -> int:
    """Convert printer state string to numeric value."""
    state_map = {
        "unknown": 0,
        "IDLE": 1,
        "RUNNING": 2,
        "PAUSE": 3,
        "FINISH": 4,
        "FAILED": 5,
        "PREPARE": 6,
        "SLICING": 7,
    }
    return state_map.get(state, 0)


@router.get("/metrics", response_class=Response)
async def get_metrics(
    db: AsyncSession = Depends(get_db),
    authorization: str | None = Header(None),
):
    """
    Prometheus metrics endpoint.

    Returns metrics in Prometheus text exposition format.
    Requires prometheus_enabled setting to be true.
    If prometheus_token is set, requires Bearer token authentication.
    """
    # Check if enabled
    enabled, token = await get_prometheus_settings(db)

    if not enabled:
        raise HTTPException(status_code=404, detail="Prometheus metrics not enabled")

    # Check authentication if token is set
    if token:
        if not authorization:
            raise HTTPException(status_code=401, detail="Authorization required")
        if not authorization.startswith("Bearer "):
            raise HTTPException(status_code=401, detail="Bearer token required")
        provided_token = authorization[7:]  # Remove "Bearer " prefix
        if provided_token != token:
            raise HTTPException(status_code=401, detail="Invalid token")

    lines: list[str] = []

    # =========================================================================
    # Printer metrics
    # =========================================================================

    # Get all printers from DB
    result = await db.execute(select(Printer).where(Printer.is_active == True))  # noqa: E712
    printers = list(result.scalars().all())

    # Build lookup for printer info
    printer_info = {p.id: p for p in printers}

    # Get all connected printer statuses
    all_statuses = printer_manager.get_all_statuses()

    # Printer connection status
    lines.append("# HELP bambuddy_printer_connected Printer connection status (1=connected, 0=disconnected)")
    lines.append("# TYPE bambuddy_printer_connected gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        connected = 1 if status and status.connected else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
            model=printer.model or "unknown",
        )
        lines.append(f"bambuddy_printer_connected{labels} {connected}")

    # Printer state
    lines.append("")
    lines.append(
        "# HELP bambuddy_printer_state Printer state (0=unknown, 1=idle, 2=running, 3=pause, 4=finish, 5=failed, 6=prepare, 7=slicing)"
    )
    lines.append("# TYPE bambuddy_printer_state gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        state_val = state_to_numeric(status.state) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_printer_state{labels} {state_val}")

    # Print progress
    lines.append("")
    lines.append("# HELP bambuddy_print_progress Current print progress (0-100)")
    lines.append("# TYPE bambuddy_print_progress gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        progress = status.progress if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_print_progress{labels} {progress:.1f}")

    # Remaining time
    lines.append("")
    lines.append("# HELP bambuddy_print_remaining_seconds Estimated remaining print time in seconds")
    lines.append("# TYPE bambuddy_print_remaining_seconds gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        remaining = status.remaining_time * 60 if status else 0  # Convert minutes to seconds
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_print_remaining_seconds{labels} {remaining}")

    # Layer progress
    lines.append("")
    lines.append("# HELP bambuddy_print_layer_current Current layer number")
    lines.append("# TYPE bambuddy_print_layer_current gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        layer = status.layer_num if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_print_layer_current{labels} {layer}")

    lines.append("")
    lines.append("# HELP bambuddy_print_layer_total Total layers in current print")
    lines.append("# TYPE bambuddy_print_layer_total gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        total = status.total_layers if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_print_layer_total{labels} {total}")

    # =========================================================================
    # Temperature metrics
    # =========================================================================

    lines.append("")
    lines.append("# HELP bambuddy_bed_temp_celsius Current bed temperature")
    lines.append("# TYPE bambuddy_bed_temp_celsius gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        temp = status.temperatures.get("bed", 0) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_bed_temp_celsius{labels} {temp:.1f}")

    lines.append("")
    lines.append("# HELP bambuddy_bed_target_celsius Target bed temperature")
    lines.append("# TYPE bambuddy_bed_target_celsius gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        temp = status.temperatures.get("bed_target", 0) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_bed_target_celsius{labels} {temp:.1f}")

    lines.append("")
    lines.append("# HELP bambuddy_nozzle_temp_celsius Current nozzle temperature")
    lines.append("# TYPE bambuddy_nozzle_temp_celsius gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        # Primary nozzle
        temp = status.temperatures.get("nozzle", 0) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
            nozzle="0",
        )
        lines.append(f"bambuddy_nozzle_temp_celsius{labels} {temp:.1f}")
        # Second nozzle if present
        if status and "nozzle_2" in status.temperatures:
            temp2 = status.temperatures.get("nozzle_2", 0)
            labels2 = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
                nozzle="1",
            )
            lines.append(f"bambuddy_nozzle_temp_celsius{labels2} {temp2:.1f}")

    lines.append("")
    lines.append("# HELP bambuddy_nozzle_target_celsius Target nozzle temperature")
    lines.append("# TYPE bambuddy_nozzle_target_celsius gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        temp = status.temperatures.get("nozzle_target", 0) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
            nozzle="0",
        )
        lines.append(f"bambuddy_nozzle_target_celsius{labels} {temp:.1f}")
        if status and "nozzle_2_target" in status.temperatures:
            temp2 = status.temperatures.get("nozzle_2_target", 0)
            labels2 = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
                nozzle="1",
            )
            lines.append(f"bambuddy_nozzle_target_celsius{labels2} {temp2:.1f}")

    lines.append("")
    lines.append(
        "# HELP bambuddy_chamber_temp_celsius Current chamber temperature (only for models with chamber sensor)"
    )
    lines.append("# TYPE bambuddy_chamber_temp_celsius gauge")
    for printer in printers:
        # Only report chamber temp for models that have a real sensor
        if not supports_chamber_temp(printer.model):
            continue
        status = all_statuses.get(printer.id)
        temp = status.temperatures.get("chamber", 0) if status else 0
        labels = format_labels(
            printer_id=str(printer.id),
            printer_name=printer.name,
            serial=printer.serial_number,
        )
        lines.append(f"bambuddy_chamber_temp_celsius{labels} {temp:.1f}")

    # =========================================================================
    # Fan speeds
    # =========================================================================

    lines.append("")
    lines.append("# HELP bambuddy_fan_speed_percent Fan speed percentage")
    lines.append("# TYPE bambuddy_fan_speed_percent gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        if not status:
            continue
        # Part cooling fan
        if "part_fan" in status.temperatures:
            val = status.temperatures["part_fan"]
            labels = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
                fan="part",
            )
            lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}")
        # Aux fan
        if "aux_fan" in status.temperatures:
            val = status.temperatures["aux_fan"]
            labels = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
                fan="aux",
            )
            lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}")
        # Chamber fan
        if "chamber_fan" in status.temperatures:
            val = status.temperatures["chamber_fan"]
            labels = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
                fan="chamber",
            )
            lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}")

    # =========================================================================
    # WiFi signal
    # =========================================================================

    lines.append("")
    lines.append("# HELP bambuddy_wifi_signal_dbm WiFi signal strength in dBm")
    lines.append("# TYPE bambuddy_wifi_signal_dbm gauge")
    for printer in printers:
        status = all_statuses.get(printer.id)
        if status and status.wifi_signal is not None:
            labels = format_labels(
                printer_id=str(printer.id),
                printer_name=printer.name,
                serial=printer.serial_number,
            )
            lines.append(f"bambuddy_wifi_signal_dbm{labels} {status.wifi_signal}")

    # =========================================================================
    # Print statistics (from database)
    # =========================================================================

    # Total prints by status
    lines.append("")
    lines.append("# HELP bambuddy_prints_total Total number of prints by result")
    lines.append("# TYPE bambuddy_prints_total counter")
    result = await db.execute(select(PrintArchive.status, func.count(PrintArchive.id)).group_by(PrintArchive.status))
    for print_result, count in result.all():
        result_label = print_result or "unknown"
        labels = format_labels(result=result_label)
        lines.append(f"bambuddy_prints_total{labels} {count}")

    # Total prints per printer
    lines.append("")
    lines.append("# HELP bambuddy_printer_prints_total Total prints per printer")
    lines.append("# TYPE bambuddy_printer_prints_total counter")
    result = await db.execute(
        select(PrintArchive.printer_id, func.count(PrintArchive.id)).group_by(PrintArchive.printer_id)
    )
    for printer_id, count in result.all():
        if printer_id and printer_id in printer_info:
            p = printer_info[printer_id]
            labels = format_labels(
                printer_id=str(printer_id),
                printer_name=p.name,
                serial=p.serial_number,
            )
            lines.append(f"bambuddy_printer_prints_total{labels} {count}")

    # Total filament used (multiply by quantity to account for multiple items printed)
    lines.append("")
    lines.append("# HELP bambuddy_filament_used_grams Total filament used in grams")
    lines.append("# TYPE bambuddy_filament_used_grams counter")
    result = await db.execute(
        select(func.coalesce(func.sum(PrintArchive.filament_used_grams * func.coalesce(PrintArchive.quantity, 1)), 0))
    )
    total_filament = result.scalar() or 0
    lines.append(f"bambuddy_filament_used_grams {total_filament:.1f}")

    # Total print time
    lines.append("")
    lines.append("# HELP bambuddy_print_time_seconds Total print time in seconds")
    lines.append("# TYPE bambuddy_print_time_seconds counter")
    result = await db.execute(select(func.coalesce(func.sum(PrintArchive.print_time_seconds), 0)))
    total_time = result.scalar() or 0
    lines.append(f"bambuddy_print_time_seconds {total_time}")

    # =========================================================================
    # Queue metrics
    # =========================================================================

    lines.append("")
    lines.append("# HELP bambuddy_queue_pending Number of pending queue items")
    lines.append("# TYPE bambuddy_queue_pending gauge")
    result = await db.execute(select(func.count(PrintQueueItem.id)).where(PrintQueueItem.status == "pending"))
    pending_count = result.scalar() or 0
    lines.append(f"bambuddy_queue_pending {pending_count}")

    lines.append("")
    lines.append("# HELP bambuddy_queue_printing Number of currently printing queue items")
    lines.append("# TYPE bambuddy_queue_printing gauge")
    result = await db.execute(select(func.count(PrintQueueItem.id)).where(PrintQueueItem.status == "printing"))
    printing_count = result.scalar() or 0
    lines.append(f"bambuddy_queue_printing {printing_count}")

    # =========================================================================
    # System metrics
    # =========================================================================

    lines.append("")
    lines.append("# HELP bambuddy_printers_connected Number of connected printers")
    lines.append("# TYPE bambuddy_printers_connected gauge")
    connected_count = sum(1 for s in all_statuses.values() if s.connected)
    lines.append(f"bambuddy_printers_connected {connected_count}")

    lines.append("")
    lines.append("# HELP bambuddy_printers_total Total number of configured printers")
    lines.append("# TYPE bambuddy_printers_total gauge")
    lines.append(f"bambuddy_printers_total {len(printers)}")

    # Add trailing newline
    lines.append("")

    content = "\n".join(lines)
    return Response(content=content, media_type="text/plain; version=0.0.4; charset=utf-8")