From 679c620fb206f7e39756a7cfc9cc0455619dabaf Mon Sep 17 00:00:00 2001 From: RNL Date: Wed, 19 Aug 2026 21:55:02 +1000 Subject: [PATCH] feat(mcp): add get_printer_status tool --- bambu_run/mcp/status.py | 139 +++++++++++++++++++++++++++++++++++++++ tests/test_mcp_status.py | 131 ++++++++++++++++++++++++++++++++++++ 2 files changed, 270 insertions(+) create mode 100644 bambu_run/mcp/status.py create mode 100644 tests/test_mcp_status.py diff --git a/bambu_run/mcp/status.py b/bambu_run/mcp/status.py new file mode 100644 index 0000000..4c6caba --- /dev/null +++ b/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) diff --git a/tests/test_mcp_status.py b/tests/test_mcp_status.py new file mode 100644 index 0000000..0c6e191 --- /dev/null +++ b/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"] == []