add: pre-gates and klippy callback system and start integrate Spoolman support

Author: gmmcosta15

BlocksScreen/devices/amu/__init__.py

@@ -10,16 +10,20 @@
 
 from .manager import AMUManager
 from .models import (
+    FilamentPos,
     GateInfo,
     GateStatus,
     MMUState,
+    SpoolmanSupport,
 )
 
 __all__: list[str] = [
     # manager
     "AMUManager",
     # models
+    "FilamentPos",
     "GateInfo",
     "GateStatus",
     "MMUState",
+    "SpoolmanSupport",
 ]

BlocksScreen/devices/amu/manager.py

@@ -36,10 +36,15 @@ class AMUManager(QtCore.QObject):
         bool, name="amu-toggled"
     )
 
+    pre_gate_changed: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
+        int, bool, name="pre-gate-changed"
+    )
+
     def __init__(self, parent: QtCore.QObject | None = None) -> None:
         super().__init__(parent)
         self._amu_state = False
         self._mmu_state: MMUState | None = None
+        self._pre_gate_sensors: dict[int, bool] = {}
         self.__setup_configfile()
 
     def __setup_configfile(self) -> None:
@@ -107,6 +112,9 @@ def get_state(self) -> MMUState | None:
         """
         return self._mmu_state
 
+    def get_pre_gate_sensors(self) -> dict[int, bool]:
+        return dict(self._pre_gate_sensors)
+
     def is_amu_active(self) -> bool:
         """Returns whether AMU includes are currently uncommented in printer.cfg"""
         return self._amu_state
@@ -171,7 +179,6 @@ def load_gate(self, gate: int) -> None:
 
     def unload(self) -> None:
         """Unload the currently loaded filament by sending MMU_UNLOAD."""
-        ...
         self.run_gcode_signal.emit("MMU_UNLOAD")
 
     def eject_gate(self, gate: int) -> None:
@@ -188,7 +195,7 @@ def eject_all_gates(self, num_gates: int) -> None:
         Args:
            num_gates: Total number of gates(from MMUState.num_gates)
         """
-        cmd = "\n".join(f"MMU_EJECT GATE={i}" for i in range(num_gates))
+        cmd: str = "\n".join(f"MMU_EJECT GATE={i}" for i in range(num_gates))
         self.run_gcode_signal.emit(cmd)
 
     def select_tool(self, tool: int) -> None:
@@ -199,6 +206,18 @@ def select_tool(self, tool: int) -> None:
         """
         self.run_gcode_signal.emit(f"MMU_CHANGE_TOOL TOOL={tool}")
 
+    def on_pre_gate_update(self, values: dict, name: str) -> None:
+        if not name.startswith("Mmu Pre Gate "):
+            return
+        try:
+            gate = int(name.removeprefix("Mmu Pre Gate "))
+        except ValueError:
+            logger.error("Failed to parse Pre-Gate: %s", name)
+            return
+        detected = bool(values.get("filament_detected", False))
+        self._pre_gate_sensors[gate] = detected
+        self.pre_gate_changed.emit(gate, detected)
+
     def update_mmu_state(self, data: dict, name: str = "") -> None:
         """Receive an MMU status dict from Moonraker and update internal state.
 
@@ -216,7 +235,7 @@ def update_mmu_state(self, data: dict, name: str = "") -> None:
             self._mmu_state = self._mmu_state.apply_diff(data)
         self.mmu_state_changed.emit(self._mmu_state)
 
-
-if __name__ == "__main__":
-    manager = AMUManager()
-    print(manager.toggle_amu_system(True))
+    def on_klippy_state(self, state: str) -> None:
+        """React to changes in klippy states"""
+        if state.lower() != "ready":
+            self._mmu_state = None

BlocksScreen/devices/amu/models.py

@@ -1,6 +1,6 @@
 import dataclasses
 from dataclasses import dataclass
-from enum import IntEnum
+from enum import IntEnum, StrEnum
 
 
 class GateStatus(IntEnum):
@@ -10,6 +10,32 @@ class GateStatus(IntEnum):
     EMPTY = -1
 
 
+class FilamentPos(IntEnum):
+    """State-machine position of the filament inside the MMU/extruder path."""
+
+    UNKNOWN = -1
+    UNLOADED = 0
+    HOMED_GATE = 1
+    START_BOWDEN = 2
+    IN_BOWDEN = 3
+    END_BOWDEN = 4
+    HOMED_ENTRY = 5
+    HOMED_EXTRUDER = 6
+    EXTRUDER_ENTRY = 7
+    HOMED_TS = 8
+    IN_EXTRUDER = 9
+    LOADED = 10
+
+
+class SpoolmanSupport(StrEnum):
+    """Level of Spoolman integration configured in Happy-Hare."""
+
+    OFF = "off"
+    READONLY = "readonly"
+    PUSH = "push"
+    PULL = "pull"
+
+
 @dataclass(frozen=True, slots=True)
 class GateInfo:
     index: int
