refactor(amu): drop heavy-spool speed override, refuse load/unload Happy-Hare would reject

Author: gmmcosta15

BlocksScreen/devices/amu/manager.py

@@ -24,13 +24,6 @@
 
 logger: logging.Logger = logging.getLogger(__name__)
 
-HEAVY_SPOOL_THRESHOLD_G: float = 1000.0
-HEAVY_SPEED_MM_S: float = 100.0
-# Must match mmu_gear max_velocity in printer.cfg.
-BASE_GEAR_SPEED_MM_S: float = 300.0
-_HEAVY_SPEED_PERCENT: int = max(1, round(HEAVY_SPEED_MM_S / BASE_GEAR_SPEED_MM_S * 100))
-_DEFAULT_SPEED_PERCENT: int = 100
-
 CONFIG_PATH: Path = Path("~/printer_data/config/printer.cfg").expanduser()
 
 
@@ -61,17 +54,13 @@ 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, ws: MoonWebSocket, parent: QtCore.QObject | None = None) -> None:
         super().__init__(parent)
         self._config_toggler = ConfigToggler(CONFIG_PATH)
         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 _gate_map(self, gate: int, **params: object) -> bool:
         """Emit MMU_GATE_MAP for one gate; False when no gate is selected yet."""
@@ -88,27 +77,6 @@ def _gate_map(self, gate: int, **params: object) -> bool:
         self.run_gcode_signal.emit(f"MMU_GATE_MAP GATE={gate}{args} QUIET=1")
         return True
 
-    def _apply_spool_data(self, gate: int, data: dict) -> None:
-        """Set the gate's gear speed from a Spoolman payload's remaining weight."""
-        if data.get("id", -1) == -1:
-            logger.warning("Spool payload for gate %d carries no id", gate)
-            return
-        remaining = data.get("remaining_weight")
-        if remaining is not None:
-            self._apply_speed_profile(gate, float(remaining))
-
-    def _apply_speed_profile(self, gate: int, remaining_weight: float) -> None:
-        """Slow the gear for a heavy spool and restore it once the spool is light."""
-        if self._mmu_state is None or not 0 <= gate < len(self._mmu_state.gates):
-            return
-        wanted = (
-            _HEAVY_SPEED_PERCENT
-            if remaining_weight > HEAVY_SPOOL_THRESHOLD_G
-            else _DEFAULT_SPEED_PERCENT
-        )
-        if self._mmu_state.gates[gate].speed_override != wanted:
-            self._gate_map(gate, SPEED=wanted)
-
     def toggle_amu_system(self, activate: bool) -> None:
         """Swap the printer.cfg variant, restarting on change; emits amu_toggled."""
         result = self._config_toggler.toggle(activate)
@@ -133,21 +101,6 @@ def is_amu_active(self) -> bool:
         state = self._mmu_state
         return self.is_amu_configured() and state is not None and state.enabled
 
