diff --git a/bambu_run/mcp/__init__.py b/bambu_run/mcp/__init__.py new file mode 100644 index 0000000..8f7ec98 --- /dev/null +++ b/bambu_run/mcp/__init__.py @@ -0,0 +1 @@ +"""MCP toolkit for Bambu-Run: read-only printer, print-history, and filament tools.""" diff --git a/bambu_run/mcp/shapes.py b/bambu_run/mcp/shapes.py new file mode 100644 index 0000000..583a1f4 --- /dev/null +++ b/bambu_run/mcp/shapes.py @@ -0,0 +1,143 @@ +"""Shared output-shaping helpers enforcing the spec's data-correction rules. + +Every rule here traces to a verified finding in "Bambu-Run MCP Tool Surface" +(the "Data Reality Findings" section) -- not a general defensive-programming +habit. Do not relax these without re-checking against real data. +""" +from __future__ import annotations + +import statistics + +STALE_THRESHOLD_SECONDS = 120 +LOW_REMAINING_THRESHOLD = 20 + + +def freshness(timestamp, now): + """(data_age_seconds, stale) for a reading timestamp relative to `now`. + + `now` is passed in rather than computed here so a caller building a + multi-field response reuses one `timezone.now()` call throughout, + keeping `generated_at` and every `data_age_seconds` mutually consistent. + """ + age_seconds = (now - timestamp).total_seconds() + return { + "data_age_seconds": int(age_seconds), + "stale": age_seconds > STALE_THRESHOLD_SECONDS, + } + + +def remaining_status(percent): + """Classify one raw remaining-percent reading, honoring the -1 sentinel. + + Returns (value, status): value is percent-or-None; status is one of + "unread" | "empty" | "low" | "ok". -1 and None both mean "no RFID + reading yet" -- the spec rules out -1 meaning "low" (jobs 62/63/70/77 + show start=-1% followed by end=90-100%, which a genuinely low reading + could not do). + """ + if percent is None or percent == -1: + return None, "unread" + percent = int(round(percent)) + if percent <= 0: + return 0, "empty" + if percent <= LOW_REMAINING_THRESHOLD: + return percent, "low" + return percent, "ok" + + +def smoothed_remaining_percent(readings): + """Median of the 5 most recent remain_percent readings for one tray. + + `readings` is most-recent-first. -1 sentinel readings are excluded + before the median, not treated as zero (spec: AMS remain_percent swings + 22-35% mid-print on real RFID noise; median-of-5 rejects single-sample + spikes while still tracking genuine change). + """ + recent = readings[:5] + valid = [r for r in recent if r is not None and r != -1] + if not valid: + return remaining_status(-1 if recent else None) + return remaining_status(statistics.median(valid)) + + +def classify_alerts(code_counts, total_rows): + """Turn {(attr, code): occurrences} window counts into the alerts array. + + Frequency heuristic (spec: HMS codes are stateful, not event-based -- + the two most common codes alone accounted for over 80% of all alert + occurrences in the verified data, in long unbroken runs, not isolated + events): a code present in >20% of window rows is "persistent" (a + likely standing advisory, surfaced once with a count); <5% is + "transient" (possibly actionable). The 5-20% band is omitted rather + than guessed at. severity is always "unknown" -- nothing in Bambu-Run + decodes these codes yet; do not assert "fine" or "faulty" from them. + """ + alerts = [] + for (attr, code), count in code_counts.items(): + ratio = count / total_rows if total_rows else 0 + if ratio > 0.20: + kind = "persistent" + elif ratio < 0.05: + kind = "transient" + else: + continue + alerts.append({ + "attr": attr, + "code": code, + "kind": kind, + "occurrences": count, + "severity": "unknown", + }) + return alerts + + +def display_name(design_title, plate_title, project_name): + """design_title -> plate_title -> project_name fallback. + + Spec finding: design_title is empty on 4 of 5 recent BambuCloudTask rows. + """ + return design_title or plate_title or project_name + + +def normalize_consumed(starting_percent, ending_percent, initial_weight_grams): + """FilamentUsage consumption, honest about the -1 sentinel and RFID jitter. + + Returns (consumed_percent, consumed_grams, data_quality). data_quality is + "ok" or "unreliable". A negative or sentinel-derived result becomes + (None, None, "unreliable") -- never a negative figure, never a + fabricated zero (spec: 24/269 usage rows go negative; a -1 sentinel + subtracted as a real value produces e.g. consumed=-101%). + """ + if starting_percent is None or starting_percent == -1 or ending_percent is None: + return None, None, "unreliable" + consumed_percent = starting_percent - ending_percent + if consumed_percent < 0: + return None, None, "unreliable" + consumed_grams = None + if initial_weight_grams: + consumed_grams = int(initial_weight_grams * (consumed_percent / 100.0)) + return consumed_percent, consumed_grams, "ok" + + +def resolve_printer(printer): + """Resolve an id, exact name, or None to a Printer queryset. + + None returns all active printers. Raises Printer.DoesNotExist if a + specific id/name was given and nothing matched -- MCP tool wrappers let + this propagate as a clear tool error rather than returning empty data + that looks like "no printers exist". + """ + from bambu_run.models import Printer + + if printer is None: + return Printer.objects.filter(is_active=True) + + if isinstance(printer, int) or (isinstance(printer, str) and printer.isdigit()): + by_id = Printer.objects.filter(pk=int(printer)) + if by_id.exists(): + return by_id + + by_name = Printer.objects.filter(name=printer) + if not by_name.exists(): + raise Printer.DoesNotExist(f"No printer matching {printer!r}") + return by_name diff --git a/tests/test_mcp_shapes.py b/tests/test_mcp_shapes.py new file mode 100644 index 0000000..06094b0 --- /dev/null +++ b/tests/test_mcp_shapes.py @@ -0,0 +1,134 @@ +import pytest +from datetime import timedelta +from django.utils import timezone + +from bambu_run.models import Printer +from bambu_run.mcp.shapes import ( + STALE_THRESHOLD_SECONDS, + freshness, + remaining_status, + smoothed_remaining_percent, + classify_alerts, + display_name, + normalize_consumed, + resolve_printer, +) + + +def test_freshness_fresh_reading(): + now = timezone.now() + ts = now - timedelta(seconds=10) + assert freshness(ts, now) == {"data_age_seconds": 10, "stale": False} + + +def test_freshness_stale_reading(): + now = timezone.now() + ts = now - timedelta(seconds=STALE_THRESHOLD_SECONDS + 30) + result = freshness(ts, now) + assert result["stale"] is True + + +def test_remaining_status_unread_sentinel(): + assert remaining_status(-1) == (None, "unread") + + +def test_remaining_status_none_is_unread(): + assert remaining_status(None) == (None, "unread") + + +def test_remaining_status_empty(): + assert remaining_status(0) == (0, "empty") + + +def test_remaining_status_low_boundary(): + assert remaining_status(20) == (20, "low") + assert remaining_status(21) == (21, "ok") + + +def test_remaining_status_ok(): + assert remaining_status(72) == (72, "ok") + + +def test_smoothed_remaining_percent_rejects_spikes(): + # Real data pattern (spec, job 122): genuine values bouncing 22-35%. + readings = [35, 22, 30, 28, 25] + assert smoothed_remaining_percent(readings) == (28, "ok") + + +def test_smoothed_remaining_percent_excludes_unread_sentinel(): + readings = [-1, -1, 90, 88, 91] + assert smoothed_remaining_percent(readings) == (90, "ok") + + +def test_smoothed_remaining_percent_all_unread(): + assert smoothed_remaining_percent([-1, -1, -1]) == (None, "unread") + + +def test_smoothed_remaining_percent_only_uses_5_most_recent(): + readings = [10, 10, 10, 10, 10, 100] + value, _ = smoothed_remaining_percent(readings) + assert value == 10 + + +def test_classify_alerts_persistent_above_20_percent(): + counts = {(83887616, 131078): 25} + assert classify_alerts(counts, total_rows=100) == [{ + "attr": 83887616, "code": 131078, + "kind": "persistent", "occurrences": 25, "severity": "unknown", + }] + + +def test_classify_alerts_transient_below_5_percent(): + counts = {(50336256, 131073): 3} + alerts = classify_alerts(counts, total_rows=100) + assert alerts[0]["kind"] == "transient" + + +def test_classify_alerts_omits_ambiguous_band(): + counts = {(1, 2): 10} # exactly 10% of 100 rows: neither >20% nor <5% + assert classify_alerts(counts, total_rows=100) == [] + + +def test_display_name_falls_back_through_chain(): + assert display_name("", "", "raw_project") == "raw_project" + assert display_name("", "plate_1", "raw_project") == "plate_1" + assert display_name("Cool Design", "plate_1", "raw_project") == "Cool Design" + + +def test_normalize_consumed_ordinary_reading(): + assert normalize_consumed(74, 68, initial_weight_grams=1000) == (6, 60, "ok") + + +def test_normalize_consumed_rfid_jitter_goes_unreliable_not_negative(): + # spec finding: start=51% end=52% -> raw consumed=-1% (ordinary jitter) + assert normalize_consumed(51, 52, initial_weight_grams=1000) == (None, None, "unreliable") + + +def test_normalize_consumed_sentinel_start_goes_unreliable(): + # spec finding: jobs 62/63/70/77, start=-1% end=100% + assert normalize_consumed(-1, 100, initial_weight_grams=1000) == (None, None, "unreliable") + + +@pytest.mark.django_db +def test_resolve_printer_by_id(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + assert list(resolve_printer(p.id)) == [p] + + +@pytest.mark.django_db +def test_resolve_printer_by_name(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + assert list(resolve_printer("RNL-H2C")) == [p] + + +@pytest.mark.django_db +def test_resolve_printer_none_returns_all_active(): + p1 = Printer.objects.create(name="A", model="H2C", is_active=True) + Printer.objects.create(name="B", model="X1C", is_active=False) + assert list(resolve_printer(None)) == [p1] + + +@pytest.mark.django_db +def test_resolve_printer_unknown_name_raises(): + with pytest.raises(Printer.DoesNotExist): + resolve_printer("nonexistent")