@@ -33,9 +59,12 @@ class MMUState:
     tool: int
     gate: int
     filament: str  # "Loaded" | "Unloaded" | "Unknown"
+    filament_pos: FilamentPos
     action: str
     print_state: str
     reason_for_pause: str
+    has_bypass: bool
+    spoolman_support: SpoolmanSupport
     gates: tuple[GateInfo, ...]
     ttg_map: tuple[int, ...]
 
@@ -87,9 +116,12 @@ def from_status(cls, data: dict) -> "MMUState":
             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", ""),
             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", [])),
         )
@@ -117,12 +149,16 @@ def apply_diff(self, diff: dict) -> "MMUState":
             "gate_spool_id",
         }
         if gate_keys.isdisjoint(diff):
-            # No changes
+            # 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"])
             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

BlocksScreen/lib/panels/mainWindow.py

@@ -202,6 +202,8 @@ def __init__(self):
         self.printer.extruder_update.connect(self.on_extruder_update)
         self.printer.heater_bed_update.connect(self.on_heater_bed_update)
         self.printer.register_callback("mmu", self.amu_manager.update_mmu_state)
+        self.printer.register_callback("filament_switch_sensor", self.amu_manager.on_pre_gate_update)
+        self.printer.register_klippy_callback(self.amu_manager.on_klippy_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

@@ -127,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._klippy_callback: typing.Callable[[str], None] | None = None
         self._callbacks: dict[str, typing.Callable[[dict, str], None]] = {}
 
     def clear_printer_objs(self) -> None:
@@ -169,6 +170,17 @@ def register_callback(
         """
         self.__inject_callback(object_type, callback)
 
+    def register_klippy_callback(self, callback: typing.Callable[[str], None]) -> None:
+        """Register a callback for klippy lifecycle state changes.
+
+        Args:
+            callback:  Callable with signature  ``(state: str) -> None``.
+        """
+        if not callable(callback):
+            logger.warning("register_klippy_callback: not callable")
+            return
+        self._klippy_callback = callback
+
     @QtCore.pyqtSlot(str, name="on_klippy_status")
     def on_klippy_status(self, state: str):
         """Handles klippy update status
@@ -183,7 +195,11 @@ def on_klippy_status(self, state: str):
                 "virtual_sdcard": None,
             }
             self.query_printer_object.emit(_query_request)
+            if self._klippy_callback is not None:
+                self._klippy_callback(state)
             return
+        if self._klippy_callback is not None:
+            self._klippy_callback(state)
         self.clear_printer_objs()  # All other states clear it
 
     @QtCore.pyqtSlot(list, name="on_object_list")
@@ -303,7 +319,6 @@ def _check_callback(self, name: str, values: dict) -> bool:
             _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):

Makefile

@@ -99,14 +99,17 @@ test: ## Unit + UI tests (excludes real D-Bus integration)
 test-fast: ## Stop on first failure, quiet output
 	$(PYTHON) -m pytest -x -q $(PYTEST_IGNORE) $(TESTS)
 
+test-fast-%: ## Stop on first failure, quiet output, scoped to tests/<folder>/
+	$(PYTHON) -m pytest -x -q $(TESTS)/$*
+
 test-unit: ## Unit tests only (*_unit.py)
 	$(PYTHON) -m pytest $(PYTEST_FLAGS) $(TESTS)/*/*_unit.py
 
 test-ui: ## UI tests only (*_ui.py)
 	$(PYTHON) -m pytest $(PYTEST_FLAGS) $(TESTS)/*/*_ui.py
 
-test-network: ## Network subsystem tests (unit + UI, no D-Bus)
-	$(PYTHON) -m pytest $(PYTEST_FLAGS) $(PYTEST_IGNORE) $(TESTS)/network/
+test-%: ## Tests all files inside the /<folder>/
+	$(PYTHON) -m pytest $(PYTEST_FLAGS) $(PYTEST_IGNORE) $(TESTS)/$*/
 
 test-integration: ## D-Bus integration tests (requires live NetworkManager)
 	$(NM_INTEGRATION) $(PYTHON) -m pytest $(PYTEST_FLAGS) $(TESTS)/*/*_integration.py