-    def fetch_spool(self, gate: int, spool_id: int) -> None:
-        """Fetch *spool_id* from Spoolman and emit spool_fetched; no-op when off."""
-        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:
-                self.spool_fetched.emit(gate, result)
-
-        self._ws.api.get_spool(spool_id, _on_result)
-
     def set_gate_info(
         self,
         gate: int,
@@ -184,22 +137,20 @@ def set_gate_color(self, gate: int, color: str) -> None:
         self._gate_map(gate, COLOR=color.lstrip("#"))
 
     def set_gate_spool(self, gate: int, spool_id: int) -> None:
-        """Bind *gate* to a Spoolman spool (-1 unbinds) and fetch its data."""
-        if self._gate_map(gate, SPOOLID=spool_id) and spool_id != -1:
-            self.fetch_spool(gate, spool_id)
+        """Bind *gate* to a Spoolman spool; -1 unbinds."""
+        self._gate_map(gate, SPOOLID=spool_id)
 
     def assign_spool(self, gate: int, spool: dict) -> None:
-        """Map *gate* from a Spoolman spool payload and apply it without a refetch."""
+        """Map *gate* from a Spoolman spool payload."""
         filament = spool.get("filament") or {}
-        if self.set_gate_info(
+        self.set_gate_info(
             gate,
             filament.get("material") or "",
             filament.get("color_hex") or "",
             spool.get("id", -1),
             filament_name=filament.get("name") or "",
             temperature=filament.get("settings_extruder_temp"),
-        ):
-            self._apply_spool_data(gate, spool)
+        )
 
     def update_spool_weight(self, gate: int, used_weight: float) -> None:
         """Push *used_weight* to Spoolman for the spool at *gate*."""
@@ -238,22 +189,34 @@ def reset_mmu(self) -> None:
         self.run_gcode_signal.emit("MMU_RESET CONFIRM=1")
 
     def _selected_gate_unloaded(self) -> GateInfo | None:
-        """Return the selected gate when it is a real gate with nothing loaded."""
+        """Return the selected gate when the MMU is enabled, homed and unloaded."""
         state = self._mmu_state
-        if state is None or state.filament_pos != FilamentPos.UNLOADED:
+        if (
+            state is None
+            or not (state.enabled and state.is_homed)
+            or state.filament_pos != FilamentPos.UNLOADED
+        ):
             return None
         return state.current_gate_info
 
     def load_gate(self) -> bool:
         """Load filament from the selected gate by sending MMU_LOAD; True if sent."""
         if self._selected_gate_unloaded() is None:
-            logger.warning("Ignoring MMU_LOAD: no unloaded gate selected")
+            logger.warning("Ignoring MMU_LOAD: no homed, unloaded gate selected")
             return False
         self.run_gcode_signal.emit("MMU_LOAD")
         return True
 
     def unload(self) -> bool:
         """Unload the currently loaded filament by sending MMU_UNLOAD; True if sent."""
+        state = self._mmu_state
+        if (
+            state is None
+            or not state.enabled
+            or state.filament_pos == FilamentPos.UNLOADED
+        ):
+            logger.warning("Ignoring MMU_UNLOAD: MMU disabled or nothing loaded")
+            return False
         self.run_gcode_signal.emit("MMU_UNLOAD")
         return True
 

BlocksScreen/lib/panels/widgets/FilamentTab/basicFilamentPanel.py

@@ -354,7 +354,12 @@ def filament_state(self):
     @filament_state.setter
     def filament_state(self, update: FilamentStates) -> None:
         self._filament_state = update
-        self.Basic_fp_load_btn.setEnabled(update is not self.FilamentStates.LOADED)
+        # Happy-Hare only loads from UNLOADED; a sensorless non-MMU printer stays UNKNOWN.
+        self.Basic_fp_load_btn.setEnabled(
+            update is self.FilamentStates.UNLOADED
+            if self.mmu_configured
+            else update is not self.FilamentStates.LOADED
+        )
         self.Basic_fp_unload_btn.setEnabled(update is not self.FilamentStates.UNLOADED)
         if self._lbl_pos.select_option(update.name.capitalize()):
             self._pos_committed = self._lbl_pos.text()

tests/amu/test_manager_unit.py

@@ -354,8 +354,16 @@ def test_load_gate(self, manager, qtbot) -> None:
             assert manager.load_gate() is True
         assert blocker.args == ["MMU_LOAD"]
 
+    # Happy-Hare drops MMU_LOAD when disabled or unhomed (mmu.py:6936).
     @pytest.mark.parametrize(
-        "overrides", [None, {"filament_pos": 10}, {"filament_pos": 0, "gate": -1}]
+        "overrides",
+        [
+            None,
+            {"filament_pos": 10},
+            {"filament_pos": 0, "gate": -1},
+            {"filament_pos": 0, "enabled": False},
+            {"filament_pos": 0, "is_homed": False},
+        ],
     )
     def test_load_gate_refused(self, manager, qtbot, overrides) -> None:
         if overrides is not None:
@@ -364,10 +372,21 @@ def test_load_gate_refused(self, manager, qtbot, overrides) -> None:
             assert manager.load_gate() is False
 
     def test_unload(self, manager, qtbot) -> None:
+        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
         with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
             assert manager.unload() is True
         assert blocker.args == ["MMU_UNLOAD"]
 
+    # Happy-Hare drops MMU_UNLOAD when disabled or already unloaded (mmu.py:6981).
+    @pytest.mark.parametrize(
+        "overrides", [None, {"filament_pos": 0}, {"enabled": False}]
+    )
+    def test_unload_refused(self, manager, qtbot, overrides) -> None:
+        if overrides is not None:
+            manager.update_mmu_state(_status(**overrides))
+        with qtbot.assertNotEmitted(manager.run_gcode_signal):
+            assert manager.unload() is False
+
     def test_eject_gate(self, manager, qtbot) -> None:
         manager.update_mmu_state(_status(filament_pos=0))
         with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
@@ -398,7 +417,13 @@ def test_check_gate_allows_an_empty_gate(self, manager, qtbot) -> None:
 
     # MMU_CHECK_GATE runs a full unload first when filament is loaded.
     @pytest.mark.parametrize(
-        "overrides", [None, {"filament_pos": 10}, {"filament_pos": 0, "gate": -2}]
+        "overrides",
+        [
+            None,
+            {"filament_pos": 10},
+            {"filament_pos": 0, "gate": -2},
+            {"filament_pos": 0, "is_homed": False},
+        ],
     )
     def test_check_gate_refused(self, manager, qtbot, overrides) -> None:
         if overrides is not None:
@@ -582,44 +607,6 @@ def test_false_when_mmu_disabled(self, manager_with_cfg, qtbot) -> None:
         assert mgr.is_amu_active() is False
 
 
-class TestSpoolManFetch:
-    def test_noop_when_mmu_state_none(self, manager, qtbot) -> None:
-        with qtbot.assertNotEmitted(manager.spool_fetched):
-            manager.fetch_spool(0, 42)
-        manager._ws.api.get_spool.assert_not_called()
-
-    def test_noop_when_spoolman_off(self, manager, qtbot) -> None:
-        status = {**_FULL_STATUS_WITH_SPOOLMAN, "spoolman_support": "off"}
-        manager.update_mmu_state(status)
-        with qtbot.assertNotEmitted(manager.spool_fetched):
-            manager.fetch_spool(0, 42)
-        manager._ws.api.get_spool.assert_not_called()
-
-    def test_emit_spool_fetched_on_sucess(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        spool_data = {"id": 42, "filament": {"name": "PLA"}, "used_weight": 50.0}
-        manager.fetch_spool(0, 42)
-        callback = manager._ws.api.get_spool.call_args.args[1]
-        with qtbot.waitSignal(manager.spool_fetched) as blocker:
-            callback(spool_data)
-        assert blocker.args[0] == 0
-        assert blocker.args[1] == spool_data
-
-    def test_no_emit_on_ws_failure(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        manager.fetch_spool(0, 42)
-        callback = manager._ws.api.get_spool.call_args.args[1]
-        with qtbot.assertNotEmitted(manager.spool_fetched):
-            callback(None)
-
-    # -1 is Happy-Hare's "no spool", not a Spoolman id.
-    def test_noop_when_spool_id_unset(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        with qtbot.assertNotEmitted(manager.spool_fetched):
-            manager.fetch_spool(0, -1)
-        manager._ws.api.get_spool.assert_not_called()
-
-
 class TestUpdateSpoolWeight:
     # gates[-1] wraps, so an unselected gate hit the last gate's spool.
     @pytest.mark.parametrize("gate", [-1, -2, 2, 99])
@@ -689,18 +676,7 @@ def test_load_cell_updates_are_ignored(self, manager, qtbot) -> None:
         assert not hasattr(manager, "gate_weight_updated")
 
 
-class TestApplySpoolData:
-    # Happy-Hare's _spoolman_sync owns the gate map; writing it here races it.
-    def test_emits_no_gcode(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        with qtbot.assertNotEmitted(manager.run_gcode_signal):
-            manager._apply_spool_data(0, _SPOOL_DATA)
-
-    # A spool callback can land after on_klippy_state() dropped the state.
-    def test_noop_when_mmu_state_none(self, manager) -> None:
-        manager._apply_spool_data(0, _SPOOL_DATA)
-        assert manager.get_state() is None
-
+class TestGcodeQuoting:
     # User-typed Spoolman text must survive Klipper's shlex parse as one value.
     @pytest.mark.parametrize(
         ("name", "color"),
@@ -726,51 +702,6 @@ def test_emitted_gcode_survives_klipper_parser(
         assert params["COLOR"] == color.lstrip("#")
         assert params["SPOOLID"] == "42"
 
-    def test_triggered_by_spool_fetched_signal(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        manager.fetch_spool(0, 42)
-        callback = manager._ws.api.get_spool.call_args.args[1]
-        with qtbot.waitSignal(manager.run_gcode_signal) as blocker:
-            callback({**_SPOOL_DATA, "remaining_weight": 1500.0})
-        assert blocker.args == ["MMU_GATE_MAP GATE=0 SPEED=33 TEMP=0 QUIET=1"]
-
-    def test_payload_without_id_is_dropped(self, manager, qtbot) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        with qtbot.assertNotEmitted(manager.run_gcode_signal):
-            manager._apply_spool_data(0, {"remaining_weight": 1500.0})
-
-    @staticmethod
-    def _speed_gcode(manager, remaining_weight, speed_override) -> list[str]:
-        manager.update_mmu_state(_status(gate_speed_override=[speed_override, 100]))
-        data = {k: v for k, v in _SPOOL_DATA.items() if k != "remaining_weight"}
-        if remaining_weight is not None:
-            data["remaining_weight"] = remaining_weight
-        signals = []
-        manager.run_gcode_signal.connect(signals.append)
-        manager._apply_spool_data(0, data)
-        return signals
-
-    @pytest.mark.parametrize("remaining_weight", [1000.1, 1500.0, 5000.0])
-    def test_heavy_spool_slows_the_gear(self, manager, remaining_weight) -> None:
-        assert self._speed_gcode(manager, remaining_weight, 100) == [
-            "MMU_GATE_MAP GATE=0 SPEED=33 TEMP=0 QUIET=1"
-        ]
-
-    @pytest.mark.parametrize("remaining_weight", [0.0, 950.0, 1000.0])
-    def test_light_spool_restores_full_speed(self, manager, remaining_weight) -> None:
-        assert self._speed_gcode(manager, remaining_weight, 33) == [
-            "MMU_GATE_MAP GATE=0 SPEED=100 TEMP=0 QUIET=1"
-        ]
-
-    @pytest.mark.parametrize(
-        ("remaining_weight", "speed_override"),
-        [(None, 100), (None, 33), (950.0, 100), (1500.0, 33)],
-    )
-    def test_speed_already_right_emits_nothing(
-        self, manager, remaining_weight, speed_override
-    ) -> None:
-        assert self._speed_gcode(manager, remaining_weight, speed_override) == []
-
 
 class TestAssignSpool:
     @staticmethod
@@ -787,50 +718,11 @@ def test_maps_gate_from_payload(self, manager) -> None:
             "SPOOLID=42 TEMP=215 QUIET=1"
         ]
 
-    # The picker already holds the payload, so no Spoolman round trip.
-    def test_applies_payload_without_refetch(self, manager) -> None:
-        self._gcode(manager, 1, _SPOOL_DATA)
-        manager._ws.api.get_spool.assert_not_called()
-
-    def test_heavy_spool_slows_the_gear(self, manager) -> None:
-        gcode = self._gcode(manager, 1, {**_SPOOL_DATA, "remaining_weight": 1500.0})
-        assert gcode[1:] == ["MMU_GATE_MAP GATE=1 SPEED=33 TEMP=0 QUIET=1"]
-
     # Spoolman sends null for an unset extruder temp.
     def test_null_temp_lets_happy_hare_default_it(self, manager) -> None:
         filament = {**_SPOOL_DATA["filament"], "settings_extruder_temp": None}
         (gcode,) = self._gcode(manager, 1, {**_SPOOL_DATA, "filament": filament})
         assert "TEMP" not in gcode
 
     def test_unselected_gate_emits_nothing(self, manager) -> None:
-        heavy = {**_SPOOL_DATA, "remaining_weight": 1500.0}
-        assert self._gcode(manager, -1, heavy) == []
-
-
-class TestSetGateSpoolAutoFetch:
-    def test_calls_fetch_spool_when_activate(self, manager) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        manager.set_gate_spool(0, 42)
-        manager._ws.api.get_spool.assert_called_once()
-        assert manager._ws.api.get_spool.call_args.args[0] == 42
-
-    def test_skips_fetch_when_spool_id_minus_one(self, manager) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        manager.set_gate_spool(0, -1)
-        manager._ws.api.get_spool.assert_not_called()
-
-    def test_skips_fetch_when_no_state(self, manager) -> None:
-        manager.set_gate_spool(0, 42)
-        manager._ws.api.get_spool.assert_not_called()
-
-    def test_skips_fetch_when_spoolman_off(self, manager) -> None:
-        status = {**_FULL_STATUS_WITH_SPOOLMAN, "spoolman_support": "off"}
-        manager.update_mmu_state(status)
-        manager.set_gate_spool(0, 42)
-        manager._ws.api.get_spool.assert_not_called()
-
-    # No MMU_GATE_MAP was sent, so caching a spool for it would lie.
-    def test_skips_fetch_when_gate_unselected(self, manager) -> None:
-        manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN)
-        manager.set_gate_spool(-1, 42)
-        manager._ws.api.get_spool.assert_not_called()
+        assert self._gcode(manager, -1, _SPOOL_DATA) == []