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/helper_methods.py

@@ -9,10 +9,10 @@
 import ctypes
 import enum
 import logging
+import math
 import os
 import pathlib
 import struct
-import typing
 
 logger = logging.getLogger(__name__)
 
@@ -21,7 +21,7 @@
     libxext = ctypes.CDLL("libXext.so.6")
 
     class DPMSState(enum.Enum):
-        """Available DPMS states"""
+        """Available DPMS states."""
 
         FAIL = -1
         ON = 0
@@ -86,7 +86,7 @@ def set_dpms_mode(mode: DPMSState) -> None:
             finally:
                 libxext.XCloseDisplay(display)
 
-    def get_dpms_timeouts() -> typing.Dict:
+    def get_dpms_timeouts() -> dict:
         """Get current DPMS timeouts"""
         _display_name = ctypes.c_char_p(b":0")
         libxext.XOpenDisplay.restype = ctypes.c_void_p
@@ -118,9 +118,7 @@ def get_dpms_timeouts() -> typing.Dict:
             "off_seconds": _off_timeout,
         }
 
-    def set_dpms_timeouts(
-        suspend: int = 0, standby: int = 0, off: int = 0
-    ) -> typing.Dict:
+    def set_dpms_timeouts(suspend: int = 0, standby: int = 0, off: int = 0) -> dict:
         """Set DPMS timeout"""
         _display_name = ctypes.c_char_p(b":0")
         libxext.XOpenDisplay.restype = ctypes.c_void_p
@@ -155,11 +153,11 @@ def set_dpms_timeouts(
             "off_seconds": _off_timeout,
         }
 
-    def get_dpms_info() -> typing.Dict:
+    def get_dpms_info() -> dict:
         """Get DPMS information
 
         Returns:
-            typing.Dict: Dpms state
+            dict: Dpms state
         """
         _dpms_state = DPMSState.FAIL
         onoff = 0
@@ -227,133 +225,122 @@ def disable_dpms() -> None:
     logger.exception(f"Unexpected exception occurred {e}")
 
 
-def convert_bytes_to_mb(self, bytes: int | float) -> float:
-    """Converts byte size to megabyte size
+def convert_bytes_to_mb(size_bytes: int | float) -> float:
+    """Converts byte size to megabyte size.
 
     Args:
-        bytes (int | float): bytes
+        size_bytes: Value in bytes.
 
     Returns:
-        mb: float that represents the number of mb
+        Equivalent value in megabytes.
     """
     _relation = 2 ** (-20)
-    return bytes * _relation
+    return size_bytes * _relation
 
 
 def calculate_current_layer(
     z_position: float,
-    object_height: float,
     layer_height: float,
     first_layer_height: float,
+    max_layers: int = 0,
 ) -> int:
-    """Calculated the current printing layer given the GCODE z position received by the
-        gcode_move object update.
-        Also updates the label where the current layer should be displayed
+    """Calculate current layer from Z position (fallback when Klipper
+    does not provide ``print_stats.info.current_layer``).
+
+    Formula ported from Mainsail ``getPrintCurrentLayer`` getter:
+    ``src/store/printer/getters.ts`` in ``mainsail-crew/mainsail``.
+
+    Uses ``ceil((z - first_layer_height) / layer_height + 1)``
+    and clamps the result to ``[0, max_layers]``.
 
     Returns:
-        int: Current layer
+        int: Current layer number (0 when not yet printing).
     """
-    if z_position == 0:
-        return -1
-    if z_position <= first_layer_height:
-        return 1
+    if layer_height <= 0 or first_layer_height < 0:
+        return 0
 
-    _current_layer = (z_position) / layer_height
+    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 int(_current_layer)
 
+def calculate_max_layers(
+    object_height: float,
+    layer_height: float,
+    first_layer_height: float,
+) -> int:
+    """Calculate total layers from metadata dimensions (fallback when
+    Klipper does not provide ``print_stats.info.total_layer``).
 
-def estimate_print_time(seconds: int) -> list:
-    """Convert time in seconds format to days, hours, minutes, seconds.
+    Formula ported from Mainsail ``getPrintMaxLayers`` getter:
+    ``src/store/printer/getters.ts`` in ``mainsail-crew/mainsail``.
 
-    Args:
-        seconds (int): Seconds
+    Uses ``ceil((object_height - first_layer_height) / layer_height + 1)``.
 
     Returns:
