update: add more backend methods, added callback calls after moonraker response and addition of tests for the new implementations

Author: gmmcosta15

BlocksScreen/devices/amu/__init__.py

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

BlocksScreen/devices/amu/config_toggler.py

@@ -0,0 +1,66 @@
+import logging
+import re
+
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+AMU_FILES: list[str] = [
+    "mmu/base/*.cfg",
+    "mmu/optional/client_macros.cfg",
+    "filament_manager.cfg",
+]
+
+_AMU_PATTERNS: list[re.Pattern[str]] = [
+    re.compile(rf"^(#?)(\[include {re.escape(f)}\])", re.MULTILINE) for f in AMU_FILES
+]
+
+
+class ConfigToggler:
+    """Manages commenting/uncommenting AMU cfg includes in printer.cfg"""
+
+    def __init__(self, config_path: Path) -> None:
+        """Store path, detects current state from file"""
+        self._path: Path | None = None
+        self._state: bool = False
+        if config_path.exists():
+            self._path = config_path
+            self._state = self._detect_state()
+        else:
+            logger.warning("Config File not found %s", config_path)
+
+    def _detect_state(self) -> bool:
+        if self._path is None:
+            return False
+        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
+
+    def toggle(self, activate: bool) -> bool:
+        """Comment or uncomment AMU includes. Returns True on success, False on no-op or error."""
+        if self._path is None:
+            logger.warning("No Config File available")
+            return False
+        if self._state == activate:
+            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
+        except OSError as e:
+            logger.error(
+                "ConfigToggler.toggle(activate=%s): read/write failed: %s", activate,e
+            )
+            return False
+    
+    def is_configured(self) -> bool:
+        """Return True if AMU includes are currently uncommented"""
+        return self._state

BlocksScreen/devices/amu/manager.py

@@ -1,24 +1,22 @@
+from __future__ import annotations
+
 import logging
-import re
 import typing
 from pathlib import Path
 
 from PyQt6 import QtCore
 
-from .models import MMUState
+from BlocksScreen.devices.amu.config_toggler import ConfigToggler
+
+from .models import MMUState, SpoolmanSupport
+
+if typing.TYPE_CHECKING:
+    from BlocksScreen.lib.moonrakerComm import MoonWebSocket
+
 
 logger: logging.Logger = logging.getLogger(__name__)
 
 CONFIG_PATH: Path = Path("~/printer_data/config/printer.cfg").expanduser()
-AMU_FILES: list[str] = [
-    "mmu/base/*.cfg",
-    "mmu/optional/client_macros.cfg",
-    "filament_manager.cfg",
-]
-
-AMU_PATTERNS: list[re.Pattern[str]] = [
-    re.compile(rf"^(#?)(\[include {re.escape(f)}\])", re.MULTILINE) for f in AMU_FILES
-]
 
 
 class AMUManager(QtCore.QObject):
@@ -39,55 +37,17 @@ 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"
+    )
 
-    def __init__(self, parent: QtCore.QObject | None = None) -> None:
+    def __init__(self, ws: MoonWebSocket, parent: QtCore.QObject | None = None) -> None:
         super().__init__(parent)
+        self._config_toggler = ConfigToggler(CONFIG_PATH)
         self._amu_state = False
+        self._ws = ws
         self._mmu_state: MMUState | None = None
         self._pre_gate_sensors: dict[int, bool] = {}
-        self.__setup_configfile()
-
-    def __setup_configfile(self) -> None:
-        """Sets up local configfile variable"""
-        self._config_filename: Path | None = CONFIG_PATH
-        if not self._config_filename.exists():
-            logger.warning("Config file not found %s", self._config_filename)
-            self._config_filename = None
-
-    def _apply_patterns(self, state: bool) -> bool:
-        """Method that comments/uncomments the AMU_FILES from the printer.cfg according with state value
-
-        Args:
-            state (bool): True: Uncomment, False: Comment
-
-        Returns:
-            bool: True: Success, False: Failed
-        """
-        if self._config_filename is None:
-            logger.warning("_apply_patterns called but no config file available")
-            return False
-
-        if self._amu_state == state:
-            return False
-
-        replacement = r"\2"
-        if not state:
-            replacement = r"#\2"
-        try:
-            text: str = self._config_filename.read_text()
-            for file in AMU_PATTERNS:
-                text = file.sub(replacement, text)
-            self._config_filename.write_text(text)
-            self._amu_state = state
-            return True
-        except OSError as e:
-            logger.error(
-                "Failed to apply(state=%s) AMU System: could not read/write %s\n%s",
-                state,
-                self._config_filename,
-                e,
-            )
-            return False
 
     def toggle_amu_system(self, activate: bool) -> None:
         """Enable or disable the AMU system by commenting/uncommenting config includes.
@@ -99,8 +59,10 @@ def toggle_amu_system(self, activate: bool) -> None:
             activate (bool): True to enable the AMU, False to disable it.
 
         """
