diff --git a/README.md b/README.md index 90f5ff4..a249b94 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,30 @@ message when it recovers. There are deliberately no repeat reminders. Note that *low filament* warning is not possible: every AMS tray reports `remain: -1` for third-party spools, so detection is reactive only. +**Error codes are explained, not just relayed.** Bambuddy hands over the number +only (its `hms_errors` entries carry no description), so an alert used to read +`⚠️ Fehler: 0x4003`. `hms.py` resolves the code against Bambu's official +catalogue — bundled offline under `hms_data/`, both languages — and the alert, +`!progress` and the failure notice now say what is actually wrong: + +``` +⚠️ 0700-2000-0002-0001 — AMS A Slot 1 Filament ist aufgebraucht. Bitte legen Sie + ein neues Filament ein. +``` + +Severity picks the marker (🛑 fatal · ⚠️ warning · ℹ️ info). Both code namespaces +the printer uses are covered — 16-char HMS codes and 8-char `print_error` codes — +and an unlisted code still goes out as a number, which is what the Bambu wiki and +support ask for. Refresh the catalogue after a firmware update with +`./.venv/bin/python scripts/refresh_hms_codes.py` and commit the result. + +The catalogue can be **scoped to one machine** by serial prefix (`?d=01P` is the +P1S). That only filters the code set — of the ~3900 codes the scoped and the +generic list share, none differ in wording. The refresh script fetches both and +merges them, so the four codes only the P1S knows (part-cooling fan, Ethernet +accessory, SD-card wear) are present without losing an explanation for anything +else a firmware might emit. + ## Localization Replies are available in **German (default)** and **English**, chosen **per diff --git a/handlers.py b/handlers.py index af2644f..9d367cc 100644 --- a/handlers.py +++ b/handlers.py @@ -19,6 +19,7 @@ import classify import colors import config +import hms import i18n import signal_client import slicing @@ -1525,15 +1526,15 @@ async def _progress(group_id): rem = s.get("remaining_time") active = state.upper() in _ACTIVE_STATES or (isinstance(prog, (int, float)) and 0 < prog < 100) paused = state.upper() in _PAUSED_STATES - hms = _hms_detail(s, lang) + hms_detail = _hms_detail(s, lang) # not `hms` — that's the module if not (name and active): text = i18n.t(lang, "progress_idle", state=state.lower()) # Nothing printing *and* the plate isn't confirmed clear → say why the # queue is stalled, so checking in an hour actually explains itself. if s.get("awaiting_plate_clear"): text += "\n" + i18n.t(lang, "progress_awaiting_clear") - if hms: - text += "\n" + i18n.t(lang, "progress_hms", detail=hms) + if hms_detail: + text += "\n" + i18n.t(lang, "progress_hms", detail=hms_detail) await signal_client.send_to_group(group_id, text, attachments=attachments) return parts = [f'{"⏸️" if paused else "🖨️"} „{name}" — {state}'] @@ -1552,9 +1553,12 @@ async def _progress(group_id): clock = (datetime.datetime.now() + datetime.timedelta(minutes=r)).strftime("%H:%M") parts.append(i18n.t(lang, "progress_remaining", dur=dur)) parts.append(i18n.t(lang, "progress_done_at", clock=clock)) - if hms: - parts.append(i18n.t(lang, "progress_hms", detail=hms)) - await signal_client.send_to_group(group_id, " · ".join(parts), attachments=attachments) + # The error goes on its own line, not into the " · " status strip: an + # explained code is a full sentence and would make that strip unreadable. + text = " · ".join(parts) + if hms_detail: + text += "\n" + i18n.t(lang, "progress_hms", detail=hms_detail) + await signal_client.send_to_group(group_id, text, attachments=attachments) async def _go(group_id): @@ -1583,40 +1587,10 @@ async def poll_completions(interval=60): # ----- intervention alerts: the printer is stopped and needs a human ----- -def _hms_entry(e): - """(code, description) from one hms_errors entry. Bambuddy returns a list, but - the element shape isn't contractually pinned — accept dicts or bare strings, - and never raise: an exception in the poller kills that whole cycle for *every* - tracker, not just this one.""" - if isinstance(e, dict): - code = (e.get("code") or e.get("hms_code") or e.get("attr") or "") - desc = (e.get("description") or e.get("desc") or e.get("message") or "") - return str(code).strip(), str(desc).strip() - return str(e).strip(), "" - - -def _hms_codes(pstatus): - """Sorted HMS error codes as strings, or [] when the printer is healthy.""" - try: - entries = (pstatus or {}).get("hms_errors") or [] - codes = [c for c, _ in (_hms_entry(e) for e in entries) if c] - return sorted(set(codes)) - except Exception: - log.warning("could not read hms_errors", exc_info=True) - return [] - - -def _hms_detail(pstatus, lang="de"): - """': 0500-4003 Filament ausgegangen' — code + text, max 3, or '' if healthy.""" - try: - entries = (pstatus or {}).get("hms_errors") or [] - except Exception: - return "" - bits = [] - for e in entries[:3]: - code, desc = _hms_entry(e) - bits.append(f"{code} {desc}".strip() or i18n.t(lang, "hms_unknown")) - return (": " + "; ".join(bits)) if bits else "" +# Reading and explaining the printer's error codes lives in hms.py — Bambuddy +# ships the number only, the sentence comes from Bambu's bundled catalogue. +_hms_codes = hms.codes +_hms_detail = hms.detail def _condition(pstatus): @@ -1626,7 +1600,7 @@ def _condition(pstatus): codes = _hms_codes(pstatus) if codes: return "hms:" + ",".join(codes) - if (pstatus or {}).get("hms_errors"): + if hms.entries(pstatus): return "hms:unknown" # errors present but unparseable — still alert if ((pstatus or {}).get("state") or "").upper() in _PAUSED_STATES: return "pause" @@ -1726,7 +1700,10 @@ async def _check_completions(): ) store.set_stage(job["id"], "done") elif status == "failed": - detail = (f": {item.get('error_message')}" if item.get("error_message") else ".") + # Bambuddy rarely fills error_message; the *why* usually sits in the + # printer's HMS codes, which are still standing at that moment. + detail = (f": {item.get('error_message')}" if item.get("error_message") + else _hms_detail(pstatus, lang) or ".") await signal_client.send_to_group( job["group_id"], i18n.t(lang, "completion_failed", name=job["model_name"], detail=detail), diff --git a/hms.py b/hms.py new file mode 100644 index 0000000..253f9bb --- /dev/null +++ b/hms.py @@ -0,0 +1,175 @@ +"""Translate the printer's error numbers into sentences a human can act on. + +Bambuddy hands us ``hms_errors`` entries that carry **no description at all** +(schema: ``code``/``attr``/``module``/``severity``/``actions``/``full_code``) — +so the bot used to relay something like „⚠️ Fehler: 0x4003", which tells nobody +anything. The sentence behind a code lives in Bambu's public catalogue (the one +Bambu Studio queries); we bundle it under ``hms_data/`` and look it up offline, +see ``scripts/refresh_hms_codes.py``. + +Two code namespaces, both delivered in the same ``hms_errors`` list: + +* **HMS** — 16 hex chars, ``f"{attr:08X}{code:08X}"``, printed as + ``0700-8004-0002-0001``. Encodes the module *and* which AMS/slot is meant, so + the catalogue text names the actual slot. +* **print_error** — 8 hex chars, printed as ``0500-8061``. Same list, shorter + key, its own catalogue table. + +Everything here is defensive on purpose: this runs inside the completion poller, +where one exception kills that cycle for *every* tracker, not just the entry +that was malformed. +""" +import functools +import gzip +import json +import logging +import os +import re + +import i18n + +log = logging.getLogger("hms") + +_DATA = os.path.join(os.path.dirname(os.path.abspath(__file__)), "hms_data") +_HEX = re.compile(r"[^0-9A-Fa-f]") + +# Severity as Bambuddy reports it ((attr >> 8) & 0xF). Bambu's own scale: +# 1 fatal, 2 serious, 3 common, 4 info. +_SEVERITY_EMOJI = {1: "🛑", 2: "⚠️", 3: "⚠️", 4: "ℹ️"} +# Fallback when severity isn't reported: the error code's top nibble carries it +# (0x4xxx fatal, 0x8xxx serious/warning, 0xCxxx prompt) — see Bambuddy's parser. +_NIBBLE_SEVERITY = {"4": 1, "8": 2, "C": 4} + + +@functools.lru_cache(maxsize=4) +def _table(lang): + """{'hms': {code: text}, 'err': {code: text}} for a language, or empty dicts. + + Cached: the file is ~4500 entries and gets hit once per poll per error.""" + path = os.path.join(_DATA, f"hms_{i18n.normalize(lang)}.json.gz") + try: + with gzip.open(path, "rt", encoding="utf-8") as fh: + data = json.load(fh) + return {"hms": data.get("hms") or {}, "err": data.get("err") or {}} + except Exception: + log.warning("HMS catalogue %s unavailable — codes stay unexplained", path, + exc_info=True) + return {"hms": {}, "err": {}} + + +def _hexkey(value): + """'0x4003' / '0500-4003' / 4003 → bare uppercase hex, or '' if nothing usable.""" + if isinstance(value, int): + return f"{value:X}" + text = str(value or "").strip() + if text[:2].lower() == "0x": + text = text[2:] + text = _HEX.sub("", text) + return text.upper() + + +def _fields(entry): + """(full_code, severity, actions) from one ``hms_errors`` element. + + Accepts every shape seen in the wild: Bambuddy's dict, a dict carrying only + the legacy ``code``/``hms_code``, or a bare string. ``full_code`` is the + catalogue key — reconstructed from ``attr`` + ``code`` when Bambuddy omits + it, since ``code`` alone is just the low 32 bits.""" + if not isinstance(entry, dict): + return _hexkey(entry), None, [] + full = _hexkey(entry.get("full_code")) + code, attr = entry.get("code"), entry.get("attr") + if len(full) not in (8, 16): + if isinstance(attr, int) and code is not None and _hexkey(code): + full = f"{attr:08X}{int(_hexkey(code), 16):08X}" + else: + full = _hexkey(code or entry.get("hms_code") or entry.get("attr")) + severity = entry.get("severity") + actions = entry.get("actions") + return full, (severity if isinstance(severity, int) else None), \ + [str(a) for a in actions] if isinstance(actions, list) else [] + + +def _display(full_code): + """'07008004' → '0700-8004'; 16-char codes get the four-group form.""" + if len(full_code) in (8, 16): + return "-".join(full_code[i:i + 4] for i in range(0, len(full_code), 4)) + return full_code or "?" + + +def _text(full_code, lang): + """The catalogue sentence for a code, or '' when it isn't listed.""" + table = _table(lang) + if len(full_code) == 16: + return table["hms"].get(full_code, "") + if len(full_code) == 8: + # print_error lives in the short table; a few short codes only exist on + # the HMS side, so try both before giving up. + return table["err"].get(full_code) or table["hms"].get(full_code, "") + return "" + + +def _severity(full_code, reported): + if reported in _SEVERITY_EMOJI: + return reported + # The error half of the code (low 16 bits) starts with the severity nibble. + nibble = full_code[8:9] if len(full_code) == 16 else full_code[4:5] + return _NIBBLE_SEVERITY.get(nibble.upper(), 2) + + +def describe(entry, lang="de"): + """One printable line: '⚠️ 0700-8004 — AMS A Slot 1: Filament ist alle …'. + + Falls back to a bare code line when the catalogue doesn't know it — the code + is still worth showing, it's what the Bambu wiki and support ask for.""" + lang = i18n.normalize(lang) + try: + full, reported, _ = _fields(entry) + # An explicit description (older Bambuddy builds, tests) wins over the + # catalogue: it's what this specific installation decided to say. + text = "" + if isinstance(entry, dict): + text = str(entry.get("description") or entry.get("desc") + or entry.get("message") or "").strip() + text = text or _text(full, lang) + emoji = _SEVERITY_EMOJI.get(_severity(full, reported), "⚠️") + code = _display(full) + if not text: + return f"{emoji} {code} — {i18n.t(lang, 'hms_unknown')}" + return f"{emoji} {code} — {text}" + except Exception: + log.warning("could not describe HMS entry %r", entry, exc_info=True) + return f"⚠️ {i18n.t(lang, 'hms_unknown')}" + + +def entries(pstatus): + """The raw ``hms_errors`` list, tolerating any junk in that field.""" + try: + raw = (pstatus or {}).get("hms_errors") + except Exception: + return [] + return list(raw) if isinstance(raw, (list, tuple)) else [] + + +def codes(pstatus): + """Sorted canonical codes — the identity used to decide 'is this the same + incident as last poll?'. Empty when the printer is healthy.""" + try: + found = {_fields(e)[0] for e in entries(pstatus)} + return sorted(c for c in found if c) + except Exception: + log.warning("could not read hms_errors", exc_info=True) + return [] + + +def detail(pstatus, lang="de", limit=3): + """':\\n\\n' for the ``{detail}`` slot of the alert templates, or + '' when nothing is wrong. Capped so a cascade of follow-up codes doesn't + bury the message.""" + found = entries(pstatus) + if not found: + return "" + lines = [describe(e, lang) for e in found[:limit]] + if len(found) > limit: + lines.append(i18n.t(i18n.normalize(lang), "hms_more", n=len(found) - limit)) + return ":\n" + "\n".join(lines) diff --git a/hms_data/hms_de.json.gz b/hms_data/hms_de.json.gz new file mode 100755 index 0000000..76ace3e Binary files /dev/null and b/hms_data/hms_de.json.gz differ diff --git a/hms_data/hms_en.json.gz b/hms_data/hms_en.json.gz new file mode 100755 index 0000000..11796e7 Binary files /dev/null and b/hms_data/hms_en.json.gz differ diff --git a/i18n.py b/i18n.py index ea4e555..22c5724 100644 --- a/i18n.py +++ b/i18n.py @@ -448,8 +448,10 @@ def normalize(lang): "en": "⏸️ paused — waiting for you", }, "progress_hms": { - "de": "⚠️ Fehler{detail}", - "en": "⚠️ error{detail}", + # {detail} already starts with ':' and carries a severity emoji per + # line (see hms.detail) — no second warning sign here. + "de": "Der Drucker meldet{detail}", + "en": "The printer reports{detail}", }, "progress_awaiting_clear": { "de": "🧹 Die Platte ist noch voll — schick !go, wenn sie frei ist.", @@ -472,8 +474,14 @@ def normalize(lang): "en": "▶️ „{name}\" is running again.", }, "hms_unknown": { - "de": "unbekannter Fehler", - "en": "unknown error", + # Shown when Bambu's catalogue doesn't list the code (new firmware). + # The number still goes out — it's what the wiki and support ask for. + "de": "unbekannter Fehlercode (in der Bambu-Wiki nachschlagen)", + "en": "unknown error code (look it up in the Bambu wiki)", + }, + "hms_more": { + "de": "… und {n} weitere Meldung(en)", + "en": "… and {n} more message(s)", }, # ----- !go ----- diff --git a/scripts/refresh_hms_codes.py b/scripts/refresh_hms_codes.py new file mode 100644 index 0000000..47854c4 --- /dev/null +++ b/scripts/refresh_hms_codes.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Re-fetch Bambu's official HMS/print-error catalogue into ``hms_data/``. + +The printer only ever sends a *number* (Bambuddy hands us ``full_code``); the +human sentence behind it lives in Bambu's public catalogue, the same one Bambu +Studio queries. We bundle it instead of calling out at runtime: the bot must be +able to explain an error while the internet is down, and the payload barely +compresses to 60 kB per language. + +``?d=`` scopes the catalogue to one machine — ``01P`` is the P1S +(our serial is ``01P00C571300842``; Bambuddy keys its action table off the same +three characters). Verified 2026-08-08: scoping only *filters* the code set, it +never changes the wording — of the ~3900 codes both lists share, **zero** differ +in text. The P1S list drops 35 codes for hardware it doesn't have and adds 4 it +alone knows (part-cooling fan ``0300-3100-…``, the Ethernet accessory, an SD-card +warning). So we fetch both and **merge**: the device list contributes its +exclusive codes, the generic one keeps an explanation available for anything an +odd firmware might emit. A wrong-but-plausible sentence isn't a risk here — the +texts are identical where they overlap. + +Run it when codes look unknown (new firmware adds some): + + ./.venv/bin/python scripts/refresh_hms_codes.py + +then commit the regenerated ``hms_data/hms_*.json.gz``. +""" +import gzip +import json +import os +import sys +import urllib.request + +URL = "https://e.bambulab.com/query.php?lang={lang}" +OUT = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "hms_data") +# Serial prefix of the machine we print on. Other known values: 00M/00W = X1 +# series, 03W/039 = H2D family. An unknown one answers `result: 201`. +DEVICE = "01P" + + +def _get(url): + # The endpoint 403s a bare urllib request — it wants a browser-ish UA. + req = urllib.request.Request( + url, headers={"User-Agent": "Mozilla/5.0 (bambu-bot hms refresh)"}) + with urllib.request.urlopen(req, timeout=60) as r: + payload = json.load(r) + # `result: 0` is success; anything else comes with an empty string as data + # (e.g. 201 for a device id Bambu doesn't publish a catalogue for). + if payload.get("result") != 0 or not isinstance(payload.get("data"), dict): + raise RuntimeError(f"{url} → result {payload.get('result')}, no catalogue") + return payload["data"] + + +def _codes(data, key, lang): + """{ECODE: text} from one catalogue section (list of {ecode, intro}).""" + return {e["ecode"].upper(): e["intro"].strip() + for e in data.get(key, {}).get(lang, []) + if e.get("ecode") and e.get("intro")} + + +def fetch(lang, device=DEVICE): + generic = _get(URL.format(lang=lang)) + scoped = _get(URL.format(lang=lang) + f"&d={device}") + # device_hms = the 16-hex HMS codes, device_error = the 8-hex print_error + # codes. Device entries last so they win on any future disagreement. + hms = {**_codes(generic, "device_hms", lang), **_codes(scoped, "device_hms", lang)} + err = {**_codes(generic, "device_error", lang), **_codes(scoped, "device_error", lang)} + return {"lang": lang, "device": device, + "ver": scoped.get("device_hms", {}).get("ver"), "hms": hms, "err": err} + + +def main(langs=("de", "en")): + os.makedirs(OUT, exist_ok=True) + for lang in langs: + table = fetch(lang) + path = os.path.join(OUT, f"hms_{lang}.json.gz") + blob = json.dumps(table, ensure_ascii=False, sort_keys=True).encode() + # mtime=0 → byte-identical output for unchanged input, so a re-run that + # changes nothing produces no git diff. + with open(path, "wb") as fh: + fh.write(gzip.compress(blob, 9, mtime=0)) + print(f"{path}: {len(table['hms'])} HMS + {len(table['err'])} print-error codes " + f"(ver {table['ver']}), {os.path.getsize(path) // 1024} kB") + + +if __name__ == "__main__": + main(tuple(sys.argv[1:]) or ("de", "en")) diff --git a/tests/test_hms.py b/tests/test_hms.py new file mode 100644 index 0000000..f7c7262 --- /dev/null +++ b/tests/test_hms.py @@ -0,0 +1,152 @@ +"""Explaining the printer's error codes. + +Bambuddy sends the *number* only — its ``hms_errors`` schema has no description +field at all — so before this the bot relayed „⚠️ Fehler: 0x4003" and left the +reader none the wiser. The catalogue under ``hms_data/`` supplies the sentence. +""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) +import hms # noqa: E402 +import i18n # noqa: E402 + + +def bambuddy_entry(full_code, severity=2, actions=()): + """The shape Bambuddy really posts (schemas/printer.py HMSErrorResponse): + ``code`` is only the low 32 bits, ``full_code`` is the catalogue key.""" + attr, code = int(full_code[:8], 16), int(full_code[8:] or "0", 16) + return {"code": f"0x{code:x}", "attr": attr, "module": (attr >> 24) & 0xFF, + "severity": severity, "actions": list(actions), "job_id": None, + "full_code": full_code} + + +# A real HMS code + its real catalogue text. Verified against Bambu's published +# table; if a firmware refresh ever renames it, that's worth noticing. +RUNOUT = "0700200000020001" +RUNOUT_TEXT = "AMS A Slot 1 Filament ist aufgebraucht" + + +def test_catalogue_is_bundled_for_both_languages(): + for lang in i18n.LANGS: + table = hms._table(lang) + assert len(table["hms"]) > 1000 and len(table["err"]) > 100 + + +def test_explains_a_real_hms_code(): + line = hms.describe(bambuddy_entry(RUNOUT)) + assert RUNOUT_TEXT in line + assert "0700-2000-0002-0001" in line # printer-screen grouping + + +def test_explains_in_english_too(): + line = hms.describe(bambuddy_entry(RUNOUT), "en") + assert "AMS A Slot 1" in line and "run out" in line + + +def test_explains_a_print_error_code(): + """print_error rides in the same list with an 8-char code and its own table.""" + line = hms.describe(bambuddy_entry("07028002"), "de") + assert "0702-8002" in line and "Schneid" in line + + +def test_code_is_reconstructed_when_full_code_is_missing(): + """`code` alone is the low half — useless as a key. attr + code rebuilds it.""" + entry = bambuddy_entry(RUNOUT) + del entry["full_code"] + assert RUNOUT_TEXT in hms.describe(entry) + + +def test_unknown_code_still_reports_the_number(): + line = hms.describe(bambuddy_entry("FFFFFFFFFFFFFFFF")) + assert "FFFF-FFFF-FFFF-FFFF" in line and "unbekannt" in line + + +def test_severity_picks_the_emoji(): + assert hms.describe(bambuddy_entry(RUNOUT, severity=1)).startswith("🛑") + assert hms.describe(bambuddy_entry(RUNOUT, severity=4)).startswith("ℹ️") + # No severity reported → derived from the code's severity nibble (0x8 = warn). + assert hms.describe({"full_code": RUNOUT}).startswith("⚠️") + + +def test_an_explicit_description_wins_over_the_catalogue(): + """Older Bambuddy builds may caption it themselves — don't overrule them.""" + line = hms.describe({"code": "0500-4003", "description": "Filament ausgegangen"}) + assert "Filament ausgegangen" in line and "0500-4003" in line + + +@pytest.mark.parametrize("entry", [None, "", "junk", {}, {"code": None}, 42, []]) +def test_junk_entries_never_raise(entry): + """This runs in the completion poller: one exception kills the whole cycle.""" + assert hms.describe(entry) + + +@pytest.mark.parametrize("status", [None, {}, {"hms_errors": None}, + {"hms_errors": "not-a-list"}, "nonsense"]) +def test_broken_status_reads_as_healthy(status): + assert hms.entries(status) == [] and hms.codes(status) == [] + assert hms.detail(status) == "" + + +def test_detail_is_one_line_per_error_and_capped(): + status = {"hms_errors": [bambuddy_entry(RUNOUT)] * 4} + out = hms.detail(status, "de", limit=3) + assert out.startswith(":\n") + assert out.count(RUNOUT_TEXT) == 3 + assert "1 weitere" in out + + +def test_codes_identify_the_incident(): + """The alert 'is this still the same problem?' key must be the full code — + two different faults in one module must not collapse into one.""" + status = {"hms_errors": [bambuddy_entry(RUNOUT), bambuddy_entry("0700200000020005")]} + assert hms.codes(status) == [RUNOUT, "0700200000020005"] + assert hms.codes({"hms_errors": []}) == [] + + +# ----- the failure message ---------------------------------------------------- + +def test_failure_message_falls_back_to_the_hms_reason(tmp_path, monkeypatch): + """Bambuddy usually leaves error_message empty; the printer's standing HMS + code is what actually says why the print died.""" + import asyncio + import config + import handlers + import store + + config.DB_PATH = str(tmp_path / "t.db") + store.init_db() + store.add_queued("group.x", "+1", "Benchy", 4, 42) + sent = [] + + async def _none(): + return None + monkeypatch.setattr(handlers.signal_client, "send_to_group", + lambda gid, msg, **kw: sent.append(msg) or _none()) + + async def _item(_id): + return {"status": "failed"} + + async def _pstatus(_pid): + return {"state": "FAILED", "hms_errors": [ + {"code": "0x20001", "attr": 0x07002000, "severity": 2, + "full_code": "0700200000020001"}]} + monkeypatch.setattr(handlers.bambuddy, "get_queue_item", _item) + monkeypatch.setattr(handlers.bambuddy, "printer_status", _pstatus) + + asyncio.run(handlers._check_completions()) + assert "fehlgeschlagen" in sent[0] + assert "AMS A Slot 1 Filament ist aufgebraucht" in sent[0] + + +def test_p1s_specific_codes_are_bundled(): + """The catalogue is scoped by serial prefix (?d=01P). Those four codes exist + only in the P1S list — if a refresh ever drops the device fetch, they vanish + and the part-cooling-fan fault reads as 'unknown' on the one printer that + can report it.""" + for lang in i18n.LANGS: + table = hms._table(lang) + assert "0300310000010001" in table["hms"] # part cooling fan stalled + assert "0500C011" in table["err"] # SD card degraded diff --git a/tests/test_intervention.py b/tests/test_intervention.py index 9c7f327..b257ed8 100644 --- a/tests/test_intervention.py +++ b/tests/test_intervention.py @@ -131,6 +131,18 @@ def test_hms_alerts_once_and_a_new_code_alerts_again(tmp_path, monkeypatch): assert len(sent) == 2 and "0300-0100" in sent[1][1] +def test_alert_explains_a_real_bambuddy_error(tmp_path, monkeypatch): + """Bambuddy sends no description — the alert must still say what happened.""" + printer = {"state": "PAUSE", "hms_errors": [ + {"code": "0x20001", "attr": 0x07002000, "module": 7, "severity": 2, + "actions": [], "job_id": None, "full_code": "0700200000020001"}]} + sent = _setup(tmp_path, monkeypatch, {"status": "printing"}, printer) + _poll(3) + assert len(sent) == 1 + assert "AMS A Slot 1 Filament ist aufgebraucht" in sent[0][1] + assert "0700-2000-0002-0001" in sent[0][1] + + def test_hms_wins_over_pause(tmp_path, monkeypatch): """A pause caused by an error should read as the error, not a bare pause.""" printer = {"state": "PAUSE", "hms_errors": [{"code": "0500-4003", "description": "x"}]} @@ -231,6 +243,20 @@ def test_progress_reports_hms(tmp_path, monkeypatch): assert "0500-4003" in sent[-1] +def test_progress_explains_a_real_bambuddy_error(tmp_path, monkeypatch): + config.DB_PATH = str(tmp_path / "t.db") + store.init_db() + sent = _progress_setup(monkeypatch, { + "state": "PAUSE", "current_print": "Box1", "progress": 42, + "hms_errors": [{"code": "0x20001", "attr": 0x07002000, "severity": 2, + "full_code": "0700200000020001"}]}) + asyncio.run(handlers._progress("group.x")) + out = sent[-1] + assert "AMS A Slot 1 Filament ist aufgebraucht" in out + # the explanation is a sentence — it must not be crammed into the " · " strip + assert "\n" in out and " · AMS" not in out + + def test_progress_idle_explains_awaiting_plate_clear(tmp_path, monkeypatch): config.DB_PATH = str(tmp_path / "t.db") store.init_db()