Backend of Happy-Hare and Spoolman is integrated

Author: gmmcosta15

BlocksScreen/devices/amu/config_toggler.py

@@ -1,6 +1,5 @@
 import logging
 import re
-
 from pathlib import Path
 
 logger = logging.getLogger(__name__)
@@ -34,7 +33,8 @@ def _detect_state(self) -> bool:
             return False
         try:
             text = self._path.read_text()
-            return all(re.search(rf"^\[include {re.escape(f)}\]", text, re.MULTILINE)
+            return all(
+                re.search(rf"^\[include {re.escape(f)}\]", text, re.MULTILINE)
                 for f in AMU_FILES
             )
         except OSError:
@@ -57,10 +57,10 @@ def toggle(self, activate: bool) -> bool:
             return True
         except OSError as e:
             logger.error(
-                "ConfigToggler.toggle(activate=%s): read/write failed: %s", activate,e
+                "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,5 +1,6 @@
 from __future__ import annotations
 
+import dataclasses
 import logging
 import typing
 from pathlib import Path
@@ -15,6 +16,15 @@
 
 logger: logging.Logger = logging.getLogger(__name__)
 
+# Spool Weight threshold for heavy-filament speed profile (grams)
+HEAVY_SPOOL_THRESHOLD_G: float = 1000.0
+# Absolute target speed (mm/s) for heavy spools - used to calculate SPEED % for MMU_GATE_MAP
+HEAVY_SPEED_MM_S: float = 100.0
+# Base gear stepper max_velocity (mm/s) - must match mmu_gear max_velocity in printer.cfg
+BASE_GEAR_SPEED_MM_S: float = 300.0
+# Precomputed speed percentage for heavy spools (avoids repeated division at runtime)
+_HEAVY_SPEED_PERCENT: int = max(1, round(HEAVY_SPEED_MM_S / BASE_GEAR_SPEED_MM_S * 100))
+
 CONFIG_PATH: Path = Path("~/printer_data/config/printer.cfg").expanduser()
 
 
@@ -39,14 +49,74 @@ class AMUManager(QtCore.QObject):
     spool_fetched: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
         int, dict, name="spool-fetched"
     )
+    gate_weight_updated: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal(
+        int, float, name="gate-weight-updated"
+    )
 
     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.spool_fetched.connect(self._apply_spool_data)
+
+    def _apply_spool_data(self, gate: int, data: dict) -> None:
+        """Apply Spoolman spool data to local gate state and sync to Klipper.
+
+        Extracts material, color, weight from the Spoolman response dict,
+        emits MMU_GATE_MAP gcode to sync Klipper, and updates the local GateInfo weight.
+        """
+        filament = data.get("filament", {})
+        material = filament.get("material", "")
+        color = filament.get("color_hex", "")
+        filament_name = filament.get("name", "")
+        temperature = filament.get("settings_extruder_temp")
+        bed_temp = filament.get("settings_bed_temp")
+        spool_id = data.get("id", -1)
+        weight = data.get("used_weight")
+        remaining_weight = data.get("remaining_weight")
+        self.set_gate_info(
+            gate,
+            material,
+            color,
+            spool_id,
+            filament_name=filament_name,
+            temperature=temperature,
+        )
+        if self._mmu_state is not None:
+            if gate >= len(self._mmu_state.gates):
+                logger.warning(
+                    "Gate index %d out of range (%d gates)",
+                    gate,
+                    len(self._mmu_state.gates),
+                )
+                return
+            gates = list(self._mmu_state.gates)
+            updates = {}
+            if weight is not None:
+                updates["weight_g"] = float(weight)
+            if remaining_weight is not None:
+                updates["remaining_weight"] = float(remaining_weight)
+                self._emit_speed_gcode(gate, remaining_weight)
+            if filament_name:
+                updates["filament_name"] = filament_name
+            if temperature is not None:
+                updates["temperature"] = float(temperature)
+            if bed_temp is not None:
+                updates["bed_temp"] = int(bed_temp)
+            if updates:
+                gates[gate] = dataclasses.replace(gates[gate], **updates)
+                self._mmu_state = dataclasses.replace(
+                    self._mmu_state, gates=tuple(gates)
+                )
+
+    def _emit_speed_gcode(self, gate: int, remaining_weight: float) -> None:
+        """Emit MMU_GATE_MAP SPEED=x for the gate based on the spool weight profile."""
+        if remaining_weight > HEAVY_SPOOL_THRESHOLD_G:
+            self.run_gcode_signal.emit(
+                f"MMU_GATE_MAP gate={gate} SPEED={_HEAVY_SPEED_PERCENT}"
+            )
 
     def toggle_amu_system(self, activate: bool) -> None:
         """Enable or disable the AMU system by commenting/uncommenting config includes.
@@ -88,12 +158,15 @@ 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.
+        Emits spool_fetched(gate, data) on success; logs and emits nothing on error.
         """
+
         if self._mmu_state is None:
             return
         if self._mmu_state.spoolman_support is SpoolmanSupport.OFF:
             return
+        if spool_id == -1:
+            return
 
         def _on_result(result: dict | None) -> None:
             if result is not None:
@@ -102,7 +175,13 @@ def _on_result(result: dict | None) -> None:
         self._ws.api.get_spool(spool_id, _on_result)
 
     def set_gate_info(
-        self, gate: int, material: str, color: str, spool_id: int
+        self,
+        gate: int,
+        material: str,
+        color: str,
+        spool_id: int,
+        filament_name: str = "",
+        temperature: int | None = None,
     ) -> None:
         """Sets all gate attributes for a single MMU_GATE
 
@@ -111,10 +190,15 @@ def set_gate_info(
             material (str): Filament material name, e.g. ``"PLA"``.
             color (str): Filament color as hex string, e.g. ``"ff56e0"``.
             spool_id (int): Spoolman spool ID, or -1 if not tracked.
+            filament_name (str): Filament display name from Spoolman.
+            temperature (int | None): Extruder temperature, omitted if None.
         """
-        self.run_gcode_signal.emit(
-            f"MMU_GATE_MAP gate={gate} MATERIAL={material} COLOR={color} SPOOLID={spool_id}"
-        )
+        gcode = f"MMU_GATE_MAP gate={gate} MATERIAL={material} COLOR={color} SPOOLID={spool_id}"
+        if filament_name:
+            gcode = f"MMU_GATE_MAP gate={gate} NAME={filament_name} MATERIAL={material} COLOR={color} SPOOLID={spool_id}"
+        if temperature is not None:
+            gcode += f" TEMP={temperature}"
+        self.run_gcode_signal.emit(gcode)
 
     def set_gate_material(self, gate: int, material: str) -> None:
         """Set the `material` at the gate `gate`
@@ -142,6 +226,8 @@ def set_gate_spool(self, gate: int, spool_id: int) -> None:
             spool_id (int): Spoolman spool ID, or -1 to clear.
         """
         self.run_gcode_signal.emit(f"MMU_GATE_MAP gate={gate} SPOOLID={spool_id}")
+        if spool_id != -1:
+            self.fetch_spool(gate, spool_id)
 
     def home_mmu(self) -> None:
         """Home the MMU selector by sending MMU_HOME."""
@@ -188,18 +274,6 @@ 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.
 
@@ -223,8 +297,45 @@ def on_object_updated(
         """Route object_updated signal from Printer to the appropriate handler."""
         if object_type == "mmu":
             self.update_mmu_state(values)
-        elif object_type == "filament_switch_detected":
+        elif object_type == "filament_switch_sensor":
             self.on_pre_gate_update(values, object_name)
+        elif object_type == "load_cell":
+            self.on_load_cell_update(values, object_name)
+
+    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 on_load_cell_update(self, values: dict, name: str) -> None:
+        """Update gate weight from a Klipper load_cell sensor reading"""
+        if self._mmu_state is None:
+            return
+        try:
+            gate = int(name.removeprefix("load_cell_mmu_"))
+        except ValueError:
+            logger.error("Failed parsing %s Load cell", name)
+            return
+
+        weight = float(values.get("force", 0))
+        if gate >= len(self._mmu_state.gates):
+            logger.warning(
+                "Gate index %d out of range (%d gates)",
+                gate,
+                len(self._mmu_state.gates),
+            )
+            return
+        gates = list(self._mmu_state.gates)
+        gates[gate] = dataclasses.replace(gates[gate], weight_g=weight)
+        self._mmu_state = dataclasses.replace(self._mmu_state, gates=tuple(gates))
+        self.gate_weight_updated.emit(gate, weight)
 
     def on_klippy_state(self, state: str) -> None:
         """React to changes in klippy states"""

BlocksScreen/devices/amu/models.py

@@ -44,8 +44,12 @@ class GateInfo:
     color: str
     color_rgb: tuple[float, float, float]
     spool_id: int
+    filament_name: str = ""
+    temperature: float | None = None
     weight_g: float | None = None
     mid_usage: bool = False
+    remaining_weight: float | None = None
+    bed_temp: int | None = None
 
     @property
     def is_available(self) -> bool:
@@ -69,6 +73,10 @@ class MMUState:
     spoolman_support: SpoolmanSupport
     gates: tuple[GateInfo, ...]
     ttg_map: tuple[int, ...]
+    pending_spool_id: int = -1
+    operation: str = ""
+    sensors: dict = dataclasses.field(default_factory=dict)
+    gate_speed_override: tuple[float, ...] = dataclasses.field(default_factory=tuple)
 
     @property
     def is_paused(self) -> bool:
@@ -99,6 +107,9 @@ def from_status(cls, data: dict) -> "MMUState":
         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)
+        filament_name = data.get("gate_filament_name", [""] * num_gates)
+        temperature = data.get("gate_temperature", [None] * num_gates)
+        speed_override = data.get("gate_speed_override", [])
 
         gates: tuple[GateInfo, ...] = tuple(
             GateInfo(
@@ -108,6 +119,8 @@ def from_status(cls, data: dict) -> "MMUState":
                 color=colors[i],
                 color_rgb=tuple(rgbs[i]),
                 spool_id=spool_ids[i],
+                filament_name=filament_name[i],
+                temperature=temperature[i],
             )
             for i in range(num_gates)
         )
@@ -123,9 +136,15 @@ def from_status(cls, data: dict) -> "MMUState":
             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)),