-        result: bool = self._apply_patterns(activate)
+        result: bool = self._config_toggler.toggle(activate)
         self.amu_toggled.emit(result)
+        if result:
+            self.run_gcode_signal.emit("FIRMWARE_RESTART")
 
     def get_state(self) -> MMUState | None:
         """Returns current MMU state, None if not yet received.
@@ -115,9 +77,30 @@ def get_state(self) -> MMUState | None:
     def get_pre_gate_sensors(self) -> dict[int, bool]:
         return dict(self._pre_gate_sensors)
 
+    def is_amu_configured(self) -> bool:
+        """Return True if AMU includes are uncommented in printer.cfg."""
+        return self._config_toggler.is_configured()
+
     def is_amu_active(self) -> bool:
         """Returns whether AMU includes are currently uncommented in printer.cfg"""
-        return self._amu_state
+        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 sucess; logs and emits nothing on error.
+        """
+        if self._mmu_state is None:
+            return
+        if self._mmu_state.spoolman_support is SpoolmanSupport.OFF:
+            return
+
+        def _on_result(result: dict | None) -> None:
+            if result is not None:
+                self.spool_fetched.emit(gate, result)
+
+        self._ws.api.get_spool(spool_id, _on_result)
 
     def set_gate_info(
         self, gate: int, material: str, color: str, spool_id: int
@@ -239,3 +222,4 @@ 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 = {}

BlocksScreen/devices/amu/models.py

@@ -44,6 +44,8 @@ class GateInfo:
     color: str
     color_rgb: tuple[float, float, float]
     spool_id: int
+    weight_g: float | None = None
+    mid_usage: bool = False
 
     @property
     def is_available(self) -> bool:

BlocksScreen/lib/moonrakerComm.py

@@ -267,6 +267,15 @@ def on_message(self, *args) -> None:
                 self.klippy_state_signal.emit(response["result"]["klippy_state"])
                 return
             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:
+                        logger.error(
+                            "WS request %s error: %s", _entry[0], response.get("error")
+                        )
+                    return
                 if "error" in response:
                     message_event = WebSocketMessageReceived(
                         method="error",
@@ -302,21 +311,23 @@ def on_message(self, *args) -> None:
         except Exception as e:
             logger.info(f"Unexpected error while creating websocket message event: {e}")
 
-    def send_request(self, method: str, params: dict = {}) -> bool:
+    def send_request(self, method: str, params: dict = {}, callback=None) -> bool:
         """Send a request over the websocket
 
         Args:
             method (str): Websocket method name
             params (dict, optional): parameters for the websocket method. Defaults to {}.
-
+            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.
         Returns:
             bool: Whether the method finished and a request was sent
         """
         if not self.connected or self.ws is None:
             return False
 
         self._request_id += 1
-        self.request_table[self._request_id] = [method, params]
+        self.request_table[self._request_id] = [method, params, callback]
         packet = {
             "jsonrpc": "2.0",
             "method": method,
@@ -809,3 +820,13 @@ def history_get_job(self, uid: str):
     def history_delete_job(self, uid: str):
         """Request delete job history"""
         raise NotImplementedError
+
+    #   ---------------------------------AMU----------------------------------
+
+    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},
+            callback=callback,
+        )

BlocksScreen/lib/moonrest.py

@@ -87,6 +87,26 @@ def get_server_info(self):
         """
         return self.get_request(method="server/info")
 
+    def get_spool(self, spool_id: int) -> dict | None:
+        """GET /server/spoolman/spool/{spool_id} via Moonraker
+
+        Returns spool dict on success, None on HTTP/network/JSON error.
+        """
+        response = self.get_request(f"server/spoolman/spool/{spool_id}")
+        if not isinstance(response, dict):
+            return None
+        return response.get("result")
+
+    def set_spool_used_weight(self, spool_id: int, weight: float) -> bool:
+        """POST /server/spoolman/spool/{spool_id} to update used_weight.
+
+        Returns True on sucess, False on any error.
+        """
+        response = self.post_request(
+            f"server/spoolman/spool/{spool_id}", json={"used_weight": weight}
+        )
+        return response is not None
+
     def firmware_restart(self):
         """firmware_restart
             POST to /printer/firmware_restart to firmware restart Klipper

BlocksScreen/lib/panels/mainWindow.py

@@ -120,8 +120,8 @@ def __init__(self):
         if usb_config:
             gdir = usb_config.get("gcodes_dir", default=None)
         self.usb_manager: USBManager = USBManager(parent=self, gcodes_dir=gdir)
-        self.amu_manager: AMUManager = AMUManager(parent=self)
         self.ws = MoonWebSocket(self)
+        self.amu_manager: AMUManager = AMUManager(ws=self.ws, parent=self)
         self.notiPage = NotificationPage(self)
         self.mc = MachineControl(self)
         self.file_data = Files(self, self.ws)
@@ -202,7 +202,9 @@ 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_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)