diff --git a/BlocksScreen/devices/amu/manager.py b/BlocksScreen/devices/amu/manager.py index 9b8b0e95..c10eb973 100644 --- a/BlocksScreen/devices/amu/manager.py +++ b/BlocksScreen/devices/amu/manager.py @@ -187,38 +187,56 @@ 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.""" + 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 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 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 state.filament_pos == FilamentPos.UNLOADED: + return "Filament is already unloaded." + return "" + def _selected_gate_unloaded(self) -> GateInfo | None: """Return the selected gate when the MMU is idle-ready: enabled, homed, unloaded.""" - state = self._mmu_state - if ( - state is None - or not (state.enabled and state.is_homed) - or state.is_printing - or state.filament_pos != FilamentPos.UNLOADED - ): + if self._mmu_state is None or self.load_refusal(): return None - return state.current_gate_info + return self._mmu_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 homed, unloaded gate selected") + if reason := self.load_refusal(): + logger.warning("Ignoring MMU_LOAD: %s", reason) 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.is_printing - or state.filament_pos == FilamentPos.UNLOADED - ): - logger.warning("Ignoring MMU_UNLOAD: MMU disabled, printing or unloaded") + def unload(self, eject: bool = False) -> bool: + """Unload via MMU_UNLOAD, or MMU_EJECT when *eject*; True if sent.""" + if reason := self.unload_refusal(): + logger.warning("Ignoring unload: %s", reason) return False - self.run_gcode_signal.emit("MMU_UNLOAD") + self.run_gcode_signal.emit("MMU_EJECT" if eject else "MMU_UNLOAD") return True def select_gate(self, gate: int) -> None: diff --git a/BlocksScreen/lib/panels/filamentTab.py b/BlocksScreen/lib/panels/filamentTab.py index e7120180..65d89a69 100644 --- a/BlocksScreen/lib/panels/filamentTab.py +++ b/BlocksScreen/lib/panels/filamentTab.py @@ -1,15 +1,15 @@ import logging import typing from collections import deque +from urllib.parse import urlencode from devices.amu import AMUManager -from devices.amu.models import GateStatus from lib.panels.widgets.Common.basePopup import BasePopup from lib.panels.widgets.Common.keyboardPage import CustomQwertyKeyboard from lib.panels.widgets.Common.loadWidget import LoadingOverlayWidget from lib.panels.widgets.Common.numpadPage import CustomNumpad from lib.panels.widgets.FilamentTab.addFilamentPage import AddFilamentPage -from lib.panels.widgets.FilamentTab.addSpoolPage import AddSpoolPage +from lib.panels.widgets.FilamentTab.addSpoolPage import AddSpoolPage, make_color_pixmap from lib.panels.widgets.FilamentTab.amuPage import AMUpage from lib.panels.widgets.FilamentTab.basicFilamentPanel import BasicFilamentPanel from lib.panels.widgets.FilamentTab.colorWheelWidget import ColorWheelWidget @@ -55,7 +55,7 @@ def __init__( self._previous_gate_states: dict[int, bool] = {} self.pre_gate_idx = {} self.popup_gates: deque = deque() - self._spool_id_map: dict[str, dict] = {} + self._spool_id_map: dict[int, dict] = {} self._current_field: QtWidgets.QLineEdit | None = None self._color_selected_callback: typing.Callable[[str], None] | None = None self._selected_spool: dict | None = None @@ -80,41 +80,55 @@ def __init__( self.spoolmanPanel.request_back.connect(self.request_back) self.spoolmanPanel.request_spools.connect( - lambda: self.ws.api.spoolman_proxy( - "GET", "/v1/spool", callback=self.spoolmanPanel.on_spools_received + lambda: self._spoolman( + self.spoolmanPanel.on_spools_received, + self.ws.api.spoolman_proxy, + "GET", + "/v1/spool", ) ) self.spoolmanPanel.request_get_spool_id.connect( lambda: self.ws.api.get_spool_id() ) self.spoolmanPanel.request_delete_spool.connect( - lambda spool_id: self.ws.api.delete_spool( - spool_id, callback=self.spoolmanPanel.on_delete_spool_result + lambda spool_id: self._spoolman( + self.spoolmanPanel.on_delete_spool_result, + self.ws.api.delete_spool, + spool_id, ) ) self.spoolmanPanel.request_filaments.connect( - lambda: self.ws.api.get_filaments( - callback=self.spoolmanPanel.on_filaments_received + lambda: self._spoolman( + self.spoolmanPanel.on_filaments_received, self.ws.api.get_filaments ) ) self.spoolmanPanel.request_add_spool.connect( - lambda filament_id, body: self.ws.api.add_spool( - filament_id, body, callback=self.spoolmanPanel.on_add_spool_result + lambda filament_id, body: self._spoolman( + self.spoolmanPanel.on_add_spool_result, + self.ws.api.add_spool, + filament_id, + body, ) ) self.spoolmanPanel.request_add_filament.connect( - lambda body: self.ws.api.add_filament( - body, callback=self.spoolmanPanel.on_add_filament_result + lambda body: self._spoolman( + self.spoolmanPanel.on_add_filament_result, + self.ws.api.add_filament, + body, ) ) self.spoolmanPanel.request_add_manufacturer.connect( - lambda body: self.ws.api.add_manufacturer( - body, callback=self.spoolmanPanel.on_add_manufacturer_result + lambda body: self._find_or_add_vendor( + body, self.spoolmanPanel.on_add_manufacturer_result ) ) self._basic_panel = BasicFilamentPanel( - self.printer, self.cfg, parent=self, load_popup=self.load_popup + self.printer, + self.cfg, + parent=self, + load_popup=self.load_popup, + amu_manager=self.amu_manager, ) self._basic_panel.run_gcode.connect(self.run_gcode) self._basic_panel.call_load_panel.connect(self.call_load_panel) @@ -133,12 +147,44 @@ def __init__( self.run_gcode.connect(self.ws.api.run_gcode) + def _spoolman(self, callback, call, *args) -> None: + """Run a Spoolman API call; a send that fails offline answers *callback* at once.""" + if not call(*args, callback=callback): + callback({"error": "not connected"}) + + def _find_or_add_vendor(self, body: dict, callback) -> None: + """Reuse the Spoolman vendor named ``body["name"]`` (any case); POST one only if none.""" + name = body["name"] + + def on_vendors(result: dict) -> None: + vendors = result.get("response") + if result.get("error") is not None or not isinstance(vendors, list): + callback({"error": result.get("error") or "bad vendor list"}) + return + # Spoolman's name filter is a case-insensitive partial match, so recheck here. + key = name.casefold() + match = next( + (v for v in vendors if (v.get("name") or "").casefold() == key), None + ) + if match is not None: + callback({"response": match, "error": None}) + else: + self._spoolman(callback, self.ws.api.add_manufacturer, body) + + self._spoolman( + on_vendors, + self.ws.api.spoolman_proxy, + "GET", + "/v1/vendor", + urlencode({"name": name}), + ) + def in_case_error(self): """Reset the popup and spool info in case of an error.""" self._reset_popup() self.reset_spool_info() self._add_spool_page.setFilter(None) - self._add_filament_page.setData("---", 0) + self._add_filament_page.reset() self.load_state = False self.load_popup.hide() @@ -479,42 +525,37 @@ def _build_spool_page(self) -> QtWidgets.QWidget: self._add_popup.add_widget(self._add_stack) self._add_spool_page.request_filaments.connect( - lambda: self.ws.api.get_filaments( - callback=self._add_spool_page.on_filaments_received + lambda: self._spoolman( + self._add_spool_page.on_filaments_received, self.ws.api.get_filaments ) ) self._add_spool_page.cancelled.connect(self._add_popup.hide) self._add_spool_page.request_add_spool.connect( - lambda filament_id, body: self.ws.api.add_spool( + lambda filament_id, body: self._spoolman( + self._add_spool_page.on_add_spool_result, + self.ws.api.add_spool, filament_id, body, - callback=self._add_spool_page.on_add_spool_result, ) ) - self._add_spool_page.accepted.connect( - lambda: { - self.ws.api.spoolman_proxy( - "GET", "/v1/spool", callback=self.on_spools_received - ), - self._add_popup.hide(), - } - ) + self._add_spool_page.accepted.connect(self._add_popup.hide) + self._add_spool_page.accepted.connect(self._request_spools) self._add_spool_page.open_add_filament.connect( lambda: self._add_stack.setCurrentIndex(1) ) self._add_filament_page.request_add_filament.connect( - lambda body: self.ws.api.add_filament( + lambda body: self._spoolman( + self._add_filament_page.on_add_filament_result, + self.ws.api.add_filament, body, - callback=self._add_filament_page.on_add_filament_result, ) ) self._add_filament_page.request_add_manufacturer.connect( - lambda body: self.ws.api.add_manufacturer( - body, - callback=self._add_filament_page.on_add_manufacturer_result, + lambda body: self._find_or_add_vendor( + body, self._add_filament_page.on_add_manufacturer_result ) ) @@ -595,36 +636,31 @@ def _add_row(title_text: str) -> QtWidgets.QLabel: return page def handle_skip_button(self): - """Handles the skip button action from the pre-gate popup to send the appropriate G-code to map the gate to no spool.""" + """Clear the popup gate's map entry so it holds no spool.""" gate = self.pre_gate_idx.get("gate", 0) self._reset_popup() - self.run_gcode.emit( - f"MMU_GATE_MAP GATE={gate} MATERIAL=N/A NAME=N/A COLOR=FFFFFF SPOOLID=-1 TEMP=250 QUIET=1" - ) + self.amu_manager.clear_gate(gate) + self._finish_popup() + + def _finish_popup(self) -> None: + """Run the pending callback, hide the popup and show the next queued gate.""" if self._popup_callback is not None: try: self._popup_callback() - except Exception as e: # noqa: BLE001 - arbitrary caller-supplied callback + except Exception as e: logger.error(f"Error executing pre-gate accept callback: {e}") finally: self._popup_callback = None - self.popup.hide() self._material_filter = None self._add_spool_page.setFilter(None) - self._add_filament_page.setData("---", 0) + self._add_filament_page.reset() self.handle_popup() def handle_popup(self, force=False): - """Handles showing the popup for pre-gate filament detection. - If multiple gates trigger, they will be queued and shown one at a time. - - Args: - force (bool): If True, forces the popup to open even if no gates are queued. - """ + """Show the next queued gate's popup; *force* opens it with an empty queue.""" if self.popup.isVisible(): return - self.ws.api.spoolman_proxy("GET", "/v1/spool", callback=self.on_spools_received) if not self.popup_gates: if force: self.pre_gate_idx = {"gate": 0} @@ -643,11 +679,13 @@ def handle_popup(self, force=False): self._popup_title_lbl.setText(f"Filament Detected — Gate {gate}") self._spoolman_title_lbl.setText(f"Filament Detected — Gate {gate}") + if self._spoolman_available: + self._request_spools() self.popup.show() @QtCore.pyqtSlot(int, str, str, "PyQt_PyObject", name="open-pregate-popup") def open_pregate_popup(self, temp, material, name, callback=None): - """Open the pre-gate popup pre-filled with a detected filament's info and *callback*.""" + """Open the pre-gate popup pre-filled with a detected filament and *callback*.""" self._popup_name.setText(name) self._popup_material.setText(material) self._popup_temp.setText(str(temp)) @@ -659,44 +697,22 @@ def open_pregate_popup(self, temp, material, name, callback=None): self.handle_popup(True) def on_popup_accept(self): - """Handles the accept action from the pre-gate popup to send the appropriate G-code to map the gate to the spool.""" + """Map the popup gate to the entered filament.""" gate = self.pre_gate_idx.get("gate", 0) - name = self._popup_name.text().strip() or "N/A" - color = self._popup_color.text().strip("#").strip() - material = self._popup_material.text().strip() try: temp = int(self._popup_temp.text().strip("°º").strip()) except ValueError: - temp = -1 - - parts = [f"MMU_GATE_MAP GATE={gate} SPOOLID=-1"] - if name: - parts.append(f'NAME="{name}"') - if material: - parts.append(f'MATERIAL="{material}"') - if color: - parts.append(f'COLOR="{color}"') - if temp > 0: - parts.append(f"TEMP={temp}") - parts.append("QUIET=1") - - self.run_gcode.emit(" ".join(parts)) - self.run_gcode.emit("MMU_GATE_MAP REFRESH=1") - - if self._popup_callback is not None: - try: - self._popup_callback() - except Exception as e: # noqa: BLE001 - arbitrary caller-supplied callback - logger.error(f"Error executing pre-gate accept callback: {e}") - finally: - self._popup_callback = None - + temp = 0 + self.amu_manager.set_gate_info( + gate, + self._popup_material.text().strip(), + self._popup_color.text().strip("#").strip(), + -1, + filament_name=self._popup_name.text().strip(), + temperature=temp if temp > 0 else None, + ) self._reset_popup() - self.popup.hide() - self._material_filter = None - self._add_spool_page.setFilter(None) - self._add_filament_page.setData("---", 0) - self.handle_popup() + self._finish_popup() def _reset_popup(self): self._popup_name.setText("") @@ -704,22 +720,35 @@ def _reset_popup(self): self._popup_material.setText("") self._popup_temp.setText("250") + def _clear_spool_list(self) -> None: + """Drop every spool row so no stale one can be mapped to a gate.""" + self._spool_id_map = {} + self._spool_model.clear() + self._spool_delegate.clear() + self.reset_spool_info() + + def _request_spools(self) -> None: + """Swap the spool list for the spinner and fetch the spools from Spoolman.""" + self._clear_spool_list() + self._spool_list_view.hide() + self._no_spools_label.hide() + self._spool_load_widget.show() + self._spoolman( + self.on_spools_received, self.ws.api.spoolman_proxy, "GET", "/v1/spool" + ) + @QtCore.pyqtSlot(dict, name="on-spools-received") def on_spools_received(self, result: dict) -> None: self._spool_load_widget.hide() - self._spool_list_view.show() - self.reset_spool_info() - if result.get("error") is not None: + self._clear_spool_list() + spools = result.get("response") + if result.get("error") is not None or not isinstance(spools, list): + self._spool_list_view.hide() self._no_spools_label.setText("Could not reach Spoolman") self._no_spools_label.show() return - spools = result.get("response") - if not isinstance(spools, list): - return - self._spool_id_map = {} - self._spool_model.clear() - self._spool_delegate.clear() + self._spool_list_view.show() self._spool_model.add_item( ListItem( text="+ Add Spool", @@ -738,17 +767,16 @@ def on_spools_received(self, result: dict) -> None: ): continue - self._spool_model.add_item( - ListItem( - text=name, - right_text=material, - left_icon=self._make_color_pixmap(filament), - _lfontsize=14, - _rfontsize=12, - height=60, - ) + item = ListItem( + text=name, + right_text=material, + left_icon=make_color_pixmap(filament), + _lfontsize=14, + _rfontsize=12, + height=60, ) - self._spool_id_map[name] = spool + self._spool_model.add_item(item) + self._spool_id_map[id(item)] = spool self.update() if self._spool_id_map: @@ -767,9 +795,10 @@ def _on_list_item_tapped(self, item: ListItem) -> None: return if item.text == "+ Add Spool": self.reset_spool_info() + self._add_spool_page.reset() self._add_popup.show() return - spool = self._spool_id_map.get(item.text) + spool = self._spool_id_map.get(id(item)) if spool is None: return self._selected_spool = spool @@ -777,47 +806,20 @@ def _on_list_item_tapped(self, item: ListItem) -> None: filament = spool.get("filament") or {} self.filament_name_label.setText(item.text) self.material_label.setText(filament.get("material", "N/A")) + remaining = spool.get("remaining_weight") self.weight_label.setText( - f"{spool.get('remaining_weight')} g" - if spool.get("remaining_weight") is not None - else "N/A" - ) - self.vendor_label.setText( - filament.get("vendor", "N/A").get("name", "N/A") - if filament.get("vendor") - else "N/A" + f"{remaining:.0f} g" if remaining is not None else "N/A" ) + self.vendor_label.setText((filament.get("vendor") or {}).get("name", "N/A")) def _on_accept_clicked(self) -> None: spool = self._selected_spool if not spool: return - filament = spool.get("filament") or {} - f_id = spool.get("id", -1) - f_name = filament.get("name", "N/A") - f_color = filament.get("color_hex", "ffffff") - f_material = filament.get("material", "") - f_temp = filament.get("settings_extruder_temp", -1) - gate = self.pre_gate_idx.get("gate", 0) - self._selected_spool = None self.accept_btn.setEnabled(False) - self.popup.hide() - self._material_filter = None - self._add_spool_page.setFilter(None) - self._add_filament_page.setData("---", 0) - self.run_gcode.emit( - f"MMU_GATE_MAP GATE={gate} SPOOLID={f_id} NAME='{f_name}' MATERIAL='{f_material}' COLOR='{f_color}' TEMP={f_temp} QUIET=1" - ) - if self._popup_callback is not None: - try: - self._popup_callback() - except Exception as e: # noqa: BLE001 - arbitrary caller-supplied callback - logger.error(f"Error executing pre-gate accept callback: {e}") - finally: - self._popup_callback = None - - self.handle_popup() + self.amu_manager.assign_spool(self.pre_gate_idx.get("gate", 0), spool) + self._finish_popup() def reset_spool_info(self): """Clear the selected-spool detail labels back to their placeholder state.""" @@ -828,38 +830,6 @@ def reset_spool_info(self): self._selected_spool = None self.accept_btn.setEnabled(False) - @staticmethod - def _make_color_pixmap(filament: dict) -> QtGui.QPixmap: - size = 32 - pixmap = QtGui.QPixmap(size, size) - pixmap.fill(QtCore.Qt.GlobalColor.transparent) - painter = QtGui.QPainter(pixmap) - painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) - multi_hexes = filament.get("multi_color_hexes") - color_hex = filament.get("color_hex") - if multi_hexes: - hexes = [h.strip() for h in multi_hexes.split(",") if h.strip()] - if hexes: - clip = QtGui.QPainterPath() - clip.addRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) - painter.setClipPath(clip) - stripe_w = size / len(hexes) - for i, h in enumerate(hexes): - painter.fillRect( - QtCore.QRectF(i * stripe_w, 0, stripe_w, size), - QtGui.QColor(f"#{h}"), - ) - elif color_hex: - painter.setBrush(QtGui.QColor(f"#{color_hex}")) - painter.setPen(QtCore.Qt.PenStyle.NoPen) - painter.drawRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) - else: - painter.setPen(QtGui.QColor(180, 180, 180)) - painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) - painter.drawRoundedRect(QtCore.QRectF(1, 1, size - 2, size - 2), 6, 6) - painter.end() - return pixmap - def _open_numpad( self, name: str, @@ -927,33 +897,17 @@ def _on_color_selected(self, hex_str: str) -> None: if callback is not None: callback(hex_str) - def _clear_gate_map(self, gate_info) -> None: - """Blank a gate's map entry when its filament runs out.""" - if gate_info.spool_id == -1 and gate_info.material in (None, "", "N/A"): - return # already blank - - self.run_gcode.emit( - f"MMU_GATE_MAP GATE={gate_info.index} MATERIAL= TEMP=-1 COLOR= SPOOLID=-1 NAME= QUIET=1" - ) - def on_mmu_state_changed(self, mmu_state): - """Handles changes in the MMU state from the AMU manager to update the UI and show the load panel when loading/unloading.""" + """Track gate fills/runouts and drive the AMU page and load popup.""" if mmu_state is None: return for gate_info in mmu_state.gates: previous_state = self._previous_gate_states.get(gate_info.index) - current_state = gate_info.status in [ - GateStatus.AVAILABLE, - GateStatus.AVAILABLE_FROM_BUFFER, - ] + current_state = gate_info.is_available self._previous_gate_states[gate_info.index] = current_state - - if previous_state is False and current_state is True: - if len(mmu_state.gates) > 1: - self.popup_gates.append({"gate": gate_info.index}) - self.handle_popup() - elif previous_state is True and gate_info.status == GateStatus.EMPTY: - self._clear_gate_map(gate_info) + if previous_state is False and current_state and len(mmu_state.gates) > 1: + self.popup_gates.append({"gate": gate_info.index}) + self.handle_popup() if not self.amu_configured: if len(mmu_state.gates) > 1: @@ -962,10 +916,12 @@ def on_mmu_state_changed(self, mmu_state): self.amu_manager, parent=self, load_popup=self.load_popup ) self.addWidget(self.amupage) - try: + if self._basic_panel is not None: self.removeWidget(self._basic_panel) self._basic_panel.deleteLater() - self.fp_button_1.disconnect() + self._basic_panel = None + try: + self.fp_button_1.clicked.disconnect() except TypeError: pass self.fp_button_1.clicked.connect( @@ -973,7 +929,6 @@ def on_mmu_state_changed(self, mmu_state): ) self.amupage.request_back.connect(self.request_back) self.amupage.request_change_tab.connect(self.request_change_tab) - self.amupage.request_gate_map.connect(self.run_gcode) self.amupage.request_numpad[ str, int, "PyQt_PyObject", int, int ].connect(self._open_numpad) @@ -997,7 +952,7 @@ def on_mmu_state_changed(self, mmu_state): if mmu_state.action == "Idle": self.load_state = False self.load_popup.hide() - if not len(mmu_state.gates) > 1: + if self._basic_panel is not None: self._basic_panel.change_page(0) elif mmu_state.action in ("Loading", "Unloading"): self.load_state = True @@ -1016,16 +971,11 @@ def on_mmu_state_changed(self, mmu_state): self.load_status_widget.set_action(mmu_state.action) gate_info = mmu_state.current_gate_info - raw_color = ( - str(gate_info.color).lstrip("#")[:6] + self.load_status_widget.set_gate_color( + QtGui.QColor.fromRgbF(*gate_info.color_rgb) if gate_info and gate_info.color - else "" + else QtGui.QColor("#ffffff") ) - parsed_color = QtGui.QColor("#" + raw_color) if raw_color else QtGui.QColor() - if parsed_color.isValid(): - self.load_status_widget.set_gate_color(parsed_color) - else: - self.load_status_widget.set_gate_color(QtGui.QColor("#ffffff")) self.load_status_label.setText(mmu_state.action) diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/addFilamentPage.py b/BlocksScreen/lib/panels/widgets/FilamentTab/addFilamentPage.py index 7f530c8c..3472327f 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/addFilamentPage.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/addFilamentPage.py @@ -1,6 +1,7 @@ import typing from lib.panels.widgets.Common.basePopup import BasePopup +from lib.panels.widgets.Common.popupDialogWidget import Popup from lib.panels.widgets.FilamentTab.colorWheelWidget import ColorWheelWidget from lib.panels.widgets.Common.keyboardPage import CustomQwertyKeyboard from lib.panels.widgets.Common.numpadPage import CustomNumpad @@ -33,6 +34,7 @@ def __init__(self, parent=None) -> None: self._color_hex = "ffffff" self._build_ui() + self._popup = Popup(self) self._keyboard = CustomQwertyKeyboard(self) self._keyboard.hide() @@ -68,6 +70,9 @@ def setData(self, material: str, extruder_temp: int): self.update() def reset(self): + """Restore the form defaults and drop any pending submit.""" + self.request_filament_body = None + self._submit_btn.setEnabled(True) self._name_field.setText("") self._material_field.setText("PLA") self._set_color("ffffff") @@ -78,11 +83,13 @@ def reset(self): @QtCore.pyqtSlot(dict, name="on-add-filament-result") def on_add_filament_result(self, result: dict) -> None: + self._submit_btn.setEnabled(True) if result.get("error") is None: self.accepted.emit() else: - ... - # TODO: some responsiveness here + self._popup.new_message( + message_type=Popup.MessageType.ERROR, message="Could not add filament" + ) def _show_keyboard( self, @@ -195,6 +202,7 @@ def _on_submit(self) -> None: body["density"] = 1.24 body["diameter"] = 1.75 # Default diameter, can be modified later if needed + self._submit_btn.setEnabled(False) if vendor_text != "---": self.request_add_manufacturer.emit({"name": vendor_text}) self.request_filament_body = body @@ -211,7 +219,11 @@ def on_add_manufacturer_result(self, result: dict) -> None: body["vendor_id"] = vendor_id self.request_add_filament.emit(body) else: - ... + self.request_filament_body = None + self._submit_btn.setEnabled(True) + self._popup.new_message( + message_type=Popup.MessageType.ERROR, message="Could not add vendor" + ) def _build_ui(self) -> None: def _f(pt: int) -> QtGui.QFont: @@ -387,12 +399,12 @@ def _make_field(parent: QtWidgets.QWidget) -> BlocksCustomLinEdit: ) root.addItem(spacer) # Submit - submit_btn = BlocksCustomButton(self) - submit_btn.setFixedSize(400, 80) - submit_btn.setFont(_f(16)) - submit_btn.setText("New Filament") - submit_btn.setPixmap( + self._submit_btn = BlocksCustomButton(self) + self._submit_btn.setFixedSize(400, 80) + self._submit_btn.setFont(_f(16)) + self._submit_btn.setText("New Filament") + self._submit_btn.setPixmap( QtGui.QPixmap(":/filament_related/media/btn_icons/add filament.svg") ) - submit_btn.clicked.connect(self._on_submit) - root.addWidget(submit_btn, alignment=QtCore.Qt.AlignmentFlag.AlignCenter) + self._submit_btn.clicked.connect(self._on_submit) + root.addWidget(self._submit_btn, alignment=QtCore.Qt.AlignmentFlag.AlignCenter) diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/addSpoolPage.py b/BlocksScreen/lib/panels/widgets/FilamentTab/addSpoolPage.py index 7a9e3296..83335df6 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/addSpoolPage.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/addSpoolPage.py @@ -4,6 +4,7 @@ from lib.panels.widgets.Common.keyboardPage import CustomQwertyKeyboard from lib.panels.widgets.Common.loadWidget import LoadingOverlayWidget from lib.panels.widgets.Common.numpadPage import CustomNumpad +from lib.panels.widgets.Common.popupDialogWidget import Popup from lib.utils.blocks_button import BlocksCustomButton from lib.utils.blocks_frame import BlocksCustomFrame from lib.utils.blocks_linedit import BlocksCustomLinEdit @@ -12,6 +13,61 @@ from PyQt6 import QtCore, QtGui, QtWidgets +def spoolman_color(hex_text: str) -> QtGui.QColor: + """Spoolman hex is rrggbb or rrggbbaa; QColor reads 8 digits as aarrggbb.""" + h = hex_text.strip().lstrip("#") + return QtGui.QColor(f"#{h[6:]}{h[:6]}") + + +def make_add_pixmap() -> QtGui.QPixmap: + """Return the 32px "+" icon of the "New ..." list rows.""" + pixmap = QtGui.QPixmap(32, 32) + pixmap.fill(QtCore.Qt.GlobalColor.transparent) + painter = QtGui.QPainter(pixmap) + painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) + pen = QtGui.QPen(QtGui.QColor(160, 160, 160)) + pen.setWidth(2) + painter.setPen(pen) + painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) + painter.drawRoundedRect(QtCore.QRectF(1, 1, 30, 30), 6, 6) + painter.drawLine(16, 8, 16, 24) + painter.drawLine(8, 16, 24, 16) + painter.end() + return pixmap + + +def make_color_pixmap(filament: dict) -> QtGui.QPixmap: + """Return a 32px swatch of a Spoolman filament's (multi) colour.""" + size = 32 + pixmap = QtGui.QPixmap(size, size) + pixmap.fill(QtCore.Qt.GlobalColor.transparent) + painter = QtGui.QPainter(pixmap) + painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) + multi_hexes: str | None = filament.get("multi_color_hexes") + color_hex: str | None = filament.get("color_hex") + if multi_hexes: + hexes = [h.strip() for h in multi_hexes.split(",") if h.strip()] + if hexes: + clip = QtGui.QPainterPath() + clip.addRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) + painter.setClipPath(clip) + stripe_w = size / len(hexes) + for i, h in enumerate(hexes): + painter.fillRect( + QtCore.QRectF(i * stripe_w, 0, stripe_w, size), spoolman_color(h) + ) + elif color_hex: + painter.setBrush(spoolman_color(color_hex)) + painter.setPen(QtCore.Qt.PenStyle.NoPen) + painter.drawRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) + else: + painter.setPen(QtGui.QColor(180, 180, 180)) + painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) + painter.drawRoundedRect(QtCore.QRectF(1, 1, size - 2, size - 2), 6, 6) + painter.end() + return pixmap + + class AddSpoolPage(QtWidgets.QWidget): request_filaments: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( name="request-filaments" @@ -33,11 +89,12 @@ class AddSpoolPage(QtWidgets.QWidget): def __init__(self, parent=None) -> None: super().__init__(parent) self._selected_filament_id: int | None = None - self._filament_id_map: dict[str, dict] = {} + self._filament_id_map: dict[int, dict] = {} self._keyboard_field: QtWidgets.QLineEdit | None = None self._material_filter: str | None = None self._build_ui() + self._popup = Popup(self) self._keyboard = CustomQwertyKeyboard(self) self._keyboard.hide() @@ -73,26 +130,27 @@ def reset(self) -> None: self.request_filaments.emit() def setFilter(self, material: str | None = None) -> None: - """Set a material filter and reload the filament list.""" + """Set the material filter applied on the next reset.""" self._material_filter = material - self.reset() @QtCore.pyqtSlot(dict, name="on-filaments-received") def on_filaments_received(self, result: dict) -> None: self._fil_load_widget.hide() - self._fil_list_view.show() - if result.get("error") is not None: - return - filaments = result.get("response") - if not isinstance(filaments, list): - return self._filament_id_map = {} self._fil_model.clear() self._fil_delegate.clear() + filaments = result.get("response") + if result.get("error") is not None or not isinstance(filaments, list): + self._fil_list_view.hide() + self._no_filaments_label.setText("Could not reach Spoolman") + self._no_filaments_label.show() + return + + self._fil_list_view.show() self._fil_model.add_item( ListItem( text=" New Filament", - left_icon=self._make_add_pixmap(), + left_icon=make_add_pixmap(), _lfontsize=14, height=60, ) @@ -107,18 +165,17 @@ def on_filaments_received(self, result: dict) -> None: and self._material_filter.lower() not in material.lower() ): continue - self._fil_model.add_item( - ListItem( - text=name, - right_text=material, - left_icon=self._make_color_pixmap(fil), - _lfontsize=14, - _rfontsize=12, - height=60, - ) + item = ListItem( + text=name, + right_text=material, + left_icon=make_color_pixmap(fil), + _lfontsize=14, + _rfontsize=12, + height=60, ) - self._filament_id_map[name] = fil - self.update() + self._fil_model.add_item(item) + self._filament_id_map[id(item)] = fil + self.update() if self._filament_id_map: self._no_filaments_label.hide() @@ -130,27 +187,15 @@ def on_filaments_received(self, result: dict) -> None: ) self._no_filaments_label.show() - @staticmethod - def _make_add_pixmap() -> QtGui.QPixmap: - pixmap = QtGui.QPixmap(32, 32) - pixmap.fill(QtCore.Qt.GlobalColor.transparent) - painter = QtGui.QPainter(pixmap) - painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) - pen = QtGui.QPen(QtGui.QColor(160, 160, 160)) - pen.setWidth(2) - painter.setPen(pen) - painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) - painter.drawRoundedRect(QtCore.QRectF(1, 1, 32 - 2, 32 - 2), 6, 6) - mid = 32 // 2 - painter.drawLine(mid, 8, mid, 32 - 8) - painter.drawLine(8, mid, 32 - 8, mid) - painter.end() - return pixmap - @QtCore.pyqtSlot(dict, name="on-add-spool-result") def on_add_spool_result(self, result: dict) -> None: + self._submit_btn.setEnabled(self._selected_filament_id is not None) if result.get("error") is None: self.accepted.emit() + else: + self._popup.new_message( + message_type=Popup.MessageType.ERROR, message="Could not add spool" + ) @QtCore.pyqtSlot(ListItem, name="on-filament-selected") def _on_filament_selected(self, item: ListItem) -> None: @@ -159,7 +204,7 @@ def _on_filament_selected(self, item: ListItem) -> None: if item.text == " New Filament": self.open_add_filament.emit() return - filament = self._filament_id_map.get(item.text) + filament = self._filament_id_map.get(id(item)) if filament: self._selected_filament_id = filament.get("id") self._lot_field.setEnabled(True) @@ -232,40 +277,9 @@ def _on_submit(self) -> None: body["initial_weight"] = float(weight) except ValueError: pass + self._submit_btn.setEnabled(False) self.request_add_spool.emit(int(self._selected_filament_id), body) - @staticmethod - def _make_color_pixmap(filament: dict) -> QtGui.QPixmap: - size = 32 - pixmap = QtGui.QPixmap(size, size) - pixmap.fill(QtCore.Qt.GlobalColor.transparent) - painter = QtGui.QPainter(pixmap) - painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) - multi_hexes: str | None = filament.get("multi_color_hexes") - color_hex: str | None = filament.get("color_hex") - if multi_hexes: - hexes = [h.strip() for h in multi_hexes.split(",") if h.strip()] - if hexes: - clip = QtGui.QPainterPath() - clip.addRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) - painter.setClipPath(clip) - stripe_w = size / len(hexes) - for i, h in enumerate(hexes): - painter.fillRect( - QtCore.QRectF(i * stripe_w, 0, stripe_w, size), - QtGui.QColor(f"#{h}"), - ) - elif color_hex: - painter.setBrush(QtGui.QColor(f"#{color_hex}")) - painter.setPen(QtCore.Qt.PenStyle.NoPen) - painter.drawRoundedRect(QtCore.QRectF(0, 0, size, size), 6, 6) - else: - painter.setPen(QtGui.QColor(180, 180, 180)) - painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) - painter.drawRoundedRect(QtCore.QRectF(1, 1, size - 2, size - 2), 6, 6) - painter.end() - return pixmap - def _build_ui(self) -> None: self.setMaximumHeight(470) diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/basicFilamentPanel.py b/BlocksScreen/lib/panels/widgets/FilamentTab/basicFilamentPanel.py index 94044239..9d88046b 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/basicFilamentPanel.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/basicFilamentPanel.py @@ -1,3 +1,4 @@ +import contextlib import enum import logging from functools import partial @@ -44,7 +45,13 @@ def __repr__(self) -> str: return f"<{self.__class__.__name__}.{self._name_}>" def __init__( - self, printer: Printer, cfg, parent=None, *, load_popup: BasePopup | None = None + self, + printer: Printer, + cfg, + parent=None, + *, + amu_manager, + load_popup: BasePopup | None = None, ) -> None: super().__init__(parent) self.printer = printer @@ -57,6 +64,7 @@ def __init__( self.filament_buttons_list = [] self.mmu_configured = False self.load_popup = load_popup + self.amu_manager = amu_manager self._mmu_state = None self._setup_ui() self.filament_state = self.FilamentStates.UNKNOWN @@ -115,10 +123,9 @@ def on_save_variables_update(self, save_variables: dict): save_variables (dict): dictionary containing the save variables from the printer """ filament_type = FilamentTypes.UNKNOWN - variables = save_variables.get("variables", {}) - filament_type = variables.get("filament_type", "") + saved = save_variables.get("variables", {}).get("filament_type", "") for i in FilamentTypes: - if filament_type and i.value.name in filament_type: + if isinstance(saved, str) and i.value.name in saved: filament_type = i break self._lbl_mat.set_right_text(filament_type.value.name) @@ -132,22 +139,14 @@ def on_print_stats_update(self, field: str, value: dict | float | str) -> None: self.state = value self._refresh_check_btn() self._refresh_pos_editable() + with contextlib.suppress(TypeError): + self.main_back_button.disconnect() if value in ("printing", "paused"): - try: - self.main_back_button.disconnect() - except TypeError: - pass - self.main_back_button.clicked.connect( lambda: self.request_change_tab.emit(0) ) - else: - try: - self.main_back_button.disconnect() - except TypeError: - pass - self.main_back_button.clicked.connect(lambda: self.request_back.emit()) + self.main_back_button.clicked.connect(self.request_back) @QtCore.pyqtSlot(str, str, bool, name="on_filament_sensor_update") def on_filament_sensor_update(self, sensor_name: str, parameter: str, value: bool): @@ -211,6 +210,12 @@ def load_filament( message="Filament is already loaded.", ) return + if self.mmu_configured and not self.amu_manager.load_gate(): + self.popup.new_message( + message_type=Popup.MessageType.ERROR, + message=self.amu_manager.load_refusal(), + ) + return if not self._in_print: self.run_gcode.emit( f"""SAVE_VARIABLE VARIABLE=filament_type VALUE='"{filament.value.name}"'""" @@ -218,9 +223,8 @@ def load_filament( if not self.mmu_configured: self.call_load_panel.emit(True, "Loading", True) self.run_gcode.emit("LOAD_FILAMENT") - return - self.load_popup.show() - self.run_gcode.emit("MMU_LOAD") + elif self.load_popup is not None: + self.load_popup.show() @QtCore.pyqtSlot(str, int, name="unload_filament") def unload_filament(self, toolhead: int = 0, temp: int = 220) -> None: @@ -239,17 +243,22 @@ def unload_filament(self, toolhead: int = 0, temp: int = 220) -> None: ) return self.find_routine_objects() + # Basic panel uses MMU_EJECT; when loaded, Happy-Hare v3 runs it as MMU_UNLOAD. + if self.mmu_configured and not self.amu_manager.unload(eject=True): + self.popup.new_message( + message_type=Popup.MessageType.ERROR, + message=self.amu_manager.unload_refusal(), + ) + return if not self._in_print: self.run_gcode.emit( f"""SAVE_VARIABLE VARIABLE=filament_type VALUE='"{FilamentTypes.UNKNOWN.value.name}"'""" ) - if not self.mmu_configured: self.call_load_panel.emit(True, "Unloading", True) self.run_gcode.emit("UNLOAD_FILAMENT") - return - self.load_popup.show() - self.run_gcode.emit("MMU_EJECT") + elif self.load_popup is not None: + self.load_popup.show() def _on_load_clicked(self) -> None: if self._in_print: @@ -347,7 +356,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() @@ -385,8 +399,7 @@ def on_pos_change(self) -> None: def _accept_pos_change(self, new_pos: str) -> None: self._pos_committed = new_pos - n = 1 if new_pos == "Loaded" else 0 - self.run_gcode.emit(f"MMU_RECOVER TOOL=0 GATE=0 LOADED={n}") + self.amu_manager.recover_single_gate(new_pos == "Loaded") def _reject_pos_change(self) -> None: self._lbl_pos.select_option(self._pos_committed) diff --git a/BlocksScreen/lib/panels/widgets/FilamentTab/spoolmanPage.py b/BlocksScreen/lib/panels/widgets/FilamentTab/spoolmanPage.py index 42984411..5d005abc 100644 --- a/BlocksScreen/lib/panels/widgets/FilamentTab/spoolmanPage.py +++ b/BlocksScreen/lib/panels/widgets/FilamentTab/spoolmanPage.py @@ -1,15 +1,23 @@ -import logging import typing from lib.panels.widgets.FilamentTab.addFilamentPage import AddFilamentPage -from lib.panels.widgets.FilamentTab.addSpoolPage import AddSpoolPage +from lib.panels.widgets.FilamentTab.addSpoolPage import ( + AddSpoolPage, + make_add_pixmap, + make_color_pixmap, + spoolman_color, +) from lib.panels.widgets.Common.basePopup import BasePopup +from lib.panels.widgets.Common.popupDialogWidget import Popup from lib.utils.blocks_frame import BlocksCustomFrame from lib.utils.icon_button import IconButton from lib.utils.list_model import EntryDelegate, EntryListModel, ListItem from PyQt6 import QtCore, QtGui, QtWidgets -logger: logging.Logger = logging.getLogger(__name__) + +def _css_color(hex_text: str) -> str: + """Qt style sheets read 8-digit hex as #aarrggbb, like QColor.""" + return spoolman_color(hex_text).name(QtGui.QColor.NameFormat.HexArgb) class SpoolmanPage(QtWidgets.QWidget): @@ -82,6 +90,7 @@ def __init__(self, parent=None) -> None: self.reload_btn.clicked.connect(self._on_reload_clicked) self.delete_btn.clicked.connect(self._on_delete_clicked) + self._popup = Popup(self) self.confirm_popup = BasePopup(self, True, True) self.confirm_popup.set_message("Do you want to delete this spool") self.confirm_popup.accepted.connect(self._on_confirm_delete) @@ -91,21 +100,27 @@ def __init__(self, parent=None) -> None: @QtCore.pyqtSlot(dict, name="on-spools-received") def on_spools_received(self, result: dict) -> None: """Handle v2 proxy response for GET /v1/spool.""" - if result.get("error") is not None: - return spools = result.get("response") - if not isinstance(spools, list): - return - self._spools = spools + if result.get("error") is None and isinstance(spools, list): + self._spools = spools + self._no_spools_label.setText("No spools found") + else: + self._spools = [] + self._selected_spool = None + self._refresh_info_box() + self._no_spools_label.setText("Could not reach Spoolman") self._build_model_list() @QtCore.pyqtSlot(dict, name="on-delete-spool-result") def on_delete_spool_result(self, result: dict) -> None: """Handle v2 proxy response for DELETE /v1/spool/{id}.""" if result.get("error") is not None: - logger.error("Delete spool failed: %s", result["error"]) + self._popup.new_message( + message_type=Popup.MessageType.ERROR, message="Could not delete spool" + ) return self._selected_spool = None + self._refresh_info_box() self._on_reload_clicked() def _spool_display_name(self, spool: dict) -> str: @@ -116,57 +131,6 @@ def _spool_display_name(self, spool: dict) -> str: name = f"{name} [Archived]" return name - @staticmethod - def _make_add_pixmap() -> QtGui.QPixmap: - pixmap = QtGui.QPixmap(32, 32) - pixmap.fill(QtCore.Qt.GlobalColor.transparent) - painter = QtGui.QPainter(pixmap) - painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) - pen = QtGui.QPen(QtGui.QColor(160, 160, 160)) - pen.setWidth(2) - painter.setPen(pen) - painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) - painter.drawRoundedRect(QtCore.QRectF(1, 1, 32 - 2, 32 - 2), 6, 6) - mid = 32 // 2 - painter.drawLine(mid, 8, mid, 32 - 8) - painter.drawLine(8, mid, 32 - 8, mid) - painter.end() - return pixmap - - @staticmethod - def _make_color_pixmap(filament: dict) -> QtGui.QPixmap: - pixmap = QtGui.QPixmap(32, 32) - pixmap.fill(QtCore.Qt.GlobalColor.transparent) - painter = QtGui.QPainter(pixmap) - painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) - - multi_hexes: str | None = filament.get("multi_color_hexes") - color_hex: str | None = filament.get("color_hex") - - if multi_hexes: - hexes = [h.strip() for h in multi_hexes.split(",") if h.strip()] - if hexes: - clip = QtGui.QPainterPath() - clip.addRoundedRect(QtCore.QRectF(0, 0, 32, 32), 6, 6) - painter.setClipPath(clip) - stripe_w = 32 / len(hexes) - for i, h in enumerate(hexes): - painter.fillRect( - QtCore.QRectF(i * stripe_w, 0, stripe_w, 32), - QtGui.QColor(f"#{h}"), - ) - elif color_hex: - painter.setBrush(QtGui.QColor(f"#{color_hex}")) - painter.setPen(QtCore.Qt.PenStyle.NoPen) - painter.drawRoundedRect(QtCore.QRectF(0, 0, 32, 32), 6, 6) - else: - painter.setPen(QtGui.QColor(180, 180, 180)) - painter.setBrush(QtCore.Qt.BrushStyle.NoBrush) - painter.drawRoundedRect(QtCore.QRectF(1, 1, 32 - 2, 32 - 2), 6, 6) - - painter.end() - return pixmap - def _on_reload_clicked(self) -> None: self.request_spools.emit() self.request_get_spool_id.emit() @@ -199,7 +163,7 @@ def _build_model_list(self) -> None: self.model.add_item( ListItem( text=" Add Spool", - left_icon=self._make_add_pixmap(), + left_icon=make_add_pixmap(), _lfontsize=14, height=60, ) @@ -230,7 +194,7 @@ def _build_model_list(self) -> None: item = ListItem( text=display_name, right_text=right_text, - left_icon=self._make_color_pixmap(filament), + left_icon=make_color_pixmap(filament), selected=False, _lfontsize=14, _rfontsize=12, @@ -257,9 +221,7 @@ def on_item_clicked(self, item: ListItem) -> None: self._refresh_info_box() def _refresh_info_box(self) -> None: - if not self._selected_spool: - return - spool = self._selected_spool + spool = self._selected_spool or {} filament = spool.get("filament") or {} vendor = filament.get("vendor") or {} @@ -302,14 +264,16 @@ def _update_color_swatches(self, filament: dict) -> None: swatch.setMinimumSize(QtCore.QSize(18, 24)) swatch.setMaximumSize(QtCore.QSize(36, 24)) swatch.setStyleSheet( - f"background-color: #{hex_val}; border-radius: 4px;" + f"background-color: {_css_color(hex_val)}; border-radius: 4px;" ) self.color_swatch_layout.addWidget(swatch) elif color_hex: swatch = QtWidgets.QFrame(self) swatch.setMinimumSize(QtCore.QSize(36, 24)) swatch.setMaximumSize(QtCore.QSize(56, 24)) - swatch.setStyleSheet(f"background-color: #{color_hex}; border-radius: 5px;") + swatch.setStyleSheet( + f"background-color: {_css_color(color_hex)}; border-radius: 5px;" + ) self.color_swatch_layout.addWidget(swatch) else: swatch = QtWidgets.QFrame(self) diff --git a/tests/amu/test_manager_unit.py b/tests/amu/test_manager_unit.py index 585471df..ffa2f779 100644 --- a/tests/amu/test_manager_unit.py +++ b/tests/amu/test_manager_unit.py @@ -402,6 +402,7 @@ def test_reset_mmu(self, manager, qtbot) -> None: def test_load_gate(self, manager, qtbot) -> None: manager.update_mmu_state(_status(filament_pos=0)) + assert manager.load_refusal() == "" with qtbot.waitSignal(manager.run_gcode_signal) as blocker: assert manager.load_gate() is True assert blocker.args == ["MMU_LOAD"] @@ -422,14 +423,19 @@ def test_load_gate(self, manager, qtbot) -> None: def test_load_gate_refused(self, manager, qtbot, overrides) -> None: if overrides is not None: manager.update_mmu_state(_status(**overrides)) + assert manager.load_refusal() with qtbot.assertNotEmitted(manager.run_gcode_signal): assert manager.load_gate() is False - def test_unload(self, manager, qtbot) -> None: + @pytest.mark.parametrize( + ("eject", "gcode"), [(False, "MMU_UNLOAD"), (True, "MMU_EJECT")] + ) + def test_unload(self, manager, qtbot, eject, gcode) -> None: manager.update_mmu_state(_FULL_STATUS_WITH_SPOOLMAN) + assert manager.unload_refusal() == "" with qtbot.waitSignal(manager.run_gcode_signal) as blocker: - assert manager.unload() is True - assert blocker.args == ["MMU_UNLOAD"] + assert manager.unload(eject=eject) is True + assert blocker.args == [gcode] # Happy-Hare drops MMU_UNLOAD when disabled or already unloaded (mmu.py:6981). @pytest.mark.parametrize( @@ -439,8 +445,9 @@ def test_unload(self, manager, qtbot) -> None: def test_unload_refused(self, manager, qtbot, overrides) -> None: if overrides is not None: manager.update_mmu_state(_status(**overrides)) + assert manager.unload_refusal() with qtbot.assertNotEmitted(manager.run_gcode_signal): - assert manager.unload() is False + assert manager.unload(eject=True) is False # A paused print is where a failed swap gets recovered by hand. @pytest.mark.parametrize("print_state", ["paused", "pause_locked"]) diff --git a/tests/widgets/test_basic_filament_panel.py b/tests/widgets/test_basic_filament_panel.py index f0cc53e8..f9f6d478 100644 --- a/tests/widgets/test_basic_filament_panel.py +++ b/tests/widgets/test_basic_filament_panel.py @@ -85,7 +85,7 @@ def _item_addrs(layout): def test_pages_do_not_share_layout_items(qtbot, mock_printer, mock_cfg): """No QLayoutItem may belong to two layouts (deterministic, no crash).""" - panel = BasicFilamentPanel(mock_printer, mock_cfg) + panel = BasicFilamentPanel(mock_printer, mock_cfg, amu_manager=MagicMock()) qtbot.addWidget(panel) shared = _item_addrs(panel.verticalLayout) & _item_addrs(panel.verticalLayout_2) assert not shared, ( @@ -96,7 +96,7 @@ def test_pages_do_not_share_layout_items(qtbot, mock_printer, mock_cfg): def test_destroying_panel_is_clean(qapp, mock_printer, mock_cfg): """Force destruction (the AMU-swap path) and confirm no double-free SIGBUS.""" - panel = BasicFilamentPanel(mock_printer, mock_cfg) + panel = BasicFilamentPanel(mock_printer, mock_cfg, amu_manager=MagicMock()) panel.deleteLater() qapp.processEvents() # Reaching here means ~BasicFilamentPanel ran without a double-free. @@ -115,7 +115,7 @@ def _mmu_state(pos, action="Idle", color="", tool=0, gate=0): @pytest.fixture def panel(qtbot, mock_printer, mock_cfg): - panel = BasicFilamentPanel(mock_printer, mock_cfg) + panel = BasicFilamentPanel(mock_printer, mock_cfg, amu_manager=MagicMock()) qtbot.addWidget(panel) return panel @@ -127,6 +127,14 @@ def test_check_button_enabled_when_loaded(panel): assert not panel.Basic_fp_check_btn.isEnabled() +# Happy-Hare refuses MMU_LOAD from a mid-path position; a sensorless printer must still load. +@pytest.mark.parametrize(("mmu", "enabled"), [(True, False), (False, True)]) +def test_load_button_in_unknown_position(panel, mmu, enabled): + panel.mmu_configured = mmu + panel.filament_state = panel.FilamentStates.UNKNOWN + assert panel.Basic_fp_load_btn.isEnabled() is enabled + + def test_sensor_without_mmu_updates_position(panel): panel.filament_sensor = "presence" panel.on_filament_sensor_update("presence", "filament_detected", True) @@ -149,14 +157,37 @@ def test_color_swatch_accepts_names(panel): assert panel._lbl_clr.text() == "" -def test_accepted_position_change_emits_mmu_recover(panel): +# A fresh save_variables file has no filament_type; that used to raise AttributeError. +@pytest.mark.parametrize( + ("variables", "expected"), + [ + ({}, "UNKNOWN"), + ({"filament_type": ""}, "UNKNOWN"), + ({"filament_type": "TPU"}, "UNKNOWN"), + ({"filament_type": 5}, "UNKNOWN"), + ({"filament_type": "PETG"}, "PETG"), + ({"filament_type": "PLA+"}, "PLA"), + ], +) +def test_save_variables_material_label(panel, variables, expected): + panel.on_save_variables_update({"variables": variables}) + assert panel._lbl_mat._right.text() == expected + + +def test_save_variables_without_variables_key(panel): + panel.on_save_variables_update({}) + assert panel._lbl_mat._right.text() == "UNKNOWN" + + +def test_accepted_position_change_recovers_through_manager(panel): panel.on_mmu_state_changed(_mmu_state(FilamentPos.LOADED)) emitted = [] panel.run_gcode.connect(emitted.append) panel._lbl_pos.select_option("Unloaded") panel.on_pos_change() panel.confirm_pos_popup.accepted.emit() - assert emitted == ["MMU_RECOVER TOOL=0 GATE=0 LOADED=0"] + panel.amu_manager.recover_single_gate.assert_called_once_with(False) + assert emitted == [] def test_rejected_position_change_reverts_without_gcode(panel): @@ -168,3 +199,55 @@ def test_rejected_position_change_reverts_without_gcode(panel): panel.confirm_pos_popup.rejected.emit() assert emitted == [] assert panel._lbl_pos.text() == "Loaded" + + +@pytest.fixture +def mmu_panel(qtbot, mock_printer, mock_cfg): + load_popup = MagicMock() + panel = BasicFilamentPanel( + mock_printer, mock_cfg, amu_manager=MagicMock(), load_popup=load_popup + ) + qtbot.addWidget(panel) + panel.mmu_configured = True + panel.show() + return panel + + +@pytest.mark.parametrize("sent", [True, False]) +def test_load_goes_through_manager(mmu_panel, monkeypatch, sent): + mmu_panel.filament_state = mmu_panel.FilamentStates.UNLOADED + mmu_panel.amu_manager.load_gate.return_value = sent + mmu_panel.amu_manager.load_refusal.return_value = "MMU is not homed." + monkeypatch.setattr(mmu_panel.popup, "new_message", MagicMock()) + emitted = [] + mmu_panel.run_gcode.connect(emitted.append) + mmu_panel.load_filament() + mmu_panel.amu_manager.load_gate.assert_called_once_with() + # A refused load tells the user why instead of doing nothing. + assert mmu_panel.popup.new_message.called is not sent + if not sent: + assert mmu_panel.popup.new_message.call_args.kwargs["message"] == ( + "MMU is not homed." + ) + # A refused load must not leave the popup up or the type label saying loaded. + assert mmu_panel.load_popup.show.called is sent + assert [g.startswith("SAVE_VARIABLE") for g in emitted] == ([True] if sent else []) + + +@pytest.mark.parametrize("sent", [True, False]) +def test_unload_goes_through_manager(mmu_panel, monkeypatch, sent): + mmu_panel.filament_state = mmu_panel.FilamentStates.LOADED + mmu_panel.amu_manager.unload.return_value = sent + mmu_panel.amu_manager.unload_refusal.return_value = "Not available while printing." + monkeypatch.setattr(mmu_panel.popup, "new_message", MagicMock()) + emitted = [] + mmu_panel.run_gcode.connect(emitted.append) + mmu_panel.unload_filament() + mmu_panel.amu_manager.unload.assert_called_once_with(eject=True) + assert mmu_panel.popup.new_message.called is not sent + if not sent: + assert mmu_panel.popup.new_message.call_args.kwargs["message"] == ( + "Not available while printing." + ) + assert mmu_panel.load_popup.show.called is sent + assert [g.startswith("SAVE_VARIABLE") for g in emitted] == ([True] if sent else []) diff --git a/tests/widgets/test_filament_tab_mmu_unit.py b/tests/widgets/test_filament_tab_mmu_unit.py index bf51a705..7ad6055e 100644 --- a/tests/widgets/test_filament_tab_mmu_unit.py +++ b/tests/widgets/test_filament_tab_mmu_unit.py @@ -22,8 +22,7 @@ "lib.panels.widgets.FilamentTab", "lib.panels.filamentTab", ) -for _stale in [m for m in sys.modules if m.startswith(_STUBBED)]: - sys.modules.pop(_stale, None) +_saved = {m: sys.modules.pop(m) for m in list(sys.modules) if m.startswith(_STUBBED)} # loadWidget imports configfile at import time; the conftest fixture only stubs it later. _had_configfile = "configfile" in sys.modules @@ -32,6 +31,10 @@ if not _had_configfile: sys.modules.pop("configfile", None) +# Put the other modules' stand-ins back so this import leaves sys.modules as found. +for _real in [m for m in sys.modules if m.startswith(_STUBBED)]: + del sys.modules[_real] +sys.modules.update(_saved) def test_none_state_after_klipper_restart_is_ignored() -> None: diff --git a/tests/widgets/test_spoolman_ui_unit.py b/tests/widgets/test_spoolman_ui_unit.py new file mode 100644 index 00000000..bd26bdfe --- /dev/null +++ b/tests/widgets/test_spoolman_ui_unit.py @@ -0,0 +1,341 @@ +"""Spoolman error surfacing, gating and stale-list handling in the filament UI.""" + +import sys +from unittest.mock import MagicMock + +import pytest +from PyQt6 import QtCore + +import devices.amu.manager as amu_manager_module +from devices.amu.models import FilamentPos, GateInfo, MMUState + +vars(sys.modules["devices.amu"]).update( + AMUManager=amu_manager_module.AMUManager, + FilamentPos=FilamentPos, + MMUState=MMUState, +) +# Other test modules leave session-wide stand-ins for these; the tab needs the real API. +_STUBBED = ( + "lib.utils.icon_button", + "lib.utils.blocks_button", + "lib.utils.blocks_label", + "lib.utils.blocks_linedit", + "lib.utils.blocks_Scrollbar", + "lib.panels.widgets.Common", + "lib.panels.widgets.FilamentTab", + "lib.panels.filamentTab", +) +_saved = {m: sys.modules.pop(m) for m in list(sys.modules) if m.startswith(_STUBBED)} + +# loadWidget imports configfile at import time; the conftest fixture only stubs it later. +_had_configfile = "configfile" in sys.modules +sys.modules.setdefault("configfile", MagicMock()) +from lib.panels.filamentTab import FilamentTab # noqa: E402 +from lib.panels.widgets.FilamentTab.addFilamentPage import AddFilamentPage # noqa: E402 +from lib.panels.widgets.FilamentTab.addSpoolPage import ( # noqa: E402 + AddSpoolPage, + spoolman_color, +) +from lib.panels.widgets.FilamentTab.spoolmanPage import SpoolmanPage # noqa: E402 +from lib.utils.list_model import ListItem # noqa: E402 + +if not _had_configfile: + sys.modules.pop("configfile", None) +# Put the other modules' stand-ins back so this import leaves sys.modules as found. +for _real in [m for m in sys.modules if m.startswith(_STUBBED)]: + del sys.modules[_real] +sys.modules.update(_saved) + +UNREACHABLE = {"error": {"code": 503, "message": "Spoolman server not available"}} +SPOOL = {"id": 7, "filament": {"name": "PLA Red", "material": "PLA"}} + + +def _spools(*spools: dict) -> dict: + return {"response": list(spools), "error": None} + + +def _rows(model) -> list[str]: + return [model.entries[i].text for i in range(model.rowCount())] + + +@pytest.fixture +def tab(qtbot): + ws = MagicMock() + widget = FilamentTab(None, MagicMock(), ws, MagicMock(), MagicMock()) + qtbot.addWidget(widget) + widget.handle_moonraker_components(["spoolman"]) + ws.api.spoolman_proxy.reset_mock() + return widget + + +class TestFilamentTabSpools: + def test_failed_fetch_drops_the_previous_list(self, tab): + tab.on_spools_received(_spools(SPOOL)) + assert tab._spool_id_map + tab.on_spools_received(UNREACHABLE) + assert tab._spool_model.rowCount() == 0 + assert tab._spool_id_map == {} + assert tab._no_spools_label.text() == "Could not reach Spoolman" + + @pytest.mark.parametrize("reply", [UNREACHABLE, {"response": None, "error": None}]) + def test_failed_fetch_hides_the_list(self, tab, reply): + tab.on_spools_received(reply) + assert tab._spool_list_view.isHidden() + assert not tab._no_spools_label.isHidden() + + def test_popup_clears_the_list_before_fetching(self, tab): + tab.on_spools_received(_spools(SPOOL)) + tab.handle_popup(True) + assert tab._spool_model.rowCount() == 0 + assert not tab._spool_load_widget.isHidden() + tab.ws.api.spoolman_proxy.assert_called_once_with( + "GET", "/v1/spool", callback=tab.on_spools_received + ) + + def test_popup_skips_spoolman_when_not_configured(self, tab): + tab.handle_moonraker_components(["history"]) + tab.handle_popup(True) + tab.ws.api.spoolman_proxy.assert_not_called() + assert tab._popup_stack.currentWidget() is tab._popup_form_page + + def test_added_spool_refetches_the_list(self, tab): + tab._add_spool_page.accepted.emit() + tab.ws.api.spoolman_proxy.assert_called_once() + + def test_add_spool_row_resets_the_add_page(self, tab, monkeypatch): + reset = MagicMock() + monkeypatch.setattr(tab._add_spool_page, "reset", reset) + tab._on_list_item_tapped(ListItem(text="+ Add Spool")) + reset.assert_called_once() + + def test_offline_fetch_shows_the_error(self, tab): + tab.ws.api.spoolman_proxy.return_value = False + tab.handle_popup(True) + assert tab._spool_load_widget.isHidden() + assert tab._no_spools_label.text() == "Could not reach Spoolman" + + def test_weight_is_whole_grams(self, tab): + tab.on_spools_received(_spools({**SPOOL, "remaining_weight": 812.6})) + tab._on_list_item_tapped(tab._spool_model.entries[1]) + assert tab.weight_label.text() == "813 g" + + def test_pregate_filter_does_not_fetch(self, tab): + requests = [] + tab._add_spool_page.request_filaments.connect(lambda: requests.append(1)) + tab._add_spool_page.setFilter("PETG") + assert requests == [] + + def test_amu_swap_rewires_control_button_quietly(self, tab, qtlog, monkeypatch): + gates = tuple(GateInfo(index=i) for i in range(2)) + state = MMUState(num_gates=2, gates=gates) + tab.amu_manager.get_state.return_value = state + tab.on_mmu_state_changed(state) + pages = [] + monkeypatch.setattr(tab, "change_page", pages.append) + tab.fp_button_1.click() + assert pages == [tab.indexOf(tab.amupage)] + assert not [r for r in qtlog.records if "disconnect" in r.message] + + def test_idle_after_amu_swap_skips_the_deleted_panel(self, tab): + two = MMUState(num_gates=2, gates=tuple(GateInfo(index=i) for i in range(2))) + tab.amu_manager.get_state.return_value = two + tab.on_mmu_state_changed(two) + QtCore.QCoreApplication.sendPostedEvents( + None, QtCore.QEvent.Type.DeferredDelete.value + ) + tab.load_state = True + tab.on_mmu_state_changed( + MMUState(num_gates=1, gates=(GateInfo(index=0),), action="Idle") + ) + assert tab._basic_panel is None + assert tab.load_state is False + + +class TestFilamentTabGates: + @staticmethod + def _state(**status) -> MMUState: + return MMUState.from_status({"num_gates": 1, "gate": 0, **status}) + + # Happy Hare owns the gate map; a runout must not wipe the user's entry. + def test_runout_keeps_the_gate_map(self, tab): + tab.on_mmu_state_changed(self._state(gate_status=[1])) + tab.on_mmu_state_changed(self._state(gate_status=[0])) + tab.amu_manager.clear_gate.assert_not_called() + + def test_load_popup_uses_happy_hare_rgb(self, tab, monkeypatch): + monkeypatch.setattr(tab.load_popup, "isVisible", lambda: True) + colors = [] + monkeypatch.setattr(tab.load_status_widget, "set_gate_color", colors.append) + tab.on_mmu_state_changed( + self._state(gate_color=["red"], gate_color_rgb=[[1.0, 0.0, 0.0]]) + ) + assert colors[-1].name() == "#ff0000" + + +class TestVendorLookup: + @staticmethod + def _reply(tab, reply: dict) -> list: + def proxy(*_args, callback): + callback(reply) + return True + + tab.ws.api.spoolman_proxy.side_effect = proxy + got = [] + tab._find_or_add_vendor({"name": "Acme"}, got.append) + return got + + def test_exact_name_is_reused(self, tab): + vendors = [{"id": 4, "name": "Acme Pro"}, {"id": 3, "name": "Acme"}] + got = self._reply(tab, {"response": vendors, "error": None}) + assert got == [{"response": {"id": 3, "name": "Acme"}, "error": None}] + assert tab.ws.api.spoolman_proxy.call_args.args == ( + "GET", + "/v1/vendor", + "name=Acme", + ) + tab.ws.api.add_manufacturer.assert_not_called() + + # Spoolman's name filter ignores case, so "acme" must not add a second vendor. + def test_name_match_ignores_case(self, tab): + got = self._reply(tab, {"response": [{"id": 3, "name": "ACME"}], "error": None}) + assert got == [{"response": {"id": 3, "name": "ACME"}, "error": None}] + tab.ws.api.add_manufacturer.assert_not_called() + + def test_missing_name_is_created(self, tab): + got = self._reply(tab, {"response": [{"id": 4, "name": "Acme Pro"}]}) + tab.ws.api.add_manufacturer.assert_called_once_with( + {"name": "Acme"}, callback=got.append + ) + + def test_failed_lookup_creates_nothing(self, tab): + got = self._reply(tab, UNREACHABLE) + assert got[0]["error"] is not None + tab.ws.api.add_manufacturer.assert_not_called() + + +@pytest.mark.parametrize( + ("hex_text", "rgba"), + [ + ("ff0000", (255, 0, 0, 255)), + ("ff000080", (255, 0, 0, 128)), + ("#00ff00", (0, 255, 0, 255)), + ], +) +def test_spoolman_hex_alpha_is_trailing(hex_text, rgba): + assert spoolman_color(hex_text).getRgb() == rgba + + +@pytest.fixture +def add_spool(qtbot): + page = AddSpoolPage() + qtbot.addWidget(page) + return page + + +class TestAddSpoolPage: + def test_unreachable_shows_error(self, add_spool): + add_spool.on_filaments_received(UNREACHABLE) + assert add_spool._fil_list_view.isHidden() + assert add_spool._no_filaments_label.text() == "Could not reach Spoolman" + + def test_same_name_filaments_select_their_own_id(self, add_spool): + fils = [{"id": 1, "name": "PLA"}, {"id": 2, "name": "PLA"}] + add_spool.on_filaments_received({"response": fils, "error": None}) + add_spool._on_filament_selected(add_spool._fil_model.entries[1]) + assert add_spool._selected_filament_id == 1 + + def test_failed_add_shows_toast(self, add_spool, monkeypatch): + toast = MagicMock() + monkeypatch.setattr(add_spool._popup, "new_message", toast) + accepted = [] + add_spool.accepted.connect(lambda: accepted.append(1)) + add_spool.on_add_spool_result(UNREACHABLE) + assert accepted == [] + assert toast.call_args.kwargs["message"] == "Could not add spool" + + def test_submit_waits_for_the_reply(self, add_spool, monkeypatch): + monkeypatch.setattr(add_spool._popup, "new_message", MagicMock()) + add_spool.on_filaments_received(_spools({"id": 1, "name": "PLA"})) + add_spool._on_filament_selected(add_spool._fil_model.entries[1]) + sent = [] + add_spool.request_add_spool.connect(lambda *a: sent.append(a)) + add_spool._submit_btn.click() + add_spool._submit_btn.click() + assert len(sent) == 1 + add_spool.on_add_spool_result(UNREACHABLE) + assert add_spool._submit_btn.isEnabled() + + +class TestAddFilamentPage: + @pytest.fixture + def page(self, qtbot, monkeypatch): + widget = AddFilamentPage() + qtbot.addWidget(widget) + monkeypatch.setattr(widget._popup, "new_message", MagicMock()) + widget.reset() + return widget + + def test_failed_add_shows_toast(self, page): + page.on_add_filament_result(UNREACHABLE) + assert page._popup.new_message.call_args.kwargs["message"] == ( + "Could not add filament" + ) + + def test_failed_vendor_drops_pending_filament(self, page): + page._vendor_field.setText("Acme") + page._submit_btn.click() + page.on_add_manufacturer_result(UNREACHABLE) + assert page.request_filament_body is None + assert page._submit_btn.isEnabled() + + def test_submit_waits_for_the_reply(self, page): + sent = [] + page.request_add_filament.connect(sent.append) + page._submit_btn.click() + page._submit_btn.click() + assert len(sent) == 1 + page.on_add_filament_result(UNREACHABLE) + assert page._submit_btn.isEnabled() + + def test_reset_drops_a_pending_submit(self, page): + page._vendor_field.setText("Acme") + page._submit_btn.click() + page.reset() + assert page.request_filament_body is None + assert page._submit_btn.isEnabled() + + +class TestSpoolmanPage: + @pytest.fixture + def page(self, qtbot): + widget = SpoolmanPage() + qtbot.addWidget(widget) + return widget + + def test_unreachable_clears_list_and_says_so(self, page): + page.on_spools_received(_spools(SPOOL)) + page.on_spools_received(UNREACHABLE) + assert _rows(page.model) == [" Add Spool"] + assert page._no_spools_label.text() == "Could not reach Spoolman" + + @pytest.mark.parametrize( + "reply", + [ + lambda page: page.on_spools_received(UNREACHABLE), + lambda page: page.on_delete_spool_result({"response": {}, "error": None}), + ], + ids=["unreachable", "deleted"], + ) + def test_reply_clears_the_selected_spool(self, page, reply): + page._selected_spool = SPOOL + page._refresh_info_box() + assert page.filament_name_label.text() == "PLA Red" + reply(page) + assert page._selected_spool is None + assert page.filament_name_label.text() == "—" + + def test_failed_delete_shows_toast(self, page, monkeypatch): + toast = MagicMock() + monkeypatch.setattr(page._popup, "new_message", toast) + page.on_delete_spool_result(UNREACHABLE) + assert toast.call_args.kwargs["message"] == "Could not delete spool"