+            spoolman_support=SpoolmanSupport(
+                data.get("spoolman_support", SpoolmanSupport.OFF)
+            ),
             gates=gates,
             ttg_map=tuple(data.get("ttg_map", [])),
+            pending_spool_id=data.get("pending_spool_id", -1),
+            operation=data.get("operation", ""),
+            sensors=dict(data.get("sensors", {})),
+            gate_speed_override=tuple(speed_override),
         )
 
     def gate_for_tool(self, tool: int) -> int:
@@ -149,6 +168,9 @@ def apply_diff(self, diff: dict) -> "MMUState":
             "gate_color",
             "gate_color_rgb",
             "gate_spool_id",
+            "gate_filament_name",
+            "gate_temperature",
+            "gate_speed_override",
         }
         if gate_keys.isdisjoint(diff):
             # No gate array changes
@@ -158,9 +180,13 @@ def apply_diff(self, diff: dict) -> "MMUState":
             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"])
+                scalar_fields["filament_pos"] = FilamentPos(
+                    scalar_fields["filament_pos"]
+                )
             if "spoolman_support" in scalar_fields:
-                scalar_fields["spoolman_support"] = SpoolmanSupport(scalar_fields["spoolman_support"])
+                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
@@ -170,6 +196,9 @@ def apply_diff(self, diff: dict) -> "MMUState":
             "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],
+            "gate_filament_name": [g.filament_name for g in self.gates],
+            "gate_speed_override": list(self.gate_speed_override),
+            "gate_temperature": [g.temperature for g in self.gates],
         }
         merged = {**dataclasses.asdict(self), **gate_data, **diff}
         return MMUState.from_status(merged)