Connected amu_manager with printer and mainwindow and lowercase AMUFolder and implement simple parts of AMU backend

Author: gmmcosta15

BlocksScreen/devices/AMU/__init__.py BlocksScreen/devices/amu/__init__.pyBlocksScreen/devices/AMU/__init__.py renamed to BlocksScreen/devices/amu/__init__.py

@@ -43,15 +43,11 @@ def __init__(self, parent: QtCore.QObject | None = None) -> None:
         self.__setup_configfile()
 
     def __setup_configfile(self) -> None:
-        """
-        Sets up local configfile variable
-
-        Raises:
-            FileNotFoundError: File Not Found
-        """
-        self._config_filename = CONFIG_PATH
+        """Sets up local configfile variable"""
+        self._config_filename: Path | None = CONFIG_PATH
         if not self._config_filename.exists():
-            raise FileNotFoundError(("Config file not found %s", self._config_filename))
+            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
@@ -62,6 +58,10 @@ def _apply_patterns(self, state: bool) -> bool:
         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
 
@@ -122,7 +122,9 @@ def set_gate_info(
             color (str): Filament color as hex string, e.g. ``"ff56e0"``.
             spool_id (int): Spoolman spool ID, or -1 if not tracked.
         """
-        ...
+        self.run_gcode_signal.emit(
+            f"MMU_GATE_MAP gate={gate} MATERIAL={material} COLOR={color} SPOOLID={spool_id}"
+        )
 
     def set_gate_material(self, gate: int, material: str) -> None:
         """Set the `material` at the gate `gate`
@@ -131,6 +133,7 @@ def set_gate_material(self, gate: int, material: str) -> None:
             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}")
 
     def set_gate_color(self, gate: int, color: str) -> None:
         """Set the `color` at the gate `gate`
@@ -139,6 +142,7 @@ def set_gate_color(self, gate: int, color: str) -> None:
             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}")
 
     def set_gate_spool(self, gate: int, spool_id: int) -> None:
         """Set the `spool_id` at the gate `gate`
@@ -147,31 +151,53 @@ def set_gate_spool(self, gate: int, spool_id: int) -> None:
             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}")
 
     def home_mmu(self) -> None:
         """Home the MMU selector by sending MMU_HOME."""
+        self.run_gcode_signal.emit("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, gate: int) -> None:
         """Load filament from the specified gate by sending MMU_LOAD
 
         Args:
             gate (int): Gate index to select (0-based)
         """
+        self.run_gcode_signal.emit(f"MMU_SELECT gate={gate}\nMMU_LOAD")
 
     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:
+        """Fully eject filament from gate, releasing from MMU gear.
+
+        Args:
+            gate: Gate index to eject from, on None to use currently selected gate.
+        """
+        self.run_gcode_signal.emit(f"MMU_EJECT GATE={gate}")
+
+    def eject_all_gates(self, num_gates: int) -> None:
+        """Fully eject filament from all gates sequentially(amu_manager.get_state().num_gates)
+
+        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))
+        self.run_gcode_signal.emit(cmd)
 
     def select_tool(self, tool: int) -> None:
         """Select a tool, triggering a filament change if needed.
 
         Args:
             tool (int): Tool index to select (0-based).
         """
-        ...
+        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:
@@ -184,7 +210,11 @@ def update_mmu_state(self, data: dict) -> None:
         Args:
             data (dict): Raw MMU status or diff dict from Moonraker
         """
-        ...
+        if self._mmu_state is None:
+            self._mmu_state = MMUState.from_status(data)
+        else:
+            self._mmu_state = self._mmu_state.apply_diff(data)
+        self.mmu_state_changed.emit(self._mmu_state)
 
 
 if __name__ == "__main__":

BlocksScreen/devices/AMU/manager.py BlocksScreen/devices/amu/manager.pyBlocksScreen/devices/AMU/manager.py renamed to BlocksScreen/devices/amu/manager.py

@@ -4,6 +4,7 @@
 
 import events
 from configfile import BlocksScreenConfig, get_configparser
+from devices.amu.manager import AMUManager
 from devices.storage import USBManager
 from lib.files import Files
 from lib.machine import MachineControl
@@ -118,8 +119,8 @@ def __init__(self):
         gdir = None
         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.notiPage = NotificationPage(self)
         self.mc = MachineControl(self)
@@ -200,6 +201,8 @@ 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.amu_manager.run_gcode_signal.connect(self.ws.api.run_gcode)
         self.run_gcode_signal.connect(self.ws.api.run_gcode)
 
         self.ui.main_content_widget.currentChanged.connect(slot=self.reset_tab_indexes)

BlocksScreen/devices/AMU/models.py BlocksScreen/devices/amu/models.pyBlocksScreen/devices/AMU/models.py renamed to BlocksScreen/devices/amu/models.py

@@ -79,6 +79,10 @@ 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],
@@ -740,3 +744,6 @@ def _unload_filament_object_updated(self, values: dict, name: str) -> None:
     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)