added testing and created STUB to start responding to moonraker updates with the corresponding AMU method

Author: gmmcosta15

BlocksScreen/devices/amu/__init__.py

@@ -1,4 +1,4 @@
-"""Network Manager Package
+"""AMU / HAPPY-Hare MMU package
 
 Architecture:
     AMUManager (manager.py)

BlocksScreen/devices/amu/manager.py

@@ -199,16 +199,16 @@ def select_tool(self, tool: int) -> None:
         """
         self.run_gcode_signal.emit(f"MMU_CHANGE_TOOL TOOL={tool}")
 
-    @QtCore.pyqtSlot(dict, name="update_mmu_state")
-    def update_mmu_state(self, data: dict) -> None:
+    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
+        emits mmu_state_changed.
 
         Args:
-            data (dict): Raw MMU status or diff dict from Moonraker
+            data: Raw MMU status or diff dict from Moonraker.
+            name: Moonraker object name suffix (always empty for ``mmu``).
         """
         if self._mmu_state is None:
             self._mmu_state = MMUState.from_status(data)

BlocksScreen/lib/panels/mainWindow.py

@@ -201,7 +201,7 @@ def __init__(self):
         self.query_object_list.connect(self.utilitiesPanel.on_object_list)
         self.printer.extruder_update.connect(self.on_extruder_update)
         self.printer.heater_bed_update.connect(self.on_heater_bed_update)
-        self.printer.mmu_updated.connect(self.amu_manager.update_mmu_state)
+        self.printer.register_callback("mmu", self.amu_manager.update_mmu_state)
         self.amu_manager.run_gcode_signal.connect(self.ws.api.run_gcode)
         self.run_gcode_signal.connect(self.ws.api.run_gcode)
 

BlocksScreen/lib/printer.py

@@ -79,10 +79,6 @@ class Printer(QtCore.QObject):
     z_tilt_update: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
         str, bool, name="z_tilt_update"
     )
-    mmu_updated: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
-        object, name="mmu-updated"
-    )
-    
     config_subscription: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
         [dict],
         [list],
@@ -107,6 +103,7 @@ class Printer(QtCore.QObject):
     printer_busy: bool = False
     current_loaded_file: str = ""
     current_loaded_file_metadata: str = ""
+    
 
     def __init__(self, parent: QtCore.QObject, ws: MoonWebSocket, /) -> None:
         super(Printer, self).__init__(parent)
@@ -130,6 +127,7 @@ def __init__(self, parent: QtCore.QObject, ws: MoonWebSocket, /) -> None:
         self.request_available_objects_signal.connect(self.ws.api.get_available_objects)
         self.request_object_subscription_signal.connect(self.ws.api.object_subscription)
         self.query_printer_object.connect(self.ws.api.object_query)
+        self._callbacks: dict[str, typing.Callable[[dict, str], None]] = {}
 
     def clear_printer_objs(self) -> None:
         """Clear all tracking of printer object"""
@@ -143,6 +141,34 @@ def clear_printer_objs(self) -> None:
         self.current_loaded_file = ""
         self.current_loaded_file_metadata = ""
 
