diff --git a/BlocksScreen/devices/storage/udisks2.py b/BlocksScreen/devices/storage/udisks2.py index d3c19a61..91af50b8 100644 --- a/BlocksScreen/devices/storage/udisks2.py +++ b/BlocksScreen/devices/storage/udisks2.py @@ -24,6 +24,8 @@ UDisks2_service: str = "org.freedesktop.UDisks2" UDisks2_obj_path: str = "org/freedesktop/UDisks2" AlreadyMountedException = "org.freedesktop.UDisks2.Error.AlreadyMounted" +# Names add_symlink can produce, the only links this module is allowed to reap. +USB_LINK_PREFIXES: tuple[str, ...] = ("USB-", "USB DRIVE") _T = typing.TypeVar(name="_T") @@ -430,7 +432,10 @@ async def _rem_interface_listener(self) -> None: device: Device = self.controlled_devs.pop(path) device.kill() del device + # Clean first: a refresh fired now would still list the symlink. + 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 +461,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: + """Publish a new USB folder, without this the file view never learns it exists.""" + 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 +487,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 +505,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( @@ -583,10 +605,21 @@ 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: + """A USB symlink is live only while its target is still a real mountpoint.""" + if not os.path.exists(link): + return False + if not link.name.startswith(USB_LINK_PREFIXES): + return True + return os.path.ismount(os.path.realpath(link)) + def _cleanup_broken_symlinks(self) -> None: + """Reap USB symlinks left behind by an unmount and tell the UI they are gone.""" 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/devices/storage/usb_controller.py b/BlocksScreen/devices/storage/usb_controller.py index c63badc3..a39ac100 100644 --- a/BlocksScreen/devices/storage/usb_controller.py +++ b/BlocksScreen/devices/storage/usb_controller.py @@ -40,7 +40,7 @@ def __init__(self, parent: QtCore.QObject, gcodes_dir: str | None) -> None: super().__init__(parent) self.gcodes_dir: pathlib.Path = ( - pathlib.Path(gcodes_dir) + pathlib.Path(gcodes_dir).expanduser() if gcodes_dir else pathlib.Path.home() / "printer_data" / "gcodes" ) diff --git a/BlocksScreen/helper_methods.py b/BlocksScreen/helper_methods.py index 114ff5f1..ca9ed443 100644 --- a/BlocksScreen/helper_methods.py +++ b/BlocksScreen/helper_methods.py @@ -16,6 +16,9 @@ logger = logging.getLogger(__name__) +# Symlink names udisks2.add_symlink can produce, kept in sync with USB_LINK_PREFIXES there. +USB_LINK_PREFIXES: tuple[str, ...] = ("USB-", "USB DRIVE") + try: ctypes.cdll.LoadLibrary("libXext.so.6") libxext = ctypes.CDLL("libXext.so.6") @@ -294,6 +297,23 @@ def estimate_print_time(seconds: int) -> list[int]: return [days, hours, mins, secs] +def format_duration(seconds: int) -> str: + """Human "1d 2h 3m" duration; sub-minute values render as seconds.""" + 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: + """Filament mass in grams, switching to 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 +364,23 @@ 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: + """Return the parent directory of *path*, POSIX-style (for root-level).""" + parent = pathlib.PurePosixPath(path.removeprefix("/")).parent + return "" if str(parent) == "." else str(parent) + + +def is_usb_mount(path: str) -> bool: + """Return True if *path* is a top-level USB mount, under either name add_symlink gives.""" + name = path.strip("/") + return bool(name) and "/" not in name and name.startswith(USB_LINK_PREFIXES) + + +def resolve_thumbnail_path( + gcode_root: pathlib.Path, requested_path: str, relative_path: str +) -> pathlib.Path: + """Resolve a thumbnails absolute path from the *requested* gcode path""" + 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 412f0648..f9927772 100644 --- a/BlocksScreen/lib/files.py +++ b/BlocksScreen/lib/files.py @@ -3,19 +3,21 @@ 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 events import ReceivedFileData 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() @@ -31,30 +33,19 @@ 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. - """ + """Data class for file metadata; thumbnails held as filesystem path strings.""" filename: str = "" - thumbnail_images: list[QtGui.QImage] = field(default_factory=list) - filament_total: typing.Union[dict, str, float] = field(default_factory=dict) + 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 total_layer: int = -1 @@ -65,56 +56,25 @@ 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" slicer_version: str = "Unknown" gcode_start_byte: int = 0 gcode_end_byte: int = 0 - print_start_time: typing.Optional[float] = None - job_id: typing.Optional[str] = None + print_start_time: float | None = None + job_id: str | None = None + 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, - } + """All fields as a plain dict for signal emission (deep-copied containers).""" + 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 @@ -126,7 +86,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), @@ -138,9 +98,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", @@ -149,28 +108,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 - """ + """Manages gcode files with event-driven updates; emits signals for directory/file changes.""" # 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") @@ -187,12 +137,16 @@ 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") + # Emitted from the websocket thread to hop the history reply onto this thread. + _history_job = QtCore.pyqtSignal(str, dict, name="history_job") + _thumbnails_ready = QtCore.pyqtSignal(str, object, name="thumbnails_ready") # Signal for preloaded USB files usb_files_loaded = QtCore.pyqtSignal( str, list, name="usb_files_loaded" ) # (usb_path, files) GCODE_EXTENSION = ".gcode" + USB_PREFIX = "USB-" GCODE_PATH = "~/printer_data/gcodes" def __init__(self, parent: QtCore.QObject, ws: MoonWebSocket) -> None: @@ -203,6 +157,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() @@ -211,18 +166,24 @@ 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() + # Client-side USB metadata: size/modified staged per path, plus lazy loader. + self._usb_meta_base: dict[str, dict] = {} + self._meta_loader: gcode_loader.GcodeMetadataLoader | None = None + # Files already probed via server.files.thumbnails (ask once). + self._thumbs_probed: set[str] = set() 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) + self._thumbnails_ready.connect(self._on_thumbnails) def _install_event_filter(self) -> None: """Install event filter on application instance.""" @@ -250,29 +211,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) -> typing.Optional[FileMetadata]: - """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") @@ -280,19 +218,16 @@ def initial_load(self) -> None: self.request_dir_info[str, bool].emit("", True) def handle_filelist_changed(self, data: typing.Union[dict, list]) -> None: - """Handle notify_filelist_changed from Moonraker.""" + """Handle notify_filelist_changed from Moonraker; 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 notification entry to its action handler.""" action_str = data.get("action", "") action = FileAction.from_string(action_str) item = data.get("item", {}) @@ -321,7 +256,7 @@ def _handle_file_created(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_created(item, {}) return @@ -342,7 +277,7 @@ 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 @@ -356,7 +291,16 @@ 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 + + # A USB symlink at the gcodes root reaches us as a file event, often modify_file. + 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._files[path] = item @@ -392,7 +336,7 @@ 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): + if helper_methods.is_usb_mount(dirname): self._preload_usb_contents(dirname) def _handle_dir_deleted(self, item: dict, _: dict) -> None: @@ -409,7 +353,7 @@ def _handle_dir_deleted(self, item: dict, _: dict) -> None: self._directories.pop(dirname, None) # Clear USB cache if this was a USB mount - if self._is_usb_mount(dirname): + if helper_methods.is_usb_mount(dirname): self._usb_files_cache.pop(dirname, None) self._pending_usb_preloads.discard(dirname) if dirname in self._usb_preload_queue: @@ -430,78 +374,147 @@ 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: + """Gcodes-root-relative dir from the request entry [method, params, callback].""" + 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: + """Full gcode-root-relative path from a bare dir-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 (thumbnails resolved from full path) 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 keeps the USB-/subdir prefix so .thumbs resolve on the drive. base_dir = (self.gcode_path / filename).parent 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(base_dir / t["relative_path"]) + for t in data.get("thumbnails", []) + if isinstance(t.get("relative_path"), str) and t["relative_path"] ] + metadata = FileMetadata.from_dict(data, thumbnail_paths) + self._files_metadata[filename] = metadata + self._metadata_retry_count.pop(filename.removeprefix("/"), None) + self.fileinfo.emit(metadata.to_dict()) + self._request_print_duration(filename, metadata) + self._request_thumbnails(filename, metadata) + logger.debug("Metadata loaded: %s", filename) - # 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) + def _request_print_duration(self, filename: str, metadata: FileMetadata) -> None: + """Ask history for how long this file's last job actually took.""" + if not metadata.job_id or metadata.print_duration is not None: + return + self.ws.api.history_get_job( + str(metadata.job_id), + lambda result, name=filename: self._history_job.emit(name, result or {}), + ) - metadata = FileMetadata.from_dict(data, thumbnail_images) - self._files_metadata[filename] = metadata + @QtCore.pyqtSlot(str, dict, name="on_history_job") + def _on_history_job(self, filename: str, result: dict) -> None: + """Merge the elapsed print time into the cached metadata and re-emit it.""" + metadata = self._files_metadata.get(filename) + job = result.get("job") or {} + duration = job.get("print_duration") + # Cancelled/errored/in-progress jobs stopped early, so their elapsed time lies. + if job.get("status") != "completed": + return + if metadata is None or not isinstance(duration, (int, float)) or duration <= 0: + return + updated = replace(metadata, print_duration=float(duration)) + self._files_metadata[filename] = updated + self.fileinfo.emit(updated.to_dict()) - # Emit updated fileinfo - self.fileinfo.emit(metadata.to_dict()) - logger.debug(f"Metadata loaded for: {filename}") + def _request_thumbnails(self, filename: str, metadata: FileMetadata) -> None: + """Ask Moonraker for extracted thumbnail paths, cheaper than a header fetch.""" + if metadata.thumbnail_paths or self._is_usb_gcode(filename): + return + if filename in self._thumbs_probed: + return + self._thumbs_probed.add(filename) + self.ws.api.get_gcode_thumbnail( + filename.removeprefix("/"), + lambda result, name=filename: self._thumbnails_ready.emit(name, result), + ) - def handle_metadata_error(self, error_data: typing.Union[str, dict]) -> None: - """ - Handle metadata request error from Moonraker. + @QtCore.pyqtSlot(str, object, name="on_thumbnails") + def _on_thumbnails(self, filename: str, result: object) -> None: + """Store thumbnail paths (gcode-root relative) and re-emit the metadata.""" + metadata = self._files_metadata.get(filename) + entries = result if isinstance(result, list) else [] + paths = [ + str(self.gcode_path / t["thumbnail_path"]) + for t in entries + if isinstance(t, dict) and isinstance(t.get("thumbnail_path"), str) + ] + if metadata is None or not paths: + return + updated = replace(metadata, thumbnail_paths=paths) + 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 an extended dir entry carries real metadata, not just a thumbnail.""" + return "estimated_time" in file_data + + def _is_usb_gcode(self, full_path: str) -> bool: + """True for gcode files living under a USB mount symlink.""" + return full_path.removeprefix("/").startswith(self.USB_PREFIX) + + def _usb_metadata_loader(self) -> gcode_loader.GcodeMetadataLoader: + """Lazily create + wire the client-side USB metadata loader (once).""" + 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: + """USB gcodes: client-parse (Moonraker can't scan them); else ask Moonraker.""" + if not self._is_usb_gcode(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 client-parsed USB metadata through the standard pipeline.""" + base = self._usb_meta_base.pop(full_path, {}) + self._process_metadata(base | meta, full_path) - Args: - error_data: The error message string or dict from Moonraker - """ + def handle_metadata_error(self, error_data: typing.Union[str, dict]) -> None: + """Handle metadata request error from Moonraker and emit signal.""" if not error_data: return @@ -518,48 +531,31 @@ def handle_metadata_error(self, error_data: typing.Union[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.""" 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) -> typing.Optional[list[dict]]: - """ - 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", "")) @@ -568,54 +564,67 @@ def _process_usb_directory_info(self, usb_path: str, data: dict) -> None: full_path = f"{usb_path}/{filename}" if filename.lower().endswith(self.GCODE_EXTENSION): - self.request_file_metadata.emit(full_path) + 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: + """Return the pending USB preload whose path matches 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 a USB symlink under files; treat it as a browsable dir. + 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: + """Use inline gcode metadata; request (full path) only what is missing.""" + 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._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: @@ -646,7 +655,7 @@ def on_request_fileinfo(self, filename: str) -> None: @QtCore.pyqtSlot(str, bool, name="get_dir_info") def get_dir_information( self, directory: str = "", extended: bool = True - ) -> typing.Optional[list]: + ) -> typing.Any: """Get directory information.""" self._current_directory = directory diff --git a/BlocksScreen/lib/moonrakerComm.py b/BlocksScreen/lib/moonrakerComm.py index de44ef0b..aca33d9e 100644 --- a/BlocksScreen/lib/moonrakerComm.py +++ b/BlocksScreen/lib/moonrakerComm.py @@ -554,12 +554,14 @@ def scan_gcode_metadata(self, filename_dir: str): ) @QtCore.pyqtSlot(name="api_get_gcode_thumbnail") - def get_gcode_thumbnail(self, filename_dir: str): - """Request gcode thumbnail""" - if isinstance(filename_dir, str) is False or filename_dir is None: + def get_gcode_thumbnail(self, filename_dir: str, callback=None) -> bool: + """Request the thumbnail paths Moonraker already extracted for a gcode.""" + if not isinstance(filename_dir, str): return False return self._ws.send_request( - method="server.files.thumbnails", params={"filename": filename_dir} + method="server.files.thumbnails", + params={"filename": filename_dir}, + callback=callback, ) @QtCore.pyqtSlot(str, str, name="api-delete-file") @@ -829,9 +831,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 past job entry; 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 e1e61bf4..cd8d97c0 100644 --- a/BlocksScreen/lib/moonrest.py +++ b/BlocksScreen/lib/moonrest.py @@ -27,6 +27,7 @@ import logging +from urllib.parse import quote import requests from requests import Request, Response @@ -44,18 +45,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 - """ + """MoonRest API for GET/POST requests to Moonraker.""" 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): @@ -63,12 +62,7 @@ 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 - """ + """Request oneshot token from Moonraker for API key authentication.""" # Response data is generally an object itself, however for some requests this may simply be an "ok" string. response = self.get_request(method="access/oneshot_token") if response is None: @@ -80,40 +74,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. - """ + """Fetch spool info from Moonraker; None on any 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. - """ + """Update spool used_weight via Moonraker; 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): @@ -135,6 +114,49 @@ 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: + """GET the first *max_bytes* of a gcode file (embedded-thumbnail parse).""" + url = f"{self.build_endpoint}/server/files/gcodes/{quote(rel_path)}" + headers = {"Range": f"bytes=0-{max_bytes - 1}"} + 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) >= max_bytes: + break + return bytes(data) + except Exception as exc: + logger.info("gcode header fetch failed for %s: %s", rel_path, exc) + return None + + def get_gcode_tail(self, rel_path: str, max_bytes: int = 65536) -> bytes | None: + """GET the last *max_bytes* of a gcode file (slicer metadata-footer parse).""" + url = f"{self.build_endpoint}/server/files/gcodes/{quote(rel_path)}" + headers = {"Range": f"bytes=-{max_bytes}"} + 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() + # 200 (Range ignored) streams whole file; keep bounded trailing window to cap RAM. + data = bytearray() + for chunk in resp.iter_content(chunk_size=65536): + data.extend(chunk) + if len(data) > 2 * max_bytes: + del data[:-max_bytes] + return bytes(data[-max_bytes:]) + except Exception as exc: + logger.info("gcode tail fetch failed for %s: %s", 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 035f27f8..6b816d14 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 + ) + # Only usb_unmounted, it fires per reaped symlink so the listing really changed. + 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 07c98bed..2de81c66 100644 --- a/BlocksScreen/lib/panels/networkWindow.py +++ b/BlocksScreen/lib/panels/networkWindow.py @@ -33,6 +33,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 @@ -44,12 +45,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] = {} @@ -143,11 +139,7 @@ def _on_text_changed(self, text: str) -> None: 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") @@ -233,13 +225,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: @@ -440,11 +426,7 @@ def _on_network_state_changed(self, state: NetworkState) -> 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 @@ -599,14 +581,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: @@ -623,16 +598,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 @@ -718,8 +684,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 @@ -744,12 +709,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): @@ -761,12 +721,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 @@ -842,11 +797,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") @@ -1125,13 +1076,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() @@ -1286,12 +1231,7 @@ def _on_wifi_static_ip_clicked(self) -> None: self.setCurrentIndex(self.indexOf(self.wifi_static_ip_page)) 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() @@ -1324,10 +1264,7 @@ def _on_wifi_static_ip_apply(self) -> None: self._nm.request_state_soon(delay_ms=3000) 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 @@ -1338,13 +1275,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] = [] @@ -1381,11 +1312,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 @@ -1394,13 +1321,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 — - no QPixmap lookup, 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): @@ -1683,11 +1604,7 @@ def _on_delete_network(self) -> None: self._nm.delete_network(ssid) 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() @@ -1948,27 +1865,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..4628ce66 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. - - """ + """Stacked widget with main, files, confirm, print, tune, babystep, and filament change 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 + # Shared embedded-thumbnail fallback for read-only USB drives. + 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,23 @@ 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.metadataPage_widget.back_btn.clicked.connect(self.back_button) self.file_data.fileinfo.connect(self.filesPage_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 +149,12 @@ 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.request_file_info.connect( self.file_data.on_request_fileinfo ) @@ -295,9 +280,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 +412,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 +429,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 dd33b52e..c8509330 100644 --- a/BlocksScreen/lib/panels/widgets/cancelPage.py +++ b/BlocksScreen/lib/panels/widgets/cancelPage.py @@ -1,5 +1,6 @@ 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 import typing @@ -8,10 +9,7 @@ class CancelPage(QtWidgets.QWidget): - """Update GUI Page, - retrieves from moonraker available clients and adds functionality - for updating or recovering them - """ + """Cancel print page with client list and recovery functionality.""" request_file_info: typing.ClassVar[QtCore.pyqtSignal] = QtCore.pyqtSignal( str, name="request_file_info" @@ -99,21 +97,26 @@ def set_file_name(self, file_name: str) -> None: def _show_screen_thumbnail(self, dict): try: - thumbnails = dict["thumbnail_images"] + thumbnails = dict["thumbnail_paths"] - last_thumb = QtGui.QPixmap.fromImage(thumbnails[-1]) + last_thumb = QtGui.QPixmap(thumbnails[-1]) if last_thumb.isNull(): - last_thumb = QtGui.QPixmap( - "BlocksScreen/lib/ui/resources/media/logoblocks400x300.png" - ) + last_thumb = self._embedded_pixmap(dict.get("filename", "")) except Exception as e: print(e) - last_thumb = QtGui.QPixmap( - "BlocksScreen/lib/ui/resources/media/logoblocks400x300.png" - ) + last_thumb = self._embedded_pixmap(dict.get("filename", "")) self.set_pixmap(last_thumb) + def _embedded_pixmap(self, gcode_path: str) -> QtGui.QPixmap: + """Cached embedded thumbnail (read-only USB fallback), else the logo placeholder.""" + pixmap = gcode_loader.cached_pixmap(gcode_path) + if pixmap is not None: + return pixmap + return QtGui.QPixmap( + "BlocksScreen/lib/ui/resources/media/logoblocks400x300.png" + ) + def _setupUI(self) -> None: """Setup widget ui""" sizePolicy = QtWidgets.QSizePolicy( diff --git a/BlocksScreen/lib/panels/widgets/confirmPage.py b/BlocksScreen/lib/panels/widgets/confirmPage.py index 0f35ba39..c05246f6 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 @@ -19,6 +20,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) @@ -27,14 +31,20 @@ def __init__(self, parent) -> None: self.setAttribute(QtCore.Qt.WidgetAttribute.WA_AcceptTouchEvents, True) self.thumbnail: QtGui.QImage = self._blocksthumbnail self._thumbnails: typing.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) ) @@ -49,53 +59,49 @@ def on_show_widget(self, text: str, filedata: dict | None = None) -> None: self.directory = directory self.filename = filename self.cf_file_name.setText(self.filename) - self._thumbnails = filedata.get("thumbnail_images", []) - if self._thumbnails: - _biggest_thumbnail = self._thumbnails[-1] # Show last which is biggest - self.thumbnail = QtGui.QImage(_biggest_thumbnail) - else: - self.thumbnail = self._blocksthumbnail - _total_filament = filedata.get("filament_weight_total") - _estimated_time = filedata.get("estimated_time") - if isinstance(_estimated_time, str): - seconds = 0 - else: - seconds = _estimated_time - - days, hours, minutes, _ = helper_methods.estimate_print_time(seconds) - if seconds <= 0: - time_str = "??" - 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(f"{filament_label}") - self.cf_info_tr.setText(f"{time_label}") + self._current_gcode = text.removeprefix("/") + self._filedata = filedata + self._thumbnails = filedata.get("thumbnail_paths", []) + self.thumbnail = self._resolve_thumbnail() + weight = filedata.get("filament_weight_total") + estimated = filedata.get("estimated_time") + seconds = int(estimated) if isinstance(estimated, (int, float)) else 0 + time_str = helper_methods.format_duration(seconds) if seconds > 0 else "??" + 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}") self.repaint() - def estimate_print_time(self, seconds: int) -> list: - """Convert time in seconds format to days, hours, minutes, seconds. - - Args: - seconds (int): Seconds + def _resolve_thumbnail(self) -> QtGui.QImage: + """Biggest on-disk thumbnail, else a cached/queued embedded one, 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: object) -> None: + """Swap in an embedded thumbnail that arrived for the shown file.""" + if gcode_path == self._current_gcode and image and not image.isNull(): + self.thumbnail = image + self.repaint() - Returns: - list: list that contains the converted information [days, hours, minutes, seconds] - """ + def estimate_print_time(self, seconds: int) -> list: + """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) @@ -164,14 +170,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, @@ -190,6 +188,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 969399ac..538eb9ab 100644 --- a/BlocksScreen/lib/panels/widgets/filesPage.py +++ b/BlocksScreen/lib/panels/widgets/filesPage.py @@ -3,6 +3,7 @@ import typing 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 @@ -15,29 +16,38 @@ 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 the fresh symlink asynchronously, so refresh once more after it. + 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", "folder": ":/ui/media/btn_icons/folderIcon.svg", - "right_arrow": ":/arrow_icons/media/btn_icons/right_arrow.svg", + "right_arrow": ":/arrow_icons/media/btn_icons/arrow_right.svg", "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: typing.Optional[QtWidgets.QWidget] = None) -> None: @@ -48,10 +58,9 @@ def __init__(self, parent: typing.Optional[QtWidgets.QWidget] = 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() @@ -78,52 +87,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.""" @@ -134,6 +97,8 @@ 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 [] + # New directory: drop the previous dir's cached metadata (bounds memory). + self._files_data.clear() logger.debug(f"Received {len(self._directories)} directories") if self.isVisible(): @@ -141,80 +106,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 file metadata and refresh the affected row in place.""" + 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 @@ -224,157 +128,44 @@ 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) -> typing.Optional[str]: - """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 to backing data and refresh the current view.""" + 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" - ) + if not path or helper_methods.get_parent_dir(path) != current: return - - # Only update UI if visible - if not self.isVisible(): - logger.debug("Widget not visible, will refresh on show") - 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}") + # Store the bare basename to match the get_directory listing convention. + 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}) + # Metadata is owned + requested by Files (_handle_file_created); no re-request. + 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 from backing data/cache 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" - ) - 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 basenames; match on basename to drop the row. + 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 + ] + current = self._curr_dir.removeprefix("/") + if self.isVisible() and helper_methods.get_parent_dir(filepath) == current: + 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 @@ -383,195 +174,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 directory to backing data and refresh.""" + path = dir_data.get("path", "").removeprefix("/") + dirname = dir_data.get("dirname", "") or helper_methods.get_parent_dir(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}" - ) + parent_dir = helper_methods.get_parent_dir(path) if path else "" + if parent_dir != self._curr_dir.removeprefix("/"): return - - # Check if already exists - if self._model_contains_item(dirname): - 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). - """ + """Remove a directory from backing data, or bail to root if it is current.""" dirname_or_path = dirname_or_path.removeprefix("/") dirname = ( - self._get_basename(dirname_or_path) + helper_methods.get_parent_dir(dirname_or_path) if "/" in dirname_or_path else dirname_or_path ) - if not dirname: 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 + "/"): 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(): - return - - removed = self._model.remove_item_by_text(dirname) - - if removed: - self._check_empty_state() - logger.debug(f"Directory removed from view: {dirname}") + self._directories = [ + d for d in self._directories if d.get("dirname", "") != dirname + ] + 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. - """ + """Navigate back to the root gcodes folder 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 directory so a newly mounted USB folder shows up.""" + 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: + """Ask Moonraker for the current directory again, used by delayed retries.""" + self.request_dir_info[str].emit(self._curr_dir) + + @QtCore.pyqtSlot(str, name="on_usb_removed") + def on_usb_removed(self, _device_path: str = "") -> None: + """Return to root when inside the USB folder, else refresh so it disappears.""" + current = self._curr_dir.removeprefix("/") + if current.startswith(self.USB_PREFIX): + 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) + # File selected (files don't have 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) + go_back_path = helper_methods.get_parent_dir(self._curr_dir) if go_back_path == "/": go_back_path = "" self._on_go_back_dir(go_back_path) @@ -591,187 +280,169 @@ 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: + """Stable identity for reconcile: dirs/back vs files, namespaced by text.""" + return f"{'d' if item.left_icon else 'f'}:{item.text}" + + def _schedule_rebuild(self) -> None: + """Coalesce bursts of metadata arrivals 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 from backing data 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]: + """Each file's bare name to its cached metadata, one lookup pass 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]: + """Ordered rows: Go Back (subdir), dirs A-Z, then files newest-first.""" + 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() - ) - 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) + 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]: + """Directory rows sorted alphabetically, excluding 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]: + """Gcode file rows ordered by the active sort key and direction.""" + files = [ + f + for f in self._file_list + if f.get("filename", f.get("path", "")) + .lower() + .endswith(self.GCODE_EXTENSION) + ] + if self._sort_key == "Import Order": + if not self._sort_descending: + files.reverse() 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, + files.sort( + key=lambda f: self._sort_value(f, meta), + reverse=self._sort_descending, ) - - if item: - self._model.add_item(item) - - def _create_file_list_item(self, filedata: dict) -> typing.Optional[ListItem]: - """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 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, + return [ + self._make_file_item(f.get("filename", f.get("path", "")), meta) + for f in files + ] + + def _sort_value(self, filedata: dict, meta: dict) -> object: + """Comparable key for the active sort column (uniform type per column).""" + name = filedata.get("filename", filedata.get("path", "")) + 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 selected sort column and rebuild the list.""" + 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, refresh the toggle icon, and rebuild.""" + self._sort_descending = not self._sort_descending + self._update_sort_order_icon() + self._build_file_list() + + def _update_sort_order_icon(self) -> None: + """Point the order-toggle button at the icon for the current 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 leading Go Back navigation row.""" + return self._row("Go Back", "", self._icons.get("back_folder"), None) - # Choose appropriate icon + def _make_directory_item(self, dir_data: dict) -> ListItem: + """A directory row; USB mounts at root 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: + """A file row from cached metadata, or an Unknown placeholder.""" + 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: typing.Optional[QtGui.QPixmap], + right_icon: typing.Optional[QtGui.QPixmap], + ) -> ListItem: + """ListItem with this page's standard sizing/flags.""" + return ListItem( + text=text, + right_text=right_text, + left_icon=left_icon, + right_icon=right_icon, callback=None, selected=False, allow_check=False, @@ -780,25 +451,52 @@ 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 - # Build full path - file_path = self._build_filepath(name) + def _format_file_meta(self, filedata: dict) -> str: + """' -