diff --git a/BlocksScreen/devices/storage/udisks2.py b/BlocksScreen/devices/storage/udisks2.py index d3c19a61..11dd3d02 100644 --- a/BlocksScreen/devices/storage/udisks2.py +++ b/BlocksScreen/devices/storage/udisks2.py @@ -4,10 +4,11 @@ import pathlib import shutil import typing -from collections.abc import Coroutine import unicodedata +from collections.abc import Coroutine import sdbus +from helper_methods import USB_FALLBACK_NAME, USB_LABEL_PREFIX, USB_LINK_PREFIXES from PyQt6 import QtCore from .device import Device @@ -54,7 +55,6 @@ def validate_label(label: str, strict: bool = True, max_length: int = 100) -> st dangerous_chars = { "\0", - "\x00", "/", "\\", ";", @@ -430,7 +430,10 @@ async def _rem_interface_listener(self) -> None: device: Device = self.controlled_devs.pop(path) device.kill() del device + # Reap first: a refresh now would still list the link. + self._cleanup_broken_symlinks() self.hardware_removed[str].emit(path) + continue self._cleanup_broken_symlinks() except sdbus.dbus_exceptions.DbusUnknownMethodError as e: logging.error( @@ -456,13 +459,22 @@ def mount(self, device: Device, label: str = ""): """Mounts the devices mountpoints""" for path, filesystem in device.file_systems.items(): _ = fire_n_forget( - coro=self._mount_filesystem(filesystem, label), + coro=self._mount_filesystem(filesystem, label, path), name=f"Mount-filesystem-{path}", task_stack=self.task_stack, ) + def _announce_mount(self, dev_path: str, symlink: str) -> str: + """Announce a new USB folder to the files view.""" + if symlink: + self.device_mounted[str, str].emit(dev_path, symlink) + return symlink + async def _mount_filesystem( - self, filesystem: UDisks2FileSystemAsyncInterface, label: str = "" + self, + filesystem: UDisks2FileSystemAsyncInterface, + label: str = "", + dev_path: str = "", ) -> str: val_label: str = validate_label(label) try: @@ -473,8 +485,13 @@ async def _mount_filesystem( "options": ("s", "rw,relatime,sync"), } mnt_path: str = await filesystem.mount(opts) - return self.add_symlink( - path=mnt_path, label=val_label, dst_path=self.gcodes_path.as_posix() + return self._announce_mount( + dev_path, + self.add_symlink( + path=mnt_path, + label=val_label, + dst_path=self.gcodes_path.as_posix(), + ), ) except sdbus.SdBusUnmappedMessageError as e: if AlreadyMountedException in e.args[0]: @@ -486,12 +503,15 @@ async def _mount_filesystem( if not mount_points: return "" mpoint: str = mount_points[0].decode("utf-8").strip("\x00") - if os.path.exists(mpoint): + if not os.path.exists(mpoint): return "" - return self.add_symlink( - path=mpoint, - dst_path=self.gcodes_path.as_posix(), - label=val_label, + return self._announce_mount( + dev_path, + self.add_symlink( + path=mpoint, + dst_path=self.gcodes_path.as_posix(), + label=val_label, + ), ) except Exception as e: logging.error( @@ -526,8 +546,8 @@ def add_symlink( """ if not _validated and label: label = validate_label(label, strict=True) - label = "USB-" + label - fallback: str = "USB DRIVE" if _index == 0 else str(f"USB DRIVE {_index}") + label = USB_LABEL_PREFIX + label + fallback = USB_FALLBACK_NAME if _index == 0 else f"{USB_FALLBACK_NAME} {_index}" dstb = pathlib.Path(dst_path).joinpath(label if label else fallback) try: if not os.path.islink(dstb): @@ -583,10 +603,22 @@ def _cleanup_symlinks(self) -> None: if os.path.islink(dir): _ = self.rem_symlink(dir.as_posix()) + def _is_symlink_live(self, link: pathlib.Path) -> bool: + """Live only while the target is still a mountpoint.""" + if not os.path.exists(link): + return False + # Never reap links we did not create. + if not link.name.startswith(USB_LINK_PREFIXES): + return True + return os.path.ismount(os.path.realpath(link)) + def _cleanup_broken_symlinks(self) -> None: + """Remove dead USB symlinks and announce each removal.""" for dir in self.gcodes_path.rglob("*"): - if os.path.islink(dir) and not os.path.exists(dir): - _ = self.rem_symlink(dir) + if not os.path.islink(dir) or self._is_symlink_live(dir): + continue + if self.rem_symlink(dir): + self.device_unmounted[str].emit(dir.as_posix()) def _resolve_symlinks( self, path: str | pathlib.Path, mount_path: str | pathlib.Path diff --git a/BlocksScreen/helper_methods.py b/BlocksScreen/helper_methods.py index 114ff5f1..390e24ea 100644 --- a/BlocksScreen/helper_methods.py +++ b/BlocksScreen/helper_methods.py @@ -16,6 +16,11 @@ logger = logging.getLogger(__name__) +# Shared by udisks2.add_symlink and the files UI USB checks. +USB_LABEL_PREFIX = "USB-" +USB_FALLBACK_NAME = "USB DRIVE" +USB_LINK_PREFIXES: tuple[str, ...] = (USB_LABEL_PREFIX, USB_FALLBACK_NAME) + try: ctypes.cdll.LoadLibrary("libXext.so.6") libxext = ctypes.CDLL("libXext.so.6") @@ -225,7 +230,7 @@ def disable_dpms() -> None: logger.exception(f"Unexpected exception occurred {e}") -def convert_bytes_to_mb(size_bytes: int | float) -> float: +def convert_bytes_to_mb(size_bytes: float) -> float: """Converts byte size to megabyte size. Args: @@ -262,7 +267,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( @@ -294,6 +299,23 @@ def estimate_print_time(seconds: int) -> list[int]: return [days, hours, mins, secs] +def format_duration(seconds: int) -> str: + """Seconds as "1d 2h 3m", or "Ns" under a minute.""" + if seconds < 60: + return f"{seconds}s" + days, hours, mins, _ = estimate_print_time(seconds) + if days > 0: + return f"{days}d {hours}h {mins}m" + if hours > 0: + return f"{hours}h {mins}m" + return f"{mins}m" + + +def format_weight(grams: float) -> str: + """Grams as "g", or "kg" past 499g.""" + return f"{grams / 1000:.2f}kg" if grams > 499 else f"{grams:.2f}g" + + def normalize( value: float, r_min: float = 0.0, @@ -344,3 +366,28 @@ def get_file_name(filename: str | None) -> str: if not filename: return "" return pathlib.PurePosixPath(filename.replace("\\", "/")).name + + +def get_parent_dir(path: str) -> str: + """POSIX parent of *path*, "" at the gcodes root.""" + parent = pathlib.PurePosixPath(path.removeprefix("/")).parent + return "" if str(parent) == "." else str(parent) + + +def is_usb_mount(path: str) -> bool: + """True for a top-level USB symlink name.""" + name = path.strip("/") + return bool(name) and "/" not in name and name.startswith(USB_LINK_PREFIXES) + + +def is_usb_path(path: str) -> bool: + """True for a USB mount or anything under one.""" + return is_usb_mount(path.strip("/").split("/", 1)[0]) + + +def resolve_thumbnail_path( + gcode_root: pathlib.Path, requested_path: str, relative_path: str +) -> pathlib.Path: + """Thumbnail path resolved against the gcode's own directory.""" + parent = pathlib.PurePosixPath(requested_path.removeprefix("/")).parent + return gcode_root / parent / relative_path diff --git a/BlocksScreen/lib/files.py b/BlocksScreen/lib/files.py index 710991a9..916b7afa 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -3,18 +3,20 @@ import logging import typing from collections import deque -from dataclasses import dataclass, field -from enum import Enum, auto +from dataclasses import asdict, dataclass, field, replace +from enum import StrEnum, auto from pathlib import Path import events +import helper_methods from lib.moonrakerComm import MoonWebSocket -from PyQt6 import QtCore, QtGui, QtWidgets +from lib.utils import gcode_loader +from PyQt6 import QtCore, QtWidgets logger = logging.getLogger(__name__) -class FileAction(Enum): +class FileAction(StrEnum): """Enumeration of possible file actions from Moonraker notifications.""" CREATE_FILE = auto() @@ -30,29 +32,18 @@ class FileAction(Enum): @classmethod def from_string(cls, action: str) -> FileAction: """Convert Moonraker action string to enum.""" - mapping = { - "create_file": cls.CREATE_FILE, - "delete_file": cls.DELETE_FILE, - "move_file": cls.MOVE_FILE, - "modify_file": cls.MODIFY_FILE, - "create_dir": cls.CREATE_DIR, - "delete_dir": cls.DELETE_DIR, - "move_dir": cls.MOVE_DIR, - "root_update": cls.ROOT_UPDATE, - } - return mapping.get(action.lower(), cls.UNKNOWN) + try: + return cls(action.lower()) + except ValueError: + return cls.UNKNOWN -@dataclass +@dataclass(frozen=True, slots=True) class FileMetadata: - """ - Data class for file metadata. - - Thumbnails are stored as QImage objects when available. - """ + """Gcode file metadata; thumbnails as filesystem paths.""" filename: str = "" - thumbnail_images: list[QtGui.QImage] = field(default_factory=list) + thumbnail_paths: list[str] = field(default_factory=list) filament_total: dict | str | float = field(default_factory=dict) estimated_time: int = 0 layer_count: int = -1 @@ -64,9 +55,8 @@ class FileMetadata: filament_weight_total: float = -1.0 layer_height: float = -1.0 first_layer_height: float = -1.0 - first_layer_extruder_temp: float = -1.0 + first_layer_extr_temp: float = -1.0 first_layer_bed_temp: float = -1.0 - chamber_temp: float = -1.0 filament_name: str = "Unknown" nozzle_diameter: float = -1.0 slicer: str = "Unknown" @@ -75,45 +65,15 @@ class FileMetadata: gcode_end_byte: int = 0 print_start_time: float | None = None job_id: str | None = None + print_duration: float | None = None def to_dict(self) -> dict: - """Convert to dictionary for signal emission.""" - return { - "filename": self.filename, - "thumbnail_images": self.thumbnail_images, - "filament_total": self.filament_total, - "estimated_time": self.estimated_time, - "layer_count": self.layer_count, - "total_layer": self.total_layer, - "object_height": self.object_height, - "size": self.size, - "modified": self.modified, - "filament_type": self.filament_type, - "filament_weight_total": self.filament_weight_total, - "layer_height": self.layer_height, - "first_layer_height": self.first_layer_height, - "first_layer_extruder_temp": self.first_layer_extruder_temp, - "first_layer_bed_temp": self.first_layer_bed_temp, - "chamber_temp": self.chamber_temp, - "filament_name": self.filament_name, - "nozzle_diameter": self.nozzle_diameter, - "slicer": self.slicer, - "slicer_version": self.slicer_version, - "gcode_start_byte": self.gcode_start_byte, - "gcode_end_byte": self.gcode_end_byte, - "print_start_time": self.print_start_time, - "job_id": self.job_id, - } + """Plain dict for signals, containers deep-copied.""" + return asdict(self) @classmethod - def from_dict( - cls, data: dict, thumbnail_images: list[QtGui.QImage] - ) -> FileMetadata: - """ - `Create FileMetadata from Moonraker API response.` - - All data comes directly from Moonraker - no local filesystem access. - """ + def from_dict(cls, data: dict, thumbnail_paths: list[str]) -> FileMetadata: + """Create FileMetadata from Moonraker API response.""" filename = data.get("filename", "") # Helper to safely get values with fallback @@ -125,7 +85,7 @@ def safe_get(key: str, default: typing.Any) -> typing.Any: return cls( filename=filename, - thumbnail_images=thumbnail_images, + thumbnail_paths=thumbnail_paths, filament_total=safe_get("filament_total", {}), estimated_time=int(safe_get("estimated_time", 0)), layer_count=safe_get("layer_count", -1), @@ -137,9 +97,8 @@ def safe_get(key: str, default: typing.Any) -> typing.Any: filament_weight_total=safe_get("filament_weight_total", -1.0), layer_height=safe_get("layer_height", -1.0), first_layer_height=safe_get("first_layer_height", -1.0), - first_layer_extruder_temp=safe_get("first_layer_extruder_temp", -1.0), + first_layer_extr_temp=safe_get("first_layer_extr_temp", -1.0), first_layer_bed_temp=safe_get("first_layer_bed_temp", -1.0), - chamber_temp=safe_get("chamber_temp", -1.0), filament_name=safe_get("filament_name", "Unknown") or "Unknown", nozzle_diameter=safe_get("nozzle_diameter", -1.0), slicer=safe_get("slicer", "Unknown") or "Unknown", @@ -148,28 +107,19 @@ def safe_get(key: str, default: typing.Any) -> typing.Any: gcode_end_byte=safe_get("gcode_end_byte", 0), print_start_time=data.get("print_start_time"), job_id=data.get("job_id"), + print_duration=data.get("print_duration"), ) class Files(QtCore.QObject): - """ - Manages gcode files with event-driven updates. - E - Signals emitted: - - on_dirs: Full directory list - - on_file_list: Full file list - - fileinfo: Single file metadata update - - file_added/removed/modified: Incremental updates - - dir_added/removed: Directory updates - - full_refresh_needed: Root changed - """ + """Gcode file and dir state, synced from Moonraker notifications.""" # Signals for API requests - request_file_list = QtCore.pyqtSignal([], [str], name="api_get_files_list") request_dir_info = QtCore.pyqtSignal( [], [str], [str, bool], name="api_get_dir_info" ) request_file_metadata = QtCore.pyqtSignal(str, name="get_file_metadata") + request_scan_metadata = QtCore.pyqtSignal(str, name="scan_file_metadata") # Signals for UI updates on_dirs = QtCore.pyqtSignal(list, name="on_dirs") @@ -186,6 +136,8 @@ class Files(QtCore.QObject): dir_added = QtCore.pyqtSignal(dict, name="dir_added") dir_removed = QtCore.pyqtSignal(str, name="dir_removed") full_refresh_needed = QtCore.pyqtSignal(name="full_refresh_needed") + # Hops history replies from the websocket thread to the Qt thread. + _history_job = QtCore.pyqtSignal(str, str, dict, name="history_job") # Signal for preloaded USB files usb_files_loaded = QtCore.pyqtSignal( @@ -202,6 +154,7 @@ def __init__(self, parent: QtCore.QObject, ws: MoonWebSocket) -> None: self._files: dict[str, dict] = {} self._directories: dict[str, dict] = {} self._files_metadata: dict[str, FileMetadata] = {} + self._metadata_retry_count: dict[str, int] = {} self._current_directory: str = "" self._initial_load_complete: bool = False self.gcode_path = Path(self.GCODE_PATH).expanduser() @@ -210,18 +163,23 @@ def __init__(self, parent: QtCore.QObject, ws: MoonWebSocket) -> None: # Track pending USB preload requests (ordered FIFO queue) self._pending_usb_preloads: set[str] = set() self._usb_preload_queue: deque[str] = deque() + # USB metadata: per-path size/modified, plus the lazy loader. + self._usb_meta_base: dict[str, dict] = {} + self._meta_loader: gcode_loader.GcodeMetadataLoader | None = None + # job_id -> completed duration; None = asked or unusable, never re-ask. + self._job_durations: dict[str, float | None] = {} self._connect_signals() self._install_event_filter() def _connect_signals(self) -> None: """Connect internal signals to websocket API.""" - self.request_file_list.connect(self.ws.api.get_file_list) - self.request_file_list[str].connect(self.ws.api.get_file_list) self.request_dir_info.connect(self.ws.api.get_dir_information) self.request_dir_info[str, bool].connect(self.ws.api.get_dir_information) self.request_dir_info[str].connect(self.ws.api.get_dir_information) self.request_file_metadata.connect(self.ws.api.get_gcode_metadata) + self.request_scan_metadata.connect(self.ws.api.scan_gcode_metadata) + self._history_job.connect(self._on_history_job) def _install_event_filter(self) -> None: """Install event filter on application instance.""" @@ -249,29 +207,6 @@ def current_directory(self, value: str) -> None: """Set current directory path.""" self._current_directory = value - @property - def is_loaded(self) -> bool: - """Check if initial load is complete.""" - return self._initial_load_complete - - def get_file_metadata(self, filename: str) -> FileMetadata | None: - """Get cached metadata for a file.""" - return self._files_metadata.get(filename.removeprefix("/")) - - def get_file_data(self, filename: str) -> dict: - """Get cached file data dict for a file.""" - clean_name = filename.removeprefix("/") - metadata = self._files_metadata.get(clean_name) - if metadata: - return metadata.to_dict() - return {} - - def refresh_directory(self, directory: str = "") -> None: - """Force refresh of a specific directory.""" - logger.debug(f"Refreshing directory: {directory or 'root'}") - self._current_directory = directory - self.request_dir_info[str, bool].emit(directory, True) - def initial_load(self) -> None: """Perform initial load of file list.""" logger.info("Performing initial file list load") @@ -279,23 +214,22 @@ def initial_load(self) -> None: self.request_dir_info[str, bool].emit("", True) def handle_filelist_changed(self, data: dict | list) -> None: - """Handle notify_filelist_changed from Moonraker.""" + """Handle notify_filelist_changed; params may batch entries.""" if isinstance(data, dict) and "params" in data: data = data.get("params", []) + entries = data if isinstance(data, list) else [data] + for entry in entries: + if isinstance(entry, dict): + self._apply_filelist_change(entry) - if isinstance(data, list): - if len(data) > 0: - data = data[0] - else: - return - - if not isinstance(data, dict): - return - + def _apply_filelist_change(self, data: dict) -> None: + """Route one filelist entry to its handler.""" action_str = data.get("action", "") action = FileAction.from_string(action_str) item = data.get("item", {}) source_item = data.get("source_item", {}) + if not (self._in_gcodes(item) or self._in_gcodes(source_item)): + return logger.debug(f"File list changed: action={action_str}, item={item}") @@ -314,13 +248,18 @@ def handle_filelist_changed(self, data: dict | list) -> None: if handler: handler(item, source_item) + @staticmethod + def _in_gcodes(item: dict) -> bool: + """True for a gcodes-root item; config/logs roots notify too.""" + return bool(item) and item.get("root", "gcodes") == "gcodes" + def _handle_file_created(self, item: dict, _: dict) -> None: """Handle new file creation.""" path = item.get("path", "") if not path: return - if self._is_usb_mount(path): + if helper_methods.is_usb_mount(path): item["dirname"] = path self._handle_dir_created(item, {}) return @@ -332,7 +271,7 @@ def _handle_file_created(self, item: dict, _: dict) -> None: self.file_added.emit(item) # Request metadata (will update later) - self.request_file_metadata.emit(path.removeprefix("/")) + self._request_gcode_metadata(path.removeprefix("/"), item) logger.info(f"File created: {path}") def _handle_file_deleted(self, item: dict, _: dict) -> None: @@ -341,13 +280,13 @@ def _handle_file_deleted(self, item: dict, _: dict) -> None: if not path: return - if self._is_usb_mount(path): + if helper_methods.is_usb_mount(path): item["dirname"] = path self._handle_dir_deleted(item, {}) return self._files.pop(path, None) - self._files_metadata.pop(path.removeprefix("/"), None) + self._forget_cached(path.removeprefix("/")) self.file_removed.emit(path) logger.info(f"File deleted: {path}") @@ -355,30 +294,38 @@ def _handle_file_deleted(self, item: dict, _: dict) -> None: def _handle_file_modified(self, item: dict, _: dict) -> None: """Handle file modification.""" path = item.get("path", "") - if not path or not path.lower().endswith(self.GCODE_EXTENSION): + if not path: return - self._files[path] = item - self._files_metadata.pop(path.removeprefix("/"), None) + # Moonraker reports a root USB symlink as a file event. + if helper_methods.is_usb_mount(path): + item["dirname"] = path + self._handle_dir_created(item, {}) + return + + if not path.lower().endswith(self.GCODE_EXTENSION): + return - self.request_file_metadata.emit(path.removeprefix("/")) + self._files[path] = item + # A same-name re-upload must not show the old parse. + self._forget_cached(path.removeprefix("/")) + # Before the request: the page drops its copy on this signal. self.file_modified.emit(item) + self._request_gcode_metadata(path.removeprefix("/"), item) logger.info(f"File modified: {path}") def _handle_file_moved(self, item: dict, source_item: dict) -> None: """Handle file move/rename.""" - old_path = source_item.get("path", "") - new_path = item.get("path", "") - - if old_path: + # A cross-root move is only a delete or a create here. + if self._in_gcodes(source_item): self._handle_file_deleted(source_item, {}) - if new_path: + if self._in_gcodes(item): self._handle_file_created(item, {}) def _handle_dir_created(self, item: dict, _: dict) -> None: """Handle directory creation.""" path = item.get("path", "") - dirname = item.get("dirname", "") + dirname = item.get("dirname", "").strip("/") if not dirname and path: dirname = path.rstrip("/").split("/")[-1] @@ -391,13 +338,14 @@ def _handle_dir_created(self, item: dict, _: dict) -> None: self.dir_added.emit(item) logger.info(f"Directory created: {dirname}") - if self._is_usb_mount(dirname): + # Full path: a subdir folder named USB-* is not a mount. + if helper_methods.is_usb_mount(path or dirname): self._preload_usb_contents(dirname) def _handle_dir_deleted(self, item: dict, _: dict) -> None: - """Handle directory deletion.""" - path = item.get("path", "") - dirname = item.get("dirname", "") + """Handle a deleted dir; emits its gcodes-relative path.""" + path = item.get("path", "").strip("/") + dirname = item.get("dirname", "").strip("/") if not dirname and path: dirname = path.rstrip("/").split("/")[-1] @@ -406,22 +354,37 @@ def _handle_dir_deleted(self, item: dict, _: dict) -> None: return self._directories.pop(dirname, None) + # A reinserted drive or recreated dir may hold different files. + self._forget_cached(path or dirname) # Clear USB cache if this was a USB mount - if self._is_usb_mount(dirname): + if helper_methods.is_usb_mount(path or dirname): self._usb_files_cache.pop(dirname, None) self._pending_usb_preloads.discard(dirname) if dirname in self._usb_preload_queue: self._usb_preload_queue.remove(dirname) logger.info(f"Cleared USB cache for: {dirname}") - self.dir_removed.emit(dirname) - logger.info(f"Directory deleted: {dirname}") + # Full path, so sub/x is not mistaken for a root x. + self.dir_removed.emit(path or dirname) + logger.info("Directory deleted: %s", path or dirname) + + def _forget_cached(self, path: str) -> None: + """Drop cached metadata and loader payloads at or under *path*.""" + root = f"{path}/" + for cache in (self._files_metadata, self._usb_meta_base): + for key in [k for k in cache if f"{k}/".startswith(root)]: + del cache[key] + for loader in (gcode_loader.get_loader(), gcode_loader.get_metadata_loader()): + if loader is not None: + loader.forget(path) def _handle_dir_moved(self, item: dict, source_item: dict) -> None: """Handle directory move/rename.""" - self._handle_dir_deleted(source_item, {}) - self._handle_dir_created(item, {}) + if self._in_gcodes(source_item): + self._handle_dir_deleted(source_item, {}) + if self._in_gcodes(item): + self._handle_dir_created(item, {}) def _handle_root_update(self, _: dict, __: dict) -> None: """Handle root update.""" @@ -429,78 +392,151 @@ def _handle_root_update(self, _: dict, __: dict) -> None: self.full_refresh_needed.emit() self.initial_load() - @staticmethod - def _is_usb_mount(path: str) -> bool: - """Check if a path is a USB mount point.""" - path = path.removeprefix("/") - return "/" not in path and path.startswith("USB-") - def handle_message_received( self, method: str, data: typing.Any, params: dict ) -> None: """Handle file-related messages received from Moonraker.""" - if "server.files.list" in method: - self._process_file_list(data) - elif "server.files.metadata" in method: + if "server.files.metadata" in method: self._process_metadata(data) elif "server.files.get_directory" in method: - self._process_directory_info(data) - - def _process_file_list(self, data: list) -> None: - """Process full file list response.""" - self._files.clear() - - for item in data: - path = item.get("path", item.get("filename", "")) - if path: - self._files[path] = item - - self._initial_load_complete = True - self.on_file_list.emit(self.file_list) - logger.info(f"Loaded {len(self._files)} files") - # Request metadata only for gcode files (async update) - for path in self._files: - if path.lower().endswith(self.GCODE_EXTENSION): - self.request_file_metadata.emit(path.removeprefix("/")) - - def _process_metadata(self, data: dict) -> None: - """Process file metadata response.""" - filename = data.get("filename") + requested_dir = self._requested_dir_from_params(params) + self._process_directory_info(data, requested_dir) + + def _requested_dir_from_params(self, params: typing.Any) -> str: + """Dir asked for by a [method, params, callback] entry.""" + try: + path = params[1].get("path", "") + except (IndexError, TypeError, AttributeError): + return "" + return path.removeprefix("gcodes/").strip("/") + + def _full_gcode_path(self, filename: str, directory: str) -> str: + """Gcodes-relative path of a bare listing filename.""" + bare = filename.removeprefix("/") + parent = directory.removeprefix("/").strip("/") + return f"{parent}/{bare}" if parent else bare + + def _process_metadata(self, data: dict, full_path: str | None = None) -> None: + """Build FileMetadata and emit fileinfo.""" + if full_path: + data = data | {"filename": full_path} + filename = data.get("filename") or data.get("path") if not filename: return - - thumbnails = data.get("thumbnails", []) - base_dir = (self.gcode_path / filename).parent + thumbs = [ + t + for t in data.get("thumbnails") or [] + if isinstance(t, dict) and isinstance(t.get("relative_path"), str) + ] + # Consumers take [-1] as largest, as in KlipperScreen. + thumbs.sort(key=lambda t: t.get("size") or 0) thumbnail_paths = [ - str(base_dir / t.get("relative_path", "")) - for t in thumbnails - if isinstance(t.get("relative_path", None), str) and t["relative_path"] + str( + helper_methods.resolve_thumbnail_path( + self.gcode_path, filename, t["relative_path"] + ) + ) + for t in thumbs + if t["relative_path"] ] - - # Load images, filtering out invalid files - thumbnail_images = [] - for path in thumbnail_paths: - image = QtGui.QImage(path) - if not image.isNull(): # skip loading errors - thumbnail_images.append(image) - - metadata = FileMetadata.from_dict(data, thumbnail_images) + metadata = FileMetadata.from_dict(data, thumbnail_paths) + duration = self._job_durations.get(str(metadata.job_id)) + if metadata.print_duration is None and duration is not None: + metadata = replace(metadata, print_duration=duration) self._files_metadata[filename] = metadata - - # Emit updated fileinfo + self._metadata_retry_count.pop(filename.removeprefix("/"), None) self.fileinfo.emit(metadata.to_dict()) - logger.debug(f"Metadata loaded for: {filename}") + logger.debug("Metadata loaded: %s", filename) + + @QtCore.pyqtSlot(str, name="request_print_duration") + def request_print_duration(self, filename: str) -> None: + """Fetch the file's last job duration from history, once.""" + filename = filename.removeprefix("/") + metadata = self._files_metadata.get(filename) + if metadata is None or not metadata.job_id: + return + job_id = str(metadata.job_id) + if job_id in self._job_durations: + return + # Mark before sending: error replies skip the callback. + self._job_durations[job_id] = None + self.ws.api.history_get_job( + job_id, + lambda result, name=filename, uid=job_id: self._history_job.emit( + name, uid, result or {} + ), + ) - def handle_metadata_error(self, error_data: str | dict) -> None: - """ - Handle metadata request error from Moonraker. + @QtCore.pyqtSlot(str, str, dict, name="on_history_job") + def _on_history_job(self, filename: str, job_id: str, result: dict) -> None: + """Cache a completed job's duration and re-emit fileinfo.""" + job = result.get("job") or {} + status = job.get("status") + if status == "in_progress": + self._job_durations.pop(job_id, None) + return + duration = job.get("print_duration") + # Cancelled/errored runs stopped early; their time misleads. + if status != "completed" or not isinstance(duration, (int, float)): + return + if duration <= 0: + return + self._job_durations[job_id] = float(duration) + metadata = self._files_metadata.get(filename) + if metadata is None or str(metadata.job_id) != job_id: + return + updated = replace(metadata, print_duration=float(duration)) + self._files_metadata[filename] = updated + self.fileinfo.emit(updated.to_dict()) - Parses the filename from the error message and emits metadata_error signal. - Called directly from MainWindow error handler. + @staticmethod + def _has_inline_metadata(file_data: dict) -> bool: + """True if a dir entry has real metadata, not only thumbnails.""" + return "estimated_time" in file_data + + def _is_cached(self, path: str, file_data: dict) -> bool: + """True if *path* is cached at this entry's size and mtime.""" + cached = self._files_metadata.get(path) + # Moonraker's own freshness test (FileManager._has_valid_data). + return cached is not None and (cached.size, cached.modified) == ( + file_data.get("size"), + file_data.get("modified"), + ) - Args: - error_data: The error message string or dict from Moonraker - """ + def _usb_metadata_loader(self) -> gcode_loader.GcodeMetadataLoader: + """Create and wire the USB metadata loader on first use.""" + if self._meta_loader is None: + loader = ( + gcode_loader.get_metadata_loader() + or gcode_loader.configure_metadata(self.ws._moonRest) + ) + loader.ready.connect(self._on_usb_metadata_ready) + self._meta_loader = loader + return self._meta_loader + + def _request_gcode_metadata( + self, full_path: str, file_data: dict | None = None + ) -> None: + """Ask Moonraker, or parse USB gcodes locally (it can't scan them).""" + if not helper_methods.is_usb_path(full_path): + self.request_file_metadata.emit(full_path) + return + rel = full_path.removeprefix("/") + if file_data: + self._usb_meta_base[rel] = { + "size": file_data.get("size", 0), + "modified": file_data.get("modified", 0.0), + } + self._usb_metadata_loader().request(rel) + + @QtCore.pyqtSlot(str, dict) + def _on_usb_metadata_ready(self, full_path: str, meta: dict) -> None: + """Feed parsed USB metadata into the normal pipeline.""" + base = self._usb_meta_base.pop(full_path, {}) + self._process_metadata(base | meta, full_path) + + def handle_metadata_error(self, error_data: str | dict) -> None: + """Retry the metadata scan named in a Moonraker error.""" if not error_data: return @@ -517,48 +553,33 @@ def handle_metadata_error(self, error_data: str | dict) -> None: end = text.find(">", start) if start > 0 and end > start: - filename = text[start:end] - clean_filename = filename.removeprefix("/") + self._retry_metadata_scan(text[start:end].removeprefix("/")) + + def _retry_metadata_scan(self, clean_filename: str) -> None: + """Force a metadata rescan up to 3 times, then give up.""" + if not clean_filename.lower().endswith(self.GCODE_EXTENSION): + return + count = self._metadata_retry_count.get(clean_filename, 0) + if count >= 3: + self._metadata_retry_count.pop(clean_filename, None) self.metadata_error.emit(clean_filename) - logger.debug(f"Metadata error for: {clean_filename}") + logger.debug("Metadata retry limit reached: %s", clean_filename) + return + self._metadata_retry_count[clean_filename] = count + 1 + self.request_scan_metadata.emit(clean_filename) + logger.debug("Metadata rescan attempt %d: %s", count + 1, clean_filename) def _preload_usb_contents(self, usb_path: str) -> None: - """ - Preload USB contents when USB is inserted. - - Requests directory info for the USB mount so files are ready - when user navigates to it. - - Args: - usb_path: The USB mount path (e.g., "USB-sda1") - """ + """Preload USB directory info when USB is inserted.""" + if usb_path in self._pending_usb_preloads: + return # a second reply would be taken as the shown listing logger.info(f"Preloading USB contents: {usb_path}") self._pending_usb_preloads.add(usb_path) 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) -> list[dict] | None: - """ - Get cached files for a USB path if available. - - Args: - usb_path: The USB mount path - - Returns: - List of file dicts if cached, None otherwise - """ - return self._usb_files_cache.get(usb_path.removeprefix("/")) - def _process_usb_directory_info(self, usb_path: str, data: dict) -> None: - """ - Process preloaded USB directory info. - - Caches the files and requests metadata for gcode files. - - Args: - usb_path: The USB mount path - data: Directory info response from Moonraker - """ + """Cache preloaded USB directory info and request metadata.""" files = [] for file_data in data.get("files", []): filename = file_data.get("filename", file_data.get("path", "")) @@ -566,55 +587,72 @@ def _process_usb_directory_info(self, usb_path: str, data: dict) -> None: files.append(file_data) full_path = f"{usb_path}/{filename}" - if filename.lower().endswith(self.GCODE_EXTENSION): - self.request_file_metadata.emit(full_path) + if filename.lower().endswith( + self.GCODE_EXTENSION + ) and not self._is_cached(full_path, file_data): + self._request_gcode_metadata(full_path, file_data) # Cache the files self._usb_files_cache[usb_path] = files self.usb_files_loaded.emit(usb_path, files) logger.info(f"Preloaded {len(files)} files from USB: {usb_path}") - 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. - matched_usb = None - - if self._usb_preload_queue: - candidate = self._usb_preload_queue.popleft() - if candidate in self._pending_usb_preloads: - matched_usb = candidate - + def _process_directory_info(self, data: dict, requested_dir: str = "") -> None: + """Publish a directory listing and dispatch its gcode metadata.""" + matched_usb = self._match_usb_preload(requested_dir) if matched_usb: self._pending_usb_preloads.discard(matched_usb) self._process_usb_directory_info(matched_usb, data) return - + self._populate_directory(data) + self.on_file_list.emit(self.file_list) + self.on_dirs.emit(self.directories) + self._initial_load_complete = True + logger.info( + "Directory loaded: %d dirs, %d files", + len(self._directories), + len(self._files), + ) + self._dispatch_metadata(requested_dir) + + def _match_usb_preload(self, requested_dir: str) -> str | None: + """Pending USB preload matching this response, else None.""" + if not requested_dir or requested_dir not in self._pending_usb_preloads: + return None + if requested_dir in self._usb_preload_queue: + self._usb_preload_queue.remove(requested_dir) + return requested_dir + + def _populate_directory(self, data: dict) -> None: + """Replace backing dir/file maps from a directory response.""" self._directories.clear() self._files.clear() - for dir_data in data.get("dirs", []): dirname = dir_data.get("dirname", "") if dirname and not dirname.startswith("."): self._directories[dirname] = dir_data - for file_data in data.get("files", []): filename = file_data.get("filename", file_data.get("path", "")) - if filename: - self._files[filename] = file_data - - self.on_file_list.emit(self.file_list) - self.on_dirs.emit(self.directories) - self._initial_load_complete = True - - logger.info( - f"Directory loaded: {len(self._directories)} dirs, {len(self._files)} files" - ) - - # Request metadata only for gcode files (async update) - for filename in self._files: - if filename.lower().endswith(self.GCODE_EXTENSION): - self.request_file_metadata.emit(filename.removeprefix("/")) + if not filename: + continue + # Moonraker lists USB symlinks as files; show them as dirs. + if helper_methods.is_usb_mount(filename): + self._directories[filename] = file_data | {"dirname": filename} + continue + self._files[filename] = file_data + + def _dispatch_metadata(self, requested_dir: str = "") -> None: + """Process each new or changed gcode once; the page keeps the rest.""" + for filename, file_data in self._files.items(): + if not filename.lower().endswith(self.GCODE_EXTENSION): + continue + full = self._full_gcode_path(filename, requested_dir) + if self._is_cached(full, file_data): + continue + if self._has_inline_metadata(file_data): + self._process_metadata(file_data, full) + else: + self._request_gcode_metadata(full, file_data) @QtCore.pyqtSlot(str, str, name="on_request_delete_file") def on_request_delete_file(self, filename: str, directory: str = "gcodes") -> None: @@ -638,14 +676,14 @@ def on_request_fileinfo(self, filename: str) -> None: if cached: self.fileinfo.emit(cached.to_dict()) else: - self.request_file_metadata.emit(clean_filename) + self._request_gcode_metadata(clean_filename) @QtCore.pyqtSlot(name="get_dir_info") @QtCore.pyqtSlot(str, name="get_dir_info") @QtCore.pyqtSlot(str, bool, name="get_dir_info") def get_dir_information( self, directory: str = "", extended: bool = True - ) -> list | None: + ) -> typing.Any: """Get directory information.""" self._current_directory = directory @@ -682,5 +720,6 @@ def _clear_all_data(self) -> None: self._usb_files_cache.clear() self._pending_usb_preloads.clear() self._usb_preload_queue.clear() + self._job_durations.clear() self._initial_load_complete = False logger.info("All file data cleared") diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index 2d792ccd..5353e43b 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -901,9 +901,11 @@ def history_reset_totals(self): """Request history reset""" raise NotImplementedError - def history_get_job(self, uid: str): - """Request job history""" - raise NotImplementedError + def history_get_job(self, uid: str, callback=None) -> bool: + """Request a history job; callback gets {"job": {...}}.""" + return self._ws.send_request( + method="server.history.get_job", params={"uid": uid}, callback=callback + ) def history_delete_job(self, uid: str): """Request delete job history""" diff --git a/BlocksScreen/lib/moonrest.py b/BlocksScreen/lib/moonrest.py index 81f46ad6..437d98e5 100644 --- a/BlocksScreen/lib/moonrest.py +++ b/BlocksScreen/lib/moonrest.py @@ -27,6 +27,7 @@ import logging +from urllib.parse import quote import requests @@ -43,18 +44,16 @@ def __init__(self, message="Unable to call method", errors=None): class MoonRest: - """MoonRest Basic API for sending end posting requests to MoonrakerAPI - - Raises: - UncallableError: An error occurred when the request type invalid - """ + """Moonraker HTTP client.""" timeout = 3 - def __init__(self, host: str = "localhost", port: int = 7125, api_key=False): + def __init__( + self, host: str = "localhost", port: int = 7125, api_key: str | None = None + ): self._host = host self._port = port - self._api_key = api_key + self._api_key: str | None = api_key @property def build_endpoint(self): @@ -62,13 +61,8 @@ def build_endpoint(self): return f"http://{self._host}:{self._port}" def get_oneshot_token(self): - """Requests Moonraker API for a oneshot token to be used on - API key authentication - - Returns: - str: A oneshot token - """ - # Response data is generally an object itself, however for some requests this may simply be an "ok" string. + """Oneshot token for API-key auth.""" + # Some requests reply with a bare "ok" string, not an object. response = self.get_request(method="access/oneshot_token") if response is None: return None @@ -79,40 +73,25 @@ def get_oneshot_token(self): ) def get_server_info(self): - """GET MoonrakerAPI /server/info - - Returns: - dict: server info from Moonraker - """ + """Fetch server info from Moonraker.""" return self.get_request(method="server/info") def get_spool(self, spool_id: int) -> dict | None: - """GET /server/spoolman/spool/{spool_id} via Moonraker - - Returns spool dict on success, None on HTTP/network/JSON error. - """ + """Spoolman spool via Moonraker, None on error.""" response = self.get_request(f"server/spoolman/spool/{spool_id}") if not isinstance(response, dict): return None return response.get("result") def set_spool_used_weight(self, spool_id: int, weight: float) -> bool: - """POST /server/spoolman/spool/{spool_id} to update used_weight. - - Returns True on sucess, False on any error. - """ + """Set a spool's used_weight; True on success.""" response = self.post_request( f"server/spoolman/spool/{spool_id}", json={"used_weight": weight} ) return response is not None def firmware_restart(self): - """firmware_restart - POST to /printer/firmware_restart to firmware restart Klipper - - Returns: - str: Returns an 'ok' from Moonraker - """ + """POST firmware_restart to Moonraker.""" return self.post_request(method="printer/firmware_restart") def post_request(self, method, data=None, json=None, json_response=True): @@ -134,6 +113,40 @@ def get_request(self, method, json=True, timeout=timeout): timeout=timeout, ) + def get_gcode_header(self, rel_path: str, max_bytes: int = 131072) -> bytes | None: + """First *max_bytes* of a gcode, for embedded thumbnails.""" + return self._stream_gcode(rel_path, f"bytes=0-{max_bytes - 1}", max_bytes) + + def get_gcode_tail(self, rel_path: str, max_bytes: int = 65536) -> bytes | None: + """Last *max_bytes* of a gcode, for the slicer footer.""" + data = self._stream_gcode(rel_path, f"bytes=-{max_bytes}", max_bytes + 1) + # More than asked for means the server ignored Range. + if data is not None and len(data) > max_bytes: + logger.info("gcode tail fetch for %s ignored Range", rel_path) + return None + return data + + def _stream_gcode(self, rel_path: str, byte_range: str, limit: int) -> bytes | None: + """Range GET a gcode, stopping once *limit* bytes arrived.""" + url = f"{self.build_endpoint}/server/files/gcodes/{quote(rel_path)}" + headers = {"Range": byte_range} + if self._api_key: + headers["x-api-key"] = self._api_key + try: + with requests.get( + url, headers=headers, stream=True, timeout=self.timeout + ) as resp: + resp.raise_for_status() + data = bytearray() + for chunk in resp.iter_content(chunk_size=65536): + data.extend(chunk) + if len(data) >= limit: + break + return bytes(data[:limit]) + except requests.RequestException as exc: + logger.info("gcode fetch (%s) failed for %s: %s", byte_range, rel_path, exc) + return None + def _request( self, request_type, diff --git a/BlocksScreen/lib/panels/mainWindow.py b/BlocksScreen/lib/panels/mainWindow.py index 452e809a..646b7c10 100644 --- a/BlocksScreen/lib/panels/mainWindow.py +++ b/BlocksScreen/lib/panels/mainWindow.py @@ -173,6 +173,13 @@ def __init__(self): self.printPanel = PrintTab( self.ui.printTab, self.file_data, self.ws, self.printer ) + self.usb_manager.usb_mounted.connect( + self.printPanel.filesPage_widget.on_usb_added + ) + # usb_unmounted only; also wiring hardware_removed refreshes twice. + self.usb_manager.usb_unmounted.connect( + self.printPanel.filesPage_widget.on_usb_removed + ) if not os.environ.get("BLOCKSCREEN_DEV"): QtWidgets.QApplication.setOverrideCursor(QtCore.Qt.CursorShape.BlankCursor) self.filamentPanel = FilamentTab( diff --git a/BlocksScreen/lib/panels/networkWindow.py b/BlocksScreen/lib/panels/networkWindow.py index b3c7ed45..2763564d 100644 --- a/BlocksScreen/lib/panels/networkWindow.py +++ b/BlocksScreen/lib/panels/networkWindow.py @@ -35,6 +35,7 @@ from lib.utils.blocks_togglebutton import NetworkWidgetbuttons from lib.utils.check_button import BlocksCustomCheckButton from lib.utils.icon_button import IconButton +from lib.utils.blocks_combobox import BlocksComboBox from lib.utils.list_model import EntryDelegate, EntryListModel, ListItem from PyQt6 import QtCore, QtGui, QtWidgets from PyQt6.QtCore import QTimer, pyqtSlot @@ -46,12 +47,7 @@ class PixmapCache: - """Process-wide cache for QPixmaps loaded from Qt resource paths. - - Every SVG is decoded exactly once. Qt's implicit sharing means the - same QPixmap can be safely referenced by any number of widgets. - Must only be called after QApplication is created. - """ + """Process-wide QPixmap cache for SVG resource paths (after QApplication init).""" _cache: dict[str, QtGui.QPixmap] = {} @@ -182,11 +178,7 @@ def _joined_target(win: "NetworkControlWindow", state: NetworkState) -> bool: class NetworkControlWindow(QtWidgets.QStackedWidget): - """Stacked-widget UI for all network control pages (Wi-Fi, Ethernet, VLAN, Hotspot). - - Owns a :class:`~BlocksScreen.lib.network.facade.NetworkManager` instance and - mediates between the UI pages and the async D-Bus worker. - """ + """Stacked-widget UI for network control (Wi-Fi, Ethernet, VLAN, Hotspot).""" update_wifi_icon = QtCore.pyqtSignal(int, name="update-wifi-icon") @@ -276,13 +268,7 @@ def _init_network_manager(self) -> None: self._prefill_ip_from_os() def _prefill_ip_from_os(self) -> None: - """Read the current IP via SIOCGIFADDR ioctl and show it immediately. - - Bypasses NetworkManager D-Bus entirely: runs on the main thread, - costs a single syscall, and completes in microseconds. Called once - during init so the user never sees "IP: --" if a connection was - already active before the UI launched. - """ + """Read and display current IP via SIOCGIFADDR ioctl (synchronous, <1us).""" _SIOCGIFADDR = 0x8915 for iface in ("eth0", "wlan0"): try: @@ -402,11 +388,7 @@ def _pending_op_outcome(self, state: NetworkState) -> bool | None: @pyqtSlot(list) def _on_scan_complete(self, networks: list[NetworkInfo]) -> None: - """Receive scan results, filter/sort them, and rebuild the SSID list view. - - Filters out the own hotspot SSID and networks with unsupported security - types before populating the list view. - """ + """Receive scan results, filter unsupported security, rebuild SSID list.""" hotspot_ssid = self._nm.hotspot_ssid filtered = [ n @@ -543,14 +525,7 @@ def _on_network_error(self, operation: str, message: str) -> None: self._show_error_popup(f"Error: {message}") def _emit_status_icon(self, state: NetworkState) -> None: - """Emit the correct header icon key based on current state. - - Ethernet -> ETHERNET, Hotspot -> HOTSPOT, - Wi-Fi connected -> signal-strength key, otherwise -> 0-bar. - - Uses self._active_signal (the single source of truth) so the - header icon always matches the list icon and panel percentage. - """ + """Emit header icon key (Ethernet/Hotspot/signal/0-bar) from _active_signal.""" if state.ethernet_connected: self.update_wifi_icon.emit(WifiIconKey.ETHERNET) elif state.hotspot_enabled: @@ -567,16 +542,7 @@ def _emit_status_icon(self, state: NetworkState) -> None: self.update_wifi_icon.emit(WifiIconKey.from_bars(0, False)) def _sync_active_network_list_icon(self, state: NetworkState) -> None: - """Rebuild the wifi list when the active network's signal bars or status changes. - - Between scans, state polling may report a different signal strength - for the connected AP. Also corrects the status label from SAVED to - ACTIVE when the connection establishes after the last scan ran. - Invalidates the item cache for that SSID so the next reconcile picks - up the new icon/label, without touching other items. - - Uses self._active_signal as the single source of truth. - """ + """Rebuild Wi-Fi list when signal/status changes; invalidate item cache.""" if not self._cached_scan_networks or not state.current_ssid: self._last_active_signal_bars = -1 return @@ -656,8 +622,7 @@ def _handle_first_run(self, state: NetworkState) -> None: self._sync_ethernet_panel(state) def _sync_toggle_states(self, state: NetworkState) -> None: - """Synchronise Wi-Fi and hotspot toggle buttons to the current NetworkState - without loops.""" + """Sync Wi-Fi/hotspot toggles to NetworkState without loops.""" if self._is_connecting: return @@ -683,12 +648,7 @@ def _sync_toggle_states(self, state: NetworkState) -> None: ) def _sync_ethernet_panel(self, state: NetworkState) -> None: - """Show/hide the ethernet panel and sync its toggle state. - - Visibility is driven by ``ethernet_carrier`` (cable physically - plugged in), while the toggle position reflects the active - connection state (``ethernet_connected``). - """ + """Show/hide ethernet panel; sync toggle to connection state (carrier + connected).""" eth_btn = self.ethernet_button.toggle_button with QtCore.QSignalBlocker(eth_btn): @@ -700,12 +660,7 @@ def _sync_ethernet_panel(self, state: NetworkState) -> None: self.ethernet_button.setVisible(state.ethernet_carrier) def _display_connected_state(self, state: NetworkState) -> None: - """Display connected network information. - - Ethernet always takes display priority: if ``ethernet_connected`` - is True we show "Ethernet" even if a Wi-Fi SSID is still lingering - (e.g. during the brief overlap before NM finishes disabling wifi). - """ + """Display connected network info (Ethernet > Wi-Fi).""" self._hide_all_info_elements() is_ethernet = state.ethernet_connected @@ -778,11 +733,7 @@ def _display_disconnected_state(self) -> None: self.update() def _display_wifi_on_no_connection(self) -> None: - """Display info panel when Wi-Fi is on but not connected. - - Uses the same layout as the connected state but shows - 'No network connected' and empty fields. - """ + """Display info panel when Wi-Fi is on but not connected.""" self._hide_all_info_elements() self.netlist_ssuid.setText("No network connected") @@ -1055,13 +1006,7 @@ def _on_hotspot_config_updated( self.hotspot_password_input_field.setText(password) def _on_hotspot_config_save(self) -> None: - """Save hotspot configuration changes. - - Reads new name/password from the UI fields, asks the worker to - delete old profiles and create a new one. If the hotspot was - active, it will be re-activated with the new config (with a - loading screen shown). - """ + """Save hotspot config and re-activate if needed.""" new_name = self.hotspot_name_input_field.text().strip() new_password = self.hotspot_password_input_field.text().strip() @@ -1220,12 +1165,7 @@ def _on_wifi_static_ip_clicked(self) -> None: @pyqtSlot() def _on_wifi_static_ip_apply(self) -> None: - """Validate static-IP fields and apply them to the current Wi-Fi connection. - - Mirrors the VLAN-creation UX: navigate to the main panel immediately, - show the loading overlay, and clear it silently once ``reconnect_complete`` - fires (no popup: the updated IP appears in the panel header instead). - """ + """Validate and apply static IP to current Wi-Fi connection.""" ssid = self.wifi_sip_title.text() ip_addr = self.wifi_sip_ip_field.text().strip() mask = self.wifi_sip_mask_field.text().strip() @@ -1259,10 +1199,7 @@ def _on_wifi_static_ip_apply(self) -> None: @pyqtSlot() def _on_wifi_reset_dhcp(self) -> None: - """Reset the current Wi-Fi connection back to DHCP via the facade. - - Same loading-screen pattern as static IP: no popup on success. - """ + """Reset current Wi-Fi connection to DHCP.""" ssid = self.wifi_sip_title.text() self.setCurrentIndex(self.indexOf(self.main_network_page)) self._pending_operation = PendingOperation.WIFI_STATIC_IP @@ -1273,13 +1210,7 @@ def _on_wifi_reset_dhcp(self) -> None: self._nm.request_state_soon(delay_ms=3000) def _build_network_list_from_scan(self, networks: list[NetworkInfo]) -> None: - """Build/update network list from scan results. - - Uses the model's built-in reconcile() with an item cache so that - ListItems are only allocated for networks whose visual state - actually changed (different signal bars or status label). - Unchanged items are reused from the cache: zero allocation. - """ + """Build/update network list from scan results via reconcile.""" self.listView.blockSignals(True) desired_items: list[ListItem] = [] @@ -1316,11 +1247,7 @@ def _build_network_list_from_scan(self, networks: list[NetworkInfo]) -> None: self.listView.update() def _patch_cached_network_status(self, ssid: str, status: NetworkStatus) -> None: - """Optimistically update one entry in the scan cache and rebuild the list. - - Called immediately after add/delete so the list reflects the change - without waiting for the next scan cycle. - """ + """Update scan cache entry and rebuild list immediately.""" self._cached_scan_networks = [ replace(n, network_status=status) if n.ssid == ssid else n for n in self._cached_scan_networks @@ -1329,13 +1256,7 @@ def _patch_cached_network_status(self, ssid: str, status: NetworkStatus) -> None self._build_network_list_from_scan(self._cached_scan_networks) def _get_or_create_item(self, network: NetworkInfo) -> ListItem | None: - """Return a cached ListItem if the network's visual state is - unchanged, otherwise create a new one and update the cache. - - Visual state = (signal_bars, status_label). When both match - the cached entry the existing ListItem is returned as-is, with - no QPixmap lookup and no allocation. - """ + """Return cached ListItem if unchanged (bars + label), else allocate new.""" if network.is_hidden or is_hidden_ssid(network.ssid): return None if not is_connectable_security(network.security_type): @@ -1620,11 +1541,7 @@ def _on_delete_network(self) -> None: @pyqtSlot() def _on_save_network_details(self) -> None: - """Save network settings changes (password / priority). - - Only performs an update if the user actually changed something. - Shows a confirmation popup on success. - """ + """Save network settings if changed; show confirmation popup.""" ssid = self.saved_connection_network_name.text() password = self.saved_connection_change_password_field.text() priority = self._get_selected_priority() @@ -1895,27 +1812,12 @@ def _setup_main_network_page(self) -> None: info_layout.addWidget(self.netlist_ip) - self.netlist_vlans_combo = QtWidgets.QComboBox( - parent=self.mn_information_layout - ) + self.netlist_vlans_combo = BlocksComboBox(parent=self.mn_information_layout) font = QtGui.QFont() font.setPointSize(11) self.netlist_vlans_combo.setFont(font) self.netlist_vlans_combo.setMinimumSize(QtCore.QSize(240, 50)) self.netlist_vlans_combo.setMaximumSize(QtCore.QSize(250, 50)) - self.netlist_vlans_combo.setStyleSheet(""" - QComboBox { - background-color: rgba(26, 143, 191, 0.05); - color: rgba(255, 255, 255, 200); - border: 1px solid rgba(255, 255, 255, 80); - border-radius: 8px; - } - QComboBox QAbstractItemView { - background-color: rgb(40, 40, 40); - color: white; - selection-background-color: rgba(26, 143, 191, 0.6); - } - """) self.netlist_vlans_combo.setVisible(False) self.netlist_vlans_combo.currentIndexChanged.connect( diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 36f62024..d2c9ec0b 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -11,11 +11,13 @@ from lib.panels.widgets.confirmPage import ConfirmWidget from lib.panels.widgets.filesPage import FilesPage from lib.panels.widgets.jobStatusPage import JobStatusWidget +from lib.panels.widgets.metadataPage import FileMetadataWidget from lib.panels.widgets.numpadPage import CustomNumpad from lib.panels.widgets.sensorsPanel import SensorsWindow from lib.panels.widgets.slider_selector_page import SliderPage from lib.panels.widgets.tunePage import TuneWidget from lib.printer import Printer +from lib.utils import gcode_loader from lib.utils.blocks_button import BlocksCustomButton from lib.utils.display_button import DisplayButton from PyQt6 import QtCore, QtGui, QtWidgets @@ -24,26 +26,7 @@ class PrintTab(QtWidgets.QStackedWidget): - """QStackedWidget that contains the following widget panels: - - - Main page: Simple page with a message field and a button to start a print; - - File list page: A file list where displayed files are selectable to be printed; - - Confirm page: A page to confirm or not if the selected file is to be printed; - - Print page: A page for controlling the ongoing job, Pause/Resume and stop functionality - - Tune page: Accessible only from the print page; - - Babystep page: Control the z_offset during a ongoing print; - - Change page: A page that permits changing the filament, stops the print -> change the filament -> resume the print; - - Args: - QStackedWidget (QStackedWidget): This class is inherited from QStackedWidget from Qt6 - - __init__: - parent (QWidget | QObject): The parent for this tab. - file_data (Files): Class object that handles printer files. - ws (MoonWebSocket): Moonraker websocket instance. - printer (Printer): Class object that handles printer objects information. - - """ + """Print tab: files, confirm, job status and tuning pages.""" request_query_print_stats: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( dict, name="request_query_print_stats" @@ -90,6 +73,8 @@ def __init__( self.setupMainPrintPage() self.ws: MoonWebSocket = ws + # Embedded-thumbnail fallback for USB gcodes. + gcode_loader.configure(ws._moonRest) self.printer: Printer = printer self.config: BlocksScreenConfig = get_configparser() # TODO: Get the gcode path from the configfile by asking the websocket first @@ -121,29 +106,28 @@ def __init__( lambda: self.change_page(self.indexOf(self.confirmPage_widget)) ) self.filesPage_widget.back_btn.clicked.connect(self.back_button) - self.filesPage_widget.request_file_info.connect( - self.file_data.on_request_fileinfo + + self.metadataPage_widget = FileMetadataWidget(self) + self.addWidget(self.metadataPage_widget) + self.confirmPage_widget.show_metadata.connect( + self.metadataPage_widget.on_show_widget ) - self.filesPage_widget.request_file_metadata.connect( - self.file_data.request_file_metadata + self.confirmPage_widget.show_metadata.connect( + lambda: self.change_page(self.indexOf(self.metadataPage_widget)) ) + self.confirmPage_widget.show_metadata.connect( + self.file_data.request_print_duration + ) + self.metadataPage_widget.back_btn.clicked.connect(self.back_button) self.file_data.fileinfo.connect(self.filesPage_widget.on_fileinfo) + self.file_data.fileinfo.connect(self.confirmPage_widget.on_fileinfo) + self.file_data.fileinfo.connect(self.metadataPage_widget.on_fileinfo) - self.filesPage_widget.request_file_list[str].connect( - self.file_data.request_file_list - ) - self.filesPage_widget.request_file_list.connect( - self.file_data.request_file_list - ) self.file_data.on_dirs.connect(self.filesPage_widget.on_directories) self.filesPage_widget.request_dir_info[str].connect( self.file_data.request_dir_info[str] ) - self.filesPage_widget.request_scan_metadata.connect( - self.ws.api.scan_gcode_metadata - ) - self.file_data.metadata_error.connect(self.filesPage_widget.on_metadata_error) self.filesPage_widget.request_dir_info.connect(self.file_data.request_dir_info) self.file_data.on_file_list.connect(self.filesPage_widget.on_file_list) self.file_data.file_added.connect(self.filesPage_widget.on_file_added) @@ -170,6 +154,15 @@ def __init__( self.jobStatusPage_widget.hide_request.connect( lambda: self.change_page(self.indexOf(self.print_page)) ) + self.jobStatusPage_widget.show_metadata.connect( + self.metadataPage_widget.on_show_widget + ) + self.jobStatusPage_widget.show_metadata.connect( + lambda: self.change_page(self.indexOf(self.metadataPage_widget)) + ) + self.jobStatusPage_widget.show_metadata.connect( + self.file_data.request_print_duration + ) self.jobStatusPage_widget.request_file_info.connect( self.file_data.on_request_fileinfo ) @@ -295,9 +288,7 @@ def __init__( @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: - """ - unblocks tabs if on standby - """ + """Unblock tabs if on standby.""" if isinstance(value, str) and "state" in field and value == "standby": self.call_load_panel.emit(False, "", False) self.on_cancel_print.emit() @@ -429,15 +420,7 @@ def _on_delete_file_confirmed(self, filename: str, directory: str) -> None: pass def setProperty(self, name: str, value: typing.Any) -> bool: - """Intercept the set property method - - Args: - name (str): Name of the dynamic property - value (typing.Any): Value for the dynamic property - - Returns: - bool: Returns to the super class - """ + """Intercept property changes.""" if name == "backgroundPixmap": self.background = value return super().setProperty(name, value) @@ -454,11 +437,7 @@ def handle_cancel_print(self) -> None: self.call_load_panel.emit(True, "Cancelling print...\nPlease wait", False) def change_page(self, index: int) -> None: - """Requests a page change page to the global manager - - Args: - index (int): page index - """ + """Request page change to global manager.""" self.request_change_page.emit(0, index) @QtCore.pyqtSlot(name="request-back") diff --git a/BlocksScreen/lib/panels/widgets/cancelPage.py b/BlocksScreen/lib/panels/widgets/cancelPage.py index 8592e578..6bebc5a8 100644 --- a/BlocksScreen/lib/panels/widgets/cancelPage.py +++ b/BlocksScreen/lib/panels/widgets/cancelPage.py @@ -3,6 +3,7 @@ from lib.utils.blocks_button import BlocksCustomButton from lib.utils.blocks_frame import BlocksCustomFrame +from lib.utils import gcode_loader from lib.utils.blocks_label import BlocksLabel from PyQt6 import QtCore, QtGui, QtWidgets @@ -99,24 +100,26 @@ def set_file_name(self, file_name: str) -> None: self.cf_file_name.setText(file_name) def _show_screen_thumbnail(self, metadata: dict | None) -> None: - """Display the largest thumbnail from file metadata. - - ``thumbnail_images`` values are pre-loaded ``QImage`` - objects produced by ``Files._process_metadata``. - """ - fallback = QtGui.QPixmap( + """Show the largest thumbnail, else the embedded one or the logo.""" + metadata = metadata or {} + meta_file = metadata.get("filename", "") + # fileinfo is global, every listed file emits it: decode only ours. + if meta_file.removeprefix("/") != self.filename.removeprefix("/"): + return + thumbnails = metadata.get("thumbnail_paths") or [] + last_thumb = QtGui.QPixmap(thumbnails[-1]) if thumbnails else QtGui.QPixmap() + if last_thumb.isNull(): + last_thumb = self._embedded_pixmap(meta_file) + self.set_pixmap(last_thumb) + + def _embedded_pixmap(self, gcode_path: str) -> QtGui.QPixmap: + """Cached embedded thumbnail, else the logo.""" + pixmap = gcode_loader.cached_pixmap(gcode_path) + if pixmap is not None: + return pixmap + return QtGui.QPixmap( "BlocksScreen/lib/ui/resources/media/logoblocks400x300.png" ) - thumbnails = metadata.get("thumbnail_images", []) if metadata else [] - if not thumbnails: - self.set_pixmap(fallback) - return - - last_thumb = thumbnails[-1] - if isinstance(last_thumb, QtGui.QImage) and not last_thumb.isNull(): - self.set_pixmap(QtGui.QPixmap.fromImage(last_thumb)) - else: - self.set_pixmap(fallback) def _setupUI(self) -> None: """Setup widget ui""" diff --git a/BlocksScreen/lib/panels/widgets/confirmPage.py b/BlocksScreen/lib/panels/widgets/confirmPage.py index ed4da506..5d7f40eb 100644 --- a/BlocksScreen/lib/panels/widgets/confirmPage.py +++ b/BlocksScreen/lib/panels/widgets/confirmPage.py @@ -2,6 +2,7 @@ import typing import helper_methods +from lib.utils import gcode_loader from lib.utils.blocks_button import BlocksCustomButton from lib.utils.blocks_frame import BlocksCustomFrame from lib.utils.blocks_label import BlocksLabel @@ -21,6 +22,9 @@ class ConfirmWidget(QtWidgets.QWidget): on_delete: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( str, str, name="delete_file" ) + show_metadata: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( + str, dict, name="show_metadata" + ) def __init__(self, parent) -> None: super().__init__(parent) @@ -28,14 +32,21 @@ def __init__(self, parent) -> None: self.setMouseTracking(True) self.setAttribute(QtCore.Qt.WidgetAttribute.WA_AcceptTouchEvents, True) self.thumbnail: QtGui.QImage = self._blocksthumbnail + self._thumbnails: list = [] + self._current_gcode: str = "" + self._loader_connected: bool = False self.directory = "gcodes" self.filename = "" + self._filedata: dict = {} self.confirm_button.clicked.connect( lambda: self.on_accept.emit( str(os.path.join(self.directory, self.filename)) ) ) self.back_btn.clicked.connect(self.request_back.emit) + self.file_info_btn.clicked.connect( + lambda: self.show_metadata.emit(self._current_gcode, self._filedata) + ) self.delete_file_button.clicked.connect( lambda: self.on_delete.emit(self.filename, self.directory) ) @@ -48,66 +59,68 @@ def on_show_widget(self, text: str, metadata: dict | None = None) -> None: self.directory = directory self.filename = filename self.cf_file_name.setText(self.filename) - self._update_metadata_labels(metadata or {}) + self._current_gcode = text.removeprefix("/") + self._filedata = metadata or {} + self._update_metadata_labels(self._filedata) self.update() def _update_metadata_labels(self, metadata: dict) -> None: """Update thumbnail and text labels from metadata.""" - self._apply_thumbnail(metadata) - raw_weight = metadata.get("filament_weight_total") - _total_filament: float | str = ( - raw_weight if isinstance(raw_weight, (int, float)) and raw_weight > 0 else 0 - ) - raw_seconds = metadata.get("estimated_time") + self._thumbnails = metadata.get("thumbnail_paths", []) + self.thumbnail = self._resolve_thumbnail() + weight = metadata.get("filament_weight_total") + estimated = metadata.get("estimated_time") seconds = ( - int(raw_seconds) - if isinstance(raw_seconds, (int, float)) and raw_seconds > 0 + int(estimated) + if isinstance(estimated, (int, float)) and estimated > 0 else 0 ) - - days, hours, minutes, _ = helper_methods.estimate_print_time(seconds) - if seconds <= 0: - time_str = "Unknown" - elif seconds < 60: - time_str = "less than 1 minute" - else: - if days > 0: - time_str = f"{days}d {hours}h {minutes}m" - elif hours > 0: - time_str = f"{hours}h {minutes}m" - else: - time_str = f"{minutes}m" - if _total_filament == 0: - _total_filament = "Unknown" - elif _total_filament > 499: - _total_filament /= 1000 - _total_filament = str("%.2f" % _total_filament) + "kg" - else: - _total_filament = str("%.2f" % _total_filament) + "g" - filament_label = f"Total Filament: {_total_filament}" - time_label = f"Slicer time: {time_str}" - self.cf_info_tf.setText(filament_label) - self.cf_info_tr.setText(time_label) - - def _apply_thumbnail(self, metadata: dict) -> None: - """Set self.thumbnail from metadata, falling back to the logo.""" - thumbnails = metadata.get("thumbnail_images", []) - if thumbnails: - last = thumbnails[-1] - if isinstance(last, QtGui.QImage) and not last.isNull(): - self.thumbnail = last - return - self.thumbnail = self._blocksthumbnail + time_str = helper_methods.format_duration(seconds) if seconds else "Unknown" + filament_str = ( + helper_methods.format_weight(weight) + if isinstance(weight, (int, float)) and weight > 0 + else "Unknown" + ) + self.cf_info_tf.setText(f"Total Filament: {filament_str}") + self.cf_info_tr.setText(f"Slicer time: {time_str}") + + @QtCore.pyqtSlot(dict, name="on_fileinfo") + def on_fileinfo(self, filedata: dict) -> None: + """Refresh the shown file's metadata, e.g. a late duration.""" + if filedata.get("filename", "").removeprefix("/") != self._current_gcode: + return + self._filedata = filedata + # A file opened before its metadata arrived shows it once it lands. + if self.isVisible(): + self._update_metadata_labels(filedata) + self.update() + + def _resolve_thumbnail(self) -> QtGui.QImage: + """Largest on-disk thumbnail, else embedded, else placeholder.""" + if self._thumbnails: + disk = QtGui.QImage(self._thumbnails[-1]) + if not disk.isNull(): + return disk + loader = gcode_loader.get_loader() + if loader is None: + return self._blocksthumbnail + if not self._loader_connected: + loader.ready.connect(self._on_embedded_ready) + self._loader_connected = True + cached = loader.cached(self._current_gcode) + if cached is not None and not cached.isNull(): + return cached + loader.request_embedded(self._current_gcode) + return self._blocksthumbnail + + def _on_embedded_ready(self, gcode_path: str, image: QtGui.QImage) -> None: + """Show a late embedded thumbnail for the shown file.""" + if gcode_path == self._current_gcode and not image.isNull(): + self.thumbnail = image + self.update() def estimate_print_time(self, seconds: int) -> list: - """Convert time in seconds format to days, hours, minutes, seconds. - - Args: - seconds (int): Seconds - - Returns: - list: list that contains the converted information [days, hours, minutes, seconds] - """ + """Convert seconds to [days, hours, minutes, seconds].""" num_min, seconds = divmod(seconds, 60) num_hours, minutes = divmod(num_min, 60) days, hours = divmod(num_hours, 24) @@ -176,14 +189,6 @@ def _setupUI(self) -> None: self.cf_header_title = QtWidgets.QHBoxLayout() self.cf_header_title.setObjectName("cf_header_title") - self.spacer = QtWidgets.QSpacerItem( - 60, - 60, - QtWidgets.QSizePolicy.Policy.Fixed, - QtWidgets.QSizePolicy.Policy.Fixed, - ) - self.spacer.setGeometry(QtCore.QRect(0, 0, 60, 60)) - self.cf_header_title.addItem(self.spacer) sizePolicy = QtWidgets.QSizePolicy( QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Expanding, @@ -202,6 +207,18 @@ def _setupUI(self) -> None: self.cf_file_name.setObjectName("cf_file_name") self.cf_header_title.addWidget(self.cf_file_name) + self.file_info_btn = IconButton(self) + self.file_info_btn.setMinimumSize(QtCore.QSize(60, 60)) + self.file_info_btn.setMaximumSize(QtCore.QSize(60, 60)) + self.file_info_btn.setFlat(True) + self.file_info_btn.setProperty( + "icon_pixmap", QtGui.QPixmap(":/files/media/btn_icons/file_icon.svg") + ) + self.file_info_btn.setObjectName("file_info_btn") + self.cf_header_title.insertWidget( + 0, self.file_info_btn, 0, QtCore.Qt.AlignmentFlag.AlignLeft + ) + self.back_btn = IconButton(self) self.back_btn.setMinimumSize(QtCore.QSize(60, 60)) self.back_btn.setMaximumSize(QtCore.QSize(60, 60)) diff --git a/BlocksScreen/lib/panels/widgets/filesPage.py b/BlocksScreen/lib/panels/widgets/filesPage.py index d731eb9c..abea5e47 100644 --- a/BlocksScreen/lib/panels/widgets/filesPage.py +++ b/BlocksScreen/lib/panels/widgets/filesPage.py @@ -2,6 +2,7 @@ import logging import helper_methods +from lib.utils.blocks_combobox import BlocksComboBox from lib.utils.blocks_Scrollbar import CustomScrollBar from lib.utils.icon_button import IconButton from lib.utils.list_model import EntryDelegate, EntryListModel, ListItem @@ -14,21 +15,27 @@ class FilesPage(QtWidgets.QWidget): # Signals request_back = QtCore.pyqtSignal(name="request_back") file_selected = QtCore.pyqtSignal(str, dict, name="file_selected") - request_file_info = QtCore.pyqtSignal(str, name="request_file_info") request_dir_info = QtCore.pyqtSignal( [], [str], [str, bool], name="api_get_dir_info" ) - request_file_list = QtCore.pyqtSignal([], [str], name="api_get_files_list") - request_file_metadata = QtCore.pyqtSignal(str, name="api_get_gcode_metadata") - request_scan_metadata = QtCore.pyqtSignal(str, name="api_scan_gcode_metadata") # Constants GCODE_EXTENSION = ".gcode" - USB_PREFIX = "USB-" + # Moonraker indexes a new symlink async; refresh again after this. + USB_SETTLE_MS = 1500 ITEM_HEIGHT = 80 LEFT_FONT_SIZE = 17 RIGHT_FONT_SIZE = 12 + SORTING_TYPES: tuple[str, ...] = ( + "Last Print", + "Print Time", + "Name", + "Filament", + "Nozzle Size", + "Import Order", + ) + # Icon paths ICON_PATHS = { "back_folder": ":/ui/media/btn_icons/back_folder.svg", @@ -37,6 +44,8 @@ class FilesPage(QtWidgets.QWidget): "usb": ":/ui/media/btn_icons/usb_icon.svg", "back": ":/ui/media/btn_icons/back.svg", "refresh": ":/ui/media/btn_icons/refresh.svg", + "sort_desc": ":/arrow_icons/media/btn_icons/sort_desc.svg", + "sort_asc": ":/arrow_icons/media/btn_icons/sort_asc.svg", } def __init__(self, parent: QtWidgets.QWidget | None = None) -> None: @@ -47,10 +56,9 @@ def __init__(self, parent: QtWidgets.QWidget | None = None) -> None: self._directories: list[dict] = [] self._curr_dir: str = "" self._pending_action: bool = False - self._pending_metadata_requests: set[str] = set() # Track pending requests - self._metadata_retry_count: dict[ - str, int - ] = {} # Track retry count per file (max 3) + self._rebuild_pending: bool = False + self._sort_key: str = self.SORTING_TYPES[0] + self._sort_descending: bool = True self._icons: dict[str, QtGui.QPixmap] = {} self._model = EntryListModel() @@ -77,52 +85,6 @@ def current_directory(self, value: str) -> None: """Set current directory path.""" self._curr_dir = value - def reload_gcodes_folder(self) -> None: - """Request reload of the gcodes folder from root.""" - logger.debug("Reloading gcodes folder") - self.request_dir_info[str].emit("") - - def clear_files_data(self) -> None: - """Clear all cached file data.""" - self._files_data.clear() - self._pending_metadata_requests.clear() - self._metadata_retry_count.clear() - - def retry_metadata_request(self, file_path: str) -> bool: - """ - Request metadata with a maximum of 3 retries per file. - Args: - file_path: Path to the file - Returns: - True if request was made, False if max retries reached - """ - clean_path = file_path.removeprefix("/") - - if not clean_path.lower().endswith(self.GCODE_EXTENSION): - return False - - current_count = self._metadata_retry_count.get(clean_path, 0) - - if current_count > 3: - # Maximum 3 force scan per file - logger.debug(f"Metadata retry limit reached for: {clean_path}") - return False - - self._metadata_retry_count[clean_path] = current_count + 1 - - if current_count == 0: - # First attempt: regular metadata request - self.request_file_metadata.emit(clean_path) - logger.debug(f"Metadata request (attempt 1) for: {clean_path}") - else: - # Subsequent attempts: force scan - self.request_scan_metadata.emit(clean_path) - logger.debug( - f"Metadata scan (attempt {current_count + 1}) for: {clean_path}" - ) - - return True - @QtCore.pyqtSlot(list, name="on_file_list") def on_file_list(self, file_list: list) -> None: """Handle receiving full files list.""" @@ -133,6 +95,7 @@ def on_file_list(self, file_list: list) -> None: def on_directories(self, directories_data: list) -> None: """Handle receiving full directories list.""" self._directories = directories_data.copy() if directories_data else [] + # _files_data persists: Files emits each file's metadata only once. logger.debug(f"Received {len(self._directories)} directories") if self.isVisible(): @@ -140,80 +103,19 @@ def on_directories(self, directories_data: list) -> None: @QtCore.pyqtSlot(dict, name="on_fileinfo") def on_fileinfo(self, filedata: dict) -> None: - """ - Handle receiving file metadata. - - Updates existing file entry in the list with better info (time, filament). - """ - if not filedata or not self.isVisible(): + """Cache metadata and refresh its row.""" + if not filedata: return - filename = filedata.get("filename", "") - if not filename or not filename.lower().endswith(self.GCODE_EXTENSION): + if not filename.lower().endswith(self.GCODE_EXTENSION): return - - # Cache the file data self._files_data[filename] = filedata - - # Remove from pending requests and reset retry count (success) - self._pending_metadata_requests.discard(filename) - self._metadata_retry_count.pop(filename, None) - - # Check if this file should be displayed in current view - file_dir = self._get_parent_directory(filename) - current = self._curr_dir.removeprefix("/") - - # Both empty = root directory, otherwise must match exactly - if file_dir != current: - return - - display_name = self._get_display_name(filename) - item = self._create_file_list_item(filedata) - if not item: - return - - self._model.remove_item_by_text(display_name) - - # Find correct position (sorted by modification time, newest first) - insert_position = self._find_file_insert_position(filedata.get("modified", 0)) - self._model.insert_item(insert_position, item) - - logger.debug(f"Updated file in list: {display_name}") - - @QtCore.pyqtSlot(str, name="on_metadata_error") - def on_metadata_error(self, filename: str) -> None: - """ - Handle metadata request failure. - - Triggers retry with scan_gcode_metadata if under retry limit. - Called when metadata request fails. - """ - if not filename: - return - - clean_filename = filename.removeprefix("/") - - if clean_filename not in self._pending_metadata_requests: - return - - # Try again (will use force scan to the max of 3 times) - if not self.retry_metadata_request(clean_filename): - # Max retries reached, remove from pending - self._pending_metadata_requests.discard(clean_filename) - logger.debug(f"Giving up on metadata for: {clean_filename}") + if self.isVisible(): + self._schedule_rebuild() @QtCore.pyqtSlot(str, list, name="on_usb_files_loaded") def on_usb_files_loaded(self, usb_path: str, files: list) -> None: - """ - Handle preloaded USB files. - - Called when USB files are preloaded in background. - If we're currently viewing this USB, update the display. - - Args: - usb_path: The USB mount path - files: List of file dicts from the USB - """ + """Update display when USB files are preloaded.""" current = self._curr_dir.removeprefix("/") # If we're currently in this USB folder, update the file list @@ -223,157 +125,45 @@ def on_usb_files_loaded(self, usb_path: str, files: list) -> None: self._build_file_list() logger.debug(f"Updated view with preloaded USB files: {usb_path}") - def _find_file_insert_position(self, modified_time: float) -> int: - """ - Find the correct position to insert a new file. - - Files should be: - 1. After all directories - 2. Sorted by modification time (newest first) - - Returns: - The index at which to insert the new file. - """ - insert_pos = 0 - - for i in range(self._model.rowCount()): - index = self._model.index(i) - item = self._model.data(index, QtCore.Qt.ItemDataRole.UserRole) - - if not item: - continue - - # Skip directories (items with left_icon) - if item.left_icon: - insert_pos = i + 1 - continue - - # This is a file - check its modification time - file_key = self._find_file_key_by_display_name(item.text) - if file_key: - file_data = self._files_data.get(file_key, {}) - file_modified = file_data.get("modified", 0) - - # Files are sorted newest first, so insert before older files - if modified_time > file_modified: - return i - - insert_pos = i + 1 - - return insert_pos - - def _find_file_key_by_display_name(self, display_name: str) -> str | None: - """Find the file key in _files_data by its display name.""" - for key in self._files_data: - if self._get_display_name(key) == display_name: - return key - return None - @QtCore.pyqtSlot(dict, name="on_file_added") def on_file_added(self, file_data: dict) -> None: - """ - Handle a single file being added. - - Called when Moonraker sends notify_filelist_changed with create_file action. - Adds file to list immediately, metadata updates later. - """ - path = file_data.get("path", file_data.get("filename", "")) - if not path: - return - - # Normalize paths - path = path.removeprefix("/") - file_dir = self._get_parent_directory(path) + """Add a created file and refresh.""" + path = file_data.get("path", file_data.get("filename", "")).removeprefix("/") current = self._curr_dir.removeprefix("/") - - # Check if file belongs to current directory - if file_dir != current: - logger.debug( - f"File '{path}' (dir: '{file_dir}') not in current directory ('{current}'), skipping" - ) - return - - # Only update UI if visible - if not self.isVisible(): - logger.debug("Widget not visible, will refresh on show") + if not path or helper_methods.get_parent_dir(path) != current: return - - display_name = self._get_display_name(path) - - if not self._model_contains_item(display_name): - # Create basic item with unknown info - modified = file_data.get("modified", 0) - - item = ListItem( - text=display_name, - right_text="Unknown Filament - Unknown time", - right_icon=self._icons.get("right_arrow"), - left_icon=None, - callback=None, - selected=False, - allow_check=False, - _lfontsize=self.LEFT_FONT_SIZE, - _rfontsize=self.RIGHT_FONT_SIZE, - height=self.ITEM_HEIGHT, - notificate=False, - ) - - # Find correct position - insert_position = self._find_file_insert_position(modified) - self._model.insert_item(insert_position, item) - self._hide_placeholder() - logger.debug(f"Added new file to list: {display_name}") - - # Request metadata for gcode files using retry mechanism - if path.lower().endswith(self.GCODE_EXTENSION): - if path not in self._pending_metadata_requests: - self._pending_metadata_requests.add(path) - self.retry_metadata_request(path) - logger.debug(f"Requested metadata for new file: {path}") + # Bare name, as get_directory lists it. + name = helper_methods.get_file_name(path) + if not any( + helper_methods.get_file_name(f.get("filename", f.get("path", ""))) == name + for f in self._file_list + ): + self._file_list.append(file_data | {"filename": name}) + # Files already requests its metadata. + if self.isVisible(): + self._build_file_list() @QtCore.pyqtSlot(str, name="on_file_removed") def on_file_removed(self, filepath: str) -> None: - """ - Handle a file being removed. - - Called when Moonraker sends notify_filelist_changed with delete_file action. - """ + """Drop a deleted file and refresh.""" filepath = filepath.removeprefix("/") - file_dir = self._get_parent_directory(filepath) - current = self._curr_dir.removeprefix("/") - - # Always clean up cache self._files_data.pop(filepath, None) - self._pending_metadata_requests.discard(filepath) - self._metadata_retry_count.pop(filepath, None) - - # Only update UI if visible and in current directory - if not self.isVisible(): - return - - if file_dir != current: - logger.debug( - f"Deleted file '{filepath}' not in current directory ('{current}'), skipping UI update" - ) + if helper_methods.get_parent_dir(filepath) != self._curr_dir.removeprefix("/"): return - - filename = self._get_basename(filepath) - display_name = self._get_display_name(filename) - - # Remove from model - removed = self._model.remove_item_by_text(display_name) - - if removed: - self._check_empty_state() - logger.debug(f"File removed from view: {filepath}") + # _file_list holds bare names. + name = helper_methods.get_file_name(filepath) + self._file_list = [ + f + for f in self._file_list + if helper_methods.get_file_name(f.get("filename", f.get("path", ""))) + != name + ] + if self.isVisible(): + self._build_file_list() @QtCore.pyqtSlot(dict, name="on_file_modified") def on_file_modified(self, file_data: dict) -> None: - """ - Handle a file being modified. - - Called when Moonraker sends notify_filelist_changed with modify_file action. - """ + """Handle file modification from Moonraker.""" path = file_data.get("path", file_data.get("filename", "")) if path: # Remove old entry and request fresh metadata @@ -382,200 +172,93 @@ def on_file_modified(self, file_data: dict) -> None: @QtCore.pyqtSlot(dict, name="on_dir_added") def on_dir_added(self, dir_data: dict) -> None: - """ - Handle a directory being added. - - Called when Moonraker sends notify_filelist_changed with create_dir action. - Inserts the directory in the correct sorted position (alphabetically, after Go Back). - """ - # Extract dirname from path or dirname field - path = dir_data.get("path", "") - dirname = dir_data.get("dirname", "") - - if not dirname and path: - dirname = self._get_basename(path) - + """Add a created dir and refresh.""" + path = dir_data.get("path", "").removeprefix("/") + dirname = dir_data.get("dirname", "") or helper_methods.get_file_name(path) if not dirname or dirname.startswith("."): return - - path = path.removeprefix("/") - parent_dir = self._get_parent_directory(path) if path else "" - current = self._curr_dir.removeprefix("/") - - if parent_dir != current: - logger.debug( - f"Directory '{dirname}' (parent: '{parent_dir}') not in current directory ('{current}'), skipping" - ) - return - - if not self.isVisible(): - logger.debug( - f"Widget not visible, skipping UI update for added dir: {dirname}" - ) - return - - # Check if already exists - if self._model_contains_item(dirname): + parent_dir = helper_methods.get_parent_dir(path) if path else "" + if parent_dir != self._curr_dir.removeprefix("/"): return - - # Ensure dirname is in dir_data - dir_data["dirname"] = dirname - - # Find the correct sorted position for this directory - insert_position = self._find_directory_insert_position(dirname) - - # Create the list item - icon = self._icons.get("folder") - if self._is_usb_directory(self._curr_dir, dirname): - icon = self._icons.get("usb") - - item = ListItem( - text=str(dirname), - left_icon=icon, - right_text="", - right_icon=None, - selected=False, - callback=None, - allow_check=False, - _lfontsize=self.LEFT_FONT_SIZE, - _rfontsize=self.RIGHT_FONT_SIZE, - height=self.ITEM_HEIGHT, - ) - - # Insert at the correct position - self._model.insert_item(insert_position, item) - - self._hide_placeholder() - logger.debug( - f"Directory added to view at position {insert_position}: {dirname}" - ) - - def _find_directory_insert_position(self, new_dirname: str) -> int: - """ - Find the correct position to insert a new directory. - - Directories should be: - 1. After "Go Back" (if present) - 2. Before all files - 3. Sorted alphabetically among other directories - - Returns: - The index at which to insert the new directory. - """ - new_dirname_lower = new_dirname.lower() - insert_pos = 0 - - for i in range(self._model.rowCount()): - index = self._model.index(i) - item = self._model.data(index, QtCore.Qt.ItemDataRole.UserRole) - - if not item: - continue - - # Skip "Go Back" - always stays at top - if item.text == "Go Back": - insert_pos = i + 1 - continue - - # If this item has a left_icon, it's a directory - if item.left_icon: - # Compare alphabetically - if item.text.lower() > new_dirname_lower: - # Found a directory that should come after the new one - return i - else: - # This directory comes before, keep looking - insert_pos = i + 1 - else: - # Hit a file - insert before it (directories come before files) - return i - - # Insert at the end of directories (or end of list if no files) - return insert_pos + if not any(d.get("dirname", "") == dirname for d in self._directories): + self._directories.append(dir_data | {"dirname": dirname}) + if self.isVisible(): + self._build_file_list() @QtCore.pyqtSlot(str, name="on_dir_removed") - def on_dir_removed(self, dirname_or_path: str) -> None: - """ - Handle a directory being removed. - - Called when Moonraker sends notify_filelist_changed with delete_dir action. - Also handles USB mount removal (which Moonraker reports as delete_file). - """ - dirname_or_path = dirname_or_path.removeprefix("/") - dirname = ( - self._get_basename(dirname_or_path) - if "/" in dirname_or_path - else dirname_or_path - ) - - if not dirname: + def on_dir_removed(self, path: str) -> None: + """Drop a deleted dir, or go to root if inside it.""" + rel = path.strip("/") + if not rel: return - - # Check if user is currently inside the removed directory (e.g., USB removed) - current = self._curr_dir.removeprefix("/") - if current == dirname or current.startswith(dirname + "/"): + # Mirror Files._forget_cached so a recreated dir gets fresh metadata. + for key in [k for k in self._files_data if k.startswith(f"{rel}/")]: + del self._files_data[key] + current = self._curr_dir.strip("/") + if current == rel or current.startswith(rel + "/"): logger.warning( - f"Current directory '{current}' was removed, returning to root" + "Current directory '%s' was removed, returning to root", current ) self.on_directory_error() return - - # Skip UI update if not visible - if not self.isVisible(): + if helper_methods.get_parent_dir(rel) != current: return - - removed = self._model.remove_item_by_text(dirname) - - if removed: - self._check_empty_state() - logger.debug(f"Directory removed from view: {dirname}") + name = helper_methods.get_file_name(rel) + self._directories = [d for d in self._directories if d.get("dirname") != name] + if self.isVisible(): + self._build_file_list() @QtCore.pyqtSlot(name="on_full_refresh_needed") def on_full_refresh_needed(self) -> None: - """ - Handle full refresh request. - - Called when Moonraker sends root_update or when major changes occur. - """ + """Refresh display on root update or major changes.""" logger.info("Full refresh requested") self._curr_dir = "" self.request_dir_info[str].emit(self._curr_dir) @QtCore.pyqtSlot(name="on_directory_error") def on_directory_error(self) -> None: - """ - Handle Directory Error. - - Immediately navigates back to root gcodes folder. - Call this from MainWindow when detecting USB removal or directory errors. - """ + """Go back to root after a directory error.""" logger.info("Directory Error - returning to root directory") - - # Reset to root directory self._curr_dir = "" - - # Clear any pending actions self._pending_action = False - self._pending_metadata_requests.clear() # Request fresh data for root directory self.request_dir_info[str].emit("") + @QtCore.pyqtSlot(str, str, name="on_usb_added") + def on_usb_added(self, _device_path: str = "", _symlink: str = "") -> None: + """Re-fetch the current dir so a new USB folder shows.""" + logger.info("USB mounted, refreshing current directory") + self.request_dir_info[str].emit(self._curr_dir) + QtCore.QTimer.singleShot(self.USB_SETTLE_MS, self._refresh_current_dir) + + def _refresh_current_dir(self) -> None: + """Re-request the current directory.""" + self.request_dir_info[str].emit(self._curr_dir) + + @QtCore.pyqtSlot(str, name="on_usb_removed") + def on_usb_removed(self, symlink: str = "") -> None: + """Leave a removed USB folder, else refresh to hide it.""" + drive = self._curr_dir.strip("/").split("/", 1)[0] + gone = helper_methods.get_file_name(symlink) + if helper_methods.is_usb_mount(drive) and gone in ("", drive): + logger.info("USB removed while inside its folder, returning to root") + self.on_directory_error() + else: + logger.info("USB removed, refreshing current directory") + self.request_dir_info[str].emit(self._curr_dir) + @QtCore.pyqtSlot(ListItem, name="on_item_selected") def _on_item_selected(self, item: ListItem) -> None: """Handle list item selection.""" if not item.left_icon: - # File selected (files don't have left icon) - filename = self._build_filepath(item.text + self.GCODE_EXTENSION) - self._on_file_item_clicked(filename) + # Only files lack a left icon. + filename = self._selected_file_path(item.text) + if filename: + self._on_file_item_clicked(filename) elif item.text == "Go Back": - # Go back selected - go_back_path = self._get_parent_directory(self._curr_dir) - if go_back_path == "/": - go_back_path = "" - self._on_go_back_dir(go_back_path) + self._on_go_back_dir(helper_methods.get_parent_dir(self._curr_dir)) else: - # Directory selected self._on_dir_item_clicked("/" + item.text) @QtCore.pyqtSlot(name="reset_dir") @@ -590,187 +273,174 @@ def showEvent(self, event: QtGui.QShowEvent) -> None: self.request_dir_info[str].emit(self._curr_dir) super().showEvent(event) - def hideEvent(self, event: QtGui.QHideEvent) -> None: - """Handle widget being hidden.""" - # Clear pending requests when hidden - self._pending_metadata_requests.clear() - super().hideEvent(event) + @staticmethod + def _item_key(item: ListItem) -> str: + """Reconcile identity: row kind plus text.""" + return f"{'d' if item.left_icon else 'f'}:{item.text}" + + def _schedule_rebuild(self) -> None: + """Coalesce metadata bursts into one deferred rebuild.""" + if self._rebuild_pending: + return + self._rebuild_pending = True + QtCore.QTimer.singleShot(0, self._flush_rebuild) + + def _flush_rebuild(self) -> None: + """Run the coalesced rebuild if still visible.""" + self._rebuild_pending = False + if self.isVisible(): + self._build_file_list() def _build_file_list(self) -> None: - """Build the complete file list display.""" - self._list_widget.blockSignals(True) - self._model.clear() - self._entry_delegate.clear() + """Rebuild the model via keyed reconcile.""" self._pending_action = False - self._pending_metadata_requests.clear() - self._metadata_retry_count.clear() - - # Determine if we're in root directory + meta = self._metadata_map() is_root = not self._curr_dir or self._curr_dir == "/" - - # Check for empty state in root directory - if not self._file_list and not self._directories and is_root: + if is_root and not self._file_list and not self._directories: + self._model.clear() + self._entry_delegate.clear() self._show_placeholder() - self._list_widget.blockSignals(False) return - - # We have content (or we're in a subdirectory), hide placeholder self._hide_placeholder() - - # Add back button if not in root + self._model.reconcile(self._desired_items(is_root, meta), self._item_key) + + def _metadata_map(self) -> dict[str, dict | None]: + """Bare name -> cached metadata, built once per rebuild.""" + meta: dict[str, dict | None] = {} + for f in self._file_list: + name = f.get("filename", f.get("path", "")) + meta[name] = self._files_data.get(self._build_filepath(name)) + return meta + + def _lookup_meta(self, filename: str, meta: dict | None) -> dict | None: + """Cached metadata from the prebuilt map, else a direct lookup.""" + if meta is not None and filename in meta: + return meta[filename] + return self._files_data.get(self._build_filepath(filename)) + + def _desired_items(self, is_root: bool, meta: dict) -> list[ListItem]: + """Rows: Go Back (subdirs), dirs A-Z, then sorted files.""" + items: list[ListItem] = [] if not is_root: - self._add_back_folder_entry() - - # Add directories (sorted alphabetically) - sorted_dirs = sorted( - self._directories, key=lambda x: x.get("dirname", "").lower() + items.append(self._make_back_folder_item()) + items.extend(self._desired_directory_items()) + items.extend(self._desired_file_items(meta)) + return items + + def _desired_directory_items(self) -> list[ListItem]: + """Dir rows A-Z, skipping dot-dirs.""" + rows = sorted(self._directories, key=lambda d: d.get("dirname", "").lower()) + return [ + self._make_directory_item(d) + for d in rows + if d.get("dirname", "") and not d["dirname"].startswith(".") + ] + + def _desired_file_items(self, meta: dict) -> list[ListItem]: + """File rows in the active sort order.""" + files = [ + f + for f in self._file_list + if f.get("filename", f.get("path", "")) + .lower() + .endswith(self.GCODE_EXTENSION) + ] + # Stable sorts: upload time orders the ties, e.g. never-printed files. + files.sort(key=self._upload_time, reverse=self._sort_descending) + files.sort( + key=lambda f: self._sort_value(f, meta), reverse=self._sort_descending ) - for dir_data in sorted_dirs: - dirname = dir_data.get("dirname", "") - if dirname and not dirname.startswith("."): - self._add_directory_list_item(dir_data) - - # Add files immediately (sorted by modification time, newest first) - sorted_files = sorted( - self._file_list, key=lambda x: x.get("modified", 0), reverse=True - ) - for file_item in sorted_files: - filename = file_item.get("filename", file_item.get("path", "")) - if not filename: - continue - - # Add file to list immediately with basic info - self._add_file_to_list(file_item) - - # Request metadata for gcode files (will update display later) - if filename.lower().endswith(self.GCODE_EXTENSION): - self._request_file_info(file_item) - - self._list_widget.blockSignals(False) - self._list_widget.update() - - def _delayed_scrollbar_update(self) -> None: - """Update scrollbar after model changes.""" - QtCore.QTimer.singleShot(10, self._setup_scrollbar) - - def _add_file_to_list(self, file_item: dict) -> None: - """Add a file entry to the list with basic info.""" - filename = file_item.get("filename", file_item.get("path", "")) - if not filename or not filename.lower().endswith(self.GCODE_EXTENSION): - return - - # Get display name - display_name = self._get_display_name(filename) - - # Check if already in list - if self._model_contains_item(display_name): - return - - # Use cached metadata if available, otherwise show unknown - full_path = self._build_filepath(filename) - cached = self._files_data.get(full_path) - - if cached: - item = self._create_file_list_item(cached) - else: - item = ListItem( - text=display_name, - right_text="Unknown Filament - Unknown time", - right_icon=self._icons.get("right_arrow"), - left_icon=None, - callback=None, - selected=False, - allow_check=False, - _lfontsize=self.LEFT_FONT_SIZE, - _rfontsize=self.RIGHT_FONT_SIZE, - height=self.ITEM_HEIGHT, - notificate=False, - ) - - if item: - self._model.add_item(item) - - def _create_file_list_item(self, filedata: dict) -> ListItem | None: - """Create a ListItem from file metadata.""" - filename = filedata.get("filename", "") - if not filename: - return None - - # Format estimated time - estimated_time = filedata.get("estimated_time", 0) - seconds = int(estimated_time) if isinstance(estimated_time, (int, float)) else 0 - time_str = self._format_print_time(seconds) - - # Get filament type - filament_type = filedata.get("filament_type") - if isinstance(filament_type, str): - text = filament_type.strip() - if text.startswith("[") and text.endswith("]"): - try: - types = json.loads(text) - except json.JSONDecodeError: - types = [text] - else: - types = [text] - else: - types = filament_type or [] - - if not isinstance(types, list): - types = [types] - - filament_type = ",".join(dict.fromkeys(types)) - - if not filament_type or filament_type == -1.0 or filament_type == "Unknown": - filament_type = "Unknown Filament" - - display_name = self._get_display_name(filename) + return [ + self._make_file_item(f.get("filename", f.get("path", "")), meta) + for f in files + ] - return ListItem( - text=display_name, - right_text=f"{filament_type} - {time_str}", - right_icon=self._icons.get("right_arrow"), - left_icon=None, # Files have no left icon - callback=None, - selected=False, - allow_check=False, - _lfontsize=self.LEFT_FONT_SIZE, - _rfontsize=self.RIGHT_FONT_SIZE, - height=self.ITEM_HEIGHT, - notificate=False, + @staticmethod + def _upload_time(filedata: dict) -> float: + """File mtime; Moonraker lists in os.listdir order, not upload order.""" + modified = filedata.get("modified", 0) + return modified if isinstance(modified, (int, float)) else 0 + + def _sort_value(self, filedata: dict, meta: dict) -> float | str: + """Sort key for the active column, one type per column.""" + name = filedata.get("filename", filedata.get("path", "")) + if self._sort_key == "Import Order": + return self._upload_time(filedata) + if self._sort_key == "Last Print": + cached = self._lookup_meta(name, meta) or {} + return cached.get("print_start_time") or 0 + if self._sort_key == "Print Time": + cached = self._lookup_meta(name, meta) or {} + est = cached.get("estimated_time", 0) + return est if isinstance(est, (int, float)) else 0 + if self._sort_key == "Nozzle Size": + cached = self._lookup_meta(name, meta) or {} + nozzle = cached.get("nozzle_diameter", -1.0) + return nozzle if isinstance(nozzle, (int, float)) else -1.0 + if self._sort_key == "Filament": + cached = self._lookup_meta(name, meta) or {} + return self._filament_label(cached.get("filament_type")).lower() + return name.lower() + + def _on_sort_key_changed(self, sort_key: str) -> None: + """Apply the chosen sort column.""" + self._sort_key = sort_key or self.SORTING_TYPES[0] + self._build_file_list() + + def _on_sort_order_toggled(self) -> None: + """Flip the sort direction.""" + self._sort_descending = not self._sort_descending + self._update_sort_order_icon() + self._build_file_list() + + def _update_sort_order_icon(self) -> None: + """Match the toggle icon to the sort direction.""" + key = "sort_desc" if self._sort_descending else "sort_asc" + self._sort_order_btn.setProperty( + "icon_pixmap", QtGui.QPixmap(self.ICON_PATHS[key]) ) + self._sort_order_btn.update() - def _add_directory_list_item(self, dir_data: dict) -> None: - """Add a directory entry to the list.""" - dir_name = dir_data.get("dirname", "") - if not dir_name: - return + def _make_back_folder_item(self) -> ListItem: + """The Go Back row.""" + return self._row("Go Back", "", self._icons.get("back_folder"), None) - # Choose appropriate icon + def _make_directory_item(self, dir_data: dict) -> ListItem: + """Dir row; root USB mounts get the USB icon.""" + name = str(dir_data.get("dirname", "")) icon = self._icons.get("folder") - if self._is_usb_directory(self._curr_dir, dir_name): + if not self._curr_dir and helper_methods.is_usb_mount(name): icon = self._icons.get("usb") - - item = ListItem( - text=str(dir_name), - left_icon=icon, - right_text="", - right_icon=None, - selected=False, - callback=None, - allow_check=False, - _lfontsize=self.LEFT_FONT_SIZE, - _rfontsize=self.RIGHT_FONT_SIZE, - height=self.ITEM_HEIGHT, + return self._row(name, "", icon, None) + + def _make_file_item(self, filename: str, meta: dict | None = None) -> ListItem: + """File row from cached metadata, else Unknown.""" + cached = self._lookup_meta(filename, meta) + right = ( + self._format_file_meta(cached) + if cached + else "Unknown Filament - Unknown time" + ) + return self._row( + self._get_display_name(filename), + right, + None, + self._icons.get("right_arrow"), ) - self._model.add_item(item) - - def _add_back_folder_entry(self) -> None: - """Add the 'Go Back' navigation entry.""" - item = ListItem( - text="Go Back", - right_text="", - right_icon=None, - left_icon=self._icons.get("back_folder"), + + def _row( + self, + text: str, + right_text: str, + left_icon: QtGui.QPixmap | None, + right_icon: QtGui.QPixmap | None, + ) -> ListItem: + """ListItem with this page's sizing and flags.""" + return ListItem( + text=text, + right_text=right_text, + left_icon=left_icon, + right_icon=right_icon, callback=None, selected=False, allow_check=False, @@ -779,25 +449,43 @@ def _add_back_folder_entry(self) -> None: height=self.ITEM_HEIGHT, notificate=False, ) - self._model.add_item(item) - - def _request_file_info(self, file_data_item: dict) -> None: - """Request metadata for a file item using retry mechanism.""" - if not file_data_item: - return - name = file_data_item.get("path", file_data_item.get("filename", "")) - if not name or not name.lower().endswith(self.GCODE_EXTENSION): - return + def _format_file_meta(self, filedata: dict) -> str: + """' -