feat(mcp): add get_printer_status tool

bambu_run/mcp/status.py

@@ -0,0 +1,139 @@
+"""get_printer_status, get_metrics_history, get_hardware_health -- the
+read_printer capability tools."""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import timedelta
+
+from django.db.models import Avg
+from django.utils import timezone
+
+from bambu_run.mcp.shapes import classify_alerts, freshness, resolve_printer, smoothed_remaining_percent
+from bambu_run.models import Hotend, PrinterMetrics
+from bambu_run.utils import strip_color_padding
+
+ALERT_WINDOW_HOURS = 24
+
+
+def get_printer_status(printer: str | int | None = None) -> dict:
+    """Current live status of one or all printers: temps, print progress, AMS slots, and alerts.
+
+    Pass a printer id or exact name to scope to one printer; omit to get every
+    active printer. Numeric percentages under ams_units[].slots[] are smoothed
+    (median of the 5 most recent readings) and paired with a remain_status
+    field ("ok"/"low"/"empty"/"unread") -- treat "unread" as "no RFID reading
+    yet", not as "empty". alerts are frequency-classified, not raw fault
+    codes; severity is always "unknown" until Bambu's HMS codes are mapped,
+    so never assert the printer is fine or faulty purely from an alert entry.
+    """
+    now = timezone.now()
+    printers = resolve_printer(printer)
+
+    entries = []
+    for p in printers:
+        metric = PrinterMetrics.objects.filter(device=p).order_by("-timestamp").first()
+        if metric is None:
+            entries.append({
+                "id": p.id, "name": p.name, "model": p.model, "location": p.location,
+                "data_age_seconds": None, "stale": True, "state": "no_data",
+                "job": None, "temperatures": None, "ams_units": [], "alerts": [],
+                "fans": None, "chamber_light": None,
+            })
+            continue
+
+        entries.append({
+            "id": p.id,
+            "name": p.name,
+            "model": p.model,
+            "location": p.location,
+            **freshness(metric.timestamp, now),
+            "state": metric.gcode_state or "UNKNOWN",
+            "job": _job_info(metric, now),
+            "temperatures": _temperatures(metric),
+            "ams_units": _ams_units(p, metric),
+            "alerts": _alerts_for_printer(p, now),
+            "fans": {
+                "cooling_percent": metric.cooling_fan_speed,
+                "aux_percent": metric.big_fan1_speed,
+                "chamber_percent": metric.big_fan2_speed,
+            },
+            "chamber_light": metric.chamber_light,
+        })
+
+    return {"generated_at": now.isoformat(), "printers": entries}
+
+
+def _job_info(metric, now):
+    if metric.gcode_state != "RUNNING":
+        return None
+    remaining = metric.remaining_time_min
+    finish_at = (now + timedelta(minutes=remaining)).isoformat() if remaining is not None else None
+    return {
+        "name": metric.subtask_name,
+        "print_percent": metric.print_percent,
+        "layer_current": metric.layer_num,
+        "layer_total": metric.total_layer_num,
+        "remaining_time_min": remaining,
+        "estimated_finish_at": finish_at,
+    }
+
+
+def _temperatures(metric):
+    def f(v):
+        return float(v) if v is not None else None
+    return {
+        "nozzle_c": f(metric.nozzle_temp), "nozzle_target_c": f(metric.nozzle_target_temp),
+        "nozzle_left_c": f(metric.nozzle_temp_left), "nozzle_left_target_c": f(metric.nozzle_target_temp_left),
+        "bed_c": f(metric.bed_temp), "bed_target_c": f(metric.bed_target_temp),
+        "chamber_c": f(metric.chamber_temp),
+    }
+
+
+def _ams_units(printer, latest_metric):
+    """Build ams_units from the latest metric's filaments, with remain_percent
+    smoothed against the printer's 5 most recent metric rows."""
+    recent_filament_lists = list(
+        PrinterMetrics.objects.filter(device=printer)
+        .order_by("-timestamp")
+        .values_list("filaments", flat=True)[:5]
+    )
+    readings = defaultdict(list)
+    for filaments in recent_filament_lists:
+        for f in filaments or []:
+            key = (f.get("ams_unit_id"), int(f.get("tray_id", -1)))
+            readings[key].append(f.get("remain_percent"))
+
+    units = {}
+    for f in latest_metric.filaments or []:
+        unit_id = f.get("ams_unit_id")
+        tray_id = int(f.get("tray_id", -1))
+        value, status = smoothed_remaining_percent(readings[(unit_id, tray_id)])
+        unit = units.setdefault(unit_id, {
+            "unit_id": unit_id,
+            "type": f.get("ams_type", ""),
+            "humidity_raw": latest_metric.ams_humidity_raw,
+            "temp_c": float(latest_metric.ams_temp) if latest_metric.ams_temp is not None else None,
+            "slots": [],
+        })
+        color = f.get("color")
+        unit["slots"].append({
+            "slot": tray_id,
+            "label": f.get("slot", ""),
+            "material": f.get("type", ""),
+            "sub_type": f.get("sub_type", ""),
+            "color_hex": f"#{strip_color_padding(color)}" if color else None,
+            "remain_percent": value,
+            "remain_status": status,
+        })
+    return list(units.values())
+
+
+def _alerts_for_printer(printer, now, window_hours=ALERT_WINDOW_HOURS):
+    since = now - timedelta(hours=window_hours)
+    window = PrinterMetrics.objects.filter(device=printer, timestamp__gte=since)
+    total = window.count()
+    counts = defaultdict(int)
+    for hms_list in window.exclude(hms=[]).values_list("hms", flat=True):
+        for entry in hms_list:
+            counts[(entry.get("attr"), entry.get("code"))] += 1
+    return classify_alerts(counts, total)