+
+    def __inject_callback(
+        self, object_type: str, callback: typing.Callable[[dict, str], None]
+    ) -> None:
+        """Internal: insert a validated callback into the dispatch table."""
+        if not object_type or not callable(callback):
+            logger.warning(
+                "register_callback: invalid args object_type=%r callable=%s",
+                object_type,
+                callable(callback),
+            )
+            return
+        self._callbacks[object_type] = callback
+
+    def register_callback(
+        self, object_type: str, callback: typing.Callable[[dict, str], None]
+    ) -> None:
+        """Register an external callback for a Moonraker object type.
+
+        When Moonraker sends an update for ``object_type``, ``callback(values, name)``
+        is called directly — no intermediate signal required.
+
+        Args:
+            object_type: Moonraker object key, e.g. ``"mmu"`` or ``"filament_switch_sensor"``.
+            callback: Callable with signature ``(values: dict, name: str) -> None``.
+        """
+        self.__inject_callback(object_type, callback)
+
     @QtCore.pyqtSlot(str, name="on_klippy_status")
     def on_klippy_status(self, state: str):
         """Handles klippy update status
@@ -275,6 +301,9 @@ def _check_callback(self, name: str, values: dict) -> bool:
         _object_type, _object_name = tuple(_split + [""] * max(0, 2 - len(_split)))
         if name.startswith("extruder"):
             _object_name = name
+        if _object_type in self._callbacks:
+            self._callbacks[_object_type](values, _object_name)
+            return True
         if hasattr(self, f"_{_object_type}_object_updated"):
             _callback = getattr(self, f"_{_object_type}_object_updated")
             if callable(_callback):
@@ -745,5 +774,5 @@ def _load_filament_object_updated(self, values: dict, name: str) -> None:
         if "state" in values.keys():
             self.load_filament_update[bool].emit(values["state"])
 
-    def _mmu_object_updated(self, values: dict, name: str = "") -> None:
-        self.mmu_updated.emit(values)
+
+    

tests/amu/__init__.py

@@ -0,0 +1,191 @@
+"""Unit tests for BlocksScreen.devices.amu.manager."""
+
+import pytest
+from unittest.mock import patch
+
+from BlocksScreen.devices.amu.manager import AMUManager
+from BlocksScreen.devices.amu.models import MMUState
+
+_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]
+"""
+
+_FULL_STATUS: dict = {
+    "enabled": True,
+    "is_homed": True,
+    "num_gates": 2,
+    "tool": 0,
+    "gate": 0,
+    "filament": "Loaded",
+    "action": "Idle",
+    "print_state": "printing",
+    "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)],
+    "gate_spool_id": [42, -1],
+    "ttg_map": [0, 1],
+}
+
+
+@pytest.fixture
+def manager(tmp_path, qapp):
+    """AMUManager with no config file (patched to a non-existent path)."""
+    missing = tmp_path / "nonexistent.cfg"
+    with patch("BlocksScreen.devices.amu.manager.CONFIG_PATH", missing):
+        yield AMUManager()
+
+
+@pytest.fixture
+def manager_with_cfg(tmp_path, qapp):
+    """AMUManager pointing at a temp printer.cfg with commented includes."""
+    cfg = tmp_path / "printer.cfg"
+    cfg.write_text(_COMMENTED_CFG)
+    with patch("BlocksScreen.devices.amu.manager.CONFIG_PATH", cfg):
+        yield AMUManager(), cfg
+
+
+class TestAMUManagerInit:
+    def test_initial_state_is_none(self, manager):
+        assert manager.get_state() is None
+
+    def test_initial_amu_inactive(self, manager):
+        assert manager.is_amu_active() is False
+
+    def test_no_config_file_sets_filename_none(self, manager):
+        assert manager._config_filename is None
+
+
+class TestToggleAMUSystem:
+    def test_no_config_emits_false(self, manager, qtbot):
+        with qtbot.waitSignal(manager.amu_toggled) as blocker:
+            manager.toggle_amu_system(True)
+        assert blocker.args == [False]
+
+    def test_same_state_emits_false(self, manager_with_cfg, qtbot):
+        mgr, _ = manager_with_cfg
+        # _amu_state starts False; toggling to False again → no change
+        with qtbot.waitSignal(mgr.amu_toggled) as blocker:
+            mgr.toggle_amu_system(False)
+        assert blocker.args == [False]
+
+    def test_activate_uncomments_includes(self, manager_with_cfg, qtbot):
+        mgr, cfg = manager_with_cfg
+        with qtbot.waitSignal(mgr.amu_toggled) as blocker:
+            mgr.toggle_amu_system(True)
+        assert blocker.args == [True]
+        assert cfg.read_text() == _UNCOMMENTED_CFG
+
+    def test_deactivate_comments_includes(self, manager_with_cfg, qtbot):
+        mgr, cfg = manager_with_cfg
+        cfg.write_text(_UNCOMMENTED_CFG)
+        mgr._amu_state = True  # align internal state with file
+        with qtbot.waitSignal(mgr.amu_toggled) as blocker:
+            mgr.toggle_amu_system(False)
+        assert blocker.args == [True]
+        assert cfg.read_text() == _COMMENTED_CFG
+
+    def test_activate_sets_amu_active(self, manager_with_cfg, qtbot):
+        mgr, _ = manager_with_cfg
+        with qtbot.waitSignal(mgr.amu_toggled):
+            mgr.toggle_amu_system(True)
+        assert mgr.is_amu_active() is True
+
+
+class TestUpdateMMUState:
+    def test_first_call_emits_state(self, manager, qtbot):
+        with qtbot.waitSignal(manager.mmu_state_changed) as blocker:
+            manager.update_mmu_state(_FULL_STATUS)
+        assert isinstance(blocker.args[0], MMUState)
+
+    def test_first_call_stores_state(self, manager):
+        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
+
+    def test_diff_updates_scalar(self, manager, qtbot):
+        manager.update_mmu_state(_FULL_STATUS)
+        with qtbot.waitSignal(manager.mmu_state_changed) as blocker:
+            manager.update_mmu_state({"tool": 1, "filament": "Unloaded"})
+        state = blocker.args[0]
+        assert state.tool == 1
+        assert state.filament == "Unloaded"
+
+    def test_diff_preserves_unchanged_fields(self, manager):
+        manager.update_mmu_state(_FULL_STATUS)
+        manager.update_mmu_state({"tool": 1})
+        assert manager.get_state().num_gates == 2
+        assert manager.get_state().gates[0].material == "PLA"
+
+    def test_each_call_emits_signal(self, manager, qtbot):
+        manager.update_mmu_state(_FULL_STATUS)
+        with qtbot.waitSignal(manager.mmu_state_changed):
+            manager.update_mmu_state({"tool": 1})
+
+
+class TestGcodeSignals:
+    def test_set_gate_info(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.set_gate_info(0, "PLA", "ff0000", 42)
+        assert blocker.args == ["MMU_GATE_MAP gate=0 MATERIAL=PLA COLOR=ff0000 SPOOLID=42"]
+
+    def test_set_gate_material(self, manager, qtbot):
+        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"]
+
+    def test_set_gate_color(self, manager, qtbot):
+        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"]
+
+    def test_set_gate_spool(self, manager, qtbot):
+        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"]
+
+    def test_home_mmu(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.home_mmu()
+        assert blocker.args == ["MMU_HOME"]
+
+    def test_reset_mmu(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.reset_mmu()
+        assert blocker.args == ["MMU_RESET"]
+
+    def test_load_gate(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.load_gate(3)
+        assert blocker.args == ["MMU_SELECT gate=3\nMMU_LOAD"]
+
+    def test_unload(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.unload()
+        assert blocker.args == ["MMU_UNLOAD"]
+
+    def test_eject_gate(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.eject_gate(1)
+        assert blocker.args == ["MMU_EJECT GATE=1"]
+
+    def test_eject_all_gates(self, manager, qtbot):
+        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"]
+
+    def test_select_tool(self, manager, qtbot):
+        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
+            manager.select_tool(2)
+        assert blocker.args == ["MMU_CHANGE_TOOL TOOL=2"]

tests/amu/test_manager_unit.py

@@ -0,0 +1,87 @@
+"""Unit tests for BlocksScreen.devices.amu.models."""
+
+from BlocksScreen.devices.amu.models import GateInfo, GateStatus, MMUState
+
+class TestGateStatus:
+    def test_values(self):
+        assert GateStatus.UNKNOWN == 0
+        assert GateStatus.AVAILABLE == 1
+        assert GateStatus.AVAILABLE_FROM_BUFFER == 2
+        assert GateStatus.EMPTY == -1
+
+    def test_is_int_enum(self):
+        assert isinstance(GateStatus.AVAILABLE, int)
+
+class TestGateInfo:
+    def _make(self, status: GateStatus) -> GateInfo:
+        return GateInfo(index=0, status=status, material="PLA", color="ff0000", color_rgb=(1.0,0.0,0.0), spool_id=1)
+    def test_is_available_true_for_available(self):
+        assert self._make(GateStatus.AVAILABLE).is_available is True
+    def test_is_available_true_for_buffer(self):
+        assert self._make(GateStatus.AVAILABLE_FROM_BUFFER).is_available is True
+    def test_is_available_false_for_empty(self):
+        assert self._make(GateStatus.EMPTY).is_available is False
+    def test_is_available_false_for_unknown(self):
+        assert self._make(GateStatus.UNKNOWN).is_available is False
+
+class TestMMUState:
+    def _full_status(self, num_gates=2) -> dict:
+        return {
+            "enabled": True,
+            "is_homed": True,
+            "num_gates": num_gates,
+            "tool": 0,
+            "gate": 0,
+            "filament": "Loaded",
+            "action": "Idle",
+            "print_state": "printing",
+            "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)],
+            "gate_spool_id": [42, -1],
+            "ttg_map": [0, 1],
+        }
+    def test_from_status_builds_correctly(self):
+        state = MMUState.from_status(self._full_status())
+        assert state.num_gates == 2
+        assert state.enabled == True
+        assert len(state.gates) == 2
+        assert state.gates[0].material == "PLA"
+        assert state.gates[1].status == GateStatus.EMPTY
+        assert state.ttg_map == (0, 1)
+    def test_from_status_empty_dict_uses_default(self):
+        state = MMUState.from_status({})
+        assert state.num_gates == 0
+        assert state.enabled ==  False
+        assert state.gates == ()
+    def test_is_paused_true(self):
+        state = MMUState.from_status({**self._full_status(),"print_state": "pause"})
+        assert state.is_paused is True
+    def test_is_paused_false(self):
+        state = MMUState.from_status(self._full_status())
+        assert state.is_paused is False
+    def test_current_gate_info(self):
+        state = MMUState.from_status(self._full_status())
+        assert state.current_gate_info is state.gates[0]
+    def test_currrent_gate_info_none_when_no_gate(self):
+        state = MMUState.from_status({**self._full_status(), "gate": -1})
+        assert state.current_gate_info is None
+    def test_gate_for_tool(self):
+        state = MMUState.from_status(self._full_status())
+        assert state.gate_for_tool(0) == 0
+        assert state.gate_for_tool(1) == 1
+    def test_gate_for_tool_out_of_range(self):
+        state = MMUState.from_status(self._full_status())
+        assert state.gate_for_tool(99) == -1
+    def test_apply_diff_scalar(self):
+        state = MMUState.from_status(self._full_status())
+        updated = state.apply_diff({"tool": 1, "filament": "Unloaded"})
+        assert updated.tool == 1
+        assert updated.filament == "Unloaded"
+        assert updated.gates == state.gates
+    def test_apply_diff_gate_keys(self):
+        state = MMUState.from_status(self._full_status())
+        updated = state.apply_diff({"gate_status": [1, 1]})
+        assert updated.gates[1].status == GateStatus.AVAILABLE