feat(amu): initial commit

Author: gmmcosta15

BlocksScreen/devices/AMU/__init__.py

@@ -0,0 +1,25 @@
+"""Network Manager Package
+
+Architecture:
+    AMUManager (manager.py)
+        └── Main AMU manager its responsible for the AMU backend
+    Models (models.py)
+        └── Data classes for type safety
+        └── Enums for states and types
+"""
+
+from .manager import AMUManager
+from .models import (
+    GateInfo,
+    GateStatus,
+    MMUState,
+)
+
+__all__: list[str] = [
+    # manager
+    "AMUManager",
+    # models
+    "GateInfo",
+    "GateStatus",
+    "MMUState",
+]

BlocksScreen/devices/AMU/manager.py

@@ -0,0 +1,86 @@
+import logging
+import re
+from pathlib import Path
+
+from PyQt6 import QtCore
+
+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):
+    """Main manager of the AMU system"""
+
+    def __init__(self, parent: QtCore.QObject | None = None) -> None:
+        super().__init__(parent)
+        self._amu_state = False
+        self.__setup_configfile()
+
+    def __setup_configfile(self) -> None:
+        """
+        Sets up local configfile variable
+
+        Raises:
+            FileNotFoundError: File Not Found
+        """
+        self._config_filename = CONFIG_PATH
+        if not self._config_filename.exists():
+            raise FileNotFoundError(("Config file not found %s", self._config_filename))
+
+    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: Sucess, False: Failed
+        """
+        # signal to say that it was changed AMU para mainwindow
+        if self._amu_state == state:
+            return
+
+        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%e",
+                state,
+                self._config_filename,
+                e,
+            )
+            return False
+
+    def toggle_amu_system(self, activate: 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: Sucess, False: Failed
+        """
+        return self._apply_patterns(activate)
+
+
+if __name__ == "__main__":
+    manager = AMUManager()
+    print(manager.toggle_amu_system(True))

BlocksScreen/devices/AMU/models.py

@@ -0,0 +1,97 @@
+import dataclasses
+from dataclasses import dataclass
+from enum import IntEnum
+
+
+class GateStatus(IntEnum):
+    UNKNOWN = 0
+    AVAILABLE = 1
+    AVAILABLE_FROM_BUFFER = 2
+    EMPTY = -1
+
+
+@dataclass(frozen=True, slots=True)
+class GateInfo:
+    index: int
+    status: GateStatus
+    material: str
+    color: str
+    color_rgb: tuple[float, float, float]
+    spool_id: int
+
+
+@dataclass(frozen=True, slots=True)
+class MMUState:
+    enabled: bool
+    is_homed: bool
+    num_gates: int
+    tool: int
+    gate: int
+    filament: str  # "Loaded" | "Unloaded" | "Unknown"
+    action: str
+    print_state: str
+    reason_for_pause: str
+    gates: tuple[GateInfo, ...]
+    ttg_map: tuple[int, ...]
+
+    @classmethod
+    def from_status(cls, data: dict) -> "MMUState":
+        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)
+
+        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],
+            )
+            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"),
+            action=data.get("action", ""),
+            print_state=data.get("print_state", ""),
+            reason_for_pause=data.get("reason_for_pause", ""),
+            gates=gates,
+            ttg_map=tuple(data.get("ttg_map", [])),
+        )
+
+    def apply_diff(self, diff: dict) -> "MMUState":
+        gate_keys: set[str] = {
+            "gate_status",
+            "gate_material",
+            "gate_color",
+            "gate_color_rgb",
+            "gate_spool_id",
+        }
+        if gate_keys.isdisjoint(diff):
+            # No changes
+            scalar_fields = {
+                k: v for k, v in diff.items() if k in MMUState.__dataclass_fields__
+            }
+            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],
+        }
+        merged = {**dataclasses.asdict(self), **gate_data, **diff}
+        return MMUState.from_status(merged)