refactor(files): files page rework with unified gcode loader, metadata page and USB mount fixes

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

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"
         )

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