perf(ui): cache paint state across widgets and add missing pyqtSlot coverage

BlocksScreen/__init__.py

@@ -1 +1 @@
-"""BlocksScreen — GUI for BLOCKS 3D Printers running Klipper."""
+"""BlocksScreen - GUI for BLOCKS 3D Printers running Klipper."""

BlocksScreen/devices/amu/manager.py

@@ -61,6 +61,7 @@ def __init__(self, ws: MoonWebSocket, parent: QtCore.QObject | None = None) -> N
         self._pre_gate_sensors: dict[int, bool] = {}
         self.spool_fetched.connect(self._apply_spool_data)
 
+    @QtCore.pyqtSlot(int, dict)
     def _apply_spool_data(self, gate: int, data: dict) -> None:
         """Apply Spoolman spool data to local gate state and sync to Klipper.
 

BlocksScreen/devices/amu/models.py

@@ -202,7 +202,7 @@ def apply_diff(self, diff: dict) -> "MMUState":
                     scalar_fields["endless_spool_groups"]
                 )
             return dataclasses.replace(self, **scalar_fields)
-        # Gate arrays changed — need full rebuild, but we lost the raw arrays
+        # Gate arrays changed - need full rebuild, but we lost the raw arrays
         # Pass current gate data + diff into from_status
         gate_data = {
             "gate_status": [g.status for g in self.gates],

BlocksScreen/devices/storage/udisks2.py

@@ -54,7 +54,6 @@ def validate_label(label: str, strict: bool = True, max_length: int = 100) -> st
 
     dangerous_chars = {
         "\0",
-        "\x00",
         "/",
         "\\",
         ";",
@@ -89,7 +88,7 @@ def validate_label(label: str, strict: bool = True, max_length: int = 100) -> st
         )
 
     final_label = clean_label.strip(" .")[:max_length]
-    return final_label if final_label else ""
+    return final_label or ""
 
 
 def fire_n_forget(
@@ -412,9 +411,9 @@ async def _properties_changed_listener(self) -> None:
         Updates tracked objects
         """
         async for (
-            path,
-            changed_properties,
-            invalid_properties,
+            _path,
+            _changed_properties,
+            _invalid_properties,
         ) in self.obj_manager.properties_changed:
             pass
 