tests/test_mcp_status.py

@@ -0,0 +1,131 @@
+import pytest
+from datetime import timedelta
+from django.utils import timezone
+
+from bambu_run.models import Printer, PrinterMetrics
+from bambu_run.mcp.status import get_printer_status
+
+
+@pytest.mark.django_db
+def test_get_printer_status_no_data_yet():
+    Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+
+    result = get_printer_status()
+
+    assert len(result["printers"]) == 1
+    entry = result["printers"][0]
+    assert entry["state"] == "no_data"
+    assert entry["stale"] is True
+    assert entry["job"] is None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_running_job_computes_finish_time():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        print_percent=63, layer_num=189, total_layer_num=301,
+        remaining_time_min=47, subtask_name="SyncGen LightQube Bracket",
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    job = result["printers"][0]["job"]
+    assert job["print_percent"] == 63
+    assert job["name"] == "SyncGen LightQube Bracket"
+    assert job["estimated_finish_at"] is not None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_idle_has_no_job():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now(), gcode_state="FINISH")
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["job"] is None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_stale_when_collector_gap_exceeds_120s():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now() - timedelta(seconds=200), gcode_state="RUNNING",
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["stale"] is True
+
+
+@pytest.mark.django_db
+def test_get_printer_status_ams_slot_carries_smoothed_percent_and_status():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    # Same tray, jittering readings across the 3 most recent polls (oldest first).
+    for i, pct in enumerate([22, 30, 35]):
+        PrinterMetrics.objects.create(
+            device=p, timestamp=now - timedelta(seconds=(2 - i) * 30), gcode_state="RUNNING",
+            filaments=[{
+                "ams_unit_id": 1, "tray_id": "2", "slot": "A01-P3",
+                "type": "PLA", "sub_type": "PLA Matte", "color": "E8AFCFFF",
+                "remain_percent": pct,
+            }],
+        )
+
+    result = get_printer_status(printer=p.id)
+
+    unit = result["printers"][0]["ams_units"][0]
+    slot = unit["slots"][0]
+    assert slot["remain_percent"] == 30  # median of [35, 30, 22] most-recent-first
+    assert slot["remain_status"] == "ok"
+    assert slot["color_hex"] == "#E8AFCF"
+
+
+@pytest.mark.django_db
+def test_get_printer_status_ams_slot_unread_sentinel():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        filaments=[{"ams_unit_id": 0, "tray_id": "0", "type": "ABS", "remain_percent": -1}],
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    slot = result["printers"][0]["ams_units"][0]["slots"][0]
+    assert slot["remain_percent"] is None
+    assert slot["remain_status"] == "unread"
+
+
+@pytest.mark.django_db
+def test_get_printer_status_never_exposes_serial_or_ip():
+    p = Printer.objects.create(
+        name="RNL-H2C", model="H2C", is_active=True,
+        serial_number="SECRET123", ip_address="192.168.1.50",
+    )
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now())
+
+    result = get_printer_status(printer=p.id)
+
+    entry_str = repr(result["printers"][0])
+    assert "SECRET123" not in entry_str
+    assert "192.168.1.50" not in entry_str
+
+
+@pytest.mark.django_db
+def test_get_printer_status_alerts_use_frequency_heuristic_not_hms_emptiness():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    # 1 of 5 rows in the alert window carries an HMS code -- below the 5%
+    # "transient" floor at this row count is not meaningful, but the point
+    # of this test is that a single non-empty hms row does NOT by itself
+    # produce an alert with unconditional certainty; it is classified, not
+    # surfaced verbatim. With 1/5 = 20%, exactly at the persistent boundary
+    # (not > 20%), so it lands in the omitted band.
+    for i in range(5):
+        hms = [{"attr": 1, "code": 2}] if i == 0 else []
+        PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(minutes=i), hms=hms)
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["alerts"] == []