From e4ac90b2b12104ce3ff205a16335405bfcca6a67 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 11:22:03 +0100 Subject: [PATCH 1/7] fix(moonraker): deliver ws callbacks on the GUI thread, fail them on disconnect, fetch spools via the Spoolman proxy --- BlocksScreen/lib/files.py | 2 +- BlocksScreen/lib/moonrakerComm.py | 50 ++++++++++++---- BlocksScreen/lib/moonrest.py | 18 +++--- tests/amu/test_moonrest_unit.py | 33 ++++------- tests/lib/test_moonraker_comm_unit.py | 85 ++++++++++++++++++++++++++- 5 files changed, 144 insertions(+), 44 deletions(-) diff --git a/BlocksScreen/lib/files.py b/BlocksScreen/lib/files.py index 916b7afa1..af5b11953 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -458,7 +458,7 @@ def request_print_duration(self, filename: str) -> None: job_id = str(metadata.job_id) if job_id in self._job_durations: return - # Mark before sending: error replies skip the callback. + # Mark before sending so an in-flight or failed lookup is not re-requested. self._job_durations[job_id] = None self.ws.api.history_get_job( job_id, diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index c59603629..ba84afd8c 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -42,6 +42,7 @@ class MoonWebSocket(QtCore.QObject, threading.Thread): klippy_state_signal = QtCore.pyqtSignal(str, name="klippy_state") query_server_info_signal = QtCore.pyqtSignal(name="query_server_information") server_components_signal = QtCore.pyqtSignal(list, name="server_components") + request_reply_signal = QtCore.pyqtSignal(object, object, name="request_reply") _KLIPPY_NOTIFY_METHODS: typing.ClassVar[frozenset[str]] = frozenset( { @@ -84,6 +85,7 @@ def __init__(self, parent: QtCore.QObject) -> None: ) self.klippy_state_signal.connect(self.api.request_printer_info) + self.request_reply_signal.connect(self._run_callback) logger.info("Websocket object initialized") @property @@ -201,6 +203,11 @@ def on_close(self, *args) -> None: _close_message = args[2] if len(args) == 3 else None with self._state_lock: self.connected = False + with self._request_lock: + callbacks = [e[2] for e in self.request_table.values() if e[2] is not None] + self.request_table.clear() + for callback in callbacks: + self.request_reply_signal.emit(callback, {"error": "disconnected"}) self.ws.keep_running = False self.connection_lost[str].emit( f"code: {_close_status_code} | message {_close_message}" @@ -302,12 +309,14 @@ def on_message(self, *args) -> None: else: _callback = _entry[2] if len(_entry) > 2 else None if _callback is not None: - if "error" not in response: - _callback(response.get("result")) - else: + if "error" in response: logger.error( - "WS request %s error: %s", _entry[0], response.get("error") + "WS request %s error: %s", _entry[0], response["error"] ) + _reply = {"error": response["error"]} + else: + _reply = response.get("result") + self.request_reply_signal.emit(_callback, _reply) return if "error" in response: message_event = WebSocketMessageReceived( @@ -349,6 +358,14 @@ def on_message(self, *args) -> None: "Unexpected error while creating websocket message event: %s", e ) + @QtCore.pyqtSlot(object, object, name="run_callback") + def _run_callback(self, callback, reply) -> None: + """Run a request callback on the GUI thread; a raise must not kill the app.""" + try: + callback(reply) + except Exception: + logger.exception("WS request callback %r failed", callback) + def send_request( self, method: str, params: dict | None = None, callback=None ) -> bool: @@ -357,9 +374,10 @@ def send_request( Args: method (str): Websocket method name params (dict, optional): parameters for the websocket method. Defaults to None. - callback (callable, optional): Called with ``response["result"]`` when the - response arrives. If None, the response is routed as a - ``WebSocketMessageReceived`` event to the parent widget. Defaults to None. + callback (callable, optional): Called on the GUI thread with + ``response["result"]``, or ``{"error": ...}`` on a JSON-RPC error. If None, + the response is routed as a ``WebSocketMessageReceived`` event to the + parent widget. Defaults to None. Returns: bool: Whether the method finished and a request was sent """ @@ -382,7 +400,14 @@ def send_request( "params": params, "id": _rid, } - _ws.send(json.dumps(packet)) + try: + _ws.send(json.dumps(packet)) + except (websocket.WebSocketException, OSError) as e: + logger.warning("WS request %s not sent: %s", method, e) + with self._request_lock: + drained = self.request_table.pop(_rid, None) is None + # on_close already failed a drained callback; False would answer it twice. + return drained and callback is not None return True @@ -923,10 +948,11 @@ def set_spool_id(self, spool_id: int | None, callback=None) -> bool: ) def get_spool(self, spool_id: int, callback) -> bool: - """Request spool data from Moonraker's Spoolman proxy""" - return self._ws.send_request( - method="server.spoolman.get_spool", - params={"spool_id": spool_id}, + """Request one spool through the Spoolman proxy; there is no get_spool RPC.""" + return self.spoolman_proxy( + request_method="GET", + path=f"/v1/spool/{spool_id}", + use_v2_response=False, callback=callback, ) diff --git a/BlocksScreen/lib/moonrest.py b/BlocksScreen/lib/moonrest.py index 437d98e51..36601d93a 100644 --- a/BlocksScreen/lib/moonrest.py +++ b/BlocksScreen/lib/moonrest.py @@ -77,19 +77,19 @@ def get_server_info(self): return self.get_request(method="server/info") def get_spool(self, spool_id: int) -> dict | None: - """Spoolman spool via Moonraker, None on error.""" - response = self.get_request(f"server/spoolman/spool/{spool_id}") + """Spoolman spool via the Moonraker proxy, None on error.""" + response = self.post_request( + "server/spoolman/proxy", + json={ + "request_method": "GET", + "path": f"/v1/spool/{spool_id}", + "use_v2_response": False, + }, + ) if not isinstance(response, dict): return None return response.get("result") - def set_spool_used_weight(self, spool_id: int, weight: float) -> bool: - """Set a spool's used_weight; True on success.""" - response = self.post_request( - f"server/spoolman/spool/{spool_id}", json={"used_weight": weight} - ) - return response is not None - def firmware_restart(self): """POST firmware_restart to Moonraker.""" return self.post_request(method="printer/firmware_restart") diff --git a/tests/amu/test_moonrest_unit.py b/tests/amu/test_moonrest_unit.py index 2291aad3d..64264cf78 100644 --- a/tests/amu/test_moonrest_unit.py +++ b/tests/amu/test_moonrest_unit.py @@ -19,35 +19,26 @@ def test_return_spool_dict_on_sucess(self, rest) -> None: "used_weight": 50.0, "remaining_weight": 200.0, } - with patch.object(rest, "get_request", return_value={"result": spool_data}): + with patch.object(rest, "post_request", return_value={"result": spool_data}): assert rest.get_spool(42) == spool_data def test_returns_non_on_http_error(self, rest) -> None: - with patch.object(rest, "get_request", return_value=None): + with patch.object(rest, "post_request", return_value=None): assert rest.get_spool(42) is None def test_returns_none_on_missing_result_key(self, rest): - with patch.object(rest, "get_request", return_value={"something": "else"}): + with patch.object(rest, "post_request", return_value={"something": "else"}): assert rest.get_spool(42) is None - def test_calls_correct_endpoint(self, rest) -> None: - with patch.object(rest, "get_request", return_value=None) as mock_get: - rest.get_spool(7) - mock_get.assert_called_once_with("server/spoolman/spool/7") - - -class TestSetSpoolUsedWeight: - def test_returns_true_on_sucess(self, rest) -> None: - with patch.object(rest, "post_request", return_value={"result": "ok"}): - assert rest.set_spool_used_weight(42, 75.5) - - def test_returns_false_on_error(self, rest) -> None: - with patch.object(rest, "post_request", return_value={"result": "ok"}): - assert rest.set_spool_used_weight(42, 75.5) - - def test_calls_correct_endpoint(self, rest) -> None: + # /server/spoolman/spool/{id} is a 404; only the proxy route exists. + def test_goes_through_the_proxy(self, rest) -> None: with patch.object(rest, "post_request", return_value=None) as mock_post: - rest.set_spool_used_weight(5, 100.0) + rest.get_spool(7) mock_post.assert_called_once_with( - "server/spoolman/spool/5", json={"used_weight": 100.0} + "server/spoolman/proxy", + json={ + "request_method": "GET", + "path": "/v1/spool/7", + "use_v2_response": False, + }, ) diff --git a/tests/lib/test_moonraker_comm_unit.py b/tests/lib/test_moonraker_comm_unit.py index 77e9608ce..2b1be9639 100644 --- a/tests/lib/test_moonraker_comm_unit.py +++ b/tests/lib/test_moonraker_comm_unit.py @@ -1,6 +1,7 @@ """Unit tests for moonrakerComm.""" import json +import threading from types import SimpleNamespace from unittest.mock import MagicMock @@ -8,7 +9,7 @@ from PyQt6 import QtCore from BlocksScreen.lib import moonrakerComm -from BlocksScreen.lib.moonrakerComm import OneShotTokenError +from BlocksScreen.lib.moonrakerComm import MoonAPI, OneShotTokenError class TestOneShotTokenError: @@ -120,6 +121,42 @@ def test_reconnect_reemits_unchanged_state(self, ws): assert states == ["ready", "ready"] +class TestRequestCallbacks: + def _reply(self, sock, callback, payload: dict) -> None: + sock.request_table[9] = ["server.spoolman.proxy", {}, callback] + sock.on_message(None, json.dumps({"jsonrpc": "2.0", "id": 9, **payload})) + + def test_result_reaches_callback(self, ws): + got = [] + self._reply(ws, got.append, {"result": {"response": [1], "error": None}}) + assert got == [{"response": [1], "error": None}] + + def test_jsonrpc_error_reaches_callback(self, ws): + got = [] + err = {"code": 503, "message": "Spoolman server not available"} + self._reply(ws, got.append, {"error": err}) + assert got == [{"error": err}] + + def test_raising_callback_is_logged_not_raised(self, ws, caplog): + def boom(_result): + raise RuntimeError("boom") + + self._reply(ws, boom, {"result": {}}) + assert "callback" in caplog.text + + def test_callback_runs_on_gui_thread(self, ws, qtbot): + threads = [] + + def record(_result): + threads.append(threading.current_thread()) + + worker = threading.Thread(target=self._reply, args=(ws, record, {"result": {}})) + worker.start() + worker.join() + qtbot.waitUntil(lambda: bool(threads)) + assert threads == [threading.main_thread()] + + class TestReconnectCycle: def test_retries_forever_until_connected(self, ws, monkeypatch): connect = MagicMock(return_value=False) @@ -192,3 +229,49 @@ def test_close_leaves_reconnect_to_watchdog(self, ws, monkeypatch, qtbot): ws.on_close(None, 1006, "gone") qtbot.wait(50) connect.assert_not_called() + + def test_close_fails_pending_requests(self, ws, qtbot): + ws.ws = MagicMock() + got = [] + ws.request_table[3] = ["server.info", {}, None] + ws.request_table[4] = ["server.spoolman.proxy", {}, got.append] + ws.on_close(None, 1006, "gone") + assert ws.request_table == {} + qtbot.waitUntil(lambda: bool(got)) + assert got == [{"error": "disconnected"}] + + def test_failed_send_returns_false_and_forgets_request(self, ws, monkeypatch): + # conftest stubs websocket; an except clause needs a real class + monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError) + ws.connected = True + ws.ws = MagicMock() + ws.ws.send.side_effect = OSError("broken pipe") + assert ws.send_request("server.info", callback=MagicMock()) is False + assert ws.request_table == {} + + +@pytest.fixture +def api(): + """MoonAPI over a stub socket; it parents itself to ws, so ws must be a QObject.""" + sock = QtCore.QObject() + sock.send_request = MagicMock(return_value=True) + return MoonAPI(sock) + + +class TestGetSpool: + def test_goes_through_the_proxy(self, api) -> None: + api.get_spool(42, None) + params = api._ws.send_request.call_args.kwargs + assert params["method"] == "server.spoolman.proxy" + assert params["params"] == { + "use_v2_response": False, + "request_method": "GET", + "path": "/v1/spool/42", + } + + def test_passes_the_callback_through(self, api) -> None: + def cb(_result): + return None + + api.get_spool(1, cb) + assert api._ws.send_request.call_args.kwargs["callback"] is cb From 70ea67064c007dea8a37249d696b807d88f067df Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 17:16:44 +0100 Subject: [PATCH 2/7] fix(moonraker): keep error replies from blanking AMU gates, unify the error shape, retry lost history lookups --- BlocksScreen/devices/amu/manager.py | 3 ++- BlocksScreen/lib/files.py | 9 ++++++--- BlocksScreen/lib/moonrakerComm.py | 24 +++++++++++++++--------- tests/amu/test_manager_unit.py | 18 ++++++++++++++++++ tests/lib/test_moonraker_comm_unit.py | 26 +++++++++++++++++++++++++- tests/widgets/test_files_data_unit.py | 14 ++++++++++++++ 6 files changed, 80 insertions(+), 14 deletions(-) diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index 010d41b6b..daeeee3e5 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -165,7 +165,8 @@ def fetch_spool(self, gate: int, spool_id: int) -> None: return def _on_result(result: dict | None) -> None: - if result is not None: + # An error reply parses as an empty spool and would blank the gate. + if result and "error" not in result: self.spool_fetched.emit(gate, result) self._ws.api.get_spool(spool_id, _on_result) diff --git a/BlocksScreen/lib/files.py b/BlocksScreen/lib/files.py index af5b11953..3f734eda5 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -458,21 +458,24 @@ def request_print_duration(self, filename: str) -> None: job_id = str(metadata.job_id) if job_id in self._job_durations: return - # Mark before sending so an in-flight or failed lookup is not re-requested. + # Mark before sending so an in-flight lookup is not re-requested. self._job_durations[job_id] = None - self.ws.api.history_get_job( + sent = self.ws.api.history_get_job( job_id, lambda result, name=filename, uid=job_id: self._history_job.emit( name, uid, result or {} ), ) + if not sent: + self._job_durations.pop(job_id, None) @QtCore.pyqtSlot(str, str, dict, name="on_history_job") def _on_history_job(self, filename: str, job_id: str, result: dict) -> None: """Cache a completed job's duration and re-emit fileinfo.""" job = result.get("job") or {} status = job.get("status") - if status == "in_progress": + # Unmarked so the next view asks again: a lost reply or a running job. + if "error" in result or status == "in_progress": self._job_durations.pop(job_id, None) return duration = job.get("print_duration") diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index ba84afd8c..fe505152e 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -207,7 +207,9 @@ def on_close(self, *args) -> None: callbacks = [e[2] for e in self.request_table.values() if e[2] is not None] self.request_table.clear() for callback in callbacks: - self.request_reply_signal.emit(callback, {"error": "disconnected"}) + # Same shape as a Moonraker JSON-RPC error; -1 marks it client-side. + error = {"code": -1, "message": "disconnected"} + self.request_reply_signal.emit(callback, {"error": error}) self.ws.keep_running = False self.connection_lost[str].emit( f"code: {_close_status_code} | message {_close_message}" @@ -374,12 +376,14 @@ def send_request( Args: method (str): Websocket method name params (dict, optional): parameters for the websocket method. Defaults to None. - callback (callable, optional): Called on the GUI thread with - ``response["result"]``, or ``{"error": ...}`` on a JSON-RPC error. If None, - the response is routed as a ``WebSocketMessageReceived`` event to the - parent widget. Defaults to None. + callback (callable, optional): Called once on the GUI thread with + ``response["result"]``, or with ``{"error": {"code": int, "message": str}}`` + on a JSON-RPC error or a disconnect (code -1); check for ``"error"`` + before reading the result. If None, the response is routed as a + ``WebSocketMessageReceived`` event to the parent widget. Defaults to None. Returns: - bool: Whether the method finished and a request was sent + bool: True if sent; the callback then runs exactly once. False if not + sent; the callback never runs, so the caller handles the failure. """ if params is None: params = {} @@ -405,9 +409,11 @@ def send_request( except (websocket.WebSocketException, OSError) as e: logger.warning("WS request %s not sent: %s", method, e) with self._request_lock: - drained = self.request_table.pop(_rid, None) is None - # on_close already failed a drained callback; False would answer it twice. - return drained and callback is not None + queued = self.request_table.pop(_rid, None) is not None + if queued or callback is None: + return False + # on_close already failed it; False would make the caller answer it twice. + return True return True diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index fe4625447..aa8d37146 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -381,6 +381,24 @@ def test_no_emit_on_ws_failure(self, manager, qtbot) -> None: with qtbot.assertNotEmitted(manager.spool_fetched): callback(None) + @pytest.mark.parametrize( + "error", + [ + {"code": 404, "message": "spool not found"}, + {"code": -1, "message": "disconnected"}, + ], + ) + def test_error_reply_leaves_the_gate_alone(self, manager, qtbot, error) -> None: + """An error read as spool data would send MATERIAL= COLOR= SPOOLID=-1.""" + manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) + manager.fetch_spool(0, 42) + callback = manager._ws.api.get_spool.call_args.args[1] + with ( + qtbot.assertNotEmitted(manager.spool_fetched), + qtbot.assertNotEmitted(manager.run_gcode_signal), + ): + callback({"error": error}) + class TestLoadCellUpdate: def test_noop_when_mmu_state_none(self, manager) -> None: diff --git a/tests/lib/test_moonraker_comm_unit.py b/tests/lib/test_moonraker_comm_unit.py index 2b1be9639..044c022bb 100644 --- a/tests/lib/test_moonraker_comm_unit.py +++ b/tests/lib/test_moonraker_comm_unit.py @@ -238,7 +238,7 @@ def test_close_fails_pending_requests(self, ws, qtbot): ws.on_close(None, 1006, "gone") assert ws.request_table == {} qtbot.waitUntil(lambda: bool(got)) - assert got == [{"error": "disconnected"}] + assert got == [{"error": {"code": -1, "message": "disconnected"}}] def test_failed_send_returns_false_and_forgets_request(self, ws, monkeypatch): # conftest stubs websocket; an except clause needs a real class @@ -249,6 +249,30 @@ def test_failed_send_returns_false_and_forgets_request(self, ws, monkeypatch): assert ws.send_request("server.info", callback=MagicMock()) is False assert ws.request_table == {} + def test_send_racing_close_answers_the_callback_once(self, ws, qtbot, monkeypatch): + """on_close already failed the callback, so True keeps the caller from a 2nd.""" + monkeypatch.setattr(moonrakerComm.websocket, "WebSocketException", OSError) + ws.connected = True + ws.ws = MagicMock() + + def close_then_fail(_packet): + ws.on_close(None, 1006, "gone") + raise OSError("broken pipe") + + ws.ws.send.side_effect = close_then_fail + got = [] + assert ws.send_request("server.info", callback=got.append) is True + qtbot.waitUntil(lambda: bool(got)) + qtbot.wait(20) + assert got == [{"error": {"code": -1, "message": "disconnected"}}] + + def test_send_while_disconnected_never_calls_back(self, ws, qtbot): + callback = MagicMock() + assert ws.send_request("server.info", callback=callback) is False + assert ws.request_table == {} + qtbot.wait(20) + callback.assert_not_called() + @pytest.fixture def api(): diff --git a/tests/widgets/test_files_data_unit.py b/tests/widgets/test_files_data_unit.py index fd01fea45..4920ec926 100644 --- a/tests/widgets/test_files_data_unit.py +++ b/tests/widgets/test_files_data_unit.py @@ -186,6 +186,20 @@ def test_in_progress_job_is_asked_again(self, files): files.request_print_duration("a.gcode") assert files.ws.api.history_get_job.call_count == 2 + def test_error_reply_is_asked_again(self, files): + """A reply lost to a disconnect must not hide the duration until restart.""" + files.request_print_duration("a.gcode") + error = {"code": -1, "message": "disconnected"} + files._on_history_job("a.gcode", "000023", {"error": error}) + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 2 + + def test_unsent_lookup_is_asked_again(self, files): + files.ws.api.history_get_job.return_value = False + files.request_print_duration("a.gcode") + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 2 + class TestFilelistNotifications: def test_every_batched_entry_is_applied(self, files): From 8cd5b58e16a21a5faf15169e313257f7152407d4 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 11:22:05 +0100 Subject: [PATCH 3/7] fix(amu): rebuild the MMU state model, guard Happy-Hare v3 gcode and make the AMU config toggle atomic --- BlocksScreen/devices/amu/__init__.py | 15 +- BlocksScreen/devices/amu/config_toggler.py | 148 +++-- BlocksScreen/devices/amu/manager.py | 452 +++++++------- BlocksScreen/devices/amu/models.py | 282 +++++---- BlocksScreen/lib/panels/mainWindow.py | 1 + tests/amu/conftest.py | 46 +- tests/amu/test_config_toggler_unit.py | 211 ++++++- tests/amu/test_manager_unit.py | 652 ++++++++++++++------- tests/amu/test_models_unit.py | 169 ++++-- 9 files changed, 1263 insertions(+), 713 deletions(-) diff --git a/BlocksScreen/devices/amu/__init__.py b/BlocksScreen/devices/amu/__init__.py index 399afa9e9..b9d5b22d2 100644 --- a/BlocksScreen/devices/amu/__init__.py +++ b/BlocksScreen/devices/amu/__init__.py @@ -1,13 +1,15 @@ -"""AMU / HAPPY-Hare MMU package +"""AMU / Happy-Hare MMU package. Architecture: AMUManager (manager.py) - └── Main AMU manager its responsible for the AMU backend + └── AMU backend Models (models.py) - └── Data classes for type safety - └── Enums for states and types + └── Typed dataclasses and state enums + ConfigToggler (config_toggler.py) + └── AMU/non-AMU printer.cfg switch """ +from .config_toggler import ConfigToggler, ToggleResult from .manager import AMUManager from .models import ( FilamentPos, @@ -16,14 +18,11 @@ MMUState, SpoolmanSupport, ) -from .config_toggler import ConfigToggler __all__: list[str] = [ - # manager "AMUManager", - # config_toggler "ConfigToggler", - # models + "ToggleResult", "FilamentPos", "GateInfo", "GateStatus", diff --git a/BlocksScreen/devices/amu/config_toggler.py b/BlocksScreen/devices/amu/config_toggler.py index 75e574f6d..9123be9de 100644 --- a/BlocksScreen/devices/amu/config_toggler.py +++ b/BlocksScreen/devices/amu/config_toggler.py @@ -1,66 +1,146 @@ +"""Swap printer.cfg between the AMU and single-filament variant includes.""" + +import contextlib import logging +import os import re +from enum import StrEnum from pathlib import Path logger = logging.getLogger(__name__) -AMU_FILES: list[str] = [ - "mmu/base/*.cfg", - "mmu/optional/client_macros.cfg", - "filament_manager.cfg", -] +AMU_INCLUDE: str = "config/variant_mmu/base/*.cfg" +SINGLE_INCLUDE: str = "config/variant_sync_single/*.cfg" + + +class ToggleResult(StrEnum): + """Outcome of a toggle; only FAILED is an error and only CHANGED needs a restart.""" + + CHANGED = "changed" + UNCHANGED = "unchanged" + FAILED = "failed" -_AMU_PATTERNS: list[re.Pattern[str]] = [ - re.compile(rf"^(#?)(\[include {re.escape(f)}\])", re.MULTILINE) for f in AMU_FILES -] + +def _include_pattern(include: str) -> re.Pattern[str]: + """Match the include commented or not, tolerating stray spaces in the brackets.""" + body = rf"\[include[ \t]+{re.escape(include)}[ \t]*\]" + return re.compile(rf"^([ \t]*)(#*)[ \t]*({body})[ \t]*(\r?)$", re.MULTILINE) + + +_PATTERNS: dict[str, re.Pattern[str]] = { + AMU_INCLUDE: _include_pattern(AMU_INCLUDE), + SINGLE_INCLUDE: _include_pattern(SINGLE_INCLUDE), +} class ConfigToggler: - """Manages commenting/uncommenting AMU cfg includes in printer.cfg""" + """Swaps printer.cfg between the AMU and single-filament variant includes.""" def __init__(self, config_path: Path) -> None: - """Store path, detects current state from file""" + """Resolve printer.cfg and read the active variant from it.""" self._path: Path | None = None self._state: bool = False if config_path.exists(): - self._path = config_path + # os.replace() would swap a symlinked printer.cfg for a regular file. + self._path = config_path.resolve() self._state = self._detect_state() else: logger.warning("Config File not found %s", config_path) - def _detect_state(self) -> bool: + def _read(self) -> str | None: + """Return printer.cfg contents, or None when there is no readable file.""" if self._path is None: - return False + return None try: - text = self._path.read_text() - return all( - re.search(rf"^\[include {re.escape(f)}\]", text, re.MULTILINE) - for f in AMU_FILES - ) - except OSError: - return False + # newline="" keeps CRLF intact; surrogateescape survives non-UTF-8 bytes. + with self._path.open( + encoding="utf-8", errors="surrogateescape", newline="" + ) as f: + return f.read() + except OSError as e: + logger.error("ConfigToggler: read of %s failed: %s", self._path, e) + return None - def toggle(self, activate: bool) -> bool: - """Comment or uncomment AMU includes. Returns True on success, False on no-op or error.""" + def _write(self, text: str) -> bool: + """Write through a sibling temp file so a crash cannot truncate printer.cfg.""" if self._path is None: - logger.warning("No Config File available") return False - if self._state == activate: + tmp = self._path.with_name(self._path.name + ".tmp") + try: + with tmp.open( + "w", encoding="utf-8", errors="surrogateescape", newline="" + ) as f: + f.write(text) + f.flush() + os.fsync(f.fileno()) + os.chmod(tmp, self._path.stat().st_mode & 0o777) + os.replace(tmp, self._path) + except OSError as e: + logger.error("ConfigToggler: write of %s failed: %s", self._path, e) + with contextlib.suppress(OSError): + tmp.unlink(missing_ok=True) return False - replacement = r"\2" if activate else r"#\2" try: - text = self._path.read_text() - for pattern in _AMU_PATTERNS: - text = pattern.sub(replacement, text) - self._path.write_text(text) - self._state = activate - return True + dir_fd = os.open(self._path.parent, os.O_RDONLY) + try: + os.fsync(dir_fd) + finally: + os.close(dir_fd) except OSError as e: + logger.warning("ConfigToggler: dir fsync of %s failed: %s", self._path, e) + return True + + @staticmethod + def _is_active(text: str, include: str) -> bool: + """True when the include is present and not commented out.""" + match = _PATTERNS[include].search(text) + return match is not None and not match.group(2) + + def _detect_state(self) -> bool: + """Read the AMU variant include state straight off disk.""" + text = self._read() + return False if text is None else self._is_active(text, AMU_INCLUDE) + + def toggle(self, activate: bool) -> ToggleResult: + """Activate one variant include and comment the other, atomically.""" + text = self._read() + if text is None: + logger.warning("No Config File available") + return ToggleResult.FAILED + self._state = self._is_active(text, AMU_INCLUDE) + if self._state == activate: + return ToggleResult.UNCHANGED + + enable = AMU_INCLUDE if activate else SINGLE_INCLUDE + disable = SINGLE_INCLUDE if activate else AMU_INCLUDE + new_text, enabled = _PATTERNS[enable].subn(r"\1\3\4", text) + new_text, disabled = _PATTERNS[disable].subn(r"\1#\3\4", new_text) + + if enabled != 1 or disabled != 1: logger.error( - "ConfigToggler.toggle(activate=%s): read/write failed: %s", activate, e + "ConfigToggler.toggle(activate=%s): expected one match each, got " + "%d for %s and %d for %s", + activate, + enabled, + enable, + disabled, + disable, ) - return False + return ToggleResult.FAILED + + if not self._is_active(new_text, enable) or self._is_active(new_text, disable): + logger.error( + "ConfigToggler.toggle(activate=%s): result would not leave exactly one " + "variant active, refusing to write", + activate, + ) + return ToggleResult.FAILED + + if not self._write(new_text): + return ToggleResult.FAILED + self._state = activate + return ToggleResult.CHANGED def is_configured(self) -> bool: - """Return True if AMU includes are currently uncommented""" + """Return True if the AMU variant include is uncommented.""" return self._state diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index daeeee3e5..928c128e6 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -1,14 +1,23 @@ +"""Qt-side facade for the AMU: owns MMU state and issues Happy-Hare gcode.""" + from __future__ import annotations -import dataclasses import logging +import math +import shlex import typing from pathlib import Path from PyQt6 import QtCore -from .config_toggler import ConfigToggler -from .models import MMUState, SpoolmanSupport +from .config_toggler import ConfigToggler, ToggleResult +from .models import ( + FilamentPos, + GateInfo, + GateStatus, + MMUState, + SpoolmanSupport, +) if typing.TYPE_CHECKING: from BlocksScreen.lib.moonrakerComm import MoonWebSocket @@ -16,20 +25,20 @@ logger: logging.Logger = logging.getLogger(__name__) -# Spool Weight threshold for heavy-filament speed profile (grams) -HEAVY_SPOOL_THRESHOLD_G: float = 1000.0 -# Absolute target speed (mm/s) for heavy spools - used to calculate SPEED % for MMU_GATE_MAP -HEAVY_SPEED_MM_S: float = 100.0 -# Base gear stepper max_velocity (mm/s) - must match mmu_gear max_velocity in printer.cfg -BASE_GEAR_SPEED_MM_S: float = 300.0 -# Precomputed speed percentage for heavy spools (avoids repeated division at runtime) -_HEAVY_SPEED_PERCENT: int = max(1, round(HEAVY_SPEED_MM_S / BASE_GEAR_SPEED_MM_S * 100)) - CONFIG_PATH: Path = Path("~/printer_data/config/printer.cfg").expanduser() +def _gcode_value(value: object) -> str: + """Quote a value for Klipper's shlex-based extended gcode parser.""" + if isinstance(value, float): + value = round(value) + # A raw newline ends the gcode line; '%' crashes save_variables' configparser. + text = " ".join(str(value).replace("%", "").split()) + return shlex.quote(text) + + class AMUManager(QtCore.QObject): - """Main manager of the AMU system""" + """Owns the MMU state and turns UI intents into guarded Happy-Hare gcode.""" run_gcode_signal: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( str, name="run-gcode" @@ -46,12 +55,6 @@ class AMUManager(QtCore.QObject): pre_gate_changed: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( int, bool, name="pre-gate-changed" ) - spool_fetched: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( - int, dict, name="spool-fetched" - ) - gate_weight_updated: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( - int, float, name="gate-weight-updated" - ) def __init__(self, ws: MoonWebSocket, parent: QtCore.QObject | None = None) -> None: super().__init__(parent) @@ -59,87 +62,43 @@ def __init__(self, ws: MoonWebSocket, parent: QtCore.QObject | None = None) -> N self._ws = ws self._mmu_state: MMUState | None = None self._pre_gate_sensors: dict[int, bool] = {} - self.spool_fetched.connect(self._apply_spool_data) - - def _apply_spool_data(self, gate: int, data: dict) -> None: - """Apply Spoolman spool data to local gate state and sync to Klipper. - - Extracts material, color, weight from the Spoolman response dict, - emits MMU_GATE_MAP gcode to sync Klipper, and updates the local GateInfo weight. - """ - if self._mmu_state is None: - return - filament = data.get("filament", {}) - material = filament.get("material", "") - color = filament.get("color_hex", "") - filament_name = filament.get("name", "") - temperature = filament.get("settings_extruder_temp") - bed_temp = filament.get("settings_bed_temp") - spool_id = data.get("id", -1) - weight = data.get("used_weight") - remaining_weight = data.get("remaining_weight") - self.set_gate_info( - gate, - material, - color, - spool_id, - filament_name=filament_name, - temperature=temperature, + self._print_state = "" + + def _gate_map(self, gate: int, **params: object) -> bool: + """Emit MMU_GATE_MAP for one gate; False when no gate is selected yet.""" + if gate < 0: + logger.warning("Ignoring gate command for unselected gate %d", gate) + return False + # TEMP=0 keeps the current temp; TEMP=None omits it, resetting to the default. + params.setdefault("TEMP", 0) + args = "".join( + f" {key}={_gcode_value(value)}" + for key, value in params.items() + if value is not None ) - gates = list(self._mmu_state.gates) - if gate >= len(gates): - logger.warning("Gate index %d out of range (%d gate)", gate, len(gates)) - return - updates = {} - if weight is not None: - updates["weight_g"] = float(weight) - if remaining_weight is not None: - updates["remaining_weight"] = float(remaining_weight) - self._emit_speed_gcode(gate, remaining_weight) - if filament_name: - updates["filament_name"] = filament_name - if temperature is not None: - updates["temperature"] = float(temperature) - if bed_temp is not None: - updates["bed_temp"] = int(bed_temp) - if updates: - gates[gate] = dataclasses.replace(gates[gate], **updates) - self._mmu_state = dataclasses.replace(self._mmu_state, gates=tuple(gates)) - self.mmu_state_changed.emit(self._mmu_state) - - def _emit_speed_gcode(self, gate: int, remaining_weight: float) -> None: - """Emit MMU_GATE_MAP SPEED=x for the gate based on the spool weight profile.""" - if remaining_weight > HEAVY_SPOOL_THRESHOLD_G: - self.run_gcode_signal.emit( - f"MMU_GATE_MAP gate={gate} SPEED={_HEAVY_SPEED_PERCENT}" - ) + self.run_gcode_signal.emit(f"MMU_GATE_MAP GATE={gate}{args} QUIET=1") + return True def toggle_amu_system(self, activate: bool) -> None: - """Enable or disable the AMU system by commenting/uncommenting config includes. - - Emits: - amu_toggled (bool): True if the operation succeeded, False otherwise. - - Args: - activate (bool): True to enable the AMU, False to disable it. - - """ - result: bool = self._config_toggler.toggle(activate) - self.amu_toggled.emit(result) - if result: - self.run_gcode_signal.emit("FIRMWARE_RESTART") + """Swap the printer.cfg variant, restarting on change; emits amu_toggled.""" + if self._print_state in ("printing", "paused"): + logger.warning( + "Ignoring AMU toggle while the print is %s", self._print_state + ) + self.amu_toggled.emit(False) + return + result = self._config_toggler.toggle(activate) + self.amu_toggled.emit(result is not ToggleResult.FAILED) + if result is ToggleResult.CHANGED: + # FIRMWARE_RESTART as gcode drops the socket and replies a false 503. + self._ws.api.firmware_restart() def get_state(self) -> MMUState | None: - """Returns current MMU state, None if not yet received. - - Returns: - MMUState: Latest state received from Moonraker - None: If no state has been received yet. - - """ + """Return the latest MMU state, or None before the first status arrives.""" return self._mmu_state def get_pre_gate_sensors(self) -> dict[int, bool]: + """Return a copy of the per-gate pre-gate switch states, keyed by gate index.""" return dict(self._pre_gate_sensors) def is_amu_configured(self) -> bool: @@ -147,29 +106,9 @@ def is_amu_configured(self) -> bool: return self._config_toggler.is_configured() def is_amu_active(self) -> bool: - """Returns whether AMU includes are currently uncommented in printer.cfg""" - return self.is_amu_configured() and self._mmu_state is not None - - def fetch_spool(self, gate: int, spool_id: int) -> None: - """Request spool data from Moonraker via WebSocket. - - No-op if MMU state not received or spoolman_support is OFF. - Emits spool_fetched(gate, data) on success; logs and emits nothing on error. - """ - - if self._mmu_state is None: - return - if self._mmu_state.spoolman_support is SpoolmanSupport.OFF: - return - if spool_id == -1: - return - - def _on_result(result: dict | None) -> None: - # An error reply parses as an empty spool and would blank the gate. - if result and "error" not in result: - self.spool_fetched.emit(gate, result) - - self._ws.api.get_spool(spool_id, _on_result) + """Return True if the AMU is configured, reporting, and not MMU-disabled.""" + state = self._mmu_state + return self.is_amu_configured() and state is not None and state.enabled def set_gate_info( self, @@ -179,147 +118,162 @@ def set_gate_info( spool_id: int, filament_name: str = "", temperature: int | None = None, - ) -> None: - """Sets all gate attributes for a single MMU_GATE - - Args: - gate (int): Gate index (0-based). - material (str): Filament material name, e.g. ``"PLA"``. - color (str): Filament color as hex string, e.g. ``"ff56e0"``. - spool_id (int): Spoolman spool ID, or -1 if not tracked. - filament_name (str): Filament display name from Spoolman. - temperature (int | None): Extruder temperature, omitted if None. - """ - gcode = f"MMU_GATE_MAP gate={gate} MATERIAL={material} COLOR={color} SPOOLID={spool_id}" - if filament_name: - gcode = f"MMU_GATE_MAP gate={gate} NAME={filament_name} MATERIAL={material} COLOR={color} SPOOLID={spool_id}" - if temperature is not None: - gcode += f" TEMP={temperature}" - self.run_gcode_signal.emit(gcode) - - def set_gate_material(self, gate: int, material: str) -> None: - """Set the `material` at the gate `gate` - - Args: - gate (int): Gate index (0-based). - material (str): Filament material name, e.g. ``"PLA"``. - """ - self.run_gcode_signal.emit(f"MMU_GATE_MAP gate={gate} MATERIAL={material}") + ) -> bool: + """Map every field of *gate*; a None temperature takes Happy-Hare's default.""" + return self._gate_map( + gate, + NAME=filament_name, + MATERIAL=material, + COLOR=color.lstrip("#"), + SPOOLID=spool_id, + TEMP=temperature, + ) - def set_gate_temp(self, gate: int, temp: int): - """Set the `temperature` at the gate `gate` + def clear_gate(self, gate: int) -> None: + """Blank a gate's map entry the way Happy-Hare resets one.""" + self._gate_map(gate, NAME="", MATERIAL="", COLOR="", SPOOLID=-1, TEMP=None) - Args: - gate (int): Gate index (0-based). - temp (int): Filament temperature, e.g. ``"220"``. - """ - self.run_gcode_signal.emit(f"MMU_GATE_MAP gate={gate} TEMP={temp}") + def set_gate_material(self, gate: int, material: str) -> None: + """Set the material of *gate*, keeping its temperature.""" + self._gate_map(gate, MATERIAL=material) + + def set_gate_temp(self, gate: int, temp: int) -> None: + """Set *gate*'s temp; skips TEMP<=0 since Happy-Hare reads 0 as "keep".""" + state = self._mmu_state + if state is not None and state.spoolman_support is SpoolmanSupport.PULL: + logger.warning("Ignoring gate %d temperature: Spoolman pull owns it", gate) + return + if not math.isfinite(temp) or round(temp) <= 0: + logger.warning("Ignoring gate %d temperature %s: not positive", gate, temp) + return + self._gate_map(gate, TEMP=temp) def set_gate_color(self, gate: int, color: str) -> None: - """Set the `color` at the gate `gate` - - Args: - gate (int): Gate index (0-based). - color (str): Filament color, e.g. ``"ff56e0"``. - """ - self.run_gcode_signal.emit(f"MMU_GATE_MAP gate={gate} COLOR={color}") + """Set the hex colour of *gate*, with or without a leading #.""" + self._gate_map(gate, COLOR=color.lstrip("#")) def set_gate_spool(self, gate: int, spool_id: int) -> None: - """Set the `spool_id` at the gate `gate` - - Args: - gate (int): Gate index (0-based). - spool_id (int): Spoolman spool ID, or -1 to clear. - """ - self.run_gcode_signal.emit(f"MMU_GATE_MAP gate={gate} SPOOLID={spool_id}") - if spool_id != -1: - self.fetch_spool(gate, spool_id) - - def update_spool_weight(self, gate: int, used_weight: float) -> None: - """Push updated used_weight to Spoolman for the spool at `gate`. - - No-op if MMU state not received, spoolman is off or read-only, gate is - out of range, of the gate has no spool assigned. - """ - if self._mmu_state is None: - return - if self._mmu_state.spoolman_support in ( - SpoolmanSupport.OFF, - SpoolmanSupport.READONLY, - ): - return - if gate >= len(self._mmu_state.gates): - logger.warning("update_spool_weight: gate %d out of range", gate) - return - spool_id = self._mmu_state.gates[gate].spool_id - if spool_id == -1: - return - self._ws.api.update_spool(spool_id, {"used_weight": used_weight}) + """Bind *gate* to a Spoolman spool; -1 unbinds.""" + self._gate_map(gate, SPOOLID=spool_id) + + def assign_spool(self, gate: int, spool: dict) -> None: + """Map *gate* from a Spoolman spool payload.""" + filament = spool.get("filament") or {} + self.set_gate_info( + gate, + filament.get("material") or "", + filament.get("color_hex") or "", + spool.get("id", -1), + filament_name=filament.get("name") or "", + temperature=filament.get("settings_extruder_temp"), + ) def home_mmu(self) -> None: - """Home the MMU selector by sending MMU_HOME.""" - self.run_gcode_signal.emit("MMU_HOME") + """Re-select the current tool via MMU_HOME; refused while filament is loaded.""" + state = self._mmu_state + if state is None: + logger.warning("Ignoring MMU_HOME: no MMU state yet") + return + # Bare MMU_HOME defaults to TOOL=0 with no loaded guard. + if state.filament_pos not in (FilamentPos.UNLOADED, FilamentPos.UNKNOWN): + logger.warning( + "Ignoring MMU_HOME while filament is %s", state.filament_pos.name + ) + return + tool = state.tool + self.run_gcode_signal.emit(f"MMU_HOME TOOL={tool}" if tool >= 0 else "MMU_HOME") def reset_mmu(self) -> None: - """Reset the MMU and clear any pause or error state by sending MMU_RESET.""" - self.run_gcode_signal.emit("MMU_RESET") - - def load_gate(self) -> None: - """Load filament from the specified gate by sending MMU_LOAD""" + """Erase all persisted MMU state (gate map, TTG map, statistics) and re-init.""" + self.run_gcode_signal.emit("MMU_RESET CONFIRM=1") + + def _selected_gate_unloaded(self) -> GateInfo | None: + """Return the selected gate when the MMU is idle-ready: enabled, homed, unloaded.""" + state = self._mmu_state + if ( + state is None + or not (state.enabled and state.is_homed) + or state.is_printing + or state.filament_pos != FilamentPos.UNLOADED + ): + return None + return state.current_gate_info + + def load_gate(self) -> bool: + """Load filament from the selected gate by sending MMU_LOAD; True if sent.""" + if self._selected_gate_unloaded() is None: + logger.warning("Ignoring MMU_LOAD: no homed, unloaded gate selected") + return False self.run_gcode_signal.emit("MMU_LOAD") - - def unload(self) -> None: - """Unload the currently loaded filament by sending MMU_UNLOAD.""" + return True + + def unload(self) -> bool: + """Unload the currently loaded filament by sending MMU_UNLOAD; True if sent.""" + state = self._mmu_state + if ( + state is None + or not state.enabled + or state.is_printing + or state.filament_pos == FilamentPos.UNLOADED + ): + logger.warning("Ignoring MMU_UNLOAD: MMU disabled, printing or unloaded") + return False self.run_gcode_signal.emit("MMU_UNLOAD") + return True def select_gate(self, gate: int) -> None: - """select the specified tool by sending MMU_SELECT - - Args: - gate (int): gate index to select (0-based) - """ + """Send MMU_SELECT unless invalid, already selected, or filament is loaded.""" + state = self._mmu_state + if ( + state is None + or not (state.enabled and state.is_homed) + or state.is_printing + or not 0 <= gate < state.num_gates + or gate == state.gate + ): + return + if state.filament_pos != FilamentPos.UNLOADED: + logger.warning( + "Ignoring MMU_SELECT while filament is %s", state.filament_pos.name + ) + return self.run_gcode_signal.emit(f"MMU_SELECT GATE={gate}") def eject_gate(self) -> None: - """Fully eject filament from gate, releasing from MMU gear.""" + """Fully eject filament from the selected gate, the only gate that can eject.""" + gate = self._selected_gate_unloaded() + if gate is None or gate.status == GateStatus.EMPTY: + logger.warning("Ignoring MMU_EJECT: no unloaded, non-empty gate selected") + return self.run_gcode_signal.emit("MMU_EJECT") def check_gate(self) -> None: - """Check the current gate for filament presence by sending MMU_CHECK_GATE.""" + """Check the selected gate for filament by sending MMU_CHECK_GATE.""" + # Happy-Hare runs a full unload first when not UNLOADED. + if self._selected_gate_unloaded() is None: + logger.warning("Ignoring MMU_CHECK_GATE: no unloaded gate selected") + return self.run_gcode_signal.emit("MMU_CHECK_GATE") - def eject_all_gates(self, num_gates: int) -> None: - """Fully eject filament from all gates sequentially - - Args: - num_gates: Total number of gates(from MMUState.num_gates) - """ - cmd: str = "\n".join(f"MMU_EJECT GATE={i}" for i in range(num_gates)) - self.run_gcode_signal.emit(cmd) + def recover_single_gate(self, loaded: bool) -> None: + """Tell Happy-Hare whether the single-gate filament path is loaded.""" + self.run_gcode_signal.emit(f"MMU_RECOVER GATE=0 LOADED={int(loaded)}") def change_tool(self, tool: int) -> None: - """Select a tool, triggering a filament change if needed. - - Args: - tool (int): Tool index to select (0-based). - """ + """Select *tool* with MMU_CHANGE_TOOL, swapping filament if needed.""" self.run_gcode_signal.emit(f"MMU_CHANGE_TOOL TOOL={tool}") - def update_mmu_state(self, data: dict, name: str = "") -> None: - """Receive an MMU status dict from Moonraker and update internal state. - - Called with either a full status response (on connect) or a diff - (from notify_status_update). Builds or updates the MMUState and - emits mmu_state_changed. - - Args: - data: Raw MMU status or diff dict from Moonraker. - name: Moonraker object name suffix (always empty for ``mmu``). - """ + def update_mmu_state(self, data: dict) -> None: + """Build state from a full status or apply a diff; emit mmu_state_changed.""" if self._mmu_state is None: + if "num_gates" not in data: + return self._mmu_state = MMUState.from_status(data) else: - self._mmu_state = self._mmu_state.apply_diff(data) + state = self._mmu_state.apply_diff(data) + if state == self._mmu_state: + return + self._mmu_state = state self.mmu_state_changed.emit(self._mmu_state) def on_object_updated( @@ -330,10 +284,11 @@ def on_object_updated( self.update_mmu_state(values) elif object_type == "filament_switch_sensor": self.on_pre_gate_update(values, object_name) - elif object_type == "load_cell": - self.on_load_cell_update(values, object_name) + elif object_type == "print_stats" and "state" in values: + self._print_state = values["state"] def on_pre_gate_update(self, values: dict, name: str) -> None: + """Track one pre-gate switch; the mmu aggregate hides unselected gates.""" if not name.startswith("mmu_pre_gate_"): return try: @@ -341,35 +296,20 @@ def on_pre_gate_update(self, values: dict, name: str) -> None: except ValueError: logger.error("Failed to parse Pre-Gate: %s", name) return - detected = bool(values.get("filament_detected", False)) + # Enabled-only diffs omit it; reading False would invent empty gates. + if "filament_detected" not in values: + return + detected = bool(values["filament_detected"]) self._pre_gate_sensors[gate] = detected self.pre_gate_changed.emit(gate, detected) - def on_load_cell_update(self, values: dict, name: str) -> None: - """Update gate weight from a Klipper load_cell sensor reading""" - if self._mmu_state is None or not name.startswith("load_cell_mmu_"): - return - try: - gate = int(name.removeprefix("load_cell_mmu_")) - except ValueError: - logger.error("Failed parsing %s Load cell", name) + @QtCore.pyqtSlot(str) + def on_klippy_state(self, state: str) -> None: + """Drop cached MMU state when klippy leaves ready and tell the UI it is gone.""" + if state.lower() == "ready": return - - weight = float(values.get("force") or 0) - if gate >= len(self._mmu_state.gates): - logger.warning( - "Gate index %d out of range (%d gates)", - gate, - len(self._mmu_state.gates), - ) + self._pre_gate_sensors = {} + if self._mmu_state is None: return - gates = list(self._mmu_state.gates) - gates[gate] = dataclasses.replace(gates[gate], weight_g=weight) - self._mmu_state = dataclasses.replace(self._mmu_state, gates=tuple(gates)) - self.gate_weight_updated.emit(gate, weight) - - def on_klippy_state(self, state: str) -> None: - """React to changes in klippy states""" - if state.lower() != "ready": - self._mmu_state = None - self._pre_gate_sensors = {} + self._mmu_state = None + self.mmu_state_changed.emit(None) diff --git a/BlocksScreen/devices/amu/models.py b/BlocksScreen/devices/amu/models.py index ddfd99b34..bf1c281fa 100644 --- a/BlocksScreen/devices/amu/models.py +++ b/BlocksScreen/devices/amu/models.py @@ -1,13 +1,35 @@ +"""Immutable value types mirroring the MMU status Happy-Hare publishes to Moonraker.""" + +from __future__ import annotations + import dataclasses +import logging +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from enum import IntEnum, StrEnum +logger = logging.getLogger(__name__) + + +def _rgb(value: Sequence[float]) -> tuple[float, float, float]: + """Coerce a colour sequence to a fixed 3-tuple, padding a short one with zeroes.""" + r, g, b = [*value, 0.0, 0.0, 0.0][:3] + return (float(r), float(g), float(b)) + class GateStatus(IntEnum): + """Gate filament availability, matching Happy-Hare's GATE_* codes (mmu.py:59).""" + + UNKNOWN = -1 EMPTY = 0 AVAILABLE = 1 AVAILABLE_FROM_BUFFER = 2 - UNKNOWN = -1 + + @classmethod + def _missing_(cls, value: object) -> GateStatus: + """Firmware can ship new codes; an unhandled ValueError would qFatal the GUI.""" + logger.warning("Unknown GateStatus %r, treating as UNKNOWN", value) + return cls.UNKNOWN class FilamentPos(IntEnum): @@ -26,6 +48,12 @@ class FilamentPos(IntEnum): IN_EXTRUDER = 9 LOADED = 10 + @classmethod + def _missing_(cls, value: object) -> FilamentPos: + """Firmware can ship new codes; an unhandled ValueError would qFatal the GUI.""" + logger.warning("Unknown FilamentPos %r, treating as UNKNOWN", value) + return cls.UNKNOWN + class SpoolmanSupport(StrEnum): """Level of Spoolman integration configured in Happy-Hare.""" @@ -35,21 +63,55 @@ class SpoolmanSupport(StrEnum): PUSH = "push" PULL = "pull" + @classmethod + def _missing_(cls, value: object) -> SpoolmanSupport: + """OFF is the safe default: it suppresses Spoolman writes, not guesses.""" + logger.warning("Unknown SpoolmanSupport %r, treating as OFF", value) + return cls.OFF + + +_GATE_ARRAYS: dict[str, str] = { + "gate_status": "status", + "gate_material": "material", + "gate_color": "color", + "gate_color_rgb": "color_rgb", + "gate_spool_id": "spool_id", + "gate_filament_name": "filament_name", + "gate_temperature": "temperature", + "gate_speed_override": "speed_override", + "drying_state": "drying_state", +} + +_GATE_COERCE: dict[str, Callable] = { + "status": GateStatus, + "color_rgb": _rgb, +} + +_GATE_DIFF_KEYS: frozenset[str] = frozenset(_GATE_ARRAYS) | {"num_gates"} + +_SCALAR_COERCE: dict[str, Callable] = { + "ttg_map": tuple, + "endless_spool_groups": tuple, + "filament_pos": FilamentPos, + "spoolman_support": SpoolmanSupport, + "sensors": dict, +} + @dataclass(frozen=True, slots=True) class GateInfo: + """One gate as Happy-Hare publishes it.""" + index: int - status: GateStatus - material: str - color: str - color_rgb: tuple[float, float, float] - spool_id: int + status: GateStatus = GateStatus.UNKNOWN + material: str = "" + color: str = "" + color_rgb: tuple[float, float, float] = (0.0, 0.0, 0.0) + spool_id: int = -1 filament_name: str = "" temperature: float | None = None - weight_g: float | None = None - mid_usage: bool = False - remaining_weight: float | None = None - bed_temp: int | None = None + speed_override: int = 100 + drying_state: str = "" @property def is_available(self) -> bool: @@ -57,168 +119,88 @@ def is_available(self) -> bool: return self.status in (GateStatus.AVAILABLE, GateStatus.AVAILABLE_FROM_BUFFER) +def _gates(arrays: Mapping[str, Sequence], num_gates: int) -> tuple[GateInfo, ...]: + """Build the gate tuple from parallel arrays, leaving short ones defaulted.""" + return tuple( + GateInfo( + index=i, + **{ + f: _GATE_COERCE[f](v[i]) if f in _GATE_COERCE else v[i] + for f, v in arrays.items() + if i < len(v) + }, + ) + for i in range(num_gates) + ) + + @dataclass(frozen=True, slots=True) class MMUState: - enabled: bool - is_homed: bool - num_gates: int - tool: int - gate: int - filament: str # "Loaded" | "Unloaded" | "Unknown" - filament_pos: FilamentPos - action: str - bowden_progress: float - print_state: str - reason_for_pause: str - has_bypass: bool - spoolman_support: SpoolmanSupport - gates: tuple[GateInfo, ...] - ttg_map: tuple[int, ...] + """A full snapshot of the MMU; rebuilt, never mutated, on every status update.""" + + enabled: bool = False + is_homed: bool = False + num_gates: int = 0 + tool: int = -1 + gate: int = -1 + filament: str = "Unknown" # "Loaded" | "Unloaded" | "Unknown" + filament_pos: FilamentPos = FilamentPos.UNKNOWN + action: str = "" + bowden_progress: float = -1.0 + print_state: str = "" + reason_for_pause: str = "" + has_bypass: bool = False + spoolman_support: SpoolmanSupport = SpoolmanSupport.OFF + gates: tuple[GateInfo, ...] = () + ttg_map: tuple[int, ...] = () pending_spool_id: int = -1 operation: str = "" - sensors: dict = dataclasses.field(default_factory=dict) - gate_speed_override: tuple[float, ...] = dataclasses.field(default_factory=tuple) + sensors: dict[str, bool | None] = dataclasses.field(default_factory=dict) endless_spool_groups: tuple[int, ...] = dataclasses.field(default_factory=tuple) @property def is_paused(self) -> bool: """Return True if the MMU is in a paused/error state.""" - return self.print_state == "pause" + return self.print_state in ("paused", "pause_locked") + + @property + def is_printing(self) -> bool: + """Return True while a print is running; a paused print allows recovery.""" + return self.print_state in ("started", "printing") @property def current_gate_info(self) -> GateInfo | None: - """Return the GateInfo for the currently selected gate, or None if no gate is selected.""" + """Return the GateInfo for the selected gate, or None when none is selected.""" if 0 <= self.gate < len(self.gates): return self.gates[self.gate] return None @classmethod - def from_status(cls, data: dict) -> "MMUState": - """Build an MMUState from a full Moonraker mmu status dict. - - Args: - data (dict): Full status dict from printer.objects.query or the initial notify_status_update payload. - - Returns: - MMUState: New instance populated from *data* - """ - num_gates = data.get("num_gates", 0) - - statuses = data.get("gate_status", [GateStatus.UNKNOWN] * num_gates) - material = data.get("gate_material", [""] * num_gates) - colors = data.get("gate_color", [""] * num_gates) - rgbs = data.get("gate_color_rgb", [(0.0, 0.0, 0.0)] * num_gates) - spool_ids = data.get("gate_spool_id", [-1] * num_gates) - filament_name = data.get("gate_filament_name", [""] * num_gates) - temperature = data.get("gate_temperature", [None] * num_gates) - speed_override = data.get("gate_speed_override", []) - weight_g = data.get("gate_weight_g", [None] * num_gates) - remaining_weight_list = data.get("gate_remaining_weight", [None] * num_gates) - bed_temperature_list = data.get("gate_bed_temp", [None] * num_gates) - mid_usage_list = data.get("gate_mid_usage", [False] * num_gates) - - gates: tuple[GateInfo, ...] = tuple( - GateInfo( - index=i, - status=GateStatus(statuses[i]), - material=material[i], - color=colors[i], - color_rgb=tuple(rgbs[i]), - spool_id=spool_ids[i], - filament_name=filament_name[i], - temperature=temperature[i], - weight_g=weight_g[i], - remaining_weight=remaining_weight_list[i], - bed_temp=bed_temperature_list[i], - mid_usage=mid_usage_list[i], - ) - for i in range(num_gates) - ) - return cls( - enabled=data.get("enabled", False), - is_homed=data.get("is_homed", False), - num_gates=num_gates, - tool=data.get("tool", -1), - gate=data.get("gate", -1), - filament=data.get("filament", "Unknown"), - filament_pos=FilamentPos(data.get("filament_pos", FilamentPos.UNKNOWN)), - action=data.get("action", ""), - bowden_progress=data.get("bowden_progress", -1), - print_state=data.get("print_state", ""), - reason_for_pause=data.get("reason_for_pause", ""), - has_bypass=data.get("has_bypass", False), - spoolman_support=SpoolmanSupport( - data.get("spoolman_support", SpoolmanSupport.OFF) - ), - gates=gates, - ttg_map=tuple(data.get("ttg_map", [])), - pending_spool_id=data.get("pending_spool_id", -1), - operation=data.get("operation", ""), - sensors=dict(data.get("sensors", {})), - gate_speed_override=tuple(speed_override), - endless_spool_groups=tuple(data.get("endless_spool_groups", [])), - ) + def from_status(cls, data: dict) -> MMUState: + """Build an MMUState from a full Moonraker mmu status dict.""" + return cls().apply_diff(data) def gate_for_tool(self, tool: int) -> int: - """Returns the gate mapped to *tool*, or -1 if unmapped.""" + """Return the gate mapped to *tool*, or -1 if unmapped.""" if 0 <= tool < len(self.ttg_map): return self.ttg_map[tool] return -1 - def apply_diff(self, diff: dict) -> "MMUState": - """Apply a Moonraker status diff and return an updated MMUState. - - Args: - diff (dict): Partial status dict from notify_status_update. - - Returns: - MMUState: New instance with updated fields - """ - gate_keys: set[str] = { - "gate_status", - "gate_material", - "gate_color", - "gate_color_rgb", - "gate_spool_id", - "gate_filament_name", - "gate_temperature", - "gate_speed_override", + def apply_diff(self, diff: dict) -> MMUState: + """Return a new MMUState with a notify_status_update diff applied.""" + changed = { + k: _SCALAR_COERCE[k](v) if k in _SCALAR_COERCE else v + for k, v in diff.items() + if k in _STATE_FIELDS } - if gate_keys.isdisjoint(diff): - # No gate array changes - scalar_fields = { - k: v for k, v in diff.items() if k in MMUState.__dataclass_fields__ - } - if "ttg_map" in scalar_fields: - scalar_fields["ttg_map"] = tuple(scalar_fields["ttg_map"]) - if "filament_pos" in scalar_fields: - scalar_fields["filament_pos"] = FilamentPos( - scalar_fields["filament_pos"] - ) - if "spoolman_support" in scalar_fields: - scalar_fields["spoolman_support"] = SpoolmanSupport( - scalar_fields["spoolman_support"] - ) - if "endless_spool_groups" in scalar_fields: - scalar_fields["endless_spool_groups"] = tuple( - scalar_fields["endless_spool_groups"] - ) - return dataclasses.replace(self, **scalar_fields) - # Gate arrays changed — need full rebuild, but we lost the raw arrays - # Pass current gate data + diff into from_status - gate_data = { - "gate_status": [g.status for g in self.gates], - "gate_material": [g.material for g in self.gates], - "gate_color": [g.color for g in self.gates], - "gate_color_rgb": [g.color_rgb for g in self.gates], - "gate_spool_id": [g.spool_id for g in self.gates], - "gate_filament_name": [g.filament_name for g in self.gates], - "gate_speed_override": list(self.gate_speed_override), - "gate_temperature": [g.temperature for g in self.gates], - "gate_weight_g": [g.weight_g for g in self.gates], - "gate_remaining_weight": [g.remaining_weight for g in self.gates], - "gate_bed_temp": [g.bed_temp for g in self.gates], - "gate_mid_usage": [g.mid_usage for g in self.gates], + if _GATE_DIFF_KEYS.isdisjoint(diff): + return dataclasses.replace(self, **changed) + arrays = { + f: diff[key] if key in diff else [getattr(g, f) for g in self.gates] + for key, f in _GATE_ARRAYS.items() } - merged = {**dataclasses.asdict(self), **gate_data, **diff} - return MMUState.from_status(merged) + num_gates = changed.get("num_gates", self.num_gates) + return dataclasses.replace(self, **changed, gates=_gates(arrays, num_gates)) + + +_STATE_FIELDS: frozenset[str] = frozenset(f.name for f in dataclasses.fields(MMUState)) diff --git a/BlocksScreen/lib/panels/mainWindow.py b/BlocksScreen/lib/panels/mainWindow.py index c27a19651..f96f11ac7 100644 --- a/BlocksScreen/lib/panels/mainWindow.py +++ b/BlocksScreen/lib/panels/mainWindow.py @@ -263,6 +263,7 @@ def __init__(self): self.printer.sensor_update.connect(self.on_temp_sensor_update) self.printer.object_updated.connect(self.amu_manager.on_object_updated) self.amu_manager.run_gcode_signal.connect(self.ws.api.run_gcode) + self.ws.klippy_state_signal.connect(self.amu_manager.on_klippy_state) self.run_gcode_signal.connect(self.ws.api.run_gcode) self.main_content_widget.currentChanged.connect(slot=self.reset_tab_indexes) diff --git a/tests/amu/conftest.py b/tests/amu/conftest.py index 647c43fcb..2419decc3 100644 --- a/tests/amu/conftest.py +++ b/tests/amu/conftest.py @@ -1,11 +1,35 @@ -COMMENTED_CFG = """\ -#[include mmu/base/*.cfg] -#[include mmu/optional/client_macros.cfg] -#[include filament_manager.cfg] -""" - -UNCOMMENTED_CFG = """\ -[include mmu/base/*.cfg] -[include mmu/optional/client_macros.cfg] -[include filament_manager.cfg] -""" +# Verbatim RF50-Klipper/printer.cfg:155-165, stray pre-bracket space included. +_BANNER = "#" * 94 + + +def _banner(title: str) -> str: + """Rebuild the 94-char banner byte-exactly; as a literal it would breach E501.""" + return ( + f"{'#' * 47} {'#' * 44}\n" + f"{'#' * 39}{title}{'#' * 37}\n" + f"{'#' * 46} {'#' * 45}\n" + ) + + +_AMU_BANNER = _banner(" Amu System ") +_SINGLE_BANNER = _banner(" Normal system ") + +# AMU off (shipped state): the single-filament include is active. +COMMENTED_CFG = ( + f"{_AMU_BANNER}{_BANNER}\n" + "#[include config/variant_mmu/base/*.cfg]\n" + f"{_BANNER}\n" + f"{_SINGLE_BANNER}{_BANNER}\n" + "[include config/variant_sync_single/*.cfg ]\n" + f"{_BANNER}\n" +) + +# AMU on: the AMU include is active, the single-filament one commented. +UNCOMMENTED_CFG = ( + f"{_AMU_BANNER}{_BANNER}\n" + "[include config/variant_mmu/base/*.cfg]\n" + f"{_BANNER}\n" + f"{_SINGLE_BANNER}{_BANNER}\n" + "#[include config/variant_sync_single/*.cfg ]\n" + f"{_BANNER}\n" +) diff --git a/tests/amu/test_config_toggler_unit.py b/tests/amu/test_config_toggler_unit.py index 81acf6f7d..3988ac25a 100644 --- a/tests/amu/test_config_toggler_unit.py +++ b/tests/amu/test_config_toggler_unit.py @@ -1,13 +1,35 @@ """Unit tests for BlocksScreen.devices.amu.config_toggler.""" -from BlocksScreen.devices.amu.config_toggler import ConfigToggler +import os +from pathlib import Path + +import pytest + +from BlocksScreen.devices.amu.config_toggler import ( + AMU_INCLUDE, + SINGLE_INCLUDE, + ConfigToggler, + ToggleResult, +) from tests.amu.conftest import COMMENTED_CFG, UNCOMMENTED_CFG +# Sibling RF50-Klipper checkout, absent in CI. +_REAL_PRINTER_CFG = Path(__file__).parents[2].parent / "RF50-Klipper" / "printer.cfg" + +# root bypasses file modes, so the chmod-based failures never happen. +_not_root = pytest.mark.skipif(os.geteuid() == 0, reason="root ignores file modes") + class TestConfigTogglerInit: def test_missing_file_sets_path_none(self, tmp_path) -> None: ct = ConfigToggler(tmp_path / "nonexistent.cfg") assert ct._path is None + assert ct.is_configured() is False + + def test_unreadable_file_is_not_configured(self, tmp_path) -> None: + unreadable = tmp_path / "printer.cfg" + unreadable.mkdir() + assert ConfigToggler(unreadable).is_configured() is False def test_commented_file_is_not_configured(self, tmp_path) -> None: cfg = tmp_path / "printer.cfg" @@ -21,36 +43,207 @@ def test_uncommented_file_is_configured(self, tmp_path) -> None: class TestConfigTogglerToggle: - def test_no_file_returns_false(self, tmp_path) -> None: + def test_no_file_fails(self, tmp_path) -> None: + ct = ConfigToggler(tmp_path / "nonexistent.cfg") + assert ct.toggle(True) is ToggleResult.FAILED + + # toggle() never reaches _write() without a path; pin the guard anyway. + def test_write_without_a_path_refuses(self, tmp_path) -> None: ct = ConfigToggler(tmp_path / "nonexistent.cfg") - assert ct.toggle(True) is False + assert ct._write("anything") is False + assert not list(tmp_path.iterdir()) def test_activate_uncomments_includes(self, tmp_path) -> None: cfg = tmp_path / "printer.cfg" cfg.write_text(COMMENTED_CFG) ct = ConfigToggler(cfg) - assert ct.toggle(True) is True + assert ct.toggle(True) is ToggleResult.CHANGED assert cfg.read_text() == UNCOMMENTED_CFG def test_deactivate_comments_includes(self, tmp_path) -> None: cfg = tmp_path / "printer.cfg" cfg.write_text(UNCOMMENTED_CFG) ct = ConfigToggler(cfg) - assert ct.toggle(False) is True + assert ct.toggle(False) is ToggleResult.CHANGED assert cfg.read_text() == COMMENTED_CFG - def test_same_state_returns_false(self, tmp_path) -> None: + # UNCHANGED, not FAILED: an already-correct config is not an error. + def test_same_state_is_unchanged(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + ct = ConfigToggler(cfg) + assert ct.toggle(False) is ToggleResult.UNCHANGED + + def test_toggle_keeps_symlink(self, tmp_path) -> None: + """os.replace() would swap a symlinked printer.cfg for a regular file.""" + real = tmp_path / "real.cfg" + real.write_text(COMMENTED_CFG) + link = tmp_path / "printer.cfg" + link.symlink_to(real) + assert ConfigToggler(link).toggle(True) is ToggleResult.CHANGED + assert link.is_symlink() + assert real.read_text() == UNCOMMENTED_CFG + + def test_non_ascii_config_round_trips(self, tmp_path) -> None: + """read_text() without encoding raises UnicodeDecodeError under LC_ALL=C.""" + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG + "# nozzle 220ºC\n", encoding="utf-8") + assert ConfigToggler(cfg).toggle(True) is ToggleResult.CHANGED + assert cfg.read_text(encoding="utf-8").endswith("# nozzle 220ºC\n") + + def test_undecodable_byte_round_trips(self, tmp_path) -> None: + """A Latin-1 byte raises ValueError, which no OSError handler would catch.""" + cfg = tmp_path / "printer.cfg" + cfg.write_bytes(COMMENTED_CFG.encode() + b"# nozzle 220\xbaC\n") + assert ConfigToggler(cfg).toggle(True) is ToggleResult.CHANGED + assert cfg.read_bytes().endswith(b"# nozzle 220\xbaC\n") + + def test_crlf_config_keeps_its_line_endings(self, tmp_path) -> None: + """Default newline handling rewrites a whole CRLF printer.cfg as LF.""" + cfg = tmp_path / "printer.cfg" + cfg.write_bytes(COMMENTED_CFG.replace("\n", "\r\n").encode()) + assert ConfigToggler(cfg).toggle(True) is ToggleResult.CHANGED + assert cfg.read_bytes() == UNCOMMENTED_CFG.replace("\n", "\r\n").encode() + + def test_external_rewrite_does_not_wedge_toggle(self, tmp_path) -> None: + """The configurator and Mainsail rewrite printer.cfg behind a live toggler.""" cfg = tmp_path / "printer.cfg" cfg.write_text(COMMENTED_CFG) ct = ConfigToggler(cfg) - assert ct.toggle(False) is False + cfg.write_text(UNCOMMENTED_CFG) + assert ct.toggle(False) is ToggleResult.CHANGED + assert cfg.read_text() == COMMENTED_CFG - def test_oserror_returns_false(self, tmp_path) -> None: + @_not_root + def test_oserror_fails(self, tmp_path) -> None: cfg = tmp_path / "printer.cfg" cfg.write_text(COMMENTED_CFG) ct = ConfigToggler(cfg) cfg.chmod(0o000) try: - assert ct.toggle(True) is False + assert ct.toggle(True) is ToggleResult.FAILED finally: cfg.chmod(0o644) + + def test_round_trip_is_byte_identical(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + ct = ConfigToggler(cfg) + ct.toggle(True) + ct.toggle(False) + assert cfg.read_text() == COMMENTED_CFG + + +class TestConfigTogglerRefusesBadWrites: + """The defects that let the wrong include paths ship unnoticed for months.""" + + def test_missing_include_returns_false_and_does_not_write(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text("[include config/variant_sync_single/*.cfg ]\n") + ct = ConfigToggler(cfg) + assert ct.toggle(True) is ToggleResult.FAILED + assert cfg.read_text() == "[include config/variant_sync_single/*.cfg ]\n" + + def test_empty_file_returns_false(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text("") + ct = ConfigToggler(cfg) + assert ct.toggle(True) is ToggleResult.FAILED + + def test_duplicate_include_returns_false(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG + f"#[include {AMU_INCLUDE}]\n") + ct = ConfigToggler(cfg) + assert ct.toggle(True) is ToggleResult.FAILED + + def test_never_leaves_both_variants_active(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + ct = ConfigToggler(cfg) + for activate in (True, False, True, False): + ct.toggle(activate) + text = cfg.read_text() + active = [ + inc + for inc in (AMU_INCLUDE, SINGLE_INCLUDE) + if ConfigToggler._is_active(text, inc) + ] + assert len(active) == 1, ( + f"variants active after toggle({activate}): {active}" + ) + + def test_write_leaves_no_temp_file(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + ConfigToggler(cfg).toggle(True) + assert list(tmp_path.iterdir()) == [cfg] + + def test_dir_sync_failure_still_changed(self, tmp_path, monkeypatch) -> None: + """The rename already landed, so FAILED here would skip the FIRMWARE_RESTART.""" + + def _boom(*_args, **_kwargs): + raise OSError("no fd") + + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + toggler = ConfigToggler(cfg) + # os.open is only reached by the parent-directory fsync. + monkeypatch.setattr(os, "open", _boom) + assert toggler.toggle(True) is ToggleResult.CHANGED + assert cfg.read_text() == UNCOMMENTED_CFG + + @_not_root + def test_preserves_file_mode(self, tmp_path) -> None: + cfg = tmp_path / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + cfg.chmod(0o640) + ConfigToggler(cfg).toggle(True) + assert cfg.stat().st_mode & 0o777 == 0o640 + + @_not_root + def test_unwritable_dir_leaves_config_intact(self, tmp_path) -> None: + """A read-only config dir must fail the toggle, not half-write printer.cfg.""" + d = tmp_path / "cfgdir" + d.mkdir() + cfg = d / "printer.cfg" + cfg.write_text(COMMENTED_CFG) + ct = ConfigToggler(cfg) + d.chmod(0o500) + try: + assert ct.toggle(True) is ToggleResult.FAILED + assert cfg.read_text() == COMMENTED_CFG + finally: + d.chmod(0o700) + assert list(d.iterdir()) == [cfg] + + +@pytest.mark.skipif( + not _REAL_PRINTER_CFG.exists(), reason="sibling RF50-Klipper checkout not present" +) +class TestAgainstRealPrinterCfg: + """Guards cross-repo coupling: fixtures only test the regex against itself.""" + + def test_both_variant_includes_resolve(self) -> None: + text = _REAL_PRINTER_CFG.read_text() + for include in (AMU_INCLUDE, SINGLE_INCLUDE): + assert ( + ConfigToggler._is_active(text, include) or f"[include {include}" in text + ) + + def test_exactly_one_variant_active_on_disk(self) -> None: + text = _REAL_PRINTER_CFG.read_text() + active = [ + inc + for inc in (AMU_INCLUDE, SINGLE_INCLUDE) + if ConfigToggler._is_active(text, inc) + ] + assert len(active) == 1, f"variants active in real printer.cfg: {active}" + + def test_toggle_round_trips_the_real_file(self, tmp_path) -> None: + original = _REAL_PRINTER_CFG.read_text() + cfg = tmp_path / "printer.cfg" + cfg.write_text(original) + ct = ConfigToggler(cfg) + assert ct.toggle(True) is ToggleResult.CHANGED + assert ct.toggle(False) is ToggleResult.CHANGED + assert cfg.read_text() == original diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index aa8d37146..8e5e8ec20 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -1,8 +1,10 @@ """Unit tests for BlocksScreen.devices.amu.manager.""" +import shlex from unittest.mock import MagicMock, patch import pytest +from PyQt6 import QtCore from BlocksScreen.devices.amu.manager import AMUManager from BlocksScreen.devices.amu.models import MMUState @@ -13,6 +15,15 @@ UNCOMMENTED_CFG as _UNCOMMENTED_CFG, ) + +def _klipper_parse(rawparams: str) -> dict[str, str]: + """Klipper's _get_extended_params (gcode.py:266); its ';' cut only derives cmd.""" + lexer = shlex.shlex(rawparams, posix=True) + lexer.whitespace_split = True + lexer.commenters = "#;" + return {k.upper(): v for k, v in (arg.split("=", 1) for arg in lexer)} + + _FULL_STATUS: dict = { "enabled": True, "is_homed": True, @@ -21,12 +32,13 @@ "gate": 0, "filament": "Loaded", "action": "Idle", - "print_state": "printing", + "print_state": "standby", "reason_for_pause": "", "gate_status": [1, -1], "gate_material": ["PLA", ""], "gate_color": ["ff0000", ""], - "gate_color_rgb": [(1.0, 0.0, 0.0), (0.0, 0.0, 0.0)], + # Lists, not tuples: the status arrives as JSON. + "gate_color_rgb": [[1.0, 0.0, 0.0], [0.0, 0.0, 0.0]], "gate_spool_id": [42, -1], "ttg_map": [0, 1], } @@ -38,6 +50,12 @@ "spoolman_support": "push", } + +def _status(**overrides: object) -> dict: + """Full MMU status with per-test overrides.""" + return {**_FULL_STATUS_WITH_SPOOLMAN, **overrides} + + _SPOOL_DATA: dict = { "id": 42, "used_weight": 50.0, @@ -52,6 +70,7 @@ } +# qapp is required: most tests take manager without qtbot. @pytest.fixture def manager(tmp_path, qapp): """AMUManager with no config file (patched to a non-existent path).""" @@ -78,6 +97,12 @@ def test_initial_state_is_none(self, manager) -> None: def test_initial_amu_not_configured(self, manager) -> None: assert manager.is_amu_active() is False + # Qt ownership keeps the manager alive once the caller drops its ref. + def test_takes_a_qt_parent(self, tmp_path, qapp) -> None: + parent = QtCore.QObject() + with patch("BlocksScreen.devices.amu.manager.CONFIG_PATH", tmp_path / "a.cfg"): + assert AMUManager(ws=MagicMock(), parent=parent).parent() is parent + class TestToggleAMUSystem: def test_no_config_emits_false(self, manager, qtbot) -> None: @@ -85,10 +110,50 @@ def test_no_config_emits_false(self, manager, qtbot) -> None: manager.toggle_amu_system(True) assert blocker.args == [False] - def test_same_state_emits_false(self, manager_with_cfg, qtbot) -> None: + # An already-correct config is a success, not a failed toggle. + def test_same_state_emits_true(self, manager_with_cfg, qtbot) -> None: mgr, _ = manager_with_cfg with qtbot.waitSignal(mgr.amu_toggled) as blocker: mgr.toggle_amu_system(False) + assert blocker.args == [True] + + def test_same_state_does_not_restart_firmware( + self, manager_with_cfg, qtbot + ) -> None: + mgr, _ = manager_with_cfg + with qtbot.assertNotEmitted(mgr.run_gcode_signal): + mgr.toggle_amu_system(False) + mgr._ws.api.firmware_restart.assert_not_called() + + # A config swap restarts the firmware, which would kill the running print. + @pytest.mark.parametrize("print_state", ["printing", "paused"]) + def test_refused_during_a_print(self, manager_with_cfg, qtbot, print_state) -> None: + mgr, cfg = manager_with_cfg + mgr.on_object_updated("print_stats", "print_stats", {"state": print_state}) + with qtbot.waitSignal(mgr.amu_toggled) as blocker: + mgr.toggle_amu_system(True) + assert blocker.args == [False] + assert cfg.read_text() == _COMMENTED_CFG + mgr._ws.api.firmware_restart.assert_not_called() + + def test_allowed_once_the_print_ends(self, manager_with_cfg, qtbot) -> None: + mgr, cfg = manager_with_cfg + mgr.on_object_updated("print_stats", "print_stats", {"state": "printing"}) + mgr.on_object_updated("print_stats", "print_stats", {"state": "complete"}) + with qtbot.waitSignal(mgr.amu_toggled) as blocker: + mgr.toggle_amu_system(True) + assert blocker.args == [True] + assert cfg.read_text() == _UNCOMMENTED_CFG + + # print_stats diffs without "state" must not reset the tracked state. + def test_print_stats_diff_without_state_is_ignored( + self, manager_with_cfg, qtbot + ) -> None: + mgr, _ = manager_with_cfg + mgr.on_object_updated("print_stats", "print_stats", {"state": "printing"}) + mgr.on_object_updated("print_stats", "print_stats", {"print_duration": 5.0}) + with qtbot.waitSignal(mgr.amu_toggled) as blocker: + mgr.toggle_amu_system(True) assert blocker.args == [False] def test_activate_uncomments_includes(self, manager_with_cfg, qtbot) -> None: @@ -101,7 +166,6 @@ def test_activate_uncomments_includes(self, manager_with_cfg, qtbot) -> None: def test_deactivate_comments_includes(self, manager_with_cfg, qtbot) -> None: mgr, cfg = manager_with_cfg cfg.write_text(_UNCOMMENTED_CFG) - mgr._config_toggler._state = True with qtbot.waitSignal(mgr.amu_toggled) as blocker: mgr.toggle_amu_system(False) assert blocker.args == [True] @@ -116,17 +180,26 @@ def test_activate_sets_amu_configured(self, manager_with_cfg, qtbot) -> None: class TestUpdateMMUState: - def test_first_call_emits_state(self, manager, qtbot) -> None: + def test_first_call_emits_and_stores_state(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.mmu_state_changed) as blocker: manager.update_mmu_state(_FULL_STATUS) + assert blocker.args[0] is manager.get_state() assert isinstance(blocker.args[0], MMUState) + assert blocker.args[0].num_gates == 2 - def test_first_call_stores_state(self, manager) -> None: + # A diff-seeded state is gateless and blanks the carousel. + def test_diff_cannot_seed_state(self, manager, qtbot) -> None: + with qtbot.assertNotEmitted(manager.mmu_state_changed): + manager.update_mmu_state({"tool": 1, "filament": "Unloaded"}) + assert manager.get_state() is None + + def test_full_status_after_clear_reseeds(self, manager) -> None: + manager.update_mmu_state(_FULL_STATUS) + manager.on_klippy_state("disconnect") + manager.update_mmu_state({"tool": 1}) + assert manager.get_state() is None manager.update_mmu_state(_FULL_STATUS) - state = manager.get_state() - assert state is not None - assert state.num_gates == 2 - assert state.enabled is True + assert manager.get_state().num_gates == 2 def test_diff_updates_scalar(self, manager, qtbot) -> None: manager.update_mmu_state(_FULL_STATUS) @@ -142,19 +215,18 @@ def test_diff_preserves_unchanged_fields(self, manager) -> None: assert manager.get_state().num_gates == 2 assert manager.get_state().gates[0].material == "PLA" - def test_each_call_emits_signal(self, manager, qtbot) -> None: + # filament_position streams during moves but is not part of the state. + @pytest.mark.parametrize("diff", [{"filament_position": 12.5}, {"tool": 0}]) + def test_no_op_diff_emits_nothing(self, manager, qtbot, diff) -> None: manager.update_mmu_state(_FULL_STATUS) - with qtbot.waitSignal(manager.mmu_state_changed): - manager.update_mmu_state({"tool": 1}) + with qtbot.assertNotEmitted(manager.mmu_state_changed): + manager.update_mmu_state(diff) - def test_from_status_parses_gate_speed_override(self, manager) -> None: - data = {**_FULL_STATUS, "gate_speed_override": [0.5, 1.0]} - manager.update_mmu_state(data) - assert manager.get_state().gate_speed_override == (0.5, 1.0) - - def test_from_status_gate_speed_override_defaults_to_empty(self, manager) -> None: + def test_same_status_after_restart_emits(self, manager, qtbot) -> None: manager.update_mmu_state(_FULL_STATUS) - assert manager.get_state().gate_speed_override == () + manager.on_klippy_state("disconnect") + with qtbot.waitSignal(manager.mmu_state_changed): + manager.update_mmu_state(_FULL_STATUS) def test_from_status_parses_endless_spool_groups(self, manager) -> None: data = {**_FULL_STATUS, "endless_spool_groups": [0, 1, 3, 0]} @@ -172,11 +244,14 @@ def test_apply_diff_updates_endless_spool_groups(self, manager) -> None: class TestGcodeSignals: - def test_set_gate_info(self, manager, qtbot) -> None: + def test_set_gate_info_without_temp_lets_happy_hare_default_it( + self, manager, qtbot + ) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.set_gate_info(0, "PLA", "ff0000", 42) + manager.set_gate_info(2, "ABS", "0000ff", 3, filament_name="ABS BLACK") assert blocker.args == [ - "MMU_GATE_MAP gate=0 MATERIAL=PLA COLOR=ff0000 SPOOLID=42" + "MMU_GATE_MAP GATE=2 NAME='ABS BLACK' MATERIAL=ABS COLOR=0000ff " + "SPOOLID=3 QUIET=1" ] def test_set_gate_info_with_temperature(self, manager, qtbot) -> None: @@ -190,89 +265,304 @@ def test_set_gate_info_with_temperature(self, manager, qtbot) -> None: temperature=240, ) assert blocker.args == [ - "MMU_GATE_MAP gate=1 NAME=PETG Transparent MATERIAL=PETG COLOR=00ff00 SPOOLID=7 TEMP=240" + "MMU_GATE_MAP GATE=1 NAME='PETG Transparent' MATERIAL=PETG COLOR=00ff00 " + "SPOOLID=7 TEMP=240 QUIET=1" ] - def test_set_gate_info_withouth_temp(self, manager, qtbot) -> None: + def test_clear_gate(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.set_gate_info(2, "ABS", "0000ff", 3, filament_name="ABS BLACK") + manager.clear_gate(1) assert blocker.args == [ - "MMU_GATE_MAP gate=2 NAME=ABS BLACK MATERIAL=ABS COLOR=0000ff SPOOLID=3" + "MMU_GATE_MAP GATE=1 NAME='' MATERIAL='' COLOR='' SPOOLID=-1 QUIET=1" ] def test_set_gate_material(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.set_gate_material(1, "PETG") - assert blocker.args == ["MMU_GATE_MAP gate=1 MATERIAL=PETG"] + assert blocker.args == ["MMU_GATE_MAP GATE=1 MATERIAL=PETG TEMP=0 QUIET=1"] def test_set_gate_color(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.set_gate_color(2, "00ff00") - assert blocker.args == ["MMU_GATE_MAP gate=2 COLOR=00ff00"] + assert blocker.args == ["MMU_GATE_MAP GATE=2 COLOR=00ff00 TEMP=0 QUIET=1"] + + @pytest.mark.parametrize(("loaded", "flag"), [(True, 1), (False, 0)]) + def test_recover_single_gate(self, manager, qtbot, loaded, flag) -> None: + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.recover_single_gate(loaded) + assert blocker.args == [f"MMU_RECOVER GATE=0 LOADED={flag}"] + + # Happy-Hare wants rrggbb or a W3C name, never a '#' prefix. + @pytest.mark.parametrize("method", ["set_gate_color", "set_gate_info"]) + def test_hash_prefixed_colour_is_stripped(self, manager, qtbot, method) -> None: + args = ( + (2, "#00ff00") if method == "set_gate_color" else (2, "PLA", "#00ff00", 3) + ) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + getattr(manager, method)(*args) + assert "COLOR=00ff00" in blocker.args[0] + + def test_set_gate_temp(self, manager, qtbot) -> None: + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.set_gate_temp(1, 240) + assert blocker.args == ["MMU_GATE_MAP GATE=1 TEMP=240 QUIET=1"] + + # Happy-Hare reads TEMP with get_int, which rejects "215.0". + @pytest.mark.parametrize("temp", [215.0, 214.6]) + def test_float_temp_is_sent_as_an_int(self, manager, qtbot, temp) -> None: + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.set_gate_temp(1, temp) + assert blocker.args == [f"MMU_GATE_MAP GATE=1 TEMP={round(temp)} QUIET=1"] + + @pytest.mark.parametrize("temp", [0, -1, 0.4, float("inf"), float("nan")]) + def test_invalid_temp_emits_no_gcode(self, manager, qtbot, temp) -> None: + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.set_gate_temp(1, temp) + + # Happy-Hare ignores TEMP when Spoolman pull mode owns the gate map. + def test_temp_skipped_in_pull_mode(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(spoolman_support="pull")) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.set_gate_temp(1, 240) + + # Omitted TEMP resets the gate to default_extruder_temp (mmu.py:8627). + @pytest.mark.parametrize( + ("method", "args"), + [ + ("set_gate_material", (1, "PETG")), + ("set_gate_color", (1, "00ff00")), + ("set_gate_spool", (1, 7)), + ], + ) + def test_partial_edit_keeps_the_gate_temp( + self, manager, qtbot, method, args + ) -> None: + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + getattr(manager, method)(*args) + assert " TEMP=0 " in blocker.args[0] def test_set_gate_spool(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.set_gate_spool(0, 7) - assert blocker.args == ["MMU_GATE_MAP gate=0 SPOOLID=7"] + assert blocker.args == ["MMU_GATE_MAP GATE=0 SPOOLID=7 TEMP=0 QUIET=1"] + + # GATE=-1 fails Happy-Hare's minval=0. + @pytest.mark.parametrize( + ("method", "args"), + [ + ("set_gate_info", (-1, "PLA", "ff0000", 42)), + ("clear_gate", (-1,)), + ("set_gate_material", (-1, "PETG")), + ("set_gate_temp", (-1, 240)), + ("set_gate_color", (-1, "00ff00")), + ("set_gate_spool", (-1, 7)), + ], + ) + def test_unselected_gate_emits_no_gcode(self, manager, qtbot, method, args) -> None: + with qtbot.assertNotEmitted(manager.run_gcode_signal): + getattr(manager, method)(*args) + + # Without state the loaded guard cannot run, and bare MMU_HOME means TOOL=0. + def test_home_mmu_without_state_sends_nothing(self, manager, qtbot) -> None: + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.home_mmu() - def test_home_mmu(self, manager, qtbot) -> None: + def test_home_mmu_without_a_tool_is_bare(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(tool=-1, filament_pos=0)) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.home_mmu() assert blocker.args == ["MMU_HOME"] + # Bare MMU_HOME defaults to TOOL=0 (mmu.py:6706). + def test_home_mmu_keeps_the_selected_tool(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(tool=1, filament_pos=0)) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.home_mmu() + assert blocker.args == ["MMU_HOME TOOL=1"] + + # home() has no check_if_loaded (mmu.py:6394), unlike MMU_SELECT. + @pytest.mark.parametrize("filament_pos", [3, 10]) + def test_home_mmu_refused_while_loaded(self, manager, qtbot, filament_pos) -> None: + manager.update_mmu_state(_status(tool=1, filament_pos=filament_pos)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.home_mmu() + + # MMU_HOME is the post-restart recovery, exactly when filament_pos is UNKNOWN. + def test_home_mmu_allowed_when_position_unknown(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(tool=1, filament_pos=-1)) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.home_mmu() + assert blocker.args == ["MMU_HOME TOOL=1"] + + # Happy-Hare ignores a bare MMU_RESET (mmu.py:4197). def test_reset_mmu(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.reset_mmu() - assert blocker.args == ["MMU_RESET"] + assert blocker.args == ["MMU_RESET CONFIRM=1"] def test_load_gate(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(filament_pos=0)) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.load_gate() + assert manager.load_gate() is True assert blocker.args == ["MMU_LOAD"] + # Happy-Hare drops MMU_LOAD when disabled or unhomed (mmu.py:6936). + @pytest.mark.parametrize( + "overrides", + [ + None, + {"filament_pos": 10}, + {"filament_pos": 0, "gate": -1}, + {"filament_pos": 0, "enabled": False}, + {"filament_pos": 0, "is_homed": False}, + {"filament_pos": 0, "print_state": "printing"}, + {"filament_pos": 0, "print_state": "started"}, + ], + ) + def test_load_gate_refused(self, manager, qtbot, overrides) -> None: + if overrides is not None: + manager.update_mmu_state(_status(**overrides)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + assert manager.load_gate() is False + def test_unload(self, manager, qtbot) -> None: + manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + assert manager.unload() is True + assert blocker.args == ["MMU_UNLOAD"] + + # Happy-Hare drops MMU_UNLOAD when disabled or already unloaded (mmu.py:6981). + @pytest.mark.parametrize( + "overrides", + [None, {"filament_pos": 0}, {"enabled": False}, {"print_state": "printing"}], + ) + def test_unload_refused(self, manager, qtbot, overrides) -> None: + if overrides is not None: + manager.update_mmu_state(_status(**overrides)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + assert manager.unload() is False + + # A paused print is where a failed swap gets recovered by hand. + @pytest.mark.parametrize("print_state", ["paused", "pause_locked"]) + def test_unload_allowed_while_paused(self, manager, qtbot, print_state) -> None: + manager.update_mmu_state(_status(print_state=print_state)) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.unload() + assert manager.unload() is True assert blocker.args == ["MMU_UNLOAD"] def test_eject_gate(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(filament_pos=0)) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: manager.eject_gate() assert blocker.args == ["MMU_EJECT"] - def test_eject_all_gates(self, manager, qtbot) -> None: + # MMU_EJECT only ejects the selected gate, and only when UNLOADED (mmu.py:7000). + @pytest.mark.parametrize( + "overrides", + [ + None, + {"filament_pos": 10}, + {"filament_pos": 0, "gate": -1}, + {"filament_pos": 0, "gate_status": [0, 1]}, + {"filament_pos": 0, "print_state": "printing"}, + ], + ) + def test_eject_gate_refused(self, manager, qtbot, overrides) -> None: + if overrides is not None: + manager.update_mmu_state(_status(**overrides)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.eject_gate() + + def test_check_gate_allows_an_empty_gate(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(filament_pos=0, gate_status=[0, 1])) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.check_gate() + assert blocker.args == ["MMU_CHECK_GATE"] + + # MMU_CHECK_GATE runs a full unload first when filament is loaded. + @pytest.mark.parametrize( + "overrides", + [ + None, + {"filament_pos": 10}, + {"filament_pos": 0, "gate": -2}, + {"filament_pos": 0, "is_homed": False}, + {"filament_pos": 0, "print_state": "printing"}, + ], + ) + def test_check_gate_refused(self, manager, qtbot, overrides) -> None: + if overrides is not None: + manager.update_mmu_state(_status(**overrides)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.check_gate() + + def test_select_gate(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(filament_pos=0)) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.eject_all_gates(3) - assert blocker.args == ["MMU_EJECT GATE=0\nMMU_EJECT GATE=1\nMMU_EJECT GATE=2"] + manager.select_gate(1) + assert blocker.args == ["MMU_SELECT GATE=1"] - def test_select_tool(self, manager, qtbot) -> None: + # Happy-Hare refuses MMU_SELECT while loaded (mmu.py:6722). + @pytest.mark.parametrize( + ("overrides", "gate"), + [ + (None, 1), + ({"filament_pos": 0}, 0), + ({"filament_pos": 0}, 2), + ({"filament_pos": 0}, -1), + ({"filament_pos": 10}, 1), + ({"filament_pos": 0, "enabled": False}, 1), + ({"filament_pos": 0, "is_homed": False}, 1), + ({"filament_pos": 0, "print_state": "printing"}, 1), + ], + ) + def test_select_gate_skipped(self, manager, qtbot, overrides, gate) -> None: + if overrides is not None: + manager.update_mmu_state(_status(**overrides)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.select_gate(gate) + + # TOOL, not GATE: the ttg map may not map tool N to gate N. + def test_change_tool(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager.select_gate(2) - assert blocker.args == ["MMU_SELECT GATE=2"] + manager.change_tool(1) + assert blocker.args == ["MMU_CHANGE_TOOL TOOL=1"] - def test_klippy_disconnect(self, manager) -> None: + def test_klippy_disconnected_to_connected(self, manager) -> None: manager.update_mmu_state(_FULL_STATUS) manager.on_klippy_state("disconnect") assert manager.get_state() is None - - def test_klippy_ready(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS) manager.on_klippy_state("ready") + manager.update_mmu_state(_FULL_STATUS) assert manager.get_state() is not None - def test_klippy_disconnected_to_connected(self, manager) -> None: + # FIRMWARE_RESTART drops the state; the panels must be told. + def test_klippy_disconnect_emits_none_state(self, manager, qtbot) -> None: + manager.update_mmu_state(_FULL_STATUS) + with qtbot.waitSignal(manager.mmu_state_changed) as blocker: + manager.on_klippy_state("disconnect") + assert blocker.args == [None] + + def test_repeated_disconnect_emits_once(self, manager, qtbot) -> None: manager.update_mmu_state(_FULL_STATUS) manager.on_klippy_state("disconnect") - assert manager.get_state() is None - manager.on_klippy_state("ready") + with qtbot.assertNotEmitted(manager.mmu_state_changed): + manager.on_klippy_state("shutdown") + manager.on_klippy_state("disconnect") + + def test_klippy_ready_emits_nothing(self, manager, qtbot) -> None: manager.update_mmu_state(_FULL_STATUS) - assert manager.get_state() is not None + with qtbot.assertNotEmitted(manager.mmu_state_changed): + manager.on_klippy_state("ready") - def test_activate_emits_firmware_restart(self, manager_with_cfg, qtbot) -> None: + # The gcode form drops the socket mid-reply and reports a false 503. + def test_activate_calls_the_firmware_restart_rpc( + self, manager_with_cfg, qtbot + ) -> None: mgr, _ = manager_with_cfg - with qtbot.waitSignal(mgr.run_gcode_signal) as blocker: + with qtbot.assertNotEmitted(mgr.run_gcode_signal): mgr.toggle_amu_system(True) - assert blocker.args == ["FIRMWARE_RESTART"] + mgr._ws.api.firmware_restart.assert_called_once_with() class TestPregateSensors: @@ -290,13 +580,28 @@ def test_pre_gate_emits_signal(self, manager, qtbot) -> None: manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_0") assert blocker.args == [0, True] + # Moonraker sends only changed keys; a missing reading is not an empty gate. + @pytest.mark.parametrize("data", [{}, {"enabled": False}, {"enabled": True}]) + def test_enabled_only_diff_is_ignored(self, manager, qtbot, data) -> None: + with qtbot.assertNotEmitted(manager.pre_gate_changed): + manager.on_pre_gate_update(data, "mmu_pre_gate_0") + assert manager.get_pre_gate_sensors() == {} + + def test_enabled_only_diff_keeps_the_last_reading(self, manager) -> None: + manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_0") + manager.on_pre_gate_update({"enabled": False}, "mmu_pre_gate_0") + assert manager.get_pre_gate_sensors() == {0: True} + def test_non_mmu_sensor_ignored(self, manager, qtbot) -> None: with qtbot.assertNotEmitted(manager.pre_gate_changed): manager.on_pre_gate_update({"filament_detected": True}, "Toolhead Sensor") - def test_pre_gate_stores_state(self, manager) -> None: - manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_2") - assert manager.get_pre_gate_sensors() == {2: True} + # int() on a bad suffix would raise into Qt. + @pytest.mark.parametrize("name", ["mmu_pre_gate_", "mmu_pre_gate_left"]) + def test_unparsable_gate_index_ignored(self, manager, qtbot, name) -> None: + with qtbot.assertNotEmitted(manager.pre_gate_changed): + manager.on_pre_gate_update({"filament_detected": True}, name) + assert manager.get_pre_gate_sensors() == {} def test_pre_gate_filament_not_detected(self, manager) -> None: manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_0") @@ -305,25 +610,14 @@ def test_pre_gate_filament_not_detected(self, manager) -> None: manager.on_pre_gate_update({"filament_detected": False}, "mmu_pre_gate_3") assert manager.get_pre_gate_sensors() == {0: True, 1: False, 2: True, 3: False} - def test_pre_gate_empty(self, manager) -> None: - assert manager.get_pre_gate_sensors() == {} - - def test_pre_gate_multiple_emits(self, manager, qtbot) -> None: - with qtbot.waitSignal(manager.pre_gate_changed) as blocker: - manager.on_pre_gate_update({"filament_detected": False}, "mmu_pre_gate_0") - assert blocker.args == [0, False] - with qtbot.waitSignal(manager.pre_gate_changed) as blocker: - manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_1") - assert blocker.args == [1, True] - with qtbot.waitSignal(manager.pre_gate_changed) as blocker: - manager.on_pre_gate_update({"filament_detected": False}, "mmu_pre_gate_2") - assert blocker.args == [2, False] - with qtbot.waitSignal(manager.pre_gate_changed) as blocker: - manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_3") - assert blocker.args == [3, True] - - def test_pre_gate_update_is_dict(self, manager) -> None: - assert isinstance(manager.get_pre_gate_sensors(), dict) + def test_pre_gate_emits_each_reading(self, manager) -> None: + seen = [] + manager.pre_gate_changed.connect(lambda *args: seen.append(args)) + for gate, detected in enumerate([False, True, False, True]): + manager.on_pre_gate_update( + {"filament_detected": detected}, f"mmu_pre_gate_{gate}" + ) + assert seen == [(0, False), (1, True), (2, False), (3, True)] def test_klippy_disconnect_clears_pre_gate_sensors(self, manager) -> None: manager.on_pre_gate_update({"filament_detected": True}, "mmu_pre_gate_0") @@ -332,9 +626,6 @@ def test_klippy_disconnect_clears_pre_gate_sensors(self, manager) -> None: class TestIsAMUActive: - def test_false_when_not_configured(self, manager) -> None: - assert manager.is_amu_active() is False - def test_false_when_configured_but_no_mmu_state( self, manager_with_cfg, qtbot ) -> None: @@ -350,36 +641,39 @@ def test_true_when_configured_and_mmu_state(self, manager_with_cfg, qtbot) -> No mgr.update_mmu_state(_FULL_STATUS) assert mgr.is_amu_active() is True + # MMU ENABLE=0 rejects every MMU_* command (mmu.py:3584). + def test_false_when_mmu_disabled(self, manager_with_cfg, qtbot) -> None: + mgr, _ = manager_with_cfg + with qtbot.waitSignal(mgr.amu_toggled): + mgr.toggle_amu_system(True) + mgr.update_mmu_state(_status(enabled=False)) + assert mgr.is_amu_active() is False -class TestSpoolManFetch: - def test_noop_when_mmu_state_none(self, manager, qtbot) -> None: - with qtbot.assertNotEmitted(manager.spool_fetched): - manager.fetch_spool(0, 42) - manager._ws.api.get_spool.assert_not_called() - def test_noop_when_spoolman_off(self, manager, qtbot) -> None: - status = {**_FULL_STATUS_WITH_SPOOLMAN, "spoolman_support": "off"} - manager.update_mmu_state(status) - with qtbot.assertNotEmitted(manager.spool_fetched): - manager.fetch_spool(0, 42) - manager._ws.api.get_spool.assert_not_called() +class TestObjectRouting: + """on_object_updated is the only entry point Printer calls; each branch matters.""" - def test_emit_spool_fetched_on_sucess(self, manager, qtbot) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - spool_data = {"id": 42, "filament": {"name": "PLA"}, "used_weight": 50.0} - manager.fetch_spool(0, 42) - callback = manager._ws.api.get_spool.call_args.args[1] - with qtbot.waitSignal(manager.spool_fetched) as blocker: - callback(spool_data) - assert blocker.args[0] == 0 - assert blocker.args[1] == spool_data + def test_mmu_type_updates_state(self, manager, qtbot) -> None: + with qtbot.waitSignal(manager.mmu_state_changed): + manager.on_object_updated("mmu", "mmu", _FULL_STATUS) + assert manager.get_state().num_gates == 2 - def test_no_emit_on_ws_failure(self, manager, qtbot) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager.fetch_spool(0, 42) - callback = manager._ws.api.get_spool.call_args.args[1] - with qtbot.assertNotEmitted(manager.spool_fetched): - callback(None) + def test_filament_switch_sensor_routes_to_pre_gate(self, manager, qtbot) -> None: + with qtbot.waitSignal(manager.pre_gate_changed) as blocker: + manager.on_object_updated( + "filament_switch_sensor", + "mmu_pre_gate_1", + {"filament_detected": True}, + ) + assert blocker.args == [1, True] + assert manager.get_pre_gate_sensors() == {1: True} + + # The object name, not the type, carries the gate index. + def test_non_pre_gate_switch_sensor_ignored(self, manager, qtbot) -> None: + with qtbot.assertNotEmitted(manager.pre_gate_changed): + manager.on_object_updated( + "filament_switch_sensor", "toolhead_sensor", {"filament_detected": True} + ) @pytest.mark.parametrize( "error", @@ -401,117 +695,69 @@ def test_error_reply_leaves_the_gate_alone(self, manager, qtbot, error) -> None: class TestLoadCellUpdate: - def test_noop_when_mmu_state_none(self, manager) -> None: - manager.on_load_cell_update({"force": 150.0}, "load_cell_mmu_0") - assert manager.get_state() is None - - def test_updates_gate_weight(self, manager) -> None: + # RF50-Klipper has no [load_cell]; the old handler also read the wrong key. + def test_load_cell_updates_are_ignored(self, manager, qtbot) -> None: manager.update_mmu_state(_FULL_STATUS) - manager.on_load_cell_update({"force": 150.0}, "load_cell_mmu_0") - assert manager.get_state().gates[0].weight_g == 150.0 - - def test_ignores_unkown_name(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS) - manager.on_load_cell_update({"force": 150.0}, "other_sensor_mmu_0") - assert manager.get_state().gates[0].weight_g is None - - def test_emits_gate_weight_updated(self, manager, qtbot) -> None: - manager.update_mmu_state(_FULL_STATUS) - with qtbot.waitSignal(manager.gate_weight_updated) as blocker: - manager.on_load_cell_update({"force": 200.0}, "load_cell_mmu_0") - assert blocker.args == [0, 200.0] - - def test_routed_via_on_object_updated(self, manager, qtbot) -> None: - manager.update_mmu_state(_FULL_STATUS) - with qtbot.waitSignal(manager.gate_weight_updated): - manager.on_object_updated("load_cell", "load_cell_mmu_0", {"force": 100.0}) + with qtbot.assertNotEmitted(manager.mmu_state_changed): + manager.on_object_updated( + "load_cell", "load_cell_mmu_0", {"force_g": 150.0} + ) -class TestApplySpoolData: - def test_emits_gcode_with_spool_fields(self, manager, qtbot) -> None: +class TestGcodeQuoting: + # User-typed Spoolman text must survive Klipper's shlex parse as one value. + @pytest.mark.parametrize( + ("name", "color"), + [ + ("PLA Basic", "ff0000"), + ("Bob's PLA", "#ff0000"), + ('He said "hi"', ""), + ("PLA", "ff0000"), + ("a;b#c", "0"), + ("multi\nline", "ff0000"), + ], + ) + def test_emitted_gcode_survives_klipper_parser( + self, manager, qtbot, name, color + ) -> None: manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - manager._apply_spool_data(0, _SPOOL_DATA) - assert blocker.args == [ - "MMU_GATE_MAP gate=0 NAME=PLA Basic MATERIAL=PLA COLOR=ff0000 SPOOLID=42 TEMP=215" - ] - - def test_updates_local_weight(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager._apply_spool_data(0, _SPOOL_DATA) - assert manager.get_state().gates[0].weight_g == 50.0 - - def test_triggered_by_spool_fetched_signal(self, manager, qtbot) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager.fetch_spool(0, 42) - callback = manager._ws.api.get_spool.call_args.args[1] - with qtbot.waitSignal(manager.run_gcode_signal): - callback(_SPOOL_DATA) - - def test_updates_local_remaining_weight(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager._apply_spool_data(0, _SPOOL_DATA) - assert manager.get_state().gates[0].remaining_weight == 950.0 - - def test_updates_local_bed_temp(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager._apply_spool_data(0, _SPOOL_DATA) - assert manager.get_state().gates[0].bed_temp == 60 - - def test_mid_usage_spool_no_remaining_weight(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - data = {k: v for k, v in _SPOOL_DATA.items() if k != "remaining_weight"} - manager._apply_spool_data(0, data) - assert manager.get_state().gates[0].remaining_weight is None - assert manager.get_state().gates[0].bed_temp == 60 - - def test_speed_gcode_emitted_for_heavy_spool(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - heavy_data = {**_SPOOL_DATA, "remaining_weight": 1500.0} - signals = [] - manager.run_gcode_signal.connect(signals.append) - manager._apply_spool_data(0, heavy_data) - speed_gcodes = [s for s in signals if "SPEED" in s] - assert len(speed_gcodes) == 1 - assert "SPEED=33" in speed_gcodes[0] + manager.set_gate_info(0, "PLA", color, 42, filament_name=name) + gcode = blocker.args[0] + assert "\n" not in gcode + params = _klipper_parse(gcode.removeprefix("MMU_GATE_MAP ")) + assert params["NAME"] == " ".join(name.split()) + assert params["COLOR"] == color.lstrip("#") + assert params["SPOOLID"] == "42" + + # Klipper's save_variables configparser shuts Klippy down on a '%'. + @pytest.mark.parametrize("name", ["50%", "%(x)s"]) + def test_percent_is_stripped(self, manager, qtbot, name) -> None: + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.set_gate_info(0, "PLA", "ff0000", 42, filament_name=name) + assert "%" not in blocker.args[0] - def test_no_speed_gcode_for_normal_spool(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - signals = [] - manager.run_gcode_signal.connect(signals.append) - manager._apply_spool_data(0, _SPOOL_DATA) - speed_gcodes = [s for s in signals if "SPEED" in s] - assert len(speed_gcodes) == 0 - def test_no_speed_gcode_when_no_remaining_weight(self, manager) -> None: +class TestAssignSpool: + @staticmethod + def _gcode(manager, gate, spool) -> list[str]: manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - data_no_weight = { - k: v for k, v in _SPOOL_DATA.items() if k != "remaining_weight" - } signals = [] manager.run_gcode_signal.connect(signals.append) - manager._apply_spool_data(0, data_no_weight) - speed_gcodes = [s for s in signals if "SPEED" in s] - assert len(speed_gcodes) == 0 + manager.assign_spool(gate, spool) + return signals + def test_maps_gate_from_payload(self, manager) -> None: + assert self._gcode(manager, 1, _SPOOL_DATA) == [ + "MMU_GATE_MAP GATE=1 NAME='PLA Basic' MATERIAL=PLA COLOR=ff0000 " + "SPOOLID=42 TEMP=215 QUIET=1" + ] -class TestSetGateSpoolAutoFetch: - def test_calls_fetch_spool_when_activate(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager.set_gate_spool(0, 42) - manager._ws.api.get_spool.assert_called_once() + # Spoolman sends null for an unset extruder temp. + def test_null_temp_lets_happy_hare_default_it(self, manager) -> None: + filament = {**_SPOOL_DATA["filament"], "settings_extruder_temp": None} + (gcode,) = self._gcode(manager, 1, {**_SPOOL_DATA, "filament": filament}) + assert "TEMP" not in gcode - def test_skips_fetch_when_spool_id_minus_one(self, manager) -> None: - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager.set_gate_spool(0, -1) - manager._ws.api.get_spool.assert_not_called() - - def test_skips_fetch_when_no_state(self, manager) -> None: - manager.set_gate_spool(0, 42) - manager._ws.api.get_spool.assert_not_called() - - def test_skips_fetch_when_spoolman_off(self, manager) -> None: - status = {**_FULL_STATUS_WITH_SPOOLMAN, "spoolman_support": "off"} - manager.update_mmu_state(status) - manager.set_gate_spool(0, 42) - manager._ws.api.get_spool.assert_not_called() + def test_unselected_gate_emits_nothing(self, manager) -> None: + assert self._gcode(manager, -1, _SPOOL_DATA) == [] diff --git a/tests/amu/test_models_unit.py b/tests/amu/test_models_unit.py index 66ce4bbdb..53046c3cf 100644 --- a/tests/amu/test_models_unit.py +++ b/tests/amu/test_models_unit.py @@ -1,5 +1,9 @@ """Unit tests for BlocksScreen.devices.amu.models.""" +import dataclasses + +import pytest + from BlocksScreen.devices.amu.models import ( FilamentPos, GateInfo, @@ -11,11 +15,11 @@ class TestGateStatus: def test_values(self) -> None: + # Must match Happy-Hare's GATE_* codes (mmu.py:59-62). + assert GateStatus.UNKNOWN == -1 assert GateStatus.EMPTY == 0 assert GateStatus.AVAILABLE == 1 assert GateStatus.AVAILABLE_FROM_BUFFER == 2 - assert GateStatus.UNKNOWN == -1 - def test_is_int_enum(self) -> None: assert isinstance(GateStatus.AVAILABLE, int) @@ -60,39 +64,20 @@ def _make(self, status: GateStatus) -> GateInfo: spool_id=1, ) - def test_is_available_true_for_available(self) -> None: - assert self._make(GateStatus.AVAILABLE).is_available is True - - def test_is_available_true_for_buffer(self) -> None: - assert self._make(GateStatus.AVAILABLE_FROM_BUFFER).is_available is True - - def test_is_available_false_for_empty(self) -> None: - assert self._make(GateStatus.EMPTY).is_available is False - - def test_is_available_false_for_unknown(self) -> None: - assert self._make(GateStatus.UNKNOWN).is_available is False - - def test_gate_info_default_weight_is_gone(self) -> None: - gate = self._make(GateStatus.AVAILABLE) - assert gate.weight_g is None - - def test_gate_info_default_mid_usage_is_false(self) -> None: - gate = self._make(GateStatus.AVAILABLE) - assert gate.mid_usage is False - - def test_gate_info_weight_and_mid_usage_set(self) -> None: - gate = GateInfo( - index=0, - status=GateStatus.AVAILABLE, - material="PLA", - color="ff0000", - color_rgb=(1.0, 0.0, 0.0), - spool_id=1, - weight_g=245.5, - mid_usage=True, - ) - assert gate.weight_g == 245.5 - assert gate.mid_usage is True + @pytest.mark.parametrize( + ("status", "available"), + [ + (GateStatus.AVAILABLE, True), + (GateStatus.AVAILABLE_FROM_BUFFER, True), + (GateStatus.EMPTY, False), + (GateStatus.UNKNOWN, False), + ], + ) + def test_is_available(self, status, available) -> None: + assert self._make(status).is_available is available + + def test_gate_info_default_speed_override_is_100(self) -> None: + assert self._make(GateStatus.AVAILABLE).speed_override == 100 def test_gate_info_default_filament_name(self) -> None: assert self._make(GateStatus.AVAILABLE).filament_name == "" @@ -130,11 +115,14 @@ def _full_status_extended(self, num_gates: int = 2) -> dict: **self._full_status(num_gates), "gate_filament_name": ["Bambu PLA Basic", ""], "gate_temperature": [215.0, 0.0], + "gate_speed_override": [100, 80], + "drying_state": ["active", ""], "pending_spool_id": 7, "operation": "loading", + # Key names from a live RF50 (mmu_sensor_manager.py:100). "sensors": { - "pre_gate_0": True, - "gate": False, + "mmu_pre_gate": True, + "mmu_gate": False, }, } @@ -147,6 +135,18 @@ def test_from_status_builds_correctly(self) -> None: assert state.gates[1].status == GateStatus.EMPTY assert state.ttg_map == (0, 1) + @pytest.mark.parametrize( + ("raw", "expected"), + [ + ([], (0.0, 0.0, 0.0)), + ([0.5], (0.5, 0.0, 0.0)), + ([0.1, 0.2, 0.3, 0.4], (0.1, 0.2, 0.3)), + ], + ) + def test_short_colour_array_pads_with_zeroes(self, raw, expected) -> None: + status = {**self._full_status(), "gate_color_rgb": [raw, (0.0, 0.0, 0.0)]} + assert MMUState.from_status(status).gates[0].color_rgb == expected + def test_from_status_empty_dict_uses_default(self) -> None: state: MMUState = MMUState.from_status({}) assert state.num_gates == 0 @@ -156,9 +156,11 @@ def test_from_status_empty_dict_uses_default(self) -> None: assert state.has_bypass is False assert state.spoolman_support is SpoolmanSupport.OFF - def test_is_paused_true(self) -> None: + # The only values Happy-Hare's is_mmu_paused() accepts (mmu.py:3040). + @pytest.mark.parametrize("print_state", ["paused", "pause_locked"]) + def test_is_paused_true(self, print_state: str) -> None: state: MMUState = MMUState.from_status( - {**self._full_status(), "print_state": "pause"} + {**self._full_status(), "print_state": print_state} ) assert state.is_paused is True @@ -202,6 +204,14 @@ def test_from_status_has_bypass_true(self) -> None: ) assert state.has_bypass is True + # Happy-Hare publishes 0-100 during a bowden move, else -1 (mmu.py:1433). + def test_bowden_progress_defaults_to_idle(self) -> None: + assert MMUState.from_status(self._full_status()).bowden_progress == -1 + + def test_apply_diff_updates_bowden_progress(self) -> None: + state: MMUState = MMUState.from_status(self._full_status()) + assert state.apply_diff({"bowden_progress": 42}).bowden_progress == 42 + def test_apply_diff_filament_pos_coerced(self) -> None: state: MMUState = MMUState.from_status(self._full_status()) updated: MMUState = state.apply_diff({"filament_pos": 0}) @@ -219,6 +229,81 @@ def test_apply_diff_gate_keys(self) -> None: updated: MMUState = state.apply_diff({"gate_status": [1, 1]}) assert updated.gates[1].status == GateStatus.AVAILABLE + def test_gate_array_diff_preserves_untouched_arrays(self) -> None: + """A diff carries only the arrays that changed; the rest must survive it.""" + state = MMUState.from_status(self._full_status_extended()) + updated = state.apply_diff({"gate_status": [2, 0]}) + assert updated.gates[0] == dataclasses.replace( + state.gates[0], status=GateStatus.AVAILABLE_FROM_BUFFER + ) + assert updated.gates[1] == state.gates[1] + + def test_from_status_parses_speed_override_per_gate(self) -> None: + state = MMUState.from_status(self._full_status_extended()) + assert (state.gates[0].speed_override, state.gates[1].speed_override) == ( + 100, + 80, + ) + + # drying_state is per-gate despite having no gate_ prefix. + def test_from_status_parses_drying_state_per_gate(self) -> None: + state = MMUState.from_status(self._full_status_extended()) + assert (state.gates[0].drying_state, state.gates[1].drying_state) == ( + "active", + "", + ) + + def test_apply_diff_updates_drying_state(self) -> None: + state = MMUState.from_status(self._full_status_extended()) + state = state.apply_diff({"drying_state": ["complete", "queued"]}) + assert [g.drying_state for g in state.gates] == ["complete", "queued"] + assert state.gates[0].speed_override == 100 + + def test_apply_diff_num_gates_resizes_gates(self) -> None: + """A variant switch resizes gates without touching any gate_* array.""" + state: MMUState = MMUState.from_status(self._full_status()) + assert len(state.apply_diff({"num_gates": 1}).gates) == 1 + grown = state.apply_diff({"num_gates": 4}) + assert len(grown.gates) == 4 + assert grown.gates[3].status is GateStatus.UNKNOWN + + # A 1->4 variant switch grows num_gates before the gate arrays catch up. + def test_from_status_short_gate_arrays_default_the_tail(self) -> None: + state: MMUState = MMUState.from_status(self._full_status_extended(4)) + assert len(state.gates) == 4 + tail = state.gates[2:] + assert [g.index for g in tail] == [2, 3] + assert all(g.status is GateStatus.UNKNOWN for g in tail) + assert all(g.spool_id == -1 for g in tail) + assert all(g.speed_override == 100 for g in tail) + + def test_from_status_long_gate_arrays_are_truncated(self) -> None: + state: MMUState = MMUState.from_status(self._full_status_extended(1)) + assert len(state.gates) == 1 + assert state.gates[0].status is GateStatus.AVAILABLE + + def test_apply_diff_does_not_alias_sensors(self) -> None: + """The frozen dataclass must not hand out a reference to the caller's dict.""" + state: MMUState = MMUState.from_status(self._full_status()) + diff = {"sensors": {"mmu_gate": True}} + updated = state.apply_diff(diff) + diff["sensors"]["extruder"] = False + assert updated.sensors == {"mmu_gate": True} + + def test_unknown_enum_values_do_not_raise(self) -> None: + """PyQt6 turns a ValueError inside a slot into qFatal, killing the whole GUI.""" + state = MMUState.from_status( + { + **self._full_status(), + "gate_status": [99, 0], + "filament_pos": 42, + "spoolman_support": "readwrite", + } + ) + assert state.gates[0].status is GateStatus.UNKNOWN + assert state.filament_pos is FilamentPos.UNKNOWN + assert state.spoolman_support is SpoolmanSupport.OFF + def test_from_status_parse_gate_filament_name(self) -> None: state = MMUState.from_status(self._full_status_extended()) assert state.gates[0].filament_name == "Bambu PLA Basic" @@ -238,7 +323,7 @@ def test_from_status_parses_operation(self) -> None: def test_from_status_parse_sensors(self) -> None: state = MMUState.from_status(self._full_status_extended()) - assert state.sensors == {"pre_gate_0": True, "gate": False} + assert state.sensors == {"mmu_pre_gate": True, "mmu_gate": False} def test_from_status_new_field_defaults(self) -> None: state = MMUState.from_status(self._full_status()) @@ -254,5 +339,5 @@ def test_apply_diff_updates_gate_filament_name(self) -> None: def test_apply_diff_replaces_sensors(self) -> None: state = MMUState.from_status(self._full_status_extended()) - updated = state.apply_diff({"sensors": {"pre_gate_0": False}}) - assert updated.sensors == {"pre_gate_0": False} + updated = state.apply_diff({"sensors": {"mmu_pre_gate": False}}) + assert updated.sensors == {"mmu_pre_gate": False} From ed5f9c59bce12dde2992aaeaf5e9126aaa2104dc Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 16:18:04 +0100 Subject: [PATCH 4/7] refactor(amu): drop the unused drying_state gate field --- BlocksScreen/devices/amu/models.py | 2 -- tests/amu/test_models_unit.py | 15 --------------- 2 files changed, 17 deletions(-) diff --git a/BlocksScreen/devices/amu/models.py b/BlocksScreen/devices/amu/models.py index bf1c281fa..68ee3982d 100644 --- a/BlocksScreen/devices/amu/models.py +++ b/BlocksScreen/devices/amu/models.py @@ -79,7 +79,6 @@ def _missing_(cls, value: object) -> SpoolmanSupport: "gate_filament_name": "filament_name", "gate_temperature": "temperature", "gate_speed_override": "speed_override", - "drying_state": "drying_state", } _GATE_COERCE: dict[str, Callable] = { @@ -111,7 +110,6 @@ class GateInfo: filament_name: str = "" temperature: float | None = None speed_override: int = 100 - drying_state: str = "" @property def is_available(self) -> bool: diff --git a/tests/amu/test_models_unit.py b/tests/amu/test_models_unit.py index 53046c3cf..f2f10594f 100644 --- a/tests/amu/test_models_unit.py +++ b/tests/amu/test_models_unit.py @@ -116,7 +116,6 @@ def _full_status_extended(self, num_gates: int = 2) -> dict: "gate_filament_name": ["Bambu PLA Basic", ""], "gate_temperature": [215.0, 0.0], "gate_speed_override": [100, 80], - "drying_state": ["active", ""], "pending_spool_id": 7, "operation": "loading", # Key names from a live RF50 (mmu_sensor_manager.py:100). @@ -245,20 +244,6 @@ def test_from_status_parses_speed_override_per_gate(self) -> None: 80, ) - # drying_state is per-gate despite having no gate_ prefix. - def test_from_status_parses_drying_state_per_gate(self) -> None: - state = MMUState.from_status(self._full_status_extended()) - assert (state.gates[0].drying_state, state.gates[1].drying_state) == ( - "active", - "", - ) - - def test_apply_diff_updates_drying_state(self) -> None: - state = MMUState.from_status(self._full_status_extended()) - state = state.apply_diff({"drying_state": ["complete", "queued"]}) - assert [g.drying_state for g in state.gates] == ["complete", "queued"] - assert state.gates[0].speed_override == 100 - def test_apply_diff_num_gates_resizes_gates(self) -> None: """A variant switch resizes gates without touching any gate_* array.""" state: MMUState = MMUState.from_status(self._full_status()) From 21e0bf163f9ed56cb916775355d410b8a0a6abbd Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 17:23:57 +0100 Subject: [PATCH 5/7] test(amu): drop the fetch_spool error test, the backend rebuild removes fetch_spool --- tests/amu/test_manager_unit.py | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index 8e5e8ec20..585471df9 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -675,24 +675,6 @@ def test_non_pre_gate_switch_sensor_ignored(self, manager, qtbot) -> None: "filament_switch_sensor", "toolhead_sensor", {"filament_detected": True} ) - @pytest.mark.parametrize( - "error", - [ - {"code": 404, "message": "spool not found"}, - {"code": -1, "message": "disconnected"}, - ], - ) - def test_error_reply_leaves_the_gate_alone(self, manager, qtbot, error) -> None: - """An error read as spool data would send MATERIAL= COLOR= SPOOLID=-1.""" - manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) - manager.fetch_spool(0, 42) - callback = manager._ws.api.get_spool.call_args.args[1] - with ( - qtbot.assertNotEmitted(manager.spool_fetched), - qtbot.assertNotEmitted(manager.run_gcode_signal), - ): - callback({"error": error}) - class TestLoadCellUpdate: # RF50-Klipper has no [load_cell]; the old handler also read the wrong key. From 6af21395f2c57105b8a91ffcc5797efe550ac740 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Tue, 13 Oct 2026 17:34:41 +0100 Subject: [PATCH 6/7] fix(files): retry a history lookup only after a disconnect, a deleted job stays final --- BlocksScreen/lib/files.py | 5 +++-- tests/widgets/test_files_data_unit.py | 10 +++++++++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/BlocksScreen/lib/files.py b/BlocksScreen/lib/files.py index 3f734eda5..392fcf8f9 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -474,8 +474,9 @@ def _on_history_job(self, filename: str, job_id: str, result: dict) -> None: """Cache a completed job's duration and re-emit fileinfo.""" job = result.get("job") or {} status = job.get("status") - # Unmarked so the next view asks again: a lost reply or a running job. - if "error" in result or status == "in_progress": + lost = (result.get("error") or {}).get("code") == -1 + # Re-ask after a disconnect or a running job; a deleted job stays final. + if lost or status == "in_progress": self._job_durations.pop(job_id, None) return duration = job.get("print_duration") diff --git a/tests/widgets/test_files_data_unit.py b/tests/widgets/test_files_data_unit.py index 4920ec926..8448dc43c 100644 --- a/tests/widgets/test_files_data_unit.py +++ b/tests/widgets/test_files_data_unit.py @@ -186,7 +186,7 @@ def test_in_progress_job_is_asked_again(self, files): files.request_print_duration("a.gcode") assert files.ws.api.history_get_job.call_count == 2 - def test_error_reply_is_asked_again(self, files): + def test_disconnect_reply_is_asked_again(self, files): """A reply lost to a disconnect must not hide the duration until restart.""" files.request_print_duration("a.gcode") error = {"code": -1, "message": "disconnected"} @@ -194,6 +194,14 @@ def test_error_reply_is_asked_again(self, files): files.request_print_duration("a.gcode") assert files.ws.api.history_get_job.call_count == 2 + def test_deleted_job_is_not_asked_again(self, files): + """A deleted job (-32601) is final; re-asking costs a request per view.""" + files.request_print_duration("a.gcode") + error = {"code": -32601, "message": "Invalid job uid: 000023"} + files._on_history_job("a.gcode", "000023", {"error": error}) + files.request_print_duration("a.gcode") + assert files.ws.api.history_get_job.call_count == 1 + def test_unsent_lookup_is_asked_again(self, files): files.ws.api.history_get_job.return_value = False files.request_print_duration("a.gcode") From 6acd261d414b9fa85b1e8b1ef80a22100883d4b3 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Wed, 14 Oct 2026 11:32:51 +0100 Subject: [PATCH 7/7] fix(amu): ignore the None MMU state sent on Klipper restart in FilamentTab --- BlocksScreen/devices/amu/manager.py | 2 +- BlocksScreen/lib/panels/filamentTab.py | 2 + tests/widgets/test_filament_tab_mmu_unit.py | 41 +++++++++++++++++++++ 3 files changed, 44 insertions(+), 1 deletion(-) create mode 100644 tests/widgets/test_filament_tab_mmu_unit.py diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index 928c128e6..9b8b0e95e 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -119,7 +119,7 @@ def set_gate_info( filament_name: str = "", temperature: int | None = None, ) -> bool: - """Map every field of *gate*; a None temperature takes Happy-Hare's default.""" + """Overwrite all gate fields; '' clears one, None temp resets to HH default.""" return self._gate_map( gate, NAME=filament_name, diff --git a/BlocksScreen/lib/panels/filamentTab.py b/BlocksScreen/lib/panels/filamentTab.py index 1ff2be060..e71201800 100644 --- a/BlocksScreen/lib/panels/filamentTab.py +++ b/BlocksScreen/lib/panels/filamentTab.py @@ -938,6 +938,8 @@ def _clear_gate_map(self, gate_info) -> None: def on_mmu_state_changed(self, mmu_state): """Handles changes in the MMU state from the AMU manager to update the UI and show the load panel when loading/unloading.""" + if mmu_state is None: + return for gate_info in mmu_state.gates: previous_state = self._previous_gate_states.get(gate_info.index) current_state = gate_info.status in [ diff --git a/tests/widgets/test_filament_tab_mmu_unit.py b/tests/widgets/test_filament_tab_mmu_unit.py new file mode 100644 index 000000000..bf51a7050 --- /dev/null +++ b/tests/widgets/test_filament_tab_mmu_unit.py @@ -0,0 +1,41 @@ +"""FilamentTab handling of the AMU manager's state stream.""" + +import sys +from unittest.mock import MagicMock + +import devices.amu.manager as amu_manager_module +from devices.amu.models import FilamentPos, MMUState + +vars(sys.modules["devices.amu"]).update( + AMUManager=amu_manager_module.AMUManager, + FilamentPos=FilamentPos, + MMUState=MMUState, +) +# Other test modules leave session-wide stand-ins for these; the tab needs the real API. +_STUBBED = ( + "lib.utils.icon_button", + "lib.utils.blocks_button", + "lib.utils.blocks_label", + "lib.utils.blocks_linedit", + "lib.utils.blocks_Scrollbar", + "lib.panels.widgets.Common", + "lib.panels.widgets.FilamentTab", + "lib.panels.filamentTab", +) +for _stale in [m for m in sys.modules if m.startswith(_STUBBED)]: + sys.modules.pop(_stale, None) + +# loadWidget imports configfile at import time; the conftest fixture only stubs it later. +_had_configfile = "configfile" in sys.modules +sys.modules.setdefault("configfile", MagicMock()) +from lib.panels.filamentTab import FilamentTab # noqa: E402 + +if not _had_configfile: + sys.modules.pop("configfile", None) + + +def test_none_state_after_klipper_restart_is_ignored() -> None: + # AMUManager.on_klippy_state emits None on every non-ready Klippy state. + tab = MagicMock() + FilamentTab.on_mmu_state_changed(tab, None) + assert tab.mock_calls == []