mirror of
https://github.com/RunLit/Bambu-Run.git
synced 2026-10-21 15:37:38 +01:00
feat(mcp): add get_hardware_health tool
This commit is contained in:
@@ -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,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -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}
|
||||
|
||||
Reference in New Issue
Block a user