@@ -528,7 +527,7 @@ def add_symlink(
             label = validate_label(label, strict=True)
             label = "USB-" + label
         fallback: str = "USB DRIVE" if _index == 0 else str(f"USB DRIVE {_index}")
-        dstb = pathlib.Path(dst_path).joinpath(label if label else fallback)
+        dstb = pathlib.Path(dst_path).joinpath(label or fallback)
         try:
             if not os.path.islink(dstb):
                 os.symlink(src=path, dst=dstb)

BlocksScreen/devices/storage/usb_controller.py

@@ -86,6 +86,7 @@ def close(self) -> None:
         self.udisks.close()
         self.deleteLater()
 
+    @QtCore.pyqtSlot()
     def _handle_full_restart(self) -> None:
         if self.need_restart:
             self.udisks.start(self.udisks.Priority.InheritPriority)
@@ -106,7 +107,7 @@ def restart_type(self, type: ResType) -> None:
         if type not in ("always", "none"):
             logging.info("Unknown restart type %s", (type,))
         if type == "always":
-            if not self._restart_type == "always":
+            if self._restart_type != "always":
                 self.udisks.finished.connect(self._handle_monitor_finished)
         else:
             try:

BlocksScreen/helper_methods.py

@@ -262,7 +262,7 @@ def calculate_current_layer(
     layer = math.ceil((z_position - first_layer_height) / layer_height + 1)
     if max_layers > 0 and layer > max_layers:
         return max_layers
-    return layer if layer > 0 else 0
+    return max(0, layer)
 
 
 def calculate_max_layers(

BlocksScreen/lib/files.py

@@ -165,6 +165,12 @@ class Files(QtCore.QObject):
         - full_refresh_needed: Root changed
     """
 
+    _EVT_WS_OPEN: typing.ClassVar[QtCore.QEvent.Type] = events.WebSocketOpen.type()
+    _EVT_KLIPPER_DISC: typing.ClassVar[QtCore.QEvent.Type] = (
+        events.KlippyDisconnected.type()
+    )
+    _EVT_FILE_DATA: typing.ClassVar[QtCore.QEvent.Type] = ReceivedFileData.type()
+
     # Signals for API requests
     request_file_list = QtCore.pyqtSignal([], [str], name="api_get_files_list")
     request_dir_info = QtCore.pyqtSignal(
@@ -225,10 +231,14 @@ def _connect_signals(self) -> None:
         self.request_file_metadata.connect(self.ws.api.get_gcode_metadata)
 
     def _install_event_filter(self) -> None:
-        """Install event filter on application instance."""
-        app = QtWidgets.QApplication.instance()
-        if app:
-            app.installEventFilter(self)
+        """Install event filter on parent to limit scope to mainWindow events only."""
+        parent = self.parent()
+        if parent is not None:
+            parent.installEventFilter(self)
+        else:
+            app = QtWidgets.QApplication.instance()
+            if app:
+                app.installEventFilter(self)
 
     @property
     def file_list(self) -> list[dict]:
@@ -578,7 +588,7 @@ def _process_usb_directory_info(self, usb_path: str, data: dict) -> None:
     def _process_directory_info(self, data: dict) -> None:
         """Process directory info response."""
         # Check if this is a USB preload response.
-        # Match by FIFO queue — Moonraker responds to get_dir_information in order.
+        # Match by FIFO queue - Moonraker responds to get_dir_information in order.
         matched_usb = None
 
         if self._usb_preload_queue:
@@ -657,19 +667,18 @@ def get_dir_information(
 
     def eventFilter(self, obj: QtCore.QObject, event: QtCore.QEvent) -> bool:
         """Handle application-level events."""
-        if event.type() == events.WebSocketOpen.type():
+        etype = event.type()
+        if etype == self._EVT_WS_OPEN:
             self.initial_load()
             return False
-
-        if event.type() == events.KlippyDisconnected.type():
+        if etype == self._EVT_KLIPPER_DISC:
             self._clear_all_data()
             return False
-
         return super().eventFilter(obj, event)
 
     def event(self, event: QtCore.QEvent) -> bool:
         """Handle object-level events."""
-        if event.type() == ReceivedFileData.type():
+        if event.type() == self._EVT_FILE_DATA:
             if isinstance(event, ReceivedFileData):
                 self.handle_message_received(event.method, event.data, event.params)
                 return True

BlocksScreen/lib/klipper_message_filter.py

@@ -40,7 +40,7 @@ def _sub(needle: str) -> Callable[[str], bool]:
 
 
 def _re(pattern: str) -> Callable[[str], bool]:
-    compiled_text = re.compile(pattern, re.I)
+    compiled_text = re.compile(pattern, re.IGNORECASE)
     return lambda text: compiled_text.search(text) is not None
 
 
@@ -74,7 +74,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Check board cooling",
         severity=Severity.WARNING,
     ),
-    # ── Gcode errors — Beacon probe ────────────────────────────────────────────
+    # ── Gcode errors - Beacon probe ────────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("sensor not in valid range"),
@@ -125,14 +125,14 @@ def _re(pattern: str) -> Callable[[str], bool]:
         display="Beacon Scan Height Invalid",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — BLTouch probe ──────────────────────────────────────────
+    # ── Gcode errors - BLTouch probe ──────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("failed to verify sensor state"),
         display="BLTouch Verify Failed",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — Eddy probe ─────────────────────────────────────────────
+    # ── Gcode errors - Eddy probe ─────────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("must calibrate probe_eddy_current"),
@@ -209,7 +209,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Recalibrate the probe",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — probe calibration ─────────────────────────────────────
+    # ── Gcode errors - probe calibration ─────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("manual probe failed"),
@@ -235,7 +235,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Clean the probe tip and retry",
         severity=Severity.WARNING,
     ),
-    # ── Gcode errors — Happy Hare MMU ──────────────────────────────────────────
+    # ── Gcode errors - Happy Hare MMU ──────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("mmu not enabled"),
@@ -292,7 +292,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Clear the tangle",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — heater / temperature ───────────────────────────────────
+    # ── Gcode errors - heater / temperature ───────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("adc out of range"),
@@ -330,7 +330,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         display="Heater Decoupled",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — MCU / timing ────────────────────────────────────────────
+    # ── Gcode errors - MCU / timing ────────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("lost communication with mcu"),
@@ -371,7 +371,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         display="Stepper Driver Error",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — homing ──────────────────────────────────────────────────
+    # ── Gcode errors - homing ──────────────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("timeout during homing"),
@@ -400,7 +400,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Check for obstructions",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — bed leveling ───────────────────────────────────────────
+    # ── Gcode errors - bed leveling ───────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_re(r"must.*z_tilt_adjust"),
@@ -420,7 +420,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         display="Leveling Failed",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — resonance tester ──────────────────────────────────────
+    # ── Gcode errors - resonance tester ──────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("no accelerometers specified"),
@@ -470,7 +470,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Check sensor connection",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — general ─────────────────────────────────────────────────
+    # ── Gcode errors - general ─────────────────────────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("filament runout"),
@@ -544,7 +544,7 @@ def _re(pattern: str) -> Callable[[str], bool]:
         hint="Reduce the LED chain length",
         severity=Severity.ERROR,
     ),
-    # ── Gcode errors — Kalico / Danger Klipper ───────────────────────────────
+    # ── Gcode errors - Kalico / Danger Klipper ───────────────────────────────
     MessageRule(
         source=MessageSource.GCODE_ERROR,
         matcher=_sub("error on unused option"),
@@ -642,8 +642,8 @@ def _re(pattern: str) -> Callable[[str], bool]:
 )
 
 _IGNORED: tuple[re.Pattern, ...] = (
-    re.compile(r"REMOVED log_path=", re.I),
-    re.compile(r"Alert: Automatically heating extruder to gatemap temp", re.I),
+    re.compile(r"REMOVED log_path=", re.IGNORECASE),
+    re.compile(r"Alert: Automatically heating extruder to gatemap temp", re.IGNORECASE),
     # add more patterns as needed,
 )
 

BlocksScreen/lib/moonrakerComm.py

@@ -325,7 +325,9 @@ def on_message(self, *args) -> None:
                 "Unexpected error while creating websocket message event: %s", e
             )
 
-    def send_request(self, method: str, params: dict = {}, callback=None) -> bool:
+    def send_request(
+        self, method: str, params: dict | None = None, callback=None
+    ) -> bool:
         """Send a request over the websocket
 
         Args:
@@ -340,6 +342,7 @@ def send_request(self, method: str, params: dict = {}, callback=None) -> bool:
         if not self.connected or self.ws is None:
             return False
 
+        params = {} if params is None else params
         self._request_id += 1
         self.request_table[self._request_id] = [method, params, callback]
         packet = {

BlocksScreen/lib/network/manager.py

@@ -17,7 +17,7 @@
 
 logger = logging.getLogger(__name__)
 
-_KEEPALIVE_POLL_MS: int = 300_000  # 5 minutes — safety net for missed signals
+_KEEPALIVE_POLL_MS: int = 300_000  # 5 minutes - safety net for missed signals
 
 
 class NetworkManager(QObject):
@@ -27,8 +27,8 @@ class NetworkManager(QObject):
     a ``NetworkManagerWorker`` that runs all D-Bus coroutines on its
     dedicated asyncio thread.
 
-    Coroutines are submitted to ``worker._asyncio_loop`` — the same loop
-    on which the D-Bus file-descriptor was registered — so signal delivery
+    Coroutines are submitted to ``worker._asyncio_loop`` - the same loop
+    on which the D-Bus file-descriptor was registered - so signal delivery
     and async I/O always occur on the correct selector.
 
     """
@@ -72,7 +72,7 @@ def __init__(self, parent: QObject | None = None) -> None:
         self._worker.reconnect_complete.connect(self.reconnect_complete)
         self._worker.initialized.connect(self._on_worker_initialized)
 
-        # Keepalive timer — safety net for any missed D-Bus signals.
+        # Keepalive timer - safety net for any missed D-Bus signals.
         self._keepalive_timer = QTimer(self)
         self._keepalive_timer.setInterval(_KEEPALIVE_POLL_MS)
         self._keepalive_timer.timeout.connect(self._on_keepalive_tick)
@@ -96,7 +96,7 @@ def _schedule(self, coro: "asyncio.Coroutine") -> None:
             future.add_done_callback(self._pending_futures.discard)
         else:
             logger.debug(
-                "Dropping early coroutine — loop not yet running: %s",
+                "Dropping early coroutine - loop not yet running: %s",
                 coro.__qualname__,
             )
             coro.close()
@@ -114,7 +114,7 @@ def _on_worker_initialized(self) -> None:
             return
         self._worker_ready = True
         logger.info(
-            "Worker initialised — starting keepalive (every %d ms)",
+            "Worker initialised - starting keepalive (every %d ms)",
             _KEEPALIVE_POLL_MS,
         )
         self._keepalive_timer.start()
@@ -185,7 +185,7 @@ def _on_hotspot_info_ready(self, ssid: str, password: str, security: str) -> Non
 
     @pyqtSlot()
     def _on_keepalive_tick(self) -> None:
-        """Safety-net refresh — runs every 5 min to catch any missed signals."""
+        """Safety-net refresh - runs every 5 min to catch any missed signals."""
         if self._shutting_down:
             return
         self._schedule(self._worker._async_get_current_state())
@@ -273,7 +273,7 @@ def update_hotspot_config(
         new_password: str,
         security: str = "wpa-psk",
     ) -> None:
-        """Change hotspot name/password/security — cleans up old profiles."""
+        """Change hotspot name/password/security - cleans up old profiles."""
         self._schedule(
             self._worker._async_update_hotspot_config(
                 old_ssid, new_ssid, new_password, security
@@ -346,17 +346,17 @@ def saved_networks(self) -> list[SavedNetwork]:
 
     @property
     def hotspot_ssid(self) -> str:
-        """Hotspot SSID — read from main-thread cache (thread-safe)."""
+        """Hotspot SSID - read from main-thread cache (thread-safe)."""
         return self._cached_hotspot_ssid
 
     @property
     def hotspot_password(self) -> str:
-        """Hotspot password — read from main-thread cache (thread-safe)."""
+        """Hotspot password - read from main-thread cache (thread-safe)."""
         return self._cached_hotspot_password
 
     @property
     def hotspot_security(self) -> str:
-        """Hotspot security type — always 'wpa-psk' (WPA2-PSK, thread-safe)."""
+        """Hotspot security type - always 'wpa-psk' (WPA2-PSK, thread-safe)."""
         return self._cached_hotspot_security
 
     def get_network_info(self, ssid: str) -> NetworkInfo | None: