From 39e4638f6a8fd8e35021198b49b49b16c96c4a8f Mon Sep 17 00:00:00 2001 From: RNL Date: Wed, 19 Aug 2026 22:04:14 +1000 Subject: [PATCH] feat(mcp): add get_hardware_health tool --- bambu_run/mcp/status.py | 58 ++++++++++++++++++++++++++++++++++++++++ tests/test_mcp_status.py | 57 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+) diff --git a/bambu_run/mcp/status.py b/bambu_run/mcp/status.py index 6f9c7b6..f7e6424 100644 --- a/bambu_run/mcp/status.py +++ b/bambu_run/mcp/status.py @@ -185,3 +185,61 @@ def get_metrics_history( "timestamps": [r.timestamp.isoformat() for r in rows], "series": series, } + + +def get_hardware_health(printer: str | int | None = None, days: int = 7) -> dict: + """Maintenance-relevant observations for one printer: wifi, AMS humidity, + alert summary, hotend inventory, and collector coverage. + + Returns observations only -- it does not render a verdict; the calling + agent decides what the numbers mean. wear_percent is deliberately absent + from each hotend entry: verified constant at 100.00 across every + hotend's entire history, so it carries no information. + """ + now = timezone.now() + target = resolve_printer(printer).first() + if target is None: + return {"printer": None} + + since = now - timedelta(days=days) + window = PrinterMetrics.objects.filter(device=target, timestamp__gte=since) + total = window.count() + latest = window.order_by("-timestamp").first() + + wifi_avg = window.exclude(wifi_signal_dbm__isnull=True).aggregate(avg=Avg("wifi_signal_dbm"))["avg"] + humidity_avg = window.exclude(ams_humidity_raw__isnull=True).aggregate(avg=Avg("ams_humidity_raw"))["avg"] + + 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 + + hotends = [ + { + "id": h.id, + "serial_number": h.serial_number, + "position": "toolhead" if h.is_toolhead else (f"slot_{h.slot_number}" if h.slot_number else "rack"), + "used_time_hours": round(h.used_time_seconds / 3600, 1), + "last_filament_profile_id": h.last_filament_profile_id or None, + } + for h in Hotend.objects.filter(printer=target) + ] + + expected_polls = int(days * 24 * 60 * 60 / 30) # 30s measured collector cadence + + return { + "printer": {"id": target.id, "name": target.name}, + "generated_at": now.isoformat(), + "wifi_signal_dbm": { + "current": latest.wifi_signal_dbm if latest else None, + "window_avg": round(wifi_avg, 1) if wifi_avg is not None else None, + }, + "ams_humidity_raw_window_avg": round(humidity_avg, 1) if humidity_avg is not None else None, + "alerts": classify_alerts(counts, total), + "hotends": hotends, + "collector": { + "polls_in_window": total, + "expected_polls": expected_polls, + "coverage_ratio": round(total / expected_polls, 2) if expected_polls else None, + }, + } diff --git a/tests/test_mcp_status.py b/tests/test_mcp_status.py index a9bf160..524dd7b 100644 --- a/tests/test_mcp_status.py +++ b/tests/test_mcp_status.py @@ -182,3 +182,60 @@ def test_get_metrics_history_hard_ceiling_500_points(): def test_get_metrics_history_no_matching_printer_returns_empty_shape(): result = get_metrics_history(printer=None) assert result == {"printer": None, "timestamps": [], "series": {}} + + +from bambu_run.models import Hotend +from bambu_run.mcp.status import get_hardware_health + + +@pytest.mark.django_db +def test_get_hardware_health_reports_wifi_and_humidity_averages(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + now = timezone.now() + PrinterMetrics.objects.create(device=p, timestamp=now, wifi_signal_dbm=-50, ams_humidity_raw=4) + PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(hours=1), wifi_signal_dbm=-60, ams_humidity_raw=6) + + result = get_hardware_health(printer=p.id, days=7) + + assert result["wifi_signal_dbm"]["current"] == -50 + assert result["wifi_signal_dbm"]["window_avg"] == -55.0 + assert result["ams_humidity_raw_window_avg"] == 5.0 + + +@pytest.mark.django_db +def test_get_hardware_health_excludes_wear_percent_field(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + PrinterMetrics.objects.create(device=p, timestamp=timezone.now()) + Hotend.objects.create( + printer=p, serial_number="SN1", raw_id=0, is_toolhead=True, + used_time_seconds=3600, wear_percent=100, + ) + + result = get_hardware_health(printer=p.id) + + hotend = result["hotends"][0] + assert hotend["used_time_hours"] == 1.0 + assert "wear_percent" not in hotend + + +@pytest.mark.django_db +def test_get_hardware_health_classifies_persistent_alert(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + now = timezone.now() + for i in range(30): + PrinterMetrics.objects.create( + device=p, timestamp=now - timedelta(minutes=i), + hms=[{"attr": 83887616, "code": 131078}], + ) + for i in range(30, 100): + PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(minutes=i), hms=[]) + + result = get_hardware_health(printer=p.id, days=1) + + assert result["alerts"][0]["kind"] == "persistent" + assert result["alerts"][0]["occurrences"] == 30 + + +@pytest.mark.django_db +def test_get_hardware_health_no_printer_returns_empty_shape(): + assert get_hardware_health(printer=None) == {"printer": None}