-        list: list that contains the converted information [days, hours, minutes, seconds]
+        int: Total layer count, or 0 if metadata is insufficient.
     """
-    num_min, seconds = divmod(seconds, 60)
-    num_hours, minutes = divmod(num_min, 60)
-    days, hours = divmod(num_hours, 24)
-    return [days, hours, minutes, seconds]
-
+    if layer_height <= 0 or object_height <= 0:
+        return 0
+    return max(1, math.ceil((object_height - first_layer_height) / layer_height + 1))
 
-def normalize(value, r_min=0.0, r_max=1.0, t_min=0.0, t_max=100):
-    """Normalize values between a rage"""
-    # https://stats.stackexchange.com/questions/281162/scale-a-number-between-a-range
-    c1 = (value - r_min) / (r_max - r_min)
-    c2 = (t_max - t_min) + t_min
-    return c1 * c2
 
+def estimate_print_time(seconds: int) -> list[int]:
+    """Convert *seconds* to ``[days, hours, minutes, seconds]``."""
+    num_min, secs = divmod(seconds, 60)
+    num_hours, mins = divmod(num_min, 60)
+    days, hours = divmod(num_hours, 24)
+    return [days, hours, mins, secs]
 
-def check_filepath_permission(filepath, access_type: int = os.R_OK) -> bool:
-    # if not isinstance(filepath, pathlib.Path):
-    """Checks for file path access
 
-    Args:
-        filepath (str | pathlib.Path): path to file
-        access_type (int, optional): _description_. Defaults to os.R_OK.
+def normalize(
+    value: float,
+    r_min: float = 0.0,
+    r_max: float = 1.0,
+    t_min: float = 0.0,
+    t_max: float = 100.0,
+) -> float:
+    """Scale *value* from range [r_min, r_max] into [t_min, t_max]."""
+    return (value - r_min) / (r_max - r_min) * (t_max - t_min) + t_min
 
-    ***
 
-    #### **Access type can be:**
+def check_filepath_permission(
+    filepath: str | pathlib.Path, access_type: int = os.R_OK
+) -> bool:
+    """Check whether *filepath* exists and has the requested access.
 
-     - F_OK -> Checks file existence on path
-     - R_OK -> Checks if file is readable
-     - W_OK -> Checks if file is Writable
-     - X_OK -> Checks if file can be executed
+    Args:
+        filepath: Path to file.
+        access_type: ``os.F_OK`` (existence), ``os.R_OK`` (read),
+            ``os.W_OK`` (write), or ``os.X_OK`` (execute).
 
-    ***
     Returns:
-        bool: _description_
-    """  #     return False
-    if not os.path.isfile(filepath):
-        return False
-    return os.access(filepath, access_type)
+        ``True`` if the file exists and satisfies *access_type*.
+    """
+    path = pathlib.Path(filepath)
+    return path.is_file() and os.access(path, access_type)
 
 
-def check_dir_existence(
-    directory: typing.Union[str, pathlib.Path],
-) -> bool:
-    """Check if a directory exists. Returns a true if it exists"""
-    if isinstance(directory, pathlib.Path):
-        return bool(directory.is_dir())
-    return bool(os.path.isdir(directory))
+def check_dir_existence(directory: str | pathlib.Path) -> bool:
+    """Return ``True`` if *directory* exists and is a directory."""
+    return pathlib.Path(directory).is_dir()
 
 
 def check_file_on_path(
-    path: typing.Union[typing.LiteralString, pathlib.Path],
-    filename: typing.Union[typing.LiteralString, pathlib.Path],
+    path: str | pathlib.Path,
+    filename: str | pathlib.Path,
 ) -> bool:
-    """Check if file exists on path. Returns true if file exists on that specified directory"""
-    _filepath = os.path.join(path, filename)
-    return os.path.exists(_filepath)
-
+    """Return ``True`` if *filename* exists under *path*."""
+    return (pathlib.Path(path) / filename).exists()
 
-def get_file_loc(filename) -> pathlib.Path: ...
 
+def get_file_name(filename: str | None) -> str:
+    """Extract the basename from a file path (handles ``/`` and ``\\``).
 
-def get_file_name(filename: typing.Optional[str]) -> str:
-    # If filename is None or empty, return empty string instead of None
+    Returns:
+        The last path component, or ``""`` if *filename* is falsy.
+    """
     if not filename:
         return ""
-    # Remove trailing slashes or backslashes
-    filename = filename.rstrip("/\\")
-
-    # Normalize Windows backslashes to forward slashes
-    filename = filename.replace("\\", "/")
-
-    parts = filename.split("/")
-
-    # Split and return the last path component
-    return parts[-1] if filename else ""
-
-
-# def get_hash(data) -> hashlib._Hash:
-#     hash = hashlib.sha256()
-#     hash.update(data.encode())
-#     hash.digest()
-#     return hash
-
-
-def digest_hash() -> None: ...
+    return pathlib.PurePosixPath(filename.replace("\\", "/")).name

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