From 66b56ebaacd5669222449daa3d2d5d26721ffec7 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 30 Oct 2026 09:19:40 +0000 Subject: [PATCH 1/5] feat(print): copy USB gcode to local storage and print the local copy --- BlocksScreen/lib/panels/printTab.py | 29 ++++ BlocksScreen/lib/utils/usb_import.py | 155 +++++++++++++++++++ tests/util/test_usb_import_unit.py | 154 ++++++++++++++++++ tests/widgets/test_print_start_guard_unit.py | 31 ++++ 4 files changed, 369 insertions(+) create mode 100644 BlocksScreen/lib/utils/usb_import.py create mode 100644 tests/util/test_usb_import_unit.py diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 35549e1f..8008603b 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -1,8 +1,10 @@ import logging import os +import pathlib import typing from functools import partial +import helper_methods from configfile import BlocksScreenConfig, get_configparser from lib.files import Files from lib.moonrakerComm import MoonWebSocket @@ -20,6 +22,7 @@ from lib.utils import gcode_loader from lib.utils.blocks_button import BlocksCustomButton from lib.utils.display_button import DisplayButton +from lib.utils.usb_import import UsbImportTask from PyQt6 import QtCore, QtGui, QtWidgets if typing.TYPE_CHECKING: @@ -78,6 +81,7 @@ def __init__( self._finish_print_handled: bool = False self._cancel_z_snapshot: float = 0.0 self._z_apply_command: str = "Z_OFFSET_APPLY_ENDSTOP" + self._usb_import_task: UsbImportTask | None = None self.setupMainPrintPage() self.ws: MoonWebSocket = ws @@ -359,8 +363,33 @@ def start_print(self, file: str) -> None: logger.warning("Print start of %s blocked: %s", file, reason) self.notify.emit("PrintTab", reason, 2, True) return + if helper_methods.is_usb_path(file): + self._start_usb_import(file) + return self.jobStatusPage_widget.on_print_start(file) + def _start_usb_import(self, file: str) -> None: + """Copy a USB gcode to the gcodes root off-thread, then print the copy.""" + # Printing from the stick would fail mid-job if it is pulled or glitches. + if self._usb_import_task is not None: + logger.info("USB import already running, ignoring %s", file) + return + self._usb_import_task = UsbImportTask(pathlib.Path(self.gcode_path), file) + self._usb_import_task.signals.done.connect(self._on_usb_import_done) + self.call_load_panel.emit(True, "Copying file from USB...\nPlease wait", False) + QtCore.QThreadPool.globalInstance().start(self._usb_import_task) + + @QtCore.pyqtSlot(str, str, str, name="on_usb_import_done") + def _on_usb_import_done(self, usb_file: str, local_file: str, error: str) -> None: + """Print the local copy, or report why the copy failed.""" + self._usb_import_task = None + self.call_load_panel.emit(False, "", False) + if error: + self.notify.emit("PrintTab", f"USB print failed: {error}", 2, True) + return + logger.info("Printing local copy %s of %s", local_file, usb_file) + self.jobStatusPage_widget.on_print_start(local_file) + @QtCore.pyqtSlot(str, str, name="delete_file") @QtCore.pyqtSlot(str, name="delete_file") def delete_file(self, filename: str, directory: str = "gcodes") -> None: diff --git a/BlocksScreen/lib/utils/usb_import.py b/BlocksScreen/lib/utils/usb_import.py new file mode 100644 index 00000000..2b43e567 --- /dev/null +++ b/BlocksScreen/lib/utils/usb_import.py @@ -0,0 +1,155 @@ +"""Copies USB gcode files into local storage so prints survive drive removal.""" + +import filecmp +import logging +import os +import pathlib +import shutil + +import helper_methods +from PyQt6 import QtCore + +logger = logging.getLogger(__name__) + +MB = 1024 * 1024 +ALLOWED_EXT = frozenset({".gcode", ".gco", ".g"}) +MIN_FREE_MARGIN = 64 * MB # free space left on the SD card after the copy +TMP_DIR_NAME = ".usb_import_tmp" + + +class UsbImportError(Exception): + """Import failure whose message is safe to show on screen.""" + + +def _tmp_dir(gcode_root: pathlib.Path) -> pathlib.Path: + """Scratch dir outside the gcodes tree, or the gcodes root if on another fs.""" + # Outside Moonraker's inotify tree the final rename reads as a plain create_file. + tmp_dir = gcode_root.parent / TMP_DIR_NAME + try: + tmp_dir.mkdir(exist_ok=True) + if os.stat(tmp_dir).st_dev == os.stat(gcode_root).st_dev: + return tmp_dir # same fs keeps os.replace atomic + except OSError: + pass + return gcode_root + + +def _is_copy_of(cand: pathlib.Path, src: pathlib.Path, src_st: os.stat_result) -> bool: + """True if *cand* already holds the bytes of *src*.""" + try: + st = cand.stat() + except FileNotFoundError: + return False + if not cand.is_file() or st.st_size != src_st.st_size: + return False + if st.st_mtime_ns == src_st.st_mtime_ns: + return True # copies keep the drive mtime, so this skips reading both files + return filecmp.cmp(src, cand, shallow=False) # stops at the first differing block + + +def _pick_destination( + src: pathlib.Path, src_st: os.stat_result, dest_dir: pathlib.Path +) -> tuple[pathlib.Path, bool]: + """(path, reused): an existing copy of *src*, else the first free name.""" + n = 0 + while True: + cand = dest_dir / (src.name if n == 0 else f"{src.stem}_{n}{src.suffix}") + if not cand.exists() and not cand.is_symlink(): + return cand, False + if _is_copy_of(cand, src, src_st): + return cand, True + n += 1 # same name, other content: never overwrite a local file + + +def _fsync(path: pathlib.Path) -> None: + """Flush *path* (file or directory) to disk.""" + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _copy_atomic( + src: pathlib.Path, + src_st: os.stat_result, + final: pathlib.Path, + tmp_dir: pathlib.Path, +) -> None: + """Copy *src* to *final* via a synced temp file, so *final* is never partial.""" + for stale in tmp_dir.glob(".*.part"): # leftovers from a crash or power loss + stale.unlink(missing_ok=True) + free = shutil.disk_usage(final.parent).free + if free < src_st.st_size + MIN_FREE_MARGIN: + raise UsbImportError(f"Not enough free space ({free // MB} MB free)") + tmp = tmp_dir / f".{final.name}.part" + try: + shutil.copyfile(src, tmp) # kernel-side copy on Linux + now = src.stat() + if (now.st_size, now.st_mtime_ns) != (src_st.st_size, src_st.st_mtime_ns): + raise UsbImportError("The file changed during the copy") + if tmp.stat().st_size != src_st.st_size: + raise UsbImportError("Copy incomplete") + os.utime(tmp, ns=(src_st.st_atime_ns, src_st.st_mtime_ns)) + _fsync(tmp) + os.replace(tmp, final) + _fsync(final.parent) # persist the rename itself across power loss + finally: + tmp.unlink(missing_ok=True) + + +def import_usb_gcode(gcode_root: pathlib.Path, filename: str) -> tuple[str, bool]: + """Copy USB *filename* into *gcode_root*; returns (gcodes-relative name, reused).""" + root = gcode_root.resolve() + rel = filename.strip("/") + if not helper_methods.is_usb_path(rel): + raise UsbImportError("Not a USB file") + usb_root = (root / rel.split("/", 1)[0]).resolve() + src = (root / rel).resolve() + if usb_root not in src.parents: # symlinks or ".." escaping the drive + raise UsbImportError("Invalid file path") + if src.suffix.lower() not in ALLOWED_EXT: + raise UsbImportError("Unsupported file type") + if not os.path.ismount(usb_root) or not src.is_file(): + raise UsbImportError("File not found, was the USB drive removed?") + try: + src_st = src.stat() + final, reused = _pick_destination(src, src_st, root) + if not reused: + _copy_atomic(src, src_st, final, _tmp_dir(root)) + except OSError as e: + raise UsbImportError( + f"Copy failed, was the USB drive removed? ({e.strerror})" + ) from e + logger.info("USB import %s -> %s (reused=%s)", rel, final.name, reused) + return final.name, reused + + +class _ImportSignals(QtCore.QObject): + """Worker result signal; QRunnable can't hold signals.""" + + done = QtCore.pyqtSignal(str, str, str) # (usb_file, local_file, error) + + +class UsbImportTask(QtCore.QRunnable): + """Runs import_usb_gcode off the UI thread.""" + + def __init__(self, gcode_root: pathlib.Path, filename: str) -> None: + super().__init__() + self.setAutoDelete(False) # the caller holds it until done fires + self._gcode_root = gcode_root + self._filename = filename + self.signals = _ImportSignals() + + @QtCore.pyqtSlot() + def run(self) -> None: + """Copy, then report (usb_file, local_file, "") or (usb_file, "", error).""" + try: + local, _ = import_usb_gcode(self._gcode_root, self._filename) + self.signals.done.emit(self._filename, local, "") + except UsbImportError as e: + logger.warning("USB import of %s failed: %s", self._filename, e) + self.signals.done.emit(self._filename, "", str(e)) + except Exception: # never leave the loadscreen up on a bug + logger.exception("USB import of %s crashed", self._filename) + self.signals.done.emit(self._filename, "", "Unexpected error copying file") diff --git a/tests/util/test_usb_import_unit.py b/tests/util/test_usb_import_unit.py new file mode 100644 index 00000000..f05a8859 --- /dev/null +++ b/tests/util/test_usb_import_unit.py @@ -0,0 +1,154 @@ +"""import_usb_gcode copies USB gcodes into the gcodes root safely.""" + +import filecmp +import os +import pathlib +from collections import namedtuple + +import pytest +from lib.utils import usb_import +from lib.utils.usb_import import UsbImportError, import_usb_gcode + +_Usage = namedtuple("_Usage", "total used free") + + +@pytest.fixture +def layout(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch): + """printer_data/gcodes with a USB-STICK symlink to a fake mounted drive.""" + gcodes = tmp_path / "printer_data" / "gcodes" + gcodes.mkdir(parents=True) + drive = tmp_path / "media" / "stick" + drive.mkdir(parents=True) + (gcodes / "USB-STICK").symlink_to(drive) + mounts = {drive.resolve()} + monkeypatch.setattr( + usb_import.os.path, "ismount", lambda p: pathlib.Path(p).resolve() in mounts + ) + return gcodes, drive, mounts + + +def test_copies_into_gcodes_root(layout) -> None: + gcodes, drive, _ = layout + (drive / "sub").mkdir() + (drive / "sub" / "cube.gcode").write_bytes(b"G28\n" * 1000) + name, reused = import_usb_gcode(gcodes, "USB-STICK/sub/cube.gcode") + assert (name, reused) == ("cube.gcode", False) + assert (gcodes / "cube.gcode").read_bytes() == b"G28\n" * 1000 + assert not list((gcodes.parent / usb_import.TMP_DIR_NAME).iterdir()) + + +def test_identical_copy_is_reused(layout) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + (gcodes / "cube.gcode").write_bytes(b"G28\n") + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", True) + assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK", "cube.gcode"] + + +def test_same_name_different_content_gets_suffix(layout) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + (gcodes / "cube.gcode").write_bytes(b"G1\n") + assert import_usb_gcode(gcodes, "/USB-STICK/cube.gcode") == ("cube_1.gcode", False) + assert (gcodes / "cube.gcode").read_bytes() == b"G1\n" + assert (gcodes / "cube_1.gcode").read_bytes() == b"G28\n" + + +def test_second_import_reuses_without_reading(layout, monkeypatch) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n" * 10) + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", False) + + def _no_read(*_args) -> bool: + raise AssertionError("matching mtime must skip the byte compare") + + monkeypatch.setattr(filecmp, "cmp", _no_read) + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", True) + + +def test_same_size_different_content_gets_suffix(layout) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + (gcodes / "cube.gcode").write_bytes(b"G29\n") + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube_1.gcode", False) + assert (gcodes / "cube.gcode").read_bytes() == b"G29\n" + + +def test_reuses_matching_suffixed_copy(layout) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + (gcodes / "cube.gcode").write_bytes(b"G1\n") + (gcodes / "cube_1.gcode").write_bytes(b"G28\n") + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube_1.gcode", True) + + +def test_stale_part_files_are_cleaned(layout) -> None: + gcodes, drive, _ = layout + tmp = gcodes.parent / usb_import.TMP_DIR_NAME + tmp.mkdir() + (tmp / ".old.gcode.part").write_bytes(b"x") + (drive / "cube.gcode").write_bytes(b"G28\n") + import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + assert not (tmp / ".old.gcode.part").exists() + + +@pytest.mark.parametrize( + ("filename", "message"), + [ + ("cube.gcode", "Not a USB file"), + ("USB-STICK/../../etc.gcode", "Invalid file path"), + ("USB-STICK/escape.gcode", "Invalid file path"), + ("USB-STICK/notes.txt", "Unsupported file type"), + ("USB-STICK/missing.gcode", "File not found"), + ], +) +def test_rejects_bad_requests(layout, filename: str, message: str) -> None: + gcodes, drive, _ = layout + (drive / "notes.txt").write_text("hi") + outside = gcodes.parent / "secret.gcode" + outside.write_text("G28") + (drive / "escape.gcode").symlink_to(outside) + with pytest.raises(UsbImportError, match=message): + import_usb_gcode(gcodes, filename) + assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK"] + + +def test_unmounted_drive_is_rejected(layout) -> None: + gcodes, drive, mounts = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + mounts.clear() + with pytest.raises(UsbImportError, match="was the USB drive removed"): + import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + + +def test_low_disk_space_is_rejected(layout, monkeypatch) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + monkeypatch.setattr(usb_import.shutil, "disk_usage", lambda _p: _Usage(0, 0, 10)) + with pytest.raises(UsbImportError, match="Not enough free space"): + import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + assert not (gcodes / "cube.gcode").exists() + + +def test_read_error_leaves_no_partial_file(layout, monkeypatch) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + + def _fail(_fd: int) -> None: + raise OSError(5, os.strerror(5)) + + monkeypatch.setattr(usb_import.os, "fsync", _fail) + with pytest.raises(UsbImportError, match="Copy failed"): + import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK"] + assert not list((gcodes.parent / usb_import.TMP_DIR_NAME).iterdir()) + + +def test_task_reports_error(layout, qtbot) -> None: + gcodes, _, _ = layout + task = usb_import.UsbImportTask(gcodes, "USB-STICK/missing.gcode") + with qtbot.waitSignal(task.signals.done) as blocker: + task.run() + assert blocker.args[0] == "USB-STICK/missing.gcode" + assert blocker.args[1] == "" + assert "File not found" in blocker.args[2] diff --git a/tests/widgets/test_print_start_guard_unit.py b/tests/widgets/test_print_start_guard_unit.py index 10e4b496..c64ecd84 100644 --- a/tests/widgets/test_print_start_guard_unit.py +++ b/tests/widgets/test_print_start_guard_unit.py @@ -66,3 +66,34 @@ def test_refusal_blocks_and_notifies(self) -> None: tab.notify.emit.assert_called_once_with( "PrintTab", "No filament at the toolhead.", 2, True ) + + def test_usb_file_is_imported_not_printed(self) -> None: + tab = _tab(None) + tab._start_usb_import = MagicMock() + PrintTab.start_print(tab, "USB-STICK/cube.gcode") + tab._start_usb_import.assert_called_once_with("USB-STICK/cube.gcode") + tab.jobStatusPage_widget.on_print_start.assert_not_called() + + +class TestUsbImportDone: + def _done_tab(self) -> SimpleNamespace: + tab = _tab(None) + tab._usb_import_task = object() + tab.call_load_panel = MagicMock() + return tab + + def test_prints_local_copy(self) -> None: + tab = self._done_tab() + PrintTab._on_usb_import_done(tab, "USB-STICK/cube.gcode", "cube.gcode", "") + assert tab._usb_import_task is None + tab.call_load_panel.emit.assert_called_once_with(False, "", False) + tab.jobStatusPage_widget.on_print_start.assert_called_once_with("cube.gcode") + + def test_error_notifies_and_does_not_print(self) -> None: + tab = self._done_tab() + PrintTab._on_usb_import_done(tab, "USB-STICK/cube.gcode", "", "Bad") + assert tab._usb_import_task is None + tab.jobStatusPage_widget.on_print_start.assert_not_called() + tab.notify.emit.assert_called_once_with( + "PrintTab", "USB print failed: Bad", 2, True + ) From 4f9445ae9ae839e0734e518c39e52d5c4de91799 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 30 Oct 2026 10:00:20 +0000 Subject: [PATCH 2/5] fix(print): wait for Moonraker processing before printing USB copy --- BlocksScreen/lib/panels/printTab.py | 65 +++++++-- BlocksScreen/lib/utils/usb_import.py | 121 ++++++++++++----- tests/util/test_usb_import_unit.py | 135 ++++++++++++++----- tests/widgets/test_print_start_guard_unit.py | 68 +++++++++- 4 files changed, 309 insertions(+), 80 deletions(-) diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 8008603b..86544d94 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -22,12 +22,14 @@ from lib.utils import gcode_loader from lib.utils.blocks_button import BlocksCustomButton from lib.utils.display_button import DisplayButton -from lib.utils.usb_import import UsbImportTask +from lib.utils.usb_import import UsbImportTask, remember_import from PyQt6 import QtCore, QtGui, QtWidgets if typing.TYPE_CHECKING: from devices.amu import AMUManager +USB_SETTLE_TIMEOUT_MS = 120_000 # Happy Hare pre-processing of a large file + logger = logging.getLogger(__name__) @@ -82,6 +84,14 @@ def __init__( self._cancel_z_snapshot: float = 0.0 self._z_apply_command: str = "Z_OFFSET_APPLY_ENDSTOP" self._usb_import_task: UsbImportTask | None = None + self._usb_key: str = "" + self._usb_local: str = "" + self._usb_created: set[str] = set() + # Moonraker announces a new gcode only once its pre-processing is done. + self._usb_settle_timer = QtCore.QTimer(self) + self._usb_settle_timer.setSingleShot(True) + self._usb_settle_timer.setInterval(USB_SETTLE_TIMEOUT_MS) + self._usb_settle_timer.timeout.connect(self._on_usb_settle_timeout) self.setupMainPrintPage() self.ws: MoonWebSocket = ws @@ -142,6 +152,7 @@ def __init__( 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) + self.file_data.file_added.connect(self._on_usb_file_added) self.file_data.file_removed.connect(self.filesPage_widget.on_file_removed) self.file_data.file_modified.connect(self.filesPage_widget.on_file_modified) self.file_data.dir_added.connect(self.filesPage_widget.on_dir_added) @@ -374,21 +385,59 @@ def _start_usb_import(self, file: str) -> None: if self._usb_import_task is not None: logger.info("USB import already running, ignoring %s", file) return + self._usb_created.clear() self._usb_import_task = UsbImportTask(pathlib.Path(self.gcode_path), file) self._usb_import_task.signals.done.connect(self._on_usb_import_done) self.call_load_panel.emit(True, "Copying file from USB...\nPlease wait", False) QtCore.QThreadPool.globalInstance().start(self._usb_import_task) - @QtCore.pyqtSlot(str, str, str, name="on_usb_import_done") - def _on_usb_import_done(self, usb_file: str, local_file: str, error: str) -> None: - """Print the local copy, or report why the copy failed.""" - self._usb_import_task = None - self.call_load_panel.emit(False, "", False) + @QtCore.pyqtSlot(str, str, str, bool, str, name="on_usb_import_done") + def _on_usb_import_done( + self, usb_file: str, local_file: str, key: str, reused: bool, error: str + ) -> None: + """Print a reused copy now, a new one once Moonraker has processed it.""" if error: + self._usb_import_task = None + self.call_load_panel.emit(False, "", False) self.notify.emit("PrintTab", f"USB print failed: {error}", 2, True) return - logger.info("Printing local copy %s of %s", local_file, usb_file) - self.jobStatusPage_widget.on_print_start(local_file) + logger.info("Local copy of %s is %s", usb_file, local_file) + self._usb_key, self._usb_local = key, local_file + # Printing before Happy Hare rewrites the file runs the unprocessed gcode. + if reused or local_file in self._usb_created: + self._usb_print_local_copy() + else: + self._usb_settle_timer.start() + + @QtCore.pyqtSlot(dict, name="on_usb_file_added") + def _on_usb_file_added(self, item: dict) -> None: + """Release a waiting USB print when Moonraker announces its local copy.""" + if self._usb_import_task is None: + return + name = str(item.get("path", "")).removeprefix("/") + self._usb_created.add(name) # may beat the worker's done signal + if name == self._usb_local: + self._usb_print_local_copy() + + @QtCore.pyqtSlot(name="on_usb_settle_timeout") + def _on_usb_settle_timeout(self) -> None: + """Print anyway if Moonraker never announced the copy.""" + logger.warning("No create_file for %s, printing it as is", self._usb_local) + self._usb_print_local_copy() + + def _usb_print_local_copy(self) -> None: + """Index the settled local copy and start printing it.""" + self._usb_settle_timer.stop() + local = self._usb_local + try: + remember_import(pathlib.Path(self.gcode_path), self._usb_key, local) + except OSError as e: + logger.warning("Could not index USB copy %s: %s", local, e) + self._usb_import_task = None + self._usb_key = self._usb_local = "" + self.call_load_panel.emit(False, "", False) + logger.info("Printing local copy %s", local) + self.jobStatusPage_widget.on_print_start(local) @QtCore.pyqtSlot(str, str, name="delete_file") @QtCore.pyqtSlot(str, name="delete_file") diff --git a/BlocksScreen/lib/utils/usb_import.py b/BlocksScreen/lib/utils/usb_import.py index 2b43e567..f4a7f72a 100644 --- a/BlocksScreen/lib/utils/usb_import.py +++ b/BlocksScreen/lib/utils/usb_import.py @@ -1,6 +1,8 @@ """Copies USB gcode files into local storage so prints survive drive removal.""" import filecmp +import hashlib +import json import logging import os import pathlib @@ -12,19 +14,80 @@ logger = logging.getLogger(__name__) MB = 1024 * 1024 -ALLOWED_EXT = frozenset({".gcode", ".gco", ".g"}) +ALLOWED_EXT = ".gcode" # the only extension lib.files lists and notifies MIN_FREE_MARGIN = 64 * MB # free space left on the SD card after the copy -TMP_DIR_NAME = ".usb_import_tmp" +SAMPLE = 64 * 1024 +STATE_DIR_NAME = ".usb_import" +INDEX_NAME = "index.json" class UsbImportError(Exception): """Import failure whose message is safe to show on screen.""" +def _state_dir(gcode_root: pathlib.Path) -> pathlib.Path: + """Index and scratch dir, beside the gcodes tree so Moonraker never lists it.""" + return gcode_root.parent / STATE_DIR_NAME + + +def _source_key(src: pathlib.Path, size: int) -> str: + """Identity of a USB file from its name, size, head and tail.""" + # Slicers stamp the header and append their config, so ends differ per slice. + h = hashlib.sha256(f"{src.name}\0{size}\0".encode()) + with src.open("rb") as f: + h.update(f.read(SAMPLE)) + if size > SAMPLE: + f.seek(max(SAMPLE, size - SAMPLE)) + h.update(f.read(SAMPLE)) + return h.hexdigest() + + +def _load_index(gcode_root: pathlib.Path) -> dict: + """Source key -> {file, size, mtime_ns} of earlier imports; empty if unreadable.""" + try: + data = json.loads((_state_dir(gcode_root) / INDEX_NAME).read_text()) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def _indexed_copy(gcode_root: pathlib.Path, key: str) -> pathlib.Path | None: + """The earlier local copy for *key*, if it is still there and untouched.""" + entry = _load_index(gcode_root).get(key) + if not isinstance(entry, dict) or not isinstance(entry.get("file"), str): + return None + local = gcode_root / entry["file"] + try: + st = local.stat() + except OSError: + return None + if (st.st_size, st.st_mtime_ns) != (entry.get("size"), entry.get("mtime_ns")): + return None # replaced or edited since, so it may no longer be this file + return local + + +def remember_import(gcode_root: pathlib.Path, key: str, local_name: str) -> None: + """Record *local_name* as the copy for *key*, once Moonraker has processed it.""" + # Moonraker and Happy Hare rewrite new gcodes, so the copy is only final now. + root = gcode_root.resolve() + st = (root / local_name).stat() + index = { + k: v + for k, v in _load_index(root).items() + if isinstance(v, dict) and (root / str(v.get("file"))).is_file() + } + index[key] = {"file": local_name, "size": st.st_size, "mtime_ns": st.st_mtime_ns} + state = _state_dir(root) + state.mkdir(exist_ok=True) + tmp = state / f".{INDEX_NAME}.part" + tmp.write_text(json.dumps(index, indent=1)) + os.replace(tmp, state / INDEX_NAME) + + def _tmp_dir(gcode_root: pathlib.Path) -> pathlib.Path: """Scratch dir outside the gcodes tree, or the gcodes root if on another fs.""" # Outside Moonraker's inotify tree the final rename reads as a plain create_file. - tmp_dir = gcode_root.parent / TMP_DIR_NAME + tmp_dir = _state_dir(gcode_root) try: tmp_dir.mkdir(exist_ok=True) if os.stat(tmp_dir).st_dev == os.stat(gcode_root).st_dev: @@ -34,29 +97,20 @@ def _tmp_dir(gcode_root: pathlib.Path) -> pathlib.Path: return gcode_root -def _is_copy_of(cand: pathlib.Path, src: pathlib.Path, src_st: os.stat_result) -> bool: - """True if *cand* already holds the bytes of *src*.""" - try: - st = cand.stat() - except FileNotFoundError: - return False - if not cand.is_file() or st.st_size != src_st.st_size: - return False - if st.st_mtime_ns == src_st.st_mtime_ns: - return True # copies keep the drive mtime, so this skips reading both files - return filecmp.cmp(src, cand, shallow=False) # stops at the first differing block - - def _pick_destination( - src: pathlib.Path, src_st: os.stat_result, dest_dir: pathlib.Path + src: pathlib.Path, size: int, dest_dir: pathlib.Path ) -> tuple[pathlib.Path, bool]: - """(path, reused): an existing copy of *src*, else the first free name.""" + """(path, reused): a byte-identical existing file, else the first free name.""" n = 0 while True: cand = dest_dir / (src.name if n == 0 else f"{src.stem}_{n}{src.suffix}") if not cand.exists() and not cand.is_symlink(): return cand, False - if _is_copy_of(cand, src, src_st): + if ( + cand.is_file() + and cand.stat().st_size == size + and filecmp.cmp(src, cand, shallow=False) + ): return cand, True n += 1 # same name, other content: never overwrite a local file @@ -90,7 +144,6 @@ def _copy_atomic( raise UsbImportError("The file changed during the copy") if tmp.stat().st_size != src_st.st_size: raise UsbImportError("Copy incomplete") - os.utime(tmp, ns=(src_st.st_atime_ns, src_st.st_mtime_ns)) _fsync(tmp) os.replace(tmp, final) _fsync(final.parent) # persist the rename itself across power loss @@ -98,8 +151,8 @@ def _copy_atomic( tmp.unlink(missing_ok=True) -def import_usb_gcode(gcode_root: pathlib.Path, filename: str) -> tuple[str, bool]: - """Copy USB *filename* into *gcode_root*; returns (gcodes-relative name, reused).""" +def import_usb_gcode(gcode_root: pathlib.Path, filename: str) -> tuple[str, bool, str]: + """Copy USB *filename* into *gcode_root*; returns (local name, reused, key).""" root = gcode_root.resolve() rel = filename.strip("/") if not helper_methods.is_usb_path(rel): @@ -108,13 +161,16 @@ def import_usb_gcode(gcode_root: pathlib.Path, filename: str) -> tuple[str, bool src = (root / rel).resolve() if usb_root not in src.parents: # symlinks or ".." escaping the drive raise UsbImportError("Invalid file path") - if src.suffix.lower() not in ALLOWED_EXT: + if src.suffix.lower() != ALLOWED_EXT: raise UsbImportError("Unsupported file type") if not os.path.ismount(usb_root) or not src.is_file(): raise UsbImportError("File not found, was the USB drive removed?") try: src_st = src.stat() - final, reused = _pick_destination(src, src_st, root) + key = _source_key(src, src_st.st_size) + final, reused = _indexed_copy(root, key), True + if final is None: + final, reused = _pick_destination(src, src_st.st_size, root) if not reused: _copy_atomic(src, src_st, final, _tmp_dir(root)) except OSError as e: @@ -122,13 +178,14 @@ def import_usb_gcode(gcode_root: pathlib.Path, filename: str) -> tuple[str, bool f"Copy failed, was the USB drive removed? ({e.strerror})" ) from e logger.info("USB import %s -> %s (reused=%s)", rel, final.name, reused) - return final.name, reused + return final.name, reused, key class _ImportSignals(QtCore.QObject): """Worker result signal; QRunnable can't hold signals.""" - done = QtCore.pyqtSignal(str, str, str) # (usb_file, local_file, error) + # (usb_file, local_file, key, reused, error) + done = QtCore.pyqtSignal(str, str, str, bool, str) class UsbImportTask(QtCore.QRunnable): @@ -143,13 +200,15 @@ def __init__(self, gcode_root: pathlib.Path, filename: str) -> None: @QtCore.pyqtSlot() def run(self) -> None: - """Copy, then report (usb_file, local_file, "") or (usb_file, "", error).""" + """Copy, then report the local copy or the error text.""" try: - local, _ = import_usb_gcode(self._gcode_root, self._filename) - self.signals.done.emit(self._filename, local, "") + local, reused, key = import_usb_gcode(self._gcode_root, self._filename) + self.signals.done.emit(self._filename, local, key, reused, "") except UsbImportError as e: logger.warning("USB import of %s failed: %s", self._filename, e) - self.signals.done.emit(self._filename, "", str(e)) + self.signals.done.emit(self._filename, "", "", False, str(e)) except Exception: # never leave the loadscreen up on a bug logger.exception("USB import of %s crashed", self._filename) - self.signals.done.emit(self._filename, "", "Unexpected error copying file") + self.signals.done.emit( + self._filename, "", "", False, "Unexpected error copying file" + ) diff --git a/tests/util/test_usb_import_unit.py b/tests/util/test_usb_import_unit.py index f05a8859..88a3a74b 100644 --- a/tests/util/test_usb_import_unit.py +++ b/tests/util/test_usb_import_unit.py @@ -1,13 +1,12 @@ """import_usb_gcode copies USB gcodes into the gcodes root safely.""" -import filecmp import os import pathlib from collections import namedtuple import pytest from lib.utils import usb_import -from lib.utils.usb_import import UsbImportError, import_usb_gcode +from lib.utils.usb_import import UsbImportError, import_usb_gcode, remember_import _Usage = namedtuple("_Usage", "total used free") @@ -27,69 +26,120 @@ def layout(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch): return gcodes, drive, mounts +def _state(gcodes: pathlib.Path) -> pathlib.Path: + return gcodes.parent / usb_import.STATE_DIR_NAME + + def test_copies_into_gcodes_root(layout) -> None: gcodes, drive, _ = layout (drive / "sub").mkdir() (drive / "sub" / "cube.gcode").write_bytes(b"G28\n" * 1000) - name, reused = import_usb_gcode(gcodes, "USB-STICK/sub/cube.gcode") + name, reused, key = import_usb_gcode(gcodes, "USB-STICK/sub/cube.gcode") assert (name, reused) == ("cube.gcode", False) + assert len(key) == 64 assert (gcodes / "cube.gcode").read_bytes() == b"G28\n" * 1000 - assert not list((gcodes.parent / usb_import.TMP_DIR_NAME).iterdir()) + assert not list(_state(gcodes).glob("*.part")) -def test_identical_copy_is_reused(layout) -> None: +def test_identical_existing_file_is_reused(layout) -> None: gcodes, drive, _ = layout (drive / "cube.gcode").write_bytes(b"G28\n") (gcodes / "cube.gcode").write_bytes(b"G28\n") - assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", True) + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode")[:2] == ("cube.gcode", True) assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK", "cube.gcode"] -def test_same_name_different_content_gets_suffix(layout) -> None: +@pytest.mark.parametrize("local", [b"G1\n", b"G29\n"]) +def test_same_name_other_content_gets_suffix(layout, local: bytes) -> None: gcodes, drive, _ = layout (drive / "cube.gcode").write_bytes(b"G28\n") - (gcodes / "cube.gcode").write_bytes(b"G1\n") - assert import_usb_gcode(gcodes, "/USB-STICK/cube.gcode") == ("cube_1.gcode", False) - assert (gcodes / "cube.gcode").read_bytes() == b"G1\n" + (gcodes / "cube.gcode").write_bytes(local) + assert import_usb_gcode(gcodes, "/USB-STICK/cube.gcode")[:2] == ( + "cube_1.gcode", + False, + ) + assert (gcodes / "cube.gcode").read_bytes() == local assert (gcodes / "cube_1.gcode").read_bytes() == b"G28\n" -def test_second_import_reuses_without_reading(layout, monkeypatch) -> None: +def test_reuses_matching_suffixed_copy(layout) -> None: gcodes, drive, _ = layout - (drive / "cube.gcode").write_bytes(b"G28\n" * 10) - assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", False) + (drive / "cube.gcode").write_bytes(b"G28\n") + (gcodes / "cube.gcode").write_bytes(b"G1\n") + (gcodes / "cube_1.gcode").write_bytes(b"G28\n") + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode")[:2] == ( + "cube_1.gcode", + True, + ) - def _no_read(*_args) -> bool: - raise AssertionError("matching mtime must skip the byte compare") - monkeypatch.setattr(filecmp, "cmp", _no_read) - assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube.gcode", True) +def test_processed_copy_is_reused_through_index(layout) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"T0\n" * 50_000) + name, _, key = import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + # Happy Hare's preprocessor rewrites the copy before it settles. + (gcodes / name).write_bytes(b"; processed by HappyHare\nMMU_CHANGE_TOOL TOOL=0\n") + remember_import(gcodes, key, name) + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == (name, True, key) + assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK", "cube.gcode"] -def test_same_size_different_content_gets_suffix(layout) -> None: +def test_edited_indexed_copy_is_not_reused(layout) -> None: gcodes, drive, _ = layout (drive / "cube.gcode").write_bytes(b"G28\n") - (gcodes / "cube.gcode").write_bytes(b"G29\n") - assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube_1.gcode", False) - assert (gcodes / "cube.gcode").read_bytes() == b"G29\n" + name, _, key = import_usb_gcode(gcodes, "USB-STICK/cube.gcode") + remember_import(gcodes, key, name) + (gcodes / name).write_bytes(b"G1 X10\n") # replaced after indexing + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode")[:2] == ( + "cube_1.gcode", + False, + ) -def test_reuses_matching_suffixed_copy(layout) -> None: +def test_index_drops_deleted_copies(layout) -> None: gcodes, drive, _ = layout + for n in ("a", "b"): + (drive / f"{n}.gcode").write_bytes(n.encode()) + name, _, key = import_usb_gcode(gcodes, f"USB-STICK/{n}.gcode") + remember_import(gcodes, key, name) + (gcodes / "a.gcode").unlink() + (drive / "c.gcode").write_bytes(b"c") + name, _, key = import_usb_gcode(gcodes, "USB-STICK/c.gcode") + remember_import(gcodes, key, name) + files = {v["file"] for v in usb_import._load_index(gcodes).values()} + assert files == {"b.gcode", "c.gcode"} + + +def test_corrupt_index_falls_back_to_copy(layout) -> None: + gcodes, drive, _ = layout + _state(gcodes).mkdir() + (_state(gcodes) / usb_import.INDEX_NAME).write_text("{not json") (drive / "cube.gcode").write_bytes(b"G28\n") - (gcodes / "cube.gcode").write_bytes(b"G1\n") - (gcodes / "cube_1.gcode").write_bytes(b"G28\n") - assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode") == ("cube_1.gcode", True) + assert import_usb_gcode(gcodes, "USB-STICK/cube.gcode")[:2] == ( + "cube.gcode", + False, + ) + + +def test_key_differs_by_tail(layout) -> None: + _, drive, _ = layout + head = b"G28\n" * 40_000 + (drive / "a").mkdir() + (drive / "b").mkdir() + a, b = drive / "a" / "cube.gcode", drive / "b" / "cube.gcode" + a.write_bytes(head + b"; layer_height = 0.2\n") + b.write_bytes(head + b"; layer_height = 0.3\n") + size = a.stat().st_size + assert usb_import._source_key(a, size) != usb_import._source_key(b, size) def test_stale_part_files_are_cleaned(layout) -> None: gcodes, drive, _ = layout - tmp = gcodes.parent / usb_import.TMP_DIR_NAME - tmp.mkdir() - (tmp / ".old.gcode.part").write_bytes(b"x") + _state(gcodes).mkdir() + (_state(gcodes) / ".old.gcode.part").write_bytes(b"x") (drive / "cube.gcode").write_bytes(b"G28\n") import_usb_gcode(gcodes, "USB-STICK/cube.gcode") - assert not (tmp / ".old.gcode.part").exists() + assert not (_state(gcodes) / ".old.gcode.part").exists() @pytest.mark.parametrize( @@ -99,12 +149,14 @@ def test_stale_part_files_are_cleaned(layout) -> None: ("USB-STICK/../../etc.gcode", "Invalid file path"), ("USB-STICK/escape.gcode", "Invalid file path"), ("USB-STICK/notes.txt", "Unsupported file type"), + ("USB-STICK/part.gco", "Unsupported file type"), ("USB-STICK/missing.gcode", "File not found"), ], ) def test_rejects_bad_requests(layout, filename: str, message: str) -> None: gcodes, drive, _ = layout (drive / "notes.txt").write_text("hi") + (drive / "part.gco").write_text("G28") outside = gcodes.parent / "secret.gcode" outside.write_text("G28") (drive / "escape.gcode").symlink_to(outside) @@ -130,7 +182,7 @@ def test_low_disk_space_is_rejected(layout, monkeypatch) -> None: assert not (gcodes / "cube.gcode").exists() -def test_read_error_leaves_no_partial_file(layout, monkeypatch) -> None: +def test_fsync_error_leaves_no_partial_file(layout, monkeypatch) -> None: gcodes, drive, _ = layout (drive / "cube.gcode").write_bytes(b"G28\n") @@ -141,7 +193,23 @@ def _fail(_fd: int) -> None: with pytest.raises(UsbImportError, match="Copy failed"): import_usb_gcode(gcodes, "USB-STICK/cube.gcode") assert sorted(p.name for p in gcodes.iterdir()) == ["USB-STICK"] - assert not list((gcodes.parent / usb_import.TMP_DIR_NAME).iterdir()) + assert not list(_state(gcodes).glob("*.part")) + + +def test_task_reports_copy(layout, qtbot) -> None: + gcodes, drive, _ = layout + (drive / "cube.gcode").write_bytes(b"G28\n") + task = usb_import.UsbImportTask(gcodes, "USB-STICK/cube.gcode") + with qtbot.waitSignal(task.signals.done) as blocker: + task.run() + usb_file, local, key, reused, error = blocker.args + assert (usb_file, local, reused, error) == ( + "USB-STICK/cube.gcode", + "cube.gcode", + False, + "", + ) + assert key def test_task_reports_error(layout, qtbot) -> None: @@ -149,6 +217,5 @@ def test_task_reports_error(layout, qtbot) -> None: task = usb_import.UsbImportTask(gcodes, "USB-STICK/missing.gcode") with qtbot.waitSignal(task.signals.done) as blocker: task.run() - assert blocker.args[0] == "USB-STICK/missing.gcode" - assert blocker.args[1] == "" - assert "File not found" in blocker.args[2] + assert blocker.args[:4] == ["USB-STICK/missing.gcode", "", "", False] + assert "File not found" in blocker.args[4] diff --git a/tests/widgets/test_print_start_guard_unit.py b/tests/widgets/test_print_start_guard_unit.py index c64ecd84..d756a729 100644 --- a/tests/widgets/test_print_start_guard_unit.py +++ b/tests/widgets/test_print_start_guard_unit.py @@ -76,22 +76,76 @@ def test_usb_file_is_imported_not_printed(self) -> None: class TestUsbImportDone: - def _done_tab(self) -> SimpleNamespace: + @pytest.fixture + def tab(self, monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: tab = _tab(None) tab._usb_import_task = object() + tab._usb_key = tab._usb_local = "" + tab._usb_created = set() + tab._usb_settle_timer = MagicMock() + tab.gcode_path = "/tmp/gcodes" tab.call_load_panel = MagicMock() + tab._usb_print_local_copy = lambda: PrintTab._usb_print_local_copy(tab) + tab.remember = MagicMock() + # printTab was evicted from sys.modules, so patch the function's globals. + monkeypatch.setitem( + PrintTab._usb_print_local_copy.__globals__, "remember_import", tab.remember + ) return tab - def test_prints_local_copy(self) -> None: - tab = self._done_tab() - PrintTab._on_usb_import_done(tab, "USB-STICK/cube.gcode", "cube.gcode", "") + def _printed(self, tab: SimpleNamespace) -> None: assert tab._usb_import_task is None tab.call_load_panel.emit.assert_called_once_with(False, "", False) tab.jobStatusPage_widget.on_print_start.assert_called_once_with("cube.gcode") + tab.remember.assert_called_once() + assert tab.remember.call_args.args[1:] == ("k", "cube.gcode") + + def test_reused_copy_prints_at_once(self, tab) -> None: + PrintTab._on_usb_import_done( + tab, "USB-STICK/cube.gcode", "cube.gcode", "k", True, "" + ) + self._printed(tab) + + def test_new_copy_waits_for_moonraker(self, tab) -> None: + PrintTab._on_usb_import_done( + tab, "USB-STICK/cube.gcode", "cube.gcode", "k", False, "" + ) + tab._usb_settle_timer.start.assert_called_once() + tab.jobStatusPage_widget.on_print_start.assert_not_called() + PrintTab._on_usb_file_added(tab, {"path": "other.gcode"}) + tab.jobStatusPage_widget.on_print_start.assert_not_called() + PrintTab._on_usb_file_added(tab, {"path": "cube.gcode"}) + self._printed(tab) + + def test_create_before_done_prints_at_once(self, tab) -> None: + PrintTab._on_usb_file_added(tab, {"path": "/cube.gcode"}) + PrintTab._on_usb_import_done( + tab, "USB-STICK/cube.gcode", "cube.gcode", "k", False, "" + ) + tab._usb_settle_timer.start.assert_not_called() + self._printed(tab) + + def test_timeout_prints_anyway(self, tab) -> None: + PrintTab._on_usb_import_done( + tab, "USB-STICK/cube.gcode", "cube.gcode", "k", False, "" + ) + PrintTab._on_usb_settle_timeout(tab) + self._printed(tab) + + def test_index_failure_still_prints(self, tab) -> None: + tab.remember.side_effect = OSError("read-only") + PrintTab._on_usb_import_done( + tab, "USB-STICK/cube.gcode", "cube.gcode", "k", True, "" + ) + tab.jobStatusPage_widget.on_print_start.assert_called_once_with("cube.gcode") + + def test_file_added_ignored_when_idle(self, tab) -> None: + tab._usb_import_task = None + PrintTab._on_usb_file_added(tab, {"path": "cube.gcode"}) + assert tab._usb_created == set() - def test_error_notifies_and_does_not_print(self) -> None: - tab = self._done_tab() - PrintTab._on_usb_import_done(tab, "USB-STICK/cube.gcode", "", "Bad") + def test_error_notifies_and_does_not_print(self, tab) -> None: + PrintTab._on_usb_import_done(tab, "USB-STICK/cube.gcode", "", "", False, "Bad") assert tab._usb_import_task is None tab.jobStatusPage_widget.on_print_start.assert_not_called() tab.notify.emit.assert_called_once_with( From db4e605fc52d4bddc3887e763c5b0b5948ed9ab1 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 30 Oct 2026 13:57:36 +0000 Subject: [PATCH 3/5] feat(print): import USB files started from Mainsail --- BlocksScreen/lib/panels/printTab.py | 12 ++++++++++++ tests/widgets/test_print_start_guard_unit.py | 20 ++++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 86544d94..0be795d6 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -29,6 +29,8 @@ from devices.amu import AMUManager USB_SETTLE_TIMEOUT_MS = 120_000 # Happy Hare pre-processing of a large file +# RF50-Klipper's SDCARD_PRINT_FILE override emits this for USB files (Mainsail starts). +USB_PRINT_ACTION = "// action:blocks_usb_print " logger = logging.getLogger(__name__) @@ -224,6 +226,7 @@ def __init__( self.jobStatusPage_widget.on_gcode_move_update ) self.printer.request_available_objects_signal.connect(self.klipper_ready_signal) + self.printer.gcode_response.connect(self._on_gcode_response) self.babystepPage = BabystepPage(self) self.babystepPage.request_back.connect(self.back_button) self.addWidget(self.babystepPage) @@ -379,6 +382,15 @@ def start_print(self, file: str) -> None: return self.jobStatusPage_widget.on_print_start(file) + @QtCore.pyqtSlot(list, name="on_gcode_response") + def _on_gcode_response(self, messages: list) -> None: + """Import a USB file whose print another client started through Klipper.""" + for line in messages: + file = str(line).removeprefix(USB_PRINT_ACTION) + # The macro sends USB paths only; a local one would bypass the touchscreen. + if file != line and helper_methods.is_usb_path(file): + self.start_print(file) + def _start_usb_import(self, file: str) -> None: """Copy a USB gcode to the gcodes root off-thread, then print the copy.""" # Printing from the stick would fail mid-job if it is pulled or glitches. diff --git a/tests/widgets/test_print_start_guard_unit.py b/tests/widgets/test_print_start_guard_unit.py index d756a729..ab2ab1bc 100644 --- a/tests/widgets/test_print_start_guard_unit.py +++ b/tests/widgets/test_print_start_guard_unit.py @@ -75,6 +75,26 @@ def test_usb_file_is_imported_not_printed(self) -> None: tab.jobStatusPage_widget.on_print_start.assert_not_called() +@pytest.mark.parametrize( + ("lines", "started"), + [ + ( + ["// action:blocks_usb_print USB-STICK/my cube.gcode"], + "USB-STICK/my cube.gcode", + ), + (["// action:blocks_usb_print cube.gcode"], None), + (["// blocks_usb_print USB-STICK/cube.gcode", "ok"], None), + ], +) +def test_usb_print_action_from_macro(lines, started) -> None: + tab = SimpleNamespace(start_print=MagicMock()) + PrintTab._on_gcode_response(tab, lines) + if started is None: + tab.start_print.assert_not_called() + else: + tab.start_print.assert_called_once_with(started) + + class TestUsbImportDone: @pytest.fixture def tab(self, monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: From c8e0c609deedef6ffed14c4db029a74d6feba5c8 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 30 Oct 2026 14:20:23 +0000 Subject: [PATCH 4/5] feat(print): enable the RF50-Klipper USB hand-off flag on Klipper ready --- BlocksScreen/lib/panels/printTab.py | 13 +++++++++++++ tests/widgets/test_print_start_guard_unit.py | 15 +++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 0be795d6..29b40f5c 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -31,6 +31,8 @@ USB_SETTLE_TIMEOUT_MS = 120_000 # Happy Hare pre-processing of a large file # RF50-Klipper's SDCARD_PRINT_FILE override emits this for USB files (Mainsail starts). USB_PRINT_ACTION = "// action:blocks_usb_print " +# The override hands USB files over only once this is True; older pairs print directly. +USB_PRINT_FLAG = "blocksscreen_usb" logger = logging.getLogger(__name__) @@ -227,6 +229,7 @@ def __init__( ) self.printer.request_available_objects_signal.connect(self.klipper_ready_signal) self.printer.gcode_response.connect(self._on_gcode_response) + self.printer.gcode_macro_update.connect(self._on_gcode_macro_update) self.babystepPage = BabystepPage(self) self.babystepPage.request_back.connect(self.back_button) self.addWidget(self.babystepPage) @@ -391,6 +394,16 @@ def _on_gcode_response(self, messages: list) -> None: if file != line and helper_methods.is_usb_path(file): self.start_print(file) + @QtCore.pyqtSlot(str, dict, name="on_gcode_macro_update") + def _on_gcode_macro_update(self, name: str, values: dict) -> None: + """Tell the RF50-Klipper USB override that BlocksScreen handles its files.""" + # False only on the override that knows the flag; Klipper resets it on restart. + if name.upper() == "SDCARD_PRINT_FILE" and values.get(USB_PRINT_FLAG) is False: + self.run_gcode_signal.emit( + f"SET_GCODE_VARIABLE MACRO=SDCARD_PRINT_FILE VARIABLE={USB_PRINT_FLAG}" + " VALUE=True" + ) + def _start_usb_import(self, file: str) -> None: """Copy a USB gcode to the gcodes root off-thread, then print the copy.""" # Printing from the stick would fail mid-job if it is pulled or glitches. diff --git a/tests/widgets/test_print_start_guard_unit.py b/tests/widgets/test_print_start_guard_unit.py index ab2ab1bc..feb42e55 100644 --- a/tests/widgets/test_print_start_guard_unit.py +++ b/tests/widgets/test_print_start_guard_unit.py @@ -95,6 +95,21 @@ def test_usb_print_action_from_macro(lines, started) -> None: tab.start_print.assert_called_once_with(started) +@pytest.mark.parametrize( + ("name", "values", "sent"), + [ + ("SDCARD_PRINT_FILE", {"blocksscreen_usb": False}, True), + ("SDCARD_PRINT_FILE", {"blocksscreen_usb": True}, False), + ("SDCARD_PRINT_FILE", {}, False), # another override, no flag to set + ("PRINT_START", {"blocksscreen_usb": False}, False), + ], +) +def test_usb_flag_set_only_on_flagged_override(name, values, sent) -> None: + tab = SimpleNamespace(run_gcode_signal=MagicMock()) + PrintTab._on_gcode_macro_update(tab, name, values) + assert tab.run_gcode_signal.emit.called is sent + + class TestUsbImportDone: @pytest.fixture def tab(self, monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: From 74737682f35ab5f773ad9570aae022b5bf2585f0 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 30 Oct 2026 16:27:26 +0000 Subject: [PATCH 5/5] fix(printTab): ack USB print hand-off so the macro can fall back --- BlocksScreen/lib/panels/printTab.py | 5 +++++ tests/widgets/test_print_start_guard_unit.py | 7 ++++++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/BlocksScreen/lib/panels/printTab.py b/BlocksScreen/lib/panels/printTab.py index 29b40f5c..005b0884 100644 --- a/BlocksScreen/lib/panels/printTab.py +++ b/BlocksScreen/lib/panels/printTab.py @@ -392,6 +392,11 @@ def _on_gcode_response(self, messages: list) -> None: file = str(line).removeprefix(USB_PRINT_ACTION) # The macro sends USB paths only; a local one would bypass the touchscreen. if file != line and helper_methods.is_usb_path(file): + # Ack first: an unacked hand-off makes the macro print from USB itself. + self.run_gcode_signal.emit( + "SET_GCODE_VARIABLE MACRO=SDCARD_PRINT_FILE" + " VARIABLE=usb_pending VALUE='\"\"'" + ) self.start_print(file) @QtCore.pyqtSlot(str, dict, name="on_gcode_macro_update") diff --git a/tests/widgets/test_print_start_guard_unit.py b/tests/widgets/test_print_start_guard_unit.py index feb42e55..9138798f 100644 --- a/tests/widgets/test_print_start_guard_unit.py +++ b/tests/widgets/test_print_start_guard_unit.py @@ -87,12 +87,17 @@ def test_usb_file_is_imported_not_printed(self) -> None: ], ) def test_usb_print_action_from_macro(lines, started) -> None: - tab = SimpleNamespace(start_print=MagicMock()) + tab = SimpleNamespace(start_print=MagicMock(), run_gcode_signal=MagicMock()) PrintTab._on_gcode_response(tab, lines) if started is None: tab.start_print.assert_not_called() + tab.run_gcode_signal.emit.assert_not_called() else: tab.start_print.assert_called_once_with(started) + tab.run_gcode_signal.emit.assert_called_once_with( + "SET_GCODE_VARIABLE MACRO=SDCARD_PRINT_FILE" + " VARIABLE=usb_pending VALUE='\"\"'" + ) @pytest.mark.parametrize(