diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index c10eb973..a458d7bb 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -187,32 +187,46 @@ def reset_mmu(self) -> None: """Erase all persisted MMU state (gate map, TTG map, statistics) and re-init.""" self.run_gcode_signal.emit("MMU_RESET CONFIRM=1") - def load_refusal(self) -> str: - """Return why the selected gate cannot load, or "" when it is idle-ready.""" + @staticmethod + def _busy_refusal(state: MMUState) -> str: + """Return why the MMU cannot take a command now, or "" when it is idle.""" + # No print guard: HH v3 parks and restores the toolhead around these mid-print. + if not state.enabled: + return "MMU is disabled." + if state.is_busy: + return f"MMU is busy ({state.action})." + return "" + + def check_refusal(self) -> str: + """Return why the selected gate cannot be checked, or "" when it can.""" state = self._mmu_state if state is None: return "MMU status not received yet." - if not state.enabled: - return "MMU is disabled." - if state.is_printing: - return "Not available while printing." + if reason := self._busy_refusal(state): + return reason if not state.is_homed: return "MMU is not homed." - if state.filament_pos != FilamentPos.UNLOADED: - return "Filament is not fully unloaded." if state.current_gate_info is None: return "No gate is selected." return "" + def load_refusal(self) -> str: + """Return why the selected gate cannot load, or "" when it is idle-ready.""" + reason = self.check_refusal() + state = self._mmu_state + if reason or state is None: + return reason + if state.filament_pos != FilamentPos.UNLOADED: + return "Filament is not fully unloaded." + return "" + def unload_refusal(self) -> str: """Return why the filament cannot be unloaded, or "" when it can.""" state = self._mmu_state if state is None: return "MMU status not received yet." - if not state.enabled: - return "MMU is disabled." - if state.is_printing: - return "Not available while printing." + if reason := self._busy_refusal(state): + return reason if state.filament_pos == FilamentPos.UNLOADED: return "Filament is already unloaded." return "" @@ -245,7 +259,7 @@ def select_gate(self, gate: int) -> None: if ( state is None or not (state.enabled and state.is_homed) - or state.is_printing + or state.is_busy or not 0 <= gate < state.num_gates or gate == state.gate ): @@ -259,6 +273,9 @@ def select_gate(self, gate: int) -> None: def eject_gate(self) -> None: """Fully eject filament from the selected gate, the only gate that can eject.""" + if self._mmu_state is not None and self._mmu_state.is_printing: + logger.warning("Ignoring MMU_EJECT: not available while printing") + return gate = self._selected_gate_unloaded() if gate is None or gate.status == GateStatus.EMPTY: logger.warning("Ignoring MMU_EJECT: no unloaded, non-empty gate selected") @@ -267,9 +284,9 @@ def eject_gate(self) -> None: def check_gate(self) -> None: """Check the selected gate for filament by sending MMU_CHECK_GATE.""" - # Happy-Hare runs a full unload first when not UNLOADED. - if self._selected_gate_unloaded() is None: - logger.warning("Ignoring MMU_CHECK_GATE: no unloaded gate selected") + # Happy-Hare v3 unloads first when not UNLOADED and reloads a LOADED tool after. + if reason := self.check_refusal(): + logger.warning("Ignoring MMU_CHECK_GATE: %s", reason) return self.run_gcode_signal.emit("MMU_CHECK_GATE") diff --git a/BlocksScreen/devices/amu/models.py b/BlocksScreen/devices/amu/models.py index 68ee3982..33c6bdb3 100644 --- a/BlocksScreen/devices/amu/models.py +++ b/BlocksScreen/devices/amu/models.py @@ -166,6 +166,11 @@ def is_printing(self) -> bool: """Return True while a print is running; a paused print allows recovery.""" return self.print_state in ("started", "printing") + @property + def is_busy(self) -> bool: + """Return True while an action runs; "Unknown" is Happy-Hare's error state.""" + return self.action not in ("", "Idle", "Unknown") + @property def current_gate_info(self) -> GateInfo | None: """Return the GateInfo for the selected gate, or None when none is selected.""" diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/amuPage.py b/BlocksScreen/lib/panels/widgets/FilamentTab/amuPage.py index 2b52b56e..b650ddb8 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/amuPage.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/amuPage.py @@ -1,6 +1,8 @@ +import contextlib +import math import typing -from devices.amu import AMUManager +from devices.amu import AMUManager, FilamentPos, MMUState from lib.panels.widgets.FilamentTab.amuWidgets import SpoolCarousel, SpoolInfoPanel from lib.panels.widgets.Common.basePopup import BasePopup from lib.utils.blocks_frame import BlocksCustomFrame @@ -12,9 +14,6 @@ class AMUpage(QtWidgets.QStackedWidget): request_back: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( name="request_back" ) - request_gate_map: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( - str, name="request-gate-map" - ) request_numpad: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( [str, int, "PyQt_PyObject"], [str, int, "PyQt_PyObject", int, int], @@ -37,53 +36,25 @@ def __init__( self.current_index = -1 self.amu_manager: AMUManager = amu_manager self.load_popup = load_popup + self.status = None self._build_ui() self.main_back_button.clicked.connect(self.request_back) self.amu_manager.mmu_state_changed.connect(self.on_mmu_state_changed) self.on_mmu_state_changed(self.amu_manager.get_state()) - self.info_panel.colorSelected.connect( - lambda hx: self.amu_manager.set_gate_color(self.current_index, hx) - ) + self.info_panel.colorSelected.connect(self._on_color_selected) self.info_panel.colorSwatchClicked.connect( lambda hx: self.request_color_wheel.emit( hx, self.info_panel.set_selected_color ) ) - self.info_panel._lbl_mat.editingFinished.connect( - lambda: self.amu_manager.set_gate_material( - self.current_index, self.info_panel._lbl_mat.text().strip("º") - ) - ) - self.info_panel._lbl_temp.editingFinished.connect( - lambda: self.amu_manager.set_gate_temp( - self.current_index, - int(self.info_panel._lbl_temp.text().strip("º") or 0), - ) - ) - self.info_panel._lbl_temp.clicked.connect( - lambda: self.request_numpad[str, int, "PyQt_PyObject", int, int].emit( - "Temperature", - int(self.info_panel._lbl_temp.text().strip("º")), - self._on_gate_temp_change, - 0, - 500, - ) - ) + self.info_panel._lbl_mat.editingFinished.connect(self._on_material_edited) + self.info_panel._lbl_temp.editingFinished.connect(self._on_temp_edited) + self.info_panel._lbl_temp.clicked.connect(self._open_temp_numpad) self.info_panel.request_keypad.connect(self.request_keyboard) - self.info_panel.loadRequested.connect( - lambda: { - self.amu_manager.load_gate(), - self.load_popup.show(), - } - ) - self.info_panel.unloadRequested.connect( - lambda: { - self.amu_manager.unload(), - self.load_popup.show(), - } - ) + self.info_panel.loadRequested.connect(self._on_load) + self.info_panel.unloadRequested.connect(self._on_unload) self.info_panel.ejectRequested.connect(self.amu_manager.eject_gate) self.info_panel.checkRequested.connect(self.amu_manager.check_gate) @@ -93,59 +64,112 @@ def __init__( @QtCore.pyqtSlot(str, float, name="on_print_stats_update") @QtCore.pyqtSlot(str, str, name="on_print_stats_update") def on_print_stats_update(self, field: str, value: dict | float | str) -> None: - """Rewire the back button between "request_back" and "change to tab 0" based on print state.""" - if isinstance(value, str) and "state" in field: - self.state = value - if value in ("printing", "pausing", "paused", "resuming"): - try: - self.main_back_button.clicked.disconnect() - except TypeError: - pass - - self.main_back_button.clicked.connect( - lambda: self.request_change_tab.emit(0) - ) - - else: - try: - self.main_back_button.clicked.disconnect() - except TypeError: - pass - self.main_back_button.clicked.connect(lambda: self.request_back.emit()) + """Point the back button at tab 0 while printing, else at request_back.""" + if not (isinstance(value, str) and "state" in field): + return + with contextlib.suppress(TypeError): + self.main_back_button.clicked.disconnect() + if value in ("printing", "pausing", "paused", "resuming"): + self.main_back_button.clicked.connect( + lambda: self.request_change_tab.emit(0) + ) + else: + self.main_back_button.clicked.connect(self.request_back) def on_mmu_state_changed(self, mmu_state): - """Refresh the carousel and the info panel's selected gate from live MMU state.""" + """Sync the carousel and info panel to live MMU state.""" + prev = self.status + self.status = mmu_state if mmu_state is None: + self.current_index = -1 + self.carousel.selectIndex(-1) + self.info_panel.clear_slot(can_unload=False) return - self.status = mmu_state - for i in range(len(mmu_state.gates)): - self.carousel.addSpool(mmu_state.gates[i], mmu_state.filament_pos) - self.update() - self._on_selection(mmu_state.gate) + # apply_diff reuses the gates tuple when no gate_* key changed. + redraw = ( + prev is None + or mmu_state.gates is not prev.gates + or mmu_state.filament_pos != prev.filament_pos + ) + if not redraw and all( + getattr(mmu_state, f) == getattr(prev, f) + for f in ("gate", "enabled", "is_homed", "is_busy", "is_printing") + ): + return + if redraw: + for gate_info in mmu_state.gates: + self.carousel.addSpool(gate_info, mmu_state.filament_pos) + self._on_selection(mmu_state) def _select_gate(self, idx: int): self.carousel.selectIndex(self.current_index) self.amu_manager.select_gate(idx) - def _on_selection(self, idx: int): - if idx < 0 or idx >= len(self.carousel.buttons): + def _on_selection(self, mmu_state: MMUState) -> None: + idx = mmu_state.gate + idle = not mmu_state.is_busy + if not 0 <= idx < len(self.carousel.buttons): + self.current_index = -1 + self.carousel.selectIndex(-1) + self.info_panel.clear_slot( + can_unload=mmu_state.enabled + and idle + and mmu_state.filament_pos != FilamentPos.UNLOADED + ) return btn = self.carousel.buttons[idx] self.current_index = idx - self.info_panel.setFilamentStatus(self.status) - self.info_panel.update_for_slot(idx, btn) + self.info_panel.update_for_slot( + btn, + enabled=mmu_state.enabled, + homed=mmu_state.is_homed, + idle=idle, + printing=mmu_state.is_printing, + ) self.carousel.selectIndex(idx) + def _gate_temp(self) -> int | None: + try: + temp = float(self.info_panel._lbl_temp.text().strip("º")) + except ValueError: + return None + return round(temp) if math.isfinite(temp) else None + def _on_gate_temp_change(self, _name: str, value: int) -> None: self.info_panel._lbl_temp.setText(str(value)) self.info_panel._lbl_temp.editingFinished.emit() + def _open_temp_numpad(self) -> None: + self.request_numpad[str, int, "PyQt_PyObject", int, int].emit( + "Temperature", self._gate_temp() or 0, self._on_gate_temp_change, 1, 500 + ) + + def _on_temp_edited(self) -> None: + temp = self._gate_temp() + if temp is not None: + self.amu_manager.set_gate_temp(self.current_index, temp) + + def _on_material_edited(self) -> None: + material = self.info_panel._lbl_mat.text().strip() + if material != "—": + self.amu_manager.set_gate_material(self.current_index, material) + + def _on_color_selected(self, hex_str: str) -> None: + self.amu_manager.set_gate_color(self.current_index, hex_str) + + def _on_load(self) -> None: + if self.amu_manager.load_gate() and self.load_popup is not None: + self.load_popup.show() + + def _on_unload(self) -> None: + if self.amu_manager.unload() and self.load_popup is not None: + self.load_popup.show() + def _build_ui(self): self.setMinimumSize(700, 420) - self.setObjectName("fans_page") + self.setObjectName("amu_page") widget = QtWidgets.QWidget(parent=self) widget.setMinimumSize(700, 420) - self.setObjectName("temperature_page") self.setLayoutDirection(QtCore.Qt.LayoutDirection.LeftToRight) widget.setObjectName("filament_control_page") self.verticalLayout = QtWidgets.QVBoxLayout() diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/amuWidgets.py b/BlocksScreen/lib/panels/widgets/FilamentTab/amuWidgets.py index 90d09ef6..500d8836 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/amuWidgets.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/amuWidgets.py @@ -89,7 +89,7 @@ def _rebuild_geometry(self) -> None: self._strip_path.addRoundedRect(QtCore.QRectF(self._strip_rect), 5, 5) def update_entry(self, gate_info: GateInfo, fm: FilamentPos): - # color_rgb is Happy Hare's parsed form, so w3c names like "red" work too + # color_rgb is Happy-Hare's parsed form, so w3c names like "red" work too. has_color = bool(gate_info.color) color = ( QtGui.QColor.fromRgbF(*gate_info.color_rgb) @@ -140,10 +140,7 @@ def paintEvent(self, a0: QtGui.QPaintEvent) -> None: painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) painter.drawPath(self._strip_path) - if is_loaded: - icon = self._icons["loaded"] - else: - icon = self._icons["unloaded"] + icon = self._icons["loaded" if is_loaded else "unloaded"] x = (self.width() - icon.width()) // 2 y = int((self.height() - icon.height()) // 1.1) @@ -195,7 +192,7 @@ def _build_ui(self): self._update_arrows() def addSpool(self, gate_info: GateInfo, fm: FilamentPos): - """Adds or updates a spool button in the carousel""" + """Add or update the spool button for a gate.""" existing = self._by_id.get(gate_info.index) if existing is not None: existing.update_entry(gate_info, fm) @@ -245,26 +242,29 @@ def _update_arrows(self): self.right_arrow.setEnabled(self._offset + self.VISIBLE < len(self.buttons)) def selectedIndex(self) -> int: - """returns the index of the currently selected button, or -1 if none is selected""" + """Return the checked button's index, or -1.""" btn = self.button_group.checkedButton() if btn: return self.buttons.index(btn) return -1 def selectIndex(self, idx: int): - """select a button by index - - Args: - idx (int): the index of the button to select - """ - if 0 <= idx < len(self.buttons): - self.buttons[idx].setChecked(True) - if idx < self._offset: - self._offset = idx - self._refresh_visible() - elif idx >= self._offset + self.VISIBLE: - self._offset = idx - self.VISIBLE + 1 - self._refresh_visible() + """Check and scroll to *idx*; out of range unchecks all.""" + if not 0 <= idx < len(self.buttons): + checked = self.button_group.checkedButton() + if checked is not None: + # Exclusive groups refuse to uncheck their checked button. + self.button_group.setExclusive(False) + checked.setChecked(False) + self.button_group.setExclusive(True) + return + self.buttons[idx].setChecked(True) + if idx < self._offset: + self._offset = idx + self._refresh_visible() + elif idx >= self._offset + self.VISIBLE: + self._offset = idx - self.VISIBLE + 1 + self._refresh_visible() class SpoolInfoPanel(QtWidgets.QWidget): @@ -274,10 +274,10 @@ class SpoolInfoPanel(QtWidgets.QWidget): checkRequested: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal() request_keypad = QtCore.pyqtSignal( "PyQt_PyObject", str, str, str, int, name="request-keyboard" - ) # value, slot index, caller widget + ) colorSwatchClicked: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( str, name="color-swatch-clicked" - ) # current color hex + ) colorSelected: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( str, name="color-selected" ) @@ -293,9 +293,7 @@ class SpoolInfoPanel(QtWidgets.QWidget): def __init__(self, amu_manager, parent=None): super().__init__(parent) self.amu_manager = amu_manager - self._slot_index = -1 self._color_hex = "" - self.Gate = -1 self._build_ui() def _build_ui(self): @@ -317,7 +315,6 @@ def _build_ui(self): info_col.setContentsMargins(0, 0, 0, 0) info_col.setSpacing(12) - # header: gate name / status header = QtWidgets.QHBoxLayout() header.setSpacing(12) @@ -386,7 +383,6 @@ def _make_field( info_col.addLayout(fields) info_col.addStretch(1) - # fixed width so status text can never push the divider/buttons around info_widget = QtWidgets.QWidget() info_widget.setLayout(info_col) info_widget.setFixedWidth(300) @@ -417,12 +413,12 @@ def _make_btn(text: str, icon_path: str, signal) -> BlocksCustomButton: ":/filament_related/media/btn_icons/unload_filament.svg", self.unloadRequested, ) - self._btn_purge = _make_btn( + self._btn_eject = _make_btn( "Eject", ":/filament_related/media/btn_icons/eject.svg", self.ejectRequested, ) - self._btn_cut = _make_btn( + self._btn_check = _make_btn( "Check\nGate", ":/filament_related/media/btn_icons/check gate 1.svg", self.checkRequested, @@ -432,34 +428,34 @@ def _make_btn(text: str, icon_path: str, signal) -> BlocksCustomButton: btn_grid.setSpacing(8) btn_grid.addWidget(self._btn_load, 0, 0) btn_grid.addWidget(self._btn_unload, 0, 1) - btn_grid.addWidget(self._btn_purge, 1, 0) - btn_grid.addWidget(self._btn_cut, 1, 1) + btn_grid.addWidget(self._btn_eject, 1, 0) + btn_grid.addWidget(self._btn_check, 1, 1) root.addLayout(btn_grid) - def setFilamentStatus(self, mmu_state): - """Updates the active gate index from the mmu state.""" - self.Gate = mmu_state.gate - - @staticmethod - def _button_states_for_status(status: GateStatus): - """Base (load, unload, purge, cut) enabled-states from gate status alone.""" - match status: - case GateStatus.AVAILABLE | GateStatus.AVAILABLE_FROM_BUFFER: - return True, False, True, True - case GateStatus.EMPTY: - return False, False, True, True - case _: # GateStatus.UNKNOWN - return True, True, False, True - - def update_for_slot(self, index: int, btn: Spoll_button) -> None: - """Update the detail panel and action buttons for the selected gate slot. - - Args: - index: The index of the selected button/gate. - btn: The selected button, carrying a GateInfo snapshot and - FilamentPos for that gate. - """ - self._slot_index = index + def clear_slot(self, *, can_unload: bool) -> None: + """Show no gate selected; Happy-Hare's bare gate commands have no target.""" + self._lbl_gate.setText("Gate —") + self._lbl_status.setText(self._UNKNOWN_TEXT) + self._lbl_temp.setText("—") + self._lbl_mat.setText("—") + self._set_color("") + for w in (self._lbl_temp, self._swatch, self._lbl_mat): + w.setEnabled(False) + self._btn_load.setEnabled(False) + self._btn_unload.setEnabled(can_unload) + self._btn_eject.setEnabled(False) + self._btn_check.setEnabled(False) + + def update_for_slot( + self, + btn: Spoll_button, + *, + enabled: bool, + homed: bool, + idle: bool, + printing: bool, + ) -> None: + """Show the selected gate *btn*; bare HH gate commands act only on it.""" color = btn.color r, g, b = color.red(), color.green(), color.blue() @@ -467,21 +463,20 @@ def update_for_slot(self, index: int, btn: Spoll_button) -> None: gate_info = btn.gate_info status = gate_info.status if gate_info is not None else GateStatus.UNKNOWN - en_load, en_unload, en_purge, en_cut = self._button_states_for_status(status) text = self._STATUS_TEXT.get(status, self._UNKNOWN_TEXT) - is_active_gate = index == self.Gate filament_pos = btn.filament_pos - - # filament_pos is MMU-wide, so it only describes the active gate - if not is_active_gate or filament_pos == FilamentPos.UNLOADED: - en_unload = False - else: - en_unload, en_load, en_purge = True, False, False - if filament_pos == FilamentPos.LOADED: - text = self._LOADED_TEXT - else: - text = self._STUCK_TEXT + loaded = filament_pos != FilamentPos.UNLOADED + # Mirrors AMUManager: idle = no HH action; only Eject waits for the print. + ready = enabled and idle + can_check = ready and homed + can_feed = can_check and not loaded and status != GateStatus.EMPTY + if loaded: + text = ( + self._LOADED_TEXT + if filament_pos == FilamentPos.LOADED + else self._STUCK_TEXT + ) spool_id = gate_info.spool_id if gate_info is not None else -1 if spool_id != -1: @@ -490,8 +485,8 @@ def update_for_slot(self, index: int, btn: Spoll_button) -> None: f" · Spool ID {spool_id}" ) - # fields are read-only while the gate is bound to a spoolman spool - editable = gate_info is not None and spool_id == -1 + # Spoolman owns the fields of a bound gate. + editable = enabled and gate_info is not None and spool_id == -1 for w in (self._lbl_temp, self._swatch, self._lbl_mat): w.setEnabled(editable) @@ -510,11 +505,10 @@ def update_for_slot(self, index: int, btn: Spoll_button) -> None: color_hex = f"{r:02X}{g:02X}{b:02X}" if gate_info and gate_info.color else "" self._set_color(color_hex) - self._btn_load.setEnabled(en_load) - self._btn_unload.setEnabled(en_unload) - self._btn_purge.setEnabled(en_purge) - self._btn_cut.setEnabled(en_cut) - self.update() + self._btn_load.setEnabled(can_feed) + self._btn_unload.setEnabled(ready and loaded) + self._btn_eject.setEnabled(can_feed and not printing) + self._btn_check.setEnabled(can_check) def _apply_swatch(self, color: QtGui.QColor | None) -> None: if color is not None: @@ -535,11 +529,7 @@ def _apply_swatch(self, color: QtGui.QColor | None) -> None: ) def set_selected_color(self, hex_str: str) -> None: - """Show a color picked for the current gate and emit ``colorSelected``. - - Args: - hex_str: The picked color as a hex string, with or without ``#``. - """ + """Show the picked colour and emit colorSelected.""" self._set_color(hex_str) self.colorSelected.emit(self._color_hex) diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index ffa2f779..07ece5cc 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -416,8 +416,7 @@ def test_load_gate(self, manager, qtbot) -> None: {"filament_pos": 0, "gate": -1}, {"filament_pos": 0, "enabled": False}, {"filament_pos": 0, "is_homed": False}, - {"filament_pos": 0, "print_state": "printing"}, - {"filament_pos": 0, "print_state": "started"}, + {"filament_pos": 0, "action": "Checking"}, ], ) def test_load_gate_refused(self, manager, qtbot, overrides) -> None: @@ -440,7 +439,12 @@ def test_unload(self, manager, qtbot, eject, gcode) -> None: # Happy-Hare drops MMU_UNLOAD when disabled or already unloaded (mmu.py:6981). @pytest.mark.parametrize( "overrides", - [None, {"filament_pos": 0}, {"enabled": False}, {"print_state": "printing"}], + [ + None, + {"filament_pos": 0}, + {"enabled": False}, + {"action": "Loading"}, + ], ) def test_unload_refused(self, manager, qtbot, overrides) -> None: if overrides is not None: @@ -457,6 +461,13 @@ def test_unload_allowed_while_paused(self, manager, qtbot, print_state) -> None: assert manager.unload() is True assert blocker.args == ["MMU_UNLOAD"] + # "Unknown" is the action Happy-Hare reports after an error, not a running action. + def test_unload_allowed_after_an_action_error(self, manager, qtbot) -> None: + manager.update_mmu_state(_status(action="Unknown")) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + assert manager.unload() is True + assert blocker.args == ["MMU_UNLOAD"] + 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: @@ -471,7 +482,7 @@ def test_eject_gate(self, manager, qtbot) -> None: {"filament_pos": 10}, {"filament_pos": 0, "gate": -1}, {"filament_pos": 0, "gate_status": [0, 1]}, - {"filament_pos": 0, "print_state": "printing"}, + {"filament_pos": 0, "action": "Homing"}, ], ) def test_eject_gate_refused(self, manager, qtbot, overrides) -> None: @@ -486,20 +497,31 @@ def test_check_gate_allows_an_empty_gate(self, manager, qtbot) -> None: manager.check_gate() assert blocker.args == ["MMU_CHECK_GATE"] - # MMU_CHECK_GATE runs a full unload first when filament is loaded. + # HH v3 unloads first, then reloads a LOADED tool after the check (mmu.py:8897). + @pytest.mark.parametrize("filament_pos", [10, 3, -1]) + def test_check_gate_allowed_from_any_position( + self, manager, qtbot, filament_pos + ) -> None: + manager.update_mmu_state(_status(filament_pos=filament_pos)) + assert manager.check_refusal() == "" + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + manager.check_gate() + assert blocker.args == ["MMU_CHECK_GATE"] + @pytest.mark.parametrize( "overrides", [ None, - {"filament_pos": 10}, - {"filament_pos": 0, "gate": -2}, - {"filament_pos": 0, "is_homed": False}, - {"filament_pos": 0, "print_state": "printing"}, + {"gate": -2}, + {"enabled": False}, + {"is_homed": False}, + {"action": "Checking"}, ], ) def test_check_gate_refused(self, manager, qtbot, overrides) -> None: if overrides is not None: manager.update_mmu_state(_status(**overrides)) + assert manager.check_refusal() with qtbot.assertNotEmitted(manager.run_gcode_signal): manager.check_gate() @@ -520,7 +542,7 @@ def test_select_gate(self, manager, qtbot) -> None: ({"filament_pos": 10}, 1), ({"filament_pos": 0, "enabled": False}, 1), ({"filament_pos": 0, "is_homed": False}, 1), - ({"filament_pos": 0, "print_state": "printing"}, 1), + ({"filament_pos": 0, "action": "Selecting"}, 1), ], ) def test_select_gate_skipped(self, manager, qtbot, overrides, gate) -> None: @@ -529,6 +551,34 @@ def test_select_gate_skipped(self, manager, qtbot, overrides, gate) -> None: with qtbot.assertNotEmitted(manager.run_gcode_signal): manager.select_gate(gate) + # HH v3 parks and restores the toolhead around these; only MMU_PRELOAD checks + # for a print (mmu.py:9040), so the manager must not block them mid-print. + @pytest.mark.parametrize("print_state", ["printing", "started"]) + @pytest.mark.parametrize( + ("filament_pos", "send", "gcode"), + [ + (0, lambda m: m.load_gate(), "MMU_LOAD"), + (10, lambda m: m.unload(), "MMU_UNLOAD"), + (10, lambda m: m.check_gate(), "MMU_CHECK_GATE"), + (0, lambda m: m.select_gate(1), "MMU_SELECT GATE=1"), + ], + ) + def test_allowed_while_printing( + self, manager, qtbot, print_state, filament_pos, send, gcode + ) -> None: + manager.update_mmu_state( + _status(filament_pos=filament_pos, print_state=print_state) + ) + with qtbot.waitSignal(manager.run_gcode_signal) as blocker: + send(manager) + assert blocker.args == [gcode] + + @pytest.mark.parametrize("print_state", ["printing", "started"]) + def test_eject_refused_while_printing(self, manager, qtbot, print_state) -> None: + manager.update_mmu_state(_status(filament_pos=0, print_state=print_state)) + with qtbot.assertNotEmitted(manager.run_gcode_signal): + manager.eject_gate() + # TOOL, not GATE: the ttg map may not map tool N to gate N. def test_change_tool(self, manager, qtbot) -> None: with qtbot.waitSignal(manager.run_gcode_signal) as blocker: diff --git a/tests/widgets/test_amu_widgets_unit.py b/tests/widgets/test_amu_widgets_unit.py new file mode 100644 index 00000000..96ac26da --- /dev/null +++ b/tests/widgets/test_amu_widgets_unit.py @@ -0,0 +1,361 @@ +"""SpoolInfoPanel action-button matrix and AMUpage gate selection.""" + +import sys +from unittest.mock import MagicMock + +import pytest + +import devices.amu.manager as amu_manager_module +from devices.amu.models import FilamentPos, GateInfo, GateStatus, MMUState + +# conftest stubs devices.amu as a bare namespace; amuPage imports these from it. +_amu = vars(sys.modules["devices.amu"]) +_AMU_NAMES = { + "AMUManager": amu_manager_module.AMUManager, + "FilamentPos": FilamentPos, + "MMUState": MMUState, +} +_amu_saved = {n: _amu[n] for n in _AMU_NAMES if n in _amu} +_amu.update(_AMU_NAMES) +# Other test modules leave session-wide stand-ins for these; the panel needs the real API. +_STUBBED = ( + "lib.utils.icon_button", + "lib.utils.blocks_button", + "lib.utils.blocks_label", + "lib.utils.blocks_linedit", + "lib.panels.widgets.FilamentTab", +) +_saved = {m: sys.modules.pop(m) for m in list(sys.modules) if m.startswith(_STUBBED)} + +from lib.panels.widgets.FilamentTab.amuPage import AMUpage # noqa: E402 +from lib.panels.widgets.FilamentTab.amuWidgets import ( # noqa: E402 + SpoolInfoPanel, + Spoll_button, +) + +# Put the stand-ins back so this import leaves sys.modules and devices.amu as found. +for _real in [m for m in sys.modules if m.startswith(_STUBBED)]: + del sys.modules[_real] +sys.modules.update(_saved) +for _name in _AMU_NAMES: + _amu.pop(_name, None) +_amu.update(_amu_saved) + + +def _button( + index: int, + status: GateStatus = GateStatus.AVAILABLE, + pos: FilamentPos = FilamentPos.UNLOADED, +) -> Spoll_button: + btn = Spoll_button() + btn.update_entry(GateInfo(index=index, status=status, material="PLA"), pos) + return btn + + +def _show( + panel: SpoolInfoPanel, + btn: Spoll_button, + *, + enabled=True, + homed=True, + idle=True, + printing=False, +) -> None: + panel.update_for_slot( + btn, enabled=enabled, homed=homed, idle=idle, printing=printing + ) + + +def _actions(panel: SpoolInfoPanel) -> dict[str, bool]: + return { + "load": panel._btn_load.isEnabled(), + "unload": panel._btn_unload.isEnabled(), + "eject": panel._btn_eject.isEnabled(), + "check": panel._btn_check.isEnabled(), + } + + +@pytest.fixture +def panel(qtbot): + widget = SpoolInfoPanel(amu_manager=MagicMock()) + qtbot.addWidget(widget) + return widget + + +class TestSpoolInfoPanelActions: + def test_active_gate_unloaded_enables_feed_and_check(self, panel): + _show(panel, _button(1)) + assert _actions(panel) == { + "load": True, + "unload": False, + "eject": True, + "check": True, + } + + def test_active_gate_empty_allows_check_only(self, panel): + # Nothing to load or eject, but MMU_CHECK_GATE is how an EMPTY gate is re-detected. + _show(panel, _button(1, status=GateStatus.EMPTY)) + assert _actions(panel) == { + "load": False, + "unload": False, + "eject": False, + "check": True, + } + + # HH v3 MMU_CHECK_GATE unloads first and reloads a LOADED tool after. + def test_active_gate_loaded_allows_unload_and_check(self, panel): + _show(panel, _button(1, pos=FilamentPos.LOADED)) + assert _actions(panel) == { + "load": False, + "unload": True, + "eject": False, + "check": True, + } + assert panel._lbl_status.text() == panel._LOADED_TEXT + + def test_mid_path_filament_shows_stuck_and_allows_unload(self, panel): + _show(panel, _button(1, pos=FilamentPos.IN_BOWDEN)) + assert _actions(panel)["unload"] is True + assert _actions(panel)["check"] is True + assert panel._lbl_status.text() == panel._STUCK_TEXT + + def test_unknown_position_counts_as_loaded(self, panel): + # UNKNOWN may hide filament in the path, so load and eject stay off. + _show(panel, _button(1, pos=FilamentPos.UNKNOWN)) + assert _actions(panel) == { + "load": False, + "unload": True, + "eject": False, + "check": True, + } + + @pytest.mark.parametrize("pos", [FilamentPos.UNLOADED, FilamentPos.LOADED]) + def test_busy_greys_actions(self, panel, pos): + _show(panel, _button(1, pos=pos), idle=False) + assert not any(_actions(panel).values()) + + # HH v3 parks around LOAD/UNLOAD/CHECK_GATE mid-print; only Eject waits. + @pytest.mark.parametrize( + ("pos", "expected"), + [ + ( + FilamentPos.UNLOADED, + {"load": True, "unload": False, "eject": False, "check": True}, + ), + ( + FilamentPos.LOADED, + {"load": False, "unload": True, "eject": False, "check": True}, + ), + ], + ) + def test_printing_greys_only_eject(self, panel, pos, expected): + _show(panel, _button(1, pos=pos), printing=True) + assert _actions(panel) == expected + + def test_spool_bound_gate_is_read_only(self, panel): + btn = Spoll_button() + btn.update_entry( + GateInfo(index=1, status=GateStatus.AVAILABLE, spool_id=7), + FilamentPos.UNLOADED, + ) + _show(panel, btn) + assert not panel._lbl_mat.isEnabled() + assert "Spool ID 7" in panel._lbl_status.text() + + @pytest.mark.parametrize("pos", [FilamentPos.UNLOADED, FilamentPos.LOADED]) + def test_disabled_mmu_greys_everything(self, panel, pos): + _show(panel, _button(1, pos=pos), enabled=False) + assert not any(_actions(panel).values()) + assert not panel._lbl_mat.isEnabled() + + def test_unhomed_mmu_allows_unload_only(self, panel): + _show(panel, _button(1), homed=False) + assert not any(_actions(panel).values()) + _show(panel, _button(1, pos=FilamentPos.LOADED), homed=False) + assert _actions(panel)["unload"] is True + + +class TestClearSlot: + @pytest.mark.parametrize("can_unload", [True, False]) + def test_only_unload_follows_the_flag(self, panel, can_unload): + _show(panel, _button(1)) + panel.clear_slot(can_unload=can_unload) + assert _actions(panel) == { + "load": False, + "unload": can_unload, + "eject": False, + "check": False, + } + assert panel._lbl_gate.text() == "Gate —" + + +def _state( + gate: int, pos: FilamentPos = FilamentPos.UNLOADED, enabled: bool = True +) -> MMUState: + return MMUState.from_status( + { + "enabled": enabled, + "is_homed": True, + "num_gates": 4, + "gate": gate, + "tool": gate, + "filament_pos": int(pos), + "gate_status": [1, 1, 1, 1], + } + ) + + +@pytest.fixture +def manager(): + mgr = MagicMock() + mgr.get_state.return_value = None + return mgr + + +@pytest.fixture +def page(qtbot, manager): + widget = AMUpage(manager) + qtbot.addWidget(widget) + widget.on_mmu_state_changed(_state(gate=1)) + return widget + + +class TestAMUpageSelect: + def test_tap_selects_the_gate_immediately(self, page, manager): + page.carousel.selectionChanged.emit(2) + manager.select_gate.assert_called_once_with(2) + + def test_tap_keeps_the_active_gate_highlighted_until_state_confirms(self, page): + page.carousel.selectionChanged.emit(2) + assert page.carousel.selectedIndex() == 1 + + def test_state_update_follows_the_selected_gate(self, page): + page.on_mmu_state_changed(_state(gate=2)) + assert page.current_index == 2 + assert page.carousel.selectedIndex() == 2 + + def test_bypass_clears_the_panel(self, page): + # gate -2 is Happy-Hare's bypass: no gate is targeted by bare commands. + page.on_mmu_state_changed(_state(gate=-2, pos=FilamentPos.LOADED)) + assert page.current_index == -1 + assert page.carousel.selectedIndex() == -1 + assert _actions(page.info_panel)["unload"] is True + assert _actions(page.info_panel)["load"] is False + + def test_bypass_on_a_disabled_mmu_cannot_unload(self, page): + page.on_mmu_state_changed( + _state(gate=-2, pos=FilamentPos.LOADED, enabled=False) + ) + assert not any(_actions(page.info_panel).values()) + + def test_no_state_clears_the_panel(self, page): + page.on_mmu_state_changed(None) + assert page.current_index == -1 + assert page.carousel.selectedIndex() == -1 + assert not any(_actions(page.info_panel).values()) + + def test_tap_with_no_active_gate_leaves_nothing_highlighted(self, page): + page.on_mmu_state_changed(_state(gate=-1)) + page.carousel.buttons[2].click() + assert page.carousel.selectedIndex() == -1 + + +class TestAMUpageActions: + def test_load_shows_popup_when_command_sent(self, manager, qtbot): + popup = MagicMock() + widget = AMUpage(manager, load_popup=popup) + qtbot.addWidget(widget) + manager.load_gate.return_value = True + widget.info_panel.loadRequested.emit() + popup.show.assert_called_once() + + def test_refused_load_shows_no_popup(self, manager, qtbot): + popup = MagicMock() + widget = AMUpage(manager, load_popup=popup) + qtbot.addWidget(widget) + manager.load_gate.return_value = False + widget.info_panel.loadRequested.emit() + popup.show.assert_not_called() + + def test_unload_without_popup_does_not_crash(self, page, manager): + manager.unload.return_value = True + page.info_panel.unloadRequested.emit() + manager.unload.assert_called_once() + + def test_temp_edit_with_placeholder_sends_nothing(self, page, manager): + page.info_panel._lbl_temp.setText("—") + page._on_temp_edited() + manager.set_gate_temp.assert_not_called() + + def test_temp_edit_strips_degree_sign(self, page, manager): + page.info_panel._lbl_temp.setText("215º") + page._on_temp_edited() + manager.set_gate_temp.assert_called_once_with(1, 215) + + @pytest.mark.parametrize("text", ["inf", "nan", "1e999"]) + def test_non_finite_temp_sends_nothing(self, page, manager, text): + page.info_panel._lbl_temp.setText(text) + page._on_temp_edited() + manager.set_gate_temp.assert_not_called() + + @pytest.mark.parametrize( + ("state", "tab"), [("printing", [0]), ("paused", [0]), ("standby", [])] + ) + def test_back_button_follows_print_state(self, page, state, tab): + tabs, backs = [], [] + page.request_change_tab.connect(tabs.append) + page.request_back.connect(lambda: backs.append(1)) + page.on_print_stats_update("state", "printing") + page.on_print_stats_update("state", state) + page.main_back_button.click() + assert (tabs, backs) == (tab, [] if tab else [1]) + + +class TestAMUpageRedraw: + @staticmethod + def _spy(page, monkeypatch) -> list: + calls = [] + monkeypatch.setattr(page.carousel, "addSpool", lambda *a: calls.append(a)) + return calls + + # A diff with no gate_* key keeps the gates tuple, so nothing to repaint. + def test_scalar_diff_skips_carousel(self, page, monkeypatch): + calls = self._spy(page, monkeypatch) + page.on_mmu_state_changed(page.status.apply_diff({"action": "Loading"})) + assert calls == [] + + def test_gate_diff_redraws_carousel(self, page, monkeypatch): + calls = self._spy(page, monkeypatch) + page.on_mmu_state_changed(page.status.apply_diff({"gate_status": [0, 1, 1, 1]})) + assert len(calls) == 4 + + # The spool buttons render filament_pos, so a position change must repaint. + def test_filament_pos_change_redraws_carousel(self, page, monkeypatch): + calls = self._spy(page, monkeypatch) + diff = {"filament_pos": int(FilamentPos.LOADED)} + page.on_mmu_state_changed(page.status.apply_diff(diff)) + assert len(calls) == 4 + + def test_unchanged_view_skips_selection(self, page, monkeypatch): + calls = [] + monkeypatch.setattr(page, "_on_selection", calls.append) + page.on_mmu_state_changed(page.status.apply_diff({"bowden_progress": 42.0})) + assert calls == [] + + def test_busy_diff_greys_then_restores(self, page): + page.on_mmu_state_changed(page.status.apply_diff({"action": "Loading"})) + assert not any(_actions(page.info_panel).values()) + page.on_mmu_state_changed(page.status.apply_diff({"action": "Idle"})) + assert _actions(page.info_panel)["load"] is True + + def test_printing_diff_greys_eject_then_restores(self, page): + page.on_mmu_state_changed(page.status.apply_diff({"print_state": "printing"})) + actions = _actions(page.info_panel) + assert actions["eject"] is False + assert actions["load"] is True + page.on_mmu_state_changed(page.status.apply_diff({"print_state": "standby"})) + assert _actions(page.info_panel)["eject"] is True + + def test_selection_follows_a_scalar_diff(self, page): + page.on_mmu_state_changed(page.status.apply_diff({"gate": 2})) + assert page.current_index == 2