diff --git a/BlocksScreen/helper_methods.py b/BlocksScreen/helper_methods.py index 25b76cac..114ff5f1 100644 --- a/BlocksScreen/helper_methods.py +++ b/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 diff --git a/BlocksScreen/lib/panels/widgets/jobStatusPage.py b/BlocksScreen/lib/panels/widgets/jobStatusPage.py index 67add6b9..ac909237 100644 --- a/BlocksScreen/lib/panels/widgets/jobStatusPage.py +++ b/BlocksScreen/lib/panels/widgets/jobStatusPage.py @@ -2,7 +2,11 @@ import typing import events -from helper_methods import calculate_current_layer, estimate_print_time +from helper_methods import ( + calculate_current_layer, + calculate_max_layers, + estimate_print_time, +) from lib.panels.widgets.basePopup import BasePopup from lib.utils.blocks_button import BlocksCustomButton from lib.utils.blocks_label import BlocksLabel @@ -57,8 +61,24 @@ class JobStatusWidget(QtWidgets.QWidget): _internal_print_status: str = "" _current_file_name: str = "" - file_metadata: dict = {} + file_metadata: dict | None = None total_layers = "?" + _print_duration: float = 0.0 + _VALID_STATES: typing.ClassVar[frozenset[str]] = frozenset({"printing", "paused"}) + _INVALID_STATES: typing.ClassVar[frozenset[str]] = frozenset( + {"cancelled", "complete", "error", "standby"} + ) + + def _post_event(self, event: QtCore.QEvent) -> None: + """Post a QEvent to the top-level window via QApplication.""" + instance = QtWidgets.QApplication.instance() + if instance: + instance.postEvent(self.window(), event) + else: + logger.error( + "QApplication.instance() is None — cannot post %s", + type(event).__name__, + ) def __init__(self, parent) -> None: super().__init__(parent) @@ -89,7 +109,6 @@ def toggle_thumbnail_expansion(self) -> None: self.printing_progress_bar.show() self.btnWidget.show() self.headerWidget.show() - self.show() def showEvent(self, a0) -> None: """Reimplemented method, handle `show` Event""" @@ -110,20 +129,16 @@ def eventFilter(self, sender_obj: QtCore.QObject, event: events.QEvent) -> bool: return super().eventFilter(sender_obj, event) def _load_thumbnails(self, *thumbnails) -> None: - """Pre-load available thumbnails for the current print object""" - self.thumbnail_graphics = list( - filter( - lambda thumb: not thumb.isNull(), - [QtGui.QPixmap(thumb) for thumb in thumbnails], - ) - ) + """Pre-load available thumbnails for the current print object.""" + self.thumbnail_graphics = [ + px for thumb in thumbnails if not (px := QtGui.QPixmap(thumb)).isNull() + ] if not self.thumbnail_graphics: logger.debug("Unable to load thumbnails, no thumbnails provided") return - self.create_thumbnail_widget() - self.thumbnail_view.installEventFilter(self) - scene = QtWidgets.QGraphicsScene() + self._ensure_thumbnail_widget() _biggest_thumb = self.thumbnail_graphics[-1] + scene = QtWidgets.QGraphicsScene() self.thumbnail_view.setSceneRect( QtCore.QRectF( self.rect().x(), @@ -132,13 +147,7 @@ def _load_thumbnails(self, *thumbnails) -> None: _biggest_thumb.height(), ) ) - scaled = QtGui.QPixmap(_biggest_thumb).scaled( - _biggest_thumb.width(), - _biggest_thumb.height(), - QtCore.Qt.AspectRatioMode.KeepAspectRatio, - QtCore.Qt.TransformationMode.SmoothTransformation, - ) - item = QtWidgets.QGraphicsPixmapItem(scaled) + item = QtWidgets.QGraphicsPixmapItem(_biggest_thumb) scene.addItem(item) self.thumbnail_view.setFrameRect( QtCore.QRect( @@ -147,9 +156,6 @@ def _load_thumbnails(self, *thumbnails) -> None: ) self.thumbnail_view.setScene(scene) self.printing_progress_bar.set_inner_pixmap(self.thumbnail_graphics[-1]) - self.printing_progress_bar.thumbnail_clicked.connect( - self.toggle_thumbnail_expansion - ) @QtCore.pyqtSlot(name="handle-cancel") def handleCancel(self) -> None: @@ -157,6 +163,10 @@ def handleCancel(self) -> None: self.cancel_print_dialog.set_message( "Are you sure you \n want to cancel \n the current print job?" ) + try: + self.cancel_print_dialog.accepted.disconnect(self.print_cancel) + except TypeError: + pass self.cancel_print_dialog.accepted.connect(self.print_cancel) self.cancel_print_dialog.open() @@ -165,34 +175,28 @@ def on_print_start(self, file: str) -> None: """Start a print job, show job status page""" self._current_file_name = file self.js_file_name_label.setText(self._current_file_name) - self.layer_display_button.setText("?") + self.layer_display_button.setText("0") self.print_time_display_button.setText("?") self.printing_progress_bar.reset() + self._print_duration = 0.0 self._internal_print_status = "printing" self.request_file_info.emit(file) self.print_start.emit(file) - print_start_event = events.PrintStart( - self._current_file_name, self.file_metadata - ) - try: - instance = QtWidgets.QApplication.instance() - if instance: - instance.postEvent(self.window(), print_start_event) - else: - raise TypeError("QApplication.instance expected non None value") - except Exception as e: - logger.debug("Unexpected error while posting print job start event: %s", e) + self._post_event(events.PrintStart(self._current_file_name, self.file_metadata)) @QtCore.pyqtSlot(dict, name="on_fileinfo") - def on_fileinfo(self, fileinfo: dict) -> None: - """Handle received file information/metadata""" - if not self.isVisible(): - return - self.total_layers = str(fileinfo.get("layer_count", "---")) - self.layer_display_button.setText("---") - self.layer_display_button.secondary_text = str(self.total_layers) - self.file_metadata = fileinfo - self._load_thumbnails(*fileinfo.get("thumbnail_images", [])) + def on_fileinfo(self, metadata: dict) -> None: + """Handle received file information/metadata. + + Loads thumbnail and layer count regardless of visibility so they + are ready when the widget is shown. + """ + layer_count = metadata.get("layer_count", -1) + self.total_layers = str(layer_count) if layer_count >= 0 else "---" + self.layer_display_button.setText("0") + self.layer_display_button.secondary_text = self.total_layers + self.file_metadata = metadata + self._load_thumbnails(*metadata.get("thumbnail_images", ())) @QtCore.pyqtSlot(name="pause_resume_print") def pause_resume_print(self) -> None: @@ -209,94 +213,88 @@ def _handle_print_state(self, state: str) -> None: """Handle print state change received from printer_status object updated """ - valid_states = {"printing", "paused"} - invalid_states = {"cancelled", "complete", "error", "standby"} lstate = state.lower() - if lstate in valid_states: + event_state = lstate + is_invalid = lstate in self._INVALID_STATES + if lstate in self._VALID_STATES: self._internal_print_status = lstate if lstate == "paused": self.pause_printing_btn.setText(" Resume") self.pause_printing_btn.setPixmap( QtGui.QPixmap(":/ui/media/btn_icons/play.svg") ) + event_state = "pause" elif lstate == "printing": self.pause_printing_btn.setText("Pause") self.pause_printing_btn.setPixmap( QtGui.QPixmap(":/ui/media/btn_icons/pause.svg") ) + event_state = "start" self.pause_printing_btn.setEnabled(True) self.request_query_print_stats.emit({"print_stats": ["filename"]}) self.call_cancel_panel.emit(False) self.show_request.emit() - lstate = "start" - elif lstate in invalid_states: - if lstate != "standby": + elif is_invalid: + if lstate == "complete": self.print_finish.emit() + self.hide_request.emit() + # Capture state before clearing so the event carries the real data. + _event_file = self._current_file_name + _event_meta = self.file_metadata + if is_invalid: self._internal_print_status = "" self._current_file_name = "" self.total_layers = "?" - self.file_metadata.clear() - self.hide_request.emit() - # if hasattr(self, "thumbnail_view"): - # getattr(self, "thumbnail_view").deleteLater() + self._print_duration = 0.0 + self.file_metadata = None # Send Event on Print state - if hasattr(events, str("Print" + lstate.capitalize())): - event_obj = getattr(events, str("Print" + lstate.capitalize())) - event = event_obj(self._current_file_name, self.file_metadata) - instance = QtWidgets.QApplication.instance() - if instance: - instance.postEvent(self.window(), event) - return - logger.error( - "QApplication.instance expected non None value,\ - Unable to post event %s", - str("Print" + lstate.capitalize()), + event_class_name = "Print" + event_state.capitalize() + if hasattr(events, event_class_name): + self._post_event( + getattr(events, event_class_name)(_event_file, _event_meta) ) @QtCore.pyqtSlot(str, dict, name="on_print_stats_update") @QtCore.pyqtSlot(str, float, name="on_print_stats_update") @QtCore.pyqtSlot(str, str, name="on_print_stats_update") def on_print_stats_update(self, field: str, value: dict | float | str) -> None: - """Processes the information that comes from the printer object "print_stats" - Displays information on the ui accordingly. + """Process updates from the ``print_stats`` printer object. Args: - field (str): The name of the updated field. - value (dict | float | str): The value for the field. + field: The name of the updated field. + value: The value for the field. """ if isinstance(value, str): if "state" in field: self._handle_print_state(value) - if "filename" in field: + elif "filename" in field: self._current_file_name = value if self.js_file_name_label.text().lower() != value.lower(): self.js_file_name_label.setText(self._current_file_name) if self.isVisible(): self.request_file_info.emit(value) - if not self.file_metadata: - return - if not self.isVisible(): - return - if isinstance(value, dict): - self.layer_fallback = False - if "total_layer" in value.keys(): - self.total_layers = value["total_layer"] + # Layer info must be processed regardless of visibility so + # Klipper's runtime values always override metadata defaults. + elif isinstance(value, dict): + if "total_layer" in value: if value["total_layer"] is not None: + self.total_layers = value["total_layer"] self.layer_display_button.secondary_text = str(self.total_layers) - else: self.total_layers = "---" - self.layer_fallback = True - if "current_layer" in value.keys(): + if "current_layer" in value: if value["current_layer"] is not None: - _current_layer = value["current_layer"] - self.layer_display_button.setText(f"{int(_current_layer)}") + self.layer_display_button.setText(f"{int(value['current_layer'])}") + self.layer_fallback = False else: self.layer_display_button.setText("---") self.layer_fallback = True elif isinstance(value, float): - if "total_duration" in field: + # print_duration tracked regardless of visibility (gates Z fallback) + if "print_duration" in field: + self._print_duration = value + if self.isVisible() and "total_duration" in field: _time = estimate_print_time(int(value)) _print_time_string = ( f"{_time[0]}Day {_time[1]}H {_time[2]}min {_time[3]} s" @@ -307,33 +305,45 @@ def on_print_stats_update(self, field: str, value: dict | float | str) -> None: @QtCore.pyqtSlot(str, list, name="on_gcode_move_update") def on_gcode_move_update(self, field: str, value: list) -> None: - """Handle gcode move""" - if not self.isVisible(): - return - if "gcode_position" in field: - if self._internal_print_status == "printing": - if self.layer_fallback: - object_height = float(self.file_metadata.get("object_height", -1.0)) - layer_height = float(self.file_metadata.get("layer_height", -1.0)) - first_layer_height = float( - self.file_metadata.get("first_layer_height", -1.0) - ) - _current_layer = calculate_current_layer( - z_position=value[2], - object_height=object_height, - layer_height=layer_height, - first_layer_height=first_layer_height, - ) + """Z-position fallback for layer count display. - total_layer = ( - (object_height) / layer_height if layer_height > 0 else -1 - ) - self.layer_display_button.secondary_text = ( - f"{int(total_layer)}" if total_layer != -1 else "---" - ) - self.layer_display_button.setText( - f"{int(_current_layer)}" if _current_layer != -1 else "---" - ) + Only runs when Klipper does not provide + ``print_stats.info.current_layer`` (``layer_fallback`` is True) + AND ``print_duration > 0``. The ``print_duration`` gate + matches Mainsail's ``getPrintCurrentLayer`` getter which + prevents layer updates during pre-print procedures (heating, + nozzle cleaning). + """ + if ( + not self.isVisible() + or "gcode_position" not in field + or self._internal_print_status != "printing" + or not self.layer_fallback + or self._print_duration <= 0 # Mainsail: skip during pre-print procedures + or len(value) <= 2 + ): + return + meta = self.file_metadata + if not meta: + return + object_height = float(meta.get("object_height", 0)) + layer_height = float(meta.get("layer_height", 0)) + first_layer_height = float(meta.get("first_layer_height", 0)) + if layer_height <= 0: + return + # Mainsail getPrintMaxLayers fallback + _max_layers = calculate_max_layers( + object_height, layer_height, first_layer_height + ) + if _max_layers > 0: + self.layer_display_button.secondary_text = str(_max_layers) + _current_layer = calculate_current_layer( + z_position=value[2], + layer_height=layer_height, + first_layer_height=first_layer_height, + max_layers=_max_layers, + ) + self.layer_display_button.setText(str(_current_layer)) @QtCore.pyqtSlot(str, float, name="virtual_sdcard_update") @QtCore.pyqtSlot(str, bool, name="virtual_sdcard_update") @@ -346,7 +356,7 @@ def virtual_sdcard_update(self, field: str, value: float | bool) -> None: """ if not self.isVisible(): return - if "progress" == field: + if field == "progress": self.printing_progress_bar.setValue(value) def _setupUI(self) -> None: @@ -488,17 +498,20 @@ def _setupUI(self) -> None: ) self.job_content_layout.addLayout(self.job_stats_display_layout) - def create_thumbnail_widget(self) -> None: - """Create thumbnail graphics view widget""" + def _ensure_thumbnail_widget(self) -> None: + """Create thumbnail graphics view widget (once).""" + if hasattr(self, "thumbnail_view"): + return self.thumbnail_view = QtWidgets.QGraphicsView() self.thumbnail_view.setMinimumSize(QtCore.QSize(48, 48)) self.thumbnail_view.setAttribute( QtCore.Qt.WidgetAttribute.WA_TranslucentBackground, True ) + self.thumbnail_view.setStyleSheet( + "QGraphicsView { background: transparent; border: none; }" + ) self.thumbnail_view.setFrameShape(QtWidgets.QFrame.Shape.NoFrame) self.thumbnail_view.setFrameShadow(QtWidgets.QFrame.Shadow.Plain) - self.thumbnail_view.setWindowFlags(QtCore.Qt.WindowType.FramelessWindowHint) - self.thumbnail_view.setObjectName("thumbnail_scene") _thumbnail_palette = QtGui.QPalette() _thumbnail_palette.setColor( QtGui.QPalette.ColorRole.Window, QtGui.QColor(0, 0, 0, 0) @@ -507,9 +520,14 @@ def create_thumbnail_widget(self) -> None: QtGui.QPalette.ColorRole.Base, QtGui.QColor(0, 0, 0, 0) ) self.thumbnail_view.setPalette(_thumbnail_palette) + self.thumbnail_view.setAutoFillBackground(False) _thumbnail_brush = QtGui.QBrush(QtGui.QColor(0, 0, 0, 0)) _thumbnail_brush.setStyle(QtCore.Qt.BrushStyle.NoBrush) self.thumbnail_view.setBackgroundBrush(_thumbnail_brush) + # Use a transparent viewport widget to prevent black background on eglfs + viewport = QtWidgets.QWidget() + viewport.setAttribute(QtCore.Qt.WidgetAttribute.WA_TranslucentBackground, True) + self.thumbnail_view.setViewport(viewport) self.thumbnail_view.setRenderHints( QtGui.QPainter.RenderHint.Antialiasing | QtGui.QPainter.RenderHint.SmoothPixmapTransform @@ -521,4 +539,8 @@ def create_thumbnail_widget(self) -> None: self.thumbnail_view.setObjectName("thumbnail_scene") self.thumbnail_view_layout = QtWidgets.QHBoxLayout(self) self.thumbnail_view_layout.addWidget(self.thumbnail_view) + self.thumbnail_view.installEventFilter(self) + self.printing_progress_bar.thumbnail_clicked.connect( + self.toggle_thumbnail_expansion + ) self.thumbnail_view.hide() diff --git a/tests/widgets/__init__.py b/tests/widgets/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/widgets/conftest.py b/tests/widgets/conftest.py new file mode 100644 index 00000000..053b1018 --- /dev/null +++ b/tests/widgets/conftest.py @@ -0,0 +1,21 @@ +"""Widget test configuration. + +Ensures ``BlocksScreen/`` is on sys.path so ``from lib.xxx`` imports +resolve correctly, and clears any empty ``lib.*`` stub packages that +the network conftest may have registered before this directory is loaded. +""" + +import sys +from pathlib import Path + +_bs_dir = Path(__file__).resolve().parent.parent.parent / "BlocksScreen" +if str(_bs_dir) not in sys.path: + sys.path.insert(0, str(_bs_dir)) + +# The network conftest registers empty namespace stubs for lib, lib.panels, +# lib.panels.widgets, and lib.utils. Clear them so the real packages +# from _bs_dir are importable. +for _pkg in ("lib", "lib.panels", "lib.panels.widgets", "lib.utils"): + mod = sys.modules.get(_pkg) + if mod is not None and not getattr(mod, "__file__", None): + del sys.modules[_pkg] diff --git a/tests/widgets/test_job_status_page_unit.py b/tests/widgets/test_job_status_page_unit.py new file mode 100644 index 00000000..bf5c4cb4 --- /dev/null +++ b/tests/widgets/test_job_status_page_unit.py @@ -0,0 +1,314 @@ +"""Unit tests for JobStatusWidget (JobStatusWidget.py) """ +import sys +import types +import pytest +from PyQt6 import QtWidgets + +# STUBS must be in sys.modules BEFORE jobStatusPage is imported so the widget +# uses lightweight stand-ins instead of the full custom classes. +def _make_stub(base): + """Return a minimal subclass of *base* usable in place of a custom widget.""" + class Stub(base): + secondary_text: str = "" + def __init__(self, *args, **kwargs): + kwargs.pop("floating", None) + super().__init__(*args, **kwargs) + def set_inner_pixmap(self, *a): pass + def set_message(self, *a): pass + def setPixmap(self, *a): pass + return Stub + + +_blocks_button = types.ModuleType("lib.utils.blocks_button") +_blocks_label = types.ModuleType("lib.utils.blocks_label") +_display_button = types.ModuleType("lib.utils.display_button") +_progress_bar = types.ModuleType("lib.utils.blocks_progressbar") +_base_popup = types.ModuleType("lib.panels.widgets.basePopup") + +_blocks_button.BlocksCustomButton = _make_stub(QtWidgets.QPushButton) +_blocks_label.BlocksLabel = _make_stub(QtWidgets.QLabel) +_display_button.DisplayButton = _make_stub(QtWidgets.QPushButton) +_progress_bar.CustomProgressBar = _make_stub(QtWidgets.QProgressBar) +_base_popup.BasePopup = _make_stub(QtWidgets.QDialog) + +for _name, _mod in [ + ("lib.utils.blocks_button", _blocks_button), + ("lib.utils.blocks_label", _blocks_label), + ("lib.utils.display_button", _display_button), + ("lib.utils.blocks_progressbar", _progress_bar), + ("lib.panels.widgets.basePopup", _base_popup), +]: + sys.modules[_name] = _mod # force-set so network conftest stubs don't win + +import events # noqa: F401, E402 # ensure events is importable before jobStatusPage loads +from lib.panels.widgets.jobStatusPage import JobStatusWidget # noqa: E402 + +@pytest.fixture() +def widget(qtbot): + """Create a JobStatusWidget with all state initialised. """ + w = JobStatusWidget(parent=None) + # initialise state that slots depend on + w._current_file_name = "" + w._print_duration = 0.0 + w._internal_print_status = "" + w.file_metadata = None + w.total_layers = "?" + qtbot.addWidget(w) + return w + +class TestOnPrintStart: + """ on_print_start sets state and emits signals""" + + def test_sets_current_file_name(self, widget): + widget.on_print_start("test.gcode") + assert widget._current_file_name == "test.gcode" + + def test_resets_print_duration(self, widget): + widget._print_duration = 99.9 + widget.on_print_start("test.gcode") + assert widget._print_duration == 0.0 + + def test_sets_status_to_printing(self, widget): + widget.on_print_start("test.gcode") + assert widget._internal_print_status == "printing" + + def test_emits_print_start_signal(self, widget, qtbot): + with qtbot.waitSignal(widget.print_start, timeout=500) as sig: + widget.on_print_start("my_file.gcode") + assert sig.args == ["my_file.gcode"] + +class TestHandlePrintState: + """ _handle_print_state drives the UI and signals correctly. """ + def test_printing_emits_show_request(self, widget, qtbot): + with qtbot.waitSignal(widget.show_request, timeout = 500): + widget._handle_print_state("printing") + + def test_paused_emits_show_request(self, widget, qtbot): + with qtbot.waitSignal(widget.show_request, timeout = 500): + widget._handle_print_state("paused") + + def test_complete_emits_print_finish(self, widget, qtbot): + with qtbot.waitSignal(widget.print_finish, timeout = 500): + widget._handle_print_state("complete") + + def test_complete_emits_hide_request(self, widget, qtbot): + with qtbot.waitSignal(widget.hide_request, timeout = 500): + widget._handle_print_state("complete") + + def test_canceller_does_not_emit_print_finish(self, widget, qtbot): + with qtbot.assertNotEmitted(widget.print_finish): + widget._handle_print_state("cancelled") + + def test_invalid_state_clears_metadata(self, widget): + widget.file_metadata = {"layer_height": 0.2} + widget._handle_print_state("complete") + assert widget.file_metadata is None + + def test_invalid_state_clears_filename(self, widget): + widget._current_file_name = "print.gcode" + widget._handle_print_state("error") + assert widget._current_file_name == "" + +class TestOnPrintStatsUpdate: + """ on_print_stats_update routes fields to the right state.""" + def test_state_field_triggers_handle_print_state(self, widget, qtbot): + with qtbot.waitSignal(widget.show_request, timeout=500): + widget.on_print_stats_update("state", "printing") + + def test_filename_field_updates_current_file(self, widget): + widget.on_print_stats_update("filename", "cube.gcode") + assert widget._current_file_name == "cube.gcode" + + def test_print_duration_stored_regardless_of_visibility(self, widget): + widget.hide() + widget.on_print_stats_update("print_duration", 42.5) + assert widget._print_duration == 42.5 + + def test_current_layer_not_none_disables_fallback(self, widget): + widget.on_print_stats_update("info", {"current_layer": 5}) + assert widget.layer_fallback is False + + def test_current_layer_none_enables_fallback(self, widget): + widget.on_print_stats_update("info", {"current_layer": None}) + assert widget.layer_fallback is True + + def test_total_layer_value_stored(self, widget): + widget.on_print_stats_update("info", {"total_layer": 120}) + assert widget.total_layers == 120 + +class TestOnGcodeMoveUpdate: + """on_gcode_move_update computes layer from Z position.""" + + def _ready_widget(self, widget): + """Put widget in the state where gcode_move_update should fire.""" + widget.show() + widget._internal_print_status = "printing" + widget.layer_fallback = True + widget._print_duration = 10.0 + widget.layer_display_button.setText("sentinel") + widget.file_metadata = { + "object_height": 10.0, + "layer_height": 0.2, + "first_layer_height": 0.2, + } + + def test_no_update_when_hidden(self, widget): + self._ready_widget(widget) + widget.hide() + widget.on_gcode_move_update("gcode_position", [0, 0, 1.0, 0]) + assert widget.layer_display_button.text() == "sentinel" + + def test_no_update_wrong_field(self, widget): + self._ready_widget(widget) + widget.on_gcode_move_update("position", [0, 0, 1.0, 0]) + assert widget.layer_display_button.text() == "sentinel" + + def test_no_update_when_not_printing(self, widget): + self._ready_widget(widget) + widget._internal_print_status = "paused" + widget.on_gcode_move_update("gcode_position", [0, 0, 1.0, 0]) + assert widget.layer_display_button.text() == "sentinel" + + def test_no_update_when_duration_zero(self, widget): + self._ready_widget(widget) + widget._print_duration = 0.0 + widget.on_gcode_move_update("gcode_position", [0, 0, 1.0, 0]) + assert widget.layer_display_button.text() == "sentinel" + + def test_calculates_layer_from_z(self, widget): + self._ready_widget(widget) + # z=0.4, layer_height=0.2, first=0.2 -> ceil ((0.4-0.2)/ 0.2 + 1) = 2 + widget.on_gcode_move_update("gcode_position", [0, 0, 0.4, 0]) + assert widget.layer_display_button.text() == "2" + + +class TestVirtualSdcardUpdate: + """virtual_sdcard_update sets progress bard, guarded by visibility.""" + + def test_no_update_when_hidden(self, widget): + from unittest.mock import patch + widget.hide() + with patch.object(widget.printing_progress_bar, 'setValue') as mock_test: + widget.virtual_sdcard_update("progress", 50) + mock_test.assert_not_called() + + def test_progress_field_sets_value(self, widget): + from unittest.mock import patch + widget.show() + with patch.object(widget.printing_progress_bar, 'setValue') as mock_test: + widget.virtual_sdcard_update("progress", 75) + mock_test.assert_called_once_with(75) + + def test_non_progress_field_ignored(self, widget): + from unittest.mock import patch + widget.hide() + with patch.object(widget.printing_progress_bar, 'setValue') as mock_test: + widget.virtual_sdcard_update("is_active", True) + mock_test.assert_not_called() + +class TestPauseResumePrint: + """ pause_resume_print toggles state and emits the right signal.""" + + def test_printing_transitions_status(self, widget): + widget._internal_print_status = "printing" + widget.pause_resume_print() + assert widget._internal_print_status == "paused" + + def test_printing_emits_print_pause(self, widget, qtbot): + widget._internal_print_status = "printing" + with qtbot.waitSignal(widget.print_pause, timeout=500): + widget.pause_resume_print() + + def test_paused_transitions_to_printing(self, widget, qtbot): + widget._internal_print_status = "paused" + with qtbot.waitSignal(widget.print_resume, timeout=500): + widget.pause_resume_print() + + def test_paused_emits_print_resume(self, widget): + widget._internal_print_status = "paused" + widget.pause_resume_print() + assert widget._internal_print_status == "printing" + + def test_disables_pause_button(self, widget): + widget._internal_print_status = "printing" + widget.pause_printing_btn.setEnabled(True) + widget.pause_resume_print() + assert not widget.pause_printing_btn.isEnabled() + + def test_unknown_state_emits_nothing(self, widget, qtbot): + widget._internal_print_status = "idle" + with qtbot.assertNotEmitted(widget.print_pause): + with qtbot.assertNotEmitted(widget.print_resume): + widget.pause_resume_print() + +class TestHandleCancel: + """handleCancel wires the cancel dialog exactly once.""" + def test_sets_cancel_message(self, widget): + from unittest.mock import patch + with patch.object(widget.cancel_print_dialog, "set_message") as m: + widget.handleCancel() + m.assert_called_once() + assert "cancel" in m.call_args[0][0].lower() + + def test_opens_dialog(self, widget): + from unittest.mock import patch + with patch.object(widget.cancel_print_dialog, "open") as m: + widget.handleCancel() + m.assert_called_once() + + def test_accepted_triggers_print_cancel(self, widget, qtbot): + widget.handleCancel() + with qtbot.waitSignal(widget.print_cancel, timeout=500): + widget.cancel_print_dialog.accepted.emit() + + def test_double_call_connects_only_once(self, widget): + widget.handleCancel() + widget.handleCancel() + emissions: list[int] = [] + widget.print_cancel.connect(lambda: emissions.append(1)) + widget.cancel_print_dialog.accepted.emit() + assert len(emissions) == 1 + +class TestOnFileInfo: + """on_fileinfo loads the thumbnail and layer count regardless of visibility""" + def _ready_widget(self, widget) -> dict: + """Put widget in the state where gcode_move_update should fire.""" + widget.show() + widget._internal_print_status = "printing" + widget.layer_fallback = True + widget._print_duration = 10.0 + return { + "layer_count": 20, + "object_height": 10.0, + "layer_height": 0.2, + "first_layer_height": 0.2, + "thumbnail_images": [], + } + + + def test_load_correct_info(self, widget): + _metadata = self._ready_widget(widget) + widget.on_fileinfo(_metadata) + assert widget.total_layers == "20" + + def test_handle_error_total_layers(self, widget): + _metadata = self._ready_widget(widget) + del _metadata['layer_count'] + widget.on_fileinfo(_metadata) + assert widget.total_layers == "---" + + def test_metadata_stored(self, widget): + _metadata = self._ready_widget(widget) + widget.on_fileinfo(_metadata) + assert widget.file_metadata is _metadata + + def test_layer_display_reset_to_zero(self, widget): + widget.layer_display_button.setText("99") + _metadata = self._ready_widget(widget) + widget.on_fileinfo(_metadata) + assert widget.layer_display_button.text() == "0" + + def test_secondary_text_set_to_total_layers(self, widget): + _metadata = self._ready_widget(widget) + widget.on_fileinfo(_metadata) + assert widget.layer_display_button.secondary_text == "20"