diff --git a/src/bambu_ai/poller.py b/src/bambu_ai/poller.py index d68e12c..7d007f2 100644 --- a/src/bambu_ai/poller.py +++ b/src/bambu_ai/poller.py @@ -14,6 +14,20 @@ from .config import load_config +def merge_delta(snap: dict, payload: dict) -> None: + """Merge an incoming partial-state payload into a long-lived snapshot. + + Bambu only re-publishes fields that *changed* in subsequent reports. So + ``nozzle_target_temper`` arrives once in the initial pushall response and + then never again until the user changes the target — the watcher must + remember it instead of falling back to 0. + + Values that are ``None`` or empty string are skipped so a transiently + missing field doesn't blow away a known-good prior value. + """ + snap.update({k: v for k, v in payload.items() if v not in (None, "")}) + + def fmt_state(p: dict) -> str: parts: list[str] = [time.strftime("[%H:%M:%S]")] @@ -46,6 +60,7 @@ def main() -> None: cfg = load_config() topic_report = f"device/{cfg['SERIAL']}/report" topic_request = f"device/{cfg['SERIAL']}/request" + snap: dict = {} # rolling snapshot — delta merges land here client = mqtt.Client( mqtt.CallbackAPIVersion.VERSION2, @@ -74,7 +89,8 @@ def on_message(_c, _u, msg): return p = payload.get("print") if p: - print(fmt_state(p)) + merge_delta(snap, p) + print(fmt_state(snap)) return for k, v in payload.items(): summary = json.dumps(v)[:140] diff --git a/tests/test_poller.py b/tests/test_poller.py new file mode 100644 index 0000000..b5977d6 --- /dev/null +++ b/tests/test_poller.py @@ -0,0 +1,102 @@ +"""Tests for the bambu-poll module — focuses on the delta-merge fix (issue #8).""" + +from __future__ import annotations + +import pytest + +from bambu_ai.poller import fmt_state, merge_delta + + +class TestMergeDelta: + def test_preserves_existing_keys_when_payload_omits_them(self) -> None: + """Bambu sends ``nozzle_target_temper`` only when it changes; subsequent + delta updates carry just ``nozzle_temper``. The snapshot must keep the + prior target value.""" + snap = {"nozzle_target_temper": 220.0, "bed_target_temper": 65.0} + merge_delta(snap, {"nozzle_temper": 219.5}) + assert snap == { + "nozzle_target_temper": 220.0, + "bed_target_temper": 65.0, + "nozzle_temper": 219.5, + } + + def test_None_and_empty_values_are_skipped(self) -> None: + snap = {"existing": "value"} + merge_delta(snap, {"a": None, "b": "", "c": "ok", "existing": None}) + assert snap == {"existing": "value", "c": "ok"} + + def test_payload_overwrites_existing_value(self) -> None: + snap = {"nozzle_temper": 220.0} + merge_delta(snap, {"nozzle_temper": 221.5}) + assert snap["nozzle_temper"] == 221.5 + + +class TestFmtStateReadsFromSnapshot: + """The bug from issue #8 — formatter would print ``nozzle 220/0`` when the + delta payload didn't include the target field. After the fix, the formatter + reads from the merged snapshot which retains the target from the prior + pushall.""" + + def test_full_state_prints_both_temps(self) -> None: + snap = { + "gcode_state": "RUNNING", + "mc_percent": 50, + "nozzle_temper": 220.0, + "nozzle_target_temper": 220.0, + "bed_temper": 65.0, + "bed_target_temper": 65.0, + } + out = fmt_state(snap) + assert "state=RUNNING" in out + assert "50%" in out + assert "nozzle 220/220" in out + assert "bed 65/65" in out + + def test_after_delta_target_temp_persists(self) -> None: + """The actual regression — simulate the message sequence the printer sends.""" + snap: dict = {} + # Initial pushall response includes target temps. + merge_delta( + snap, + { + "gcode_state": "RUNNING", + "nozzle_temper": 220.0, + "nozzle_target_temper": 220.0, + "bed_temper": 65.0, + "bed_target_temper": 65.0, + }, + ) + # Subsequent delta: only the measured nozzle temp updates. + merge_delta(snap, {"nozzle_temper": 219.5}) + out = fmt_state(snap) + assert "nozzle 220/220" in out, f"target should persist, got: {out}" + assert "bed 65/65" in out, f"bed target should persist, got: {out}" + + def test_partial_state_works_without_pct_or_layer(self) -> None: + snap = {"gcode_state": "IDLE"} + out = fmt_state(snap) + assert "state=IDLE" in out + + def test_missing_target_falls_back_to_zero(self) -> None: + """Backward-compat: if a target was never seen, we still print measured.""" + snap = {"nozzle_temper": 25.0} + out = fmt_state(snap) + # Without the merged target, format shows 25/0. + assert "nozzle 25/0" in out + + +@pytest.mark.parametrize( + "layer,total,expected", + [ + (0, 0, False), # zero total — skipped + (5, 100, True), + (None, 100, False), + ], +) +def test_layer_formatting_handles_edge_cases(layer, total, expected) -> None: + snap = {"layer_num": layer, "total_layer_num": total} if layer is not None else {"total_layer_num": total} + out = fmt_state(snap) + if expected: + assert f"layer {layer}/{total}" in out + else: + assert "layer" not in out