bugfix: resolve 40+ crashes, threading races, and Klipper protocol issues

BlocksScreen/BlocksScreen.py

@@ -1,6 +1,5 @@
 import logging
 import sys
-import typing
 
 from logger import CrashHandler, LogManager, install_crash_handler, setup_logging
 
@@ -39,7 +38,7 @@ def notify(self, a0: QtCore.QObject, a1: QtCore.QEvent) -> bool:  # type: ignore
 RESET = "\033[0m"
 
 
-def show_splash(window: typing.Optional[QtWidgets.QWidget] = None):
+def show_splash(window: QtWidgets.QWidget | None = None):
     """Show splash screen on app initialization"""
     logo = QtGui.QPixmap("BlocksScreen/BlocksScreen/lib/ui/resources/logoblocks.png")
     splash = QtWidgets.QSplashScreen(pixmap=logo)

BlocksScreen/lib/filament.py

@@ -1,6 +1,7 @@
 # Class that represents a filament spool
 
-from typing import Optional
+from __future__ import annotations
+
 import enum
 
 
@@ -29,17 +30,18 @@ def __init__(
         self,
         name: str,
         temperature: int,
-        brand: Optional[str] = None,
-        spool_type: Optional[SpoolMaterial] = None,
-        spool_weight: Optional[float] = None,
+        brand: str | None = None,
+        spool_type: SpoolMaterial | None = None,
+        spool_weight: float | None = None,
     ):
         if not isinstance(name, str) or not isinstance(temperature, int):
             raise TypeError("__init__() invalid argument type")
 
         self._name: str = name
         self._temperature: int = temperature
-        self._weight: Optional[float] = None
-        self._brand: Optional[str] = brand
+        self._weight: float | None = None
+        self._brand: str | None = brand
+        self._spool_type: Filament.SpoolMaterial | None = None
 
         if spool_type is not None and spool_type in self.SpoolMaterial:
             self._spool_type = spool_type
@@ -55,7 +57,7 @@ def temperature(self) -> int:
         return self._temperature
 
     @property
-    def weight(self) -> Optional[float]:
+    def weight(self) -> float | None:
         if self._weight is None:
             return
         return self._weight
@@ -65,22 +67,19 @@ def weight(self, new_value: float):
         self._weight = new_value
 
     @property
-    def brand(self) -> Optional[str]:
+    def brand(self) -> str | None:
         return self._brand
 
     @brand.setter
-    def brand(self, new_value: str) -> Optional[str]:
+    def brand(self, new_value: str) -> None:
         self._brand = new_value
 
     @property
-    def spool_type(self) -> Optional[SpoolMaterial]:
+    def spool_type(self) -> SpoolMaterial | None:
         return self._spool_type
 
     @spool_type.setter
-    def spool_type(self, new):
-        if new not in self.SpoolMaterial:
-            if isinstance(new, self.SpoolMaterial):
-                raise ValueError(
-                    "Spool Material type is invalid"
-                )  # Correct type but invalid option
+    def spool_type(self, new) -> None:
+        if new is not None and not isinstance(new, self.SpoolMaterial):
+            raise ValueError(f"Spool Material type is invalid: {new!r}")
         self._spool_type = new

BlocksScreen/lib/files.py

@@ -8,7 +8,6 @@
 from pathlib import Path
 
 import events
-from events import ReceivedFileData
 from lib.moonrakerComm import MoonWebSocket
 from PyQt6 import QtCore, QtGui, QtWidgets
 
@@ -54,7 +53,7 @@ class FileMetadata:
 
     filename: str = ""
     thumbnail_images: list[QtGui.QImage] = field(default_factory=list)
-    filament_total: typing.Union[dict, str, float] = field(default_factory=dict)
+    filament_total: dict | str | float = field(default_factory=dict)
     estimated_time: int = 0
     layer_count: int = -1
     total_layer: int = -1
@@ -74,8 +73,8 @@ class FileMetadata:
     slicer_version: str = "Unknown"
     gcode_start_byte: int = 0
     gcode_end_byte: int = 0
-    print_start_time: typing.Optional[float] = None
-    job_id: typing.Optional[str] = None
+    print_start_time: float | None = None
+    job_id: str | None = None
 
     def to_dict(self) -> dict:
         """Convert to dictionary for signal emission."""
@@ -255,7 +254,7 @@ def is_loaded(self) -> bool:
         """Check if initial load is complete."""
         return self._initial_load_complete
 
-    def get_file_metadata(self, filename: str) -> typing.Optional[FileMetadata]:
+    def get_file_metadata(self, filename: str) -> FileMetadata | None:
         """Get cached metadata for a file."""
         return self._files_metadata.get(filename.removeprefix("/"))
 
@@ -279,7 +278,7 @@ def initial_load(self) -> None:
         self._initial_load_complete = False
         self.request_dir_info[str, bool].emit("", True)
 
-    def handle_filelist_changed(self, data: typing.Union[dict, list]) -> None:
+    def handle_filelist_changed(self, data: dict | list) -> None:
         """Handle notify_filelist_changed from Moonraker."""
         if isinstance(data, dict) and "params" in data:
             data = data.get("params", [])
@@ -492,7 +491,7 @@ def _process_metadata(self, data: dict) -> None:
         self.fileinfo.emit(metadata.to_dict())
         logger.debug(f"Metadata loaded for: {filename}")
 
-    def handle_metadata_error(self, error_data: typing.Union[str, dict]) -> None:
+    def handle_metadata_error(self, error_data: str | dict) -> None:
         """
         Handle metadata request error from Moonraker.
 
@@ -538,7 +537,7 @@ def _preload_usb_contents(self, usb_path: str) -> None:
         self._usb_preload_queue.append(usb_path)
         self.ws.api.get_dir_information(usb_path, True)
 
-    def get_cached_usb_files(self, usb_path: str) -> typing.Optional[list[dict]]:
+    def get_cached_usb_files(self, usb_path: str) -> list[dict] | None:
         """
         Get cached files for a USB path if available.
 
@@ -646,7 +645,7 @@ def on_request_fileinfo(self, filename: str) -> None:
     @QtCore.pyqtSlot(str, bool, name="get_dir_info")
     def get_dir_information(
         self, directory: str = "", extended: bool = True
-    ) -> typing.Optional[list]:
+    ) -> list | None:
         """Get directory information."""
         self._current_directory = directory
 
@@ -669,8 +668,8 @@ def eventFilter(self, obj: QtCore.QObject, event: QtCore.QEvent) -> bool:
 
     def event(self, event: QtCore.QEvent) -> bool:
         """Handle object-level events."""
-        if event.type() == ReceivedFileData.type():
-            if isinstance(event, ReceivedFileData):
+        if event.type() == events.ReceivedFileData.type():
+            if isinstance(event, events.ReceivedFileData):
                 self.handle_message_received(event.method, event.data, event.params)
                 return True
         return super().event(event)