From 6e42a4b7ed506898d9a90e15ace321c808ba25f3 Mon Sep 17 00:00:00 2001 From: Rohan Pandula Date: Sat, 18 Jul 2026 22:32:25 -0700 Subject: [PATCH 1/5] Add coolscan-roll optional dependency group coolscanpy is a standalone, SANE-free library for whole-roll scanning on a Nikon Coolscan LS-5000 with a roll feeder. Add it as a new dependency-group, following the same pattern as the existing "scanner" group for python-sane: opt-in, not a hard dependency of the rest of the application. --- pyproject.toml | 3 +++ uv.lock | 30 ++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 368661ee..c61807a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,6 +55,9 @@ camera = [ # Camera Scanning tab with them — are macOS and Linux only. "gphoto2>=2.5 ; sys_platform != 'win32'", ] +coolscan-roll = [ + "coolscanpy>=0.1.2", +] [project.urls] Homepage = "https://github.com/marcinz606/NegPy" diff --git a/uv.lock b/uv.lock index cfbb8304..400d686c 100644 --- a/uv.lock +++ b/uv.lock @@ -65,6 +65,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "coolscanpy" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "imagecodecs" }, + { name = "jinja2" }, + { name = "numpy" }, + { name = "opencv-python-headless" }, + { name = "pyusb" }, + { name = "tifffile" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c6/d0/2ab023bcd243daa927f20e3cdaac0db36a8a4b5df02d3424640c7863e7bd/coolscanpy-0.1.2.tar.gz", hash = "sha256:1ff98ea9689b19ad765bb7109ac18ab9d71509d266e87cbdf8434fa14577808e", size = 289030, upload-time = "2026-07-19T02:58:42.101Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/66/18473737fcba32a57fbef3d2668b6f15d94c869810d1a7dbeb4fff57a7bd/coolscanpy-0.1.2-py3-none-any.whl", hash = "sha256:a780eb3613b30e1299913c3c5f33f6b3255382ac4c3309e46691a071ec477a24", size = 296086, upload-time = "2026-07-19T02:58:40.819Z" }, +] + [[package]] name = "coverage" version = "7.14.0" @@ -315,6 +332,9 @@ dependencies = [ camera = [ { name = "gphoto2", marker = "sys_platform != 'win32'" }, ] +coolscan-roll = [ + { name = "coolscanpy" }, +] dev = [ { name = "pyinstaller" }, { name = "pytest" }, @@ -347,6 +367,7 @@ requires-dist = [ [package.metadata.requires-dev] camera = [{ name = "gphoto2", marker = "sys_platform != 'win32'", specifier = ">=2.5" }] +coolscan-roll = [{ name = "coolscanpy", specifier = ">=0.1.2" }] dev = [ { name = "pyinstaller", specifier = ">=6.18.0" }, { name = "pytest", specifier = "==9.0.2" }, @@ -725,6 +746,15 @@ version = "2.9.2" source = { registry = "https://pypi.org/simple" } sdist = { url = "https://files.pythonhosted.org/packages/45/e9/e8baff69fc2347606c547201204d4b4843c7ad8ecb9164eceee42016eff6/python_sane-2.9.2.tar.gz", hash = "sha256:50ab8e0b033cececad26c7231a7254f80ad8fe9ec6b5c25add2493d7e2a07bbe", size = 22513, upload-time = "2025-07-21T21:20:21.735Z" } +[[package]] +name = "pyusb" +version = "1.3.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/00/6b/ce3727395e52b7b76dfcf0c665e37d223b680b9becc60710d4bc08b7b7cb/pyusb-1.3.1.tar.gz", hash = "sha256:3af070b607467c1c164f49d5b0caabe8ac78dbed9298d703a8dbf9df4052d17e", size = 77281, upload-time = "2025-01-08T23:45:01.866Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/28/b8/27e6312e86408a44fe16bd28ee12dd98608b39f7e7e57884a24e8f29b573/pyusb-1.3.1-py3-none-any.whl", hash = "sha256:bf9b754557af4717fe80c2b07cc2b923a9151f5c08d17bdb5345dac09d6a0430", size = 58465, upload-time = "2025-01-08T23:45:00.029Z" }, +] + [[package]] name = "pywin32-ctypes" version = "0.2.3" From cb209167298be1ebc23afd2905bcead5341bc321 Mon Sep 17 00:00:00 2001 From: Rohan Pandula Date: Sat, 18 Jul 2026 22:32:32 -0700 Subject: [PATCH 2/5] Add coolscanpy roll-scanning backend adapter CoolscanRoll wraps coolscanpy's Device/Roll workflow (preview, spacing correction, approval, batch fine-scan, safe stop) behind a small RollHandle, translating coolscanpy's typed PyCoolscanError hierarchy to plain RuntimeErrors the same way SaneBackend already reports failures elsewhere in this package. available() is a cheap importlib.util.find_spec() check, mirroring ScanSidebar's existing _sane_available() gate, so the feature degrades to fully absent when coolscanpy isn't installed. open_roll() is the single place this adapter resolves a device handle, marked as the intended re-point target once the maintainer's own generic SANE-based coolscan route defines a real backend-selection seam on ScannerService. Tests inject a fake coolscanpy module (tests/scanners/conftest.py) so none of this requires real hardware or the real package installed. --- .../scanners/coolscanpy_roll.py | 175 ++++++++++++++++++ tests/scanners/conftest.py | 175 ++++++++++++++++++ tests/scanners/test_coolscanpy_roll.py | 166 +++++++++++++++++ 3 files changed, 516 insertions(+) create mode 100644 negpy/infrastructure/scanners/coolscanpy_roll.py create mode 100644 tests/scanners/conftest.py create mode 100644 tests/scanners/test_coolscanpy_roll.py diff --git a/negpy/infrastructure/scanners/coolscanpy_roll.py b/negpy/infrastructure/scanners/coolscanpy_roll.py new file mode 100644 index 00000000..bd466550 --- /dev/null +++ b/negpy/infrastructure/scanners/coolscanpy_roll.py @@ -0,0 +1,175 @@ +"""Optional coolscanpy-backed roll-scanning adapter for the Nikon Coolscan LS-5000. + +coolscanpy (https://github.com/rohanpandula/coolscanpy) is a standalone, +SANE-free library that talks to an LS-5000 plus SA-21/SA-30 roll feeder +directly over USB: whole-roll preview, per-slot spacing correction, and +batch fine-scanning with receipts. NegPy consumes it exactly like +python-sane or gphoto2 elsewhere in this package -- an optional dependency, +imported lazily, entirely absent by default. See `available()`. + +This module owns every place coolscanpy itself gets imported or driven for +the roll workflow: opening a device, opening its roll extension, and +translating coolscanpy's exceptions. `negpy.services.scanning.roll_service` +builds the file-writing workflow on top of it but never imports coolscanpy +directly. + +-------------------------------------------------------------------------- +INTEGRATION POINT for a future re-point +-------------------------------------------------------------------------- +`open_roll()` below is the one place a device handle gets resolved for the +roll workflow. It currently goes straight to `coolscanpy.open()`. NegPy's +maintainer has a generic SANE-based coolscan route planned upstream, which +is expected to add a real backend-selection seam to +`negpy.services.scanning.service.ScannerService` (today `_get_backend()` +just hardcodes `SaneBackend()`). When that seam lands, re-pointing this +adapter at it should only mean changing how `open_roll()` resolves a +device -- everything built on top of the returned `RollHandle` (this +module's exception translation, and all of +`negpy/services/scanning/roll_service.py`) is unaffected, since it never +talks to coolscanpy except through this one function and the `RollHandle` +it returns. +-------------------------------------------------------------------------- +""" + +from __future__ import annotations + +import importlib.util +from typing import TYPE_CHECKING, Iterable, Iterator + +if TYPE_CHECKING: + import coolscanpy + + +def available() -> bool: + """True if the optional `coolscanpy` dependency is importable. + + A cheap, side-effect-free presence check -- mirrors the + `ScanSidebar._sane_available()` check the plain Scan panel already uses + for python-sane, so callers (a future roll panel, a service, a test) can + degrade gracefully without ever importing coolscanpy just to ask whether + it exists. + """ + return importlib.util.find_spec("coolscanpy") is not None + + +class RollHandle: + """One open coolscanpy `Device` + `Roll`, released together on `close()`. + + Construct via `open_roll()`, not directly. Every method here is a thin + forward to the underlying `coolscanpy.Roll`, translating coolscanpy's + typed exceptions (rooted at `PyCoolscanError`) to plain `RuntimeError`s + -- matching how `SaneBackend.scan()` already reports failures elsewhere + in this package (a message string the scan worker forwards verbatim to + the GUI's status label). NegPy's scanner layer has no typed exception + vocabulary of its own for a caller to catch selectively; a caller that + wants coolscanpy's own richer hierarchy can still recover it from + `__cause__`, since every translation chains `from error`. + """ + + def __init__(self, device: "coolscanpy.Device", roll: "coolscanpy.Roll") -> None: + self._device = device + self._roll = roll + + def preview(self, slots: Iterable[int] | None = None, *, on_progress=None) -> list: + """One whole-roll transport read. See `coolscanpy.Roll.preview`.""" + try: + return self._roll.preview(slots, on_progress=on_progress) + except Exception as error: + raise _translate(error) from error + + def set_spacing_offset(self, slot: int, offset_rows: int) -> None: + try: + self._roll.set_spacing_offset(slot, offset_rows) + except Exception as error: + raise _translate(error) from error + + def approve(self, slot: int) -> None: + try: + self._roll.approve(slot) + except Exception as error: + raise _translate(error) from error + + def needs_approval(self, slot: int) -> bool: + try: + return self._roll.needs_approval(slot) + except Exception as error: + raise _translate(error) from error + + def scan_many(self, slots: Iterable[int], *, on_progress=None) -> Iterator["coolscanpy.Frame"]: + """Batch fine-scan `slots` in one transport reservation. + + Note for a future caller that wires this to a cancel affordance + (mirroring how `ScanWorker.run_scan` swallows a cancelled plain scan + rather than reporting it as an error): a `RuntimeError` raised here + whose `__cause__` is a `coolscanpy.SafeStopRequested` means + `safe_stop()` was already called deliberately and should be treated + the same way, not surfaced as a failure. + """ + try: + yield from self._roll.scan_many(slots, on_progress=on_progress) + except Exception as error: + raise _translate(error) from error + + def safe_stop(self) -> None: + """Request a graceful stop; the frame already in flight still finishes.""" + self._roll.safe_stop() + + def close(self) -> None: + """Idempotent. Releases the roll extension, then the device.""" + try: + self._roll.close() + finally: + self._device.close() + + def __enter__(self) -> "RollHandle": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + +def list_devices() -> "list[coolscanpy.DeviceInfo]": + """Enumerate attached LS-5000 units. Empty list, never raises, if none found.""" + import coolscanpy + + return coolscanpy.get_devices() + + +def open_roll(device_id: str | None = None, *, material: "coolscanpy.Material | None" = None) -> RollHandle: + """Open one LS-5000 and its roll-feeder extension. + + `device_id` follows `coolscanpy.open()`: omitted (`None`) picks "the one + attached unit", an explicit id (from `list_devices()`) disambiguates + when more than one is attached. `material` defaults to + `coolscanpy.Material.COLOR_NEGATIVE`, the only material coolscanpy's + roll engine fine-scans end to end today; black-and-white preview works + but its batch scan raises `NotImplementedError` (unchanged, surfaced as + a plain `RuntimeError` here like everything else). + """ + import coolscanpy + + resolved_material = material if material is not None else coolscanpy.Material.COLOR_NEGATIVE + try: + device = coolscanpy.open(device_id or "ls5000") + except Exception as error: + raise _translate(error) from error + + try: + roll = device.roll(material=resolved_material) + except Exception as error: + device.close() + raise _translate(error) from error + + return RollHandle(device, roll) + + +def _translate(error: BaseException) -> RuntimeError: + """Flatten any coolscanpy failure to a plain `RuntimeError`. + + Covers the typed `PyCoolscanError` hierarchy (`RollMismatch`, + `FingerprintRefused`, `ManualReviewRequired`, `SafeStopRequested`, ...) + and the handful of plain `ValueError`/`RuntimeError` coolscanpy itself + raises (e.g. "no roll adapter is attached"), so nothing coolscanpy-shaped + ever crosses out of this module. + """ + return RuntimeError(str(error)) diff --git a/tests/scanners/conftest.py b/tests/scanners/conftest.py new file mode 100644 index 00000000..5890566a --- /dev/null +++ b/tests/scanners/conftest.py @@ -0,0 +1,175 @@ +"""Shared fixtures for tests/scanners/: a fake `coolscanpy` module. + +`coolscanpy_roll.py` / `roll_service.py` treat coolscanpy as optional and +import it lazily, so these tests never need the real package installed -- +`fake_coolscanpy` injects a minimal stand-in module into `sys.modules` +instead, exercising the same `import coolscanpy` path the production code +takes. +""" + +from __future__ import annotations + +import dataclasses +import importlib.machinery +import sys +import types +from typing import Any + +import pytest + + +class FakePyCoolscanError(Exception): + """Stand-in for coolscanpy.PyCoolscanError.""" + + +class FakeDeviceNotFound(FakePyCoolscanError): + pass + + +class FakeManualReviewRequired(FakePyCoolscanError): + def __init__(self, message: str, *, slot: int) -> None: + super().__init__(message) + self.slot = slot + + +class FakeSafeStopRequested(FakePyCoolscanError): + pass + + +class FakeMaterial: + COLOR_NEGATIVE = "color-negative" + BLACK_AND_WHITE_NEGATIVE = "black-and-white-negative" + + +@dataclasses.dataclass(frozen=True) +class FakeThumbnail: + slot: int + image: Any + boundary_rows: tuple + spacing_offset: int + needs_approval: bool + warnings: tuple = () + + +@dataclasses.dataclass(frozen=True) +class FakeReceipt: + """Deliberately smaller than the real coolscanpy Receipt -- write_frame() + only needs *a* dataclass instance to round-trip through dataclasses.asdict().""" + + version: int + slot: int + dpi: int + depth: int + device_id: str + transport_smear_verdict: str + + +@dataclasses.dataclass(frozen=True) +class FakeFrame: + slot: int + rgb: Any + ir: Any + ir_validity: Any + receipt: Any + + +class FakeRoll: + """Scripted in-memory stand-in for `coolscanpy.Roll`.""" + + def __init__(self, thumbnails=None, frames=None, raise_on=None) -> None: + self._thumbnails = list(thumbnails or []) + self._frames = {frame.slot: frame for frame in (frames or [])} + self._raise_on = raise_on or {} + self.approved: list[int] = [] + self.spacing_offsets: dict[int, int] = {} + self.safe_stop_called = False + self.closed = False + + def preview(self, slots=None, *, on_progress=None): + if "preview" in self._raise_on: + raise self._raise_on["preview"] + wanted = None if slots is None else set(slots) + return [t for t in self._thumbnails if wanted is None or t.slot in wanted] + + def set_spacing_offset(self, slot, offset_rows) -> None: + self.spacing_offsets[slot] = offset_rows + + def approve(self, slot) -> None: + if "approve" in self._raise_on: + raise self._raise_on["approve"] + self.approved.append(slot) + + def needs_approval(self, slot) -> bool: + return slot in self._raise_on.get("needs_approval_slots", ()) + + def scan_many(self, slots, *, on_progress=None): + for slot in slots: + error = self._raise_on.get("scan_many_slots", {}).get(slot) + if error is not None: + raise error + yield self._frames[slot] + + def safe_stop(self) -> None: + self.safe_stop_called = True + + def close(self) -> None: + self.closed = True + + +class FakeDevice: + def __init__(self, roll) -> None: + self._roll = roll + self.closed = False + self.roll_called_with = None + + def roll(self, *, material=None): + self.roll_called_with = material + return self._roll + + def close(self) -> None: + self.closed = True + + +@pytest.fixture +def fake_coolscanpy(monkeypatch): + """Install a fake `coolscanpy` module; returns an object to script it. + + Usage:: + + fake_coolscanpy.state["open_device"] = fake_coolscanpy.Device(fake_coolscanpy.Roll()) + handle = coolscanpy_roll.open_roll() + """ + module = types.ModuleType("coolscanpy") + # importlib.util.find_spec() raises ValueError on a sys.modules entry + # with no __spec__, so give the fake one like a real imported module has. + module.__spec__ = importlib.machinery.ModuleSpec("coolscanpy", loader=None) + + state: dict[str, Any] = {"devices": [], "open_device": None, "open_error": None} + + def fake_get_devices(local_only: bool = False): + return state["devices"] + + def fake_open(devname: str): + if state["open_error"] is not None: + raise state["open_error"] + return state["open_device"] + + module.get_devices = fake_get_devices + module.open = fake_open + module.Material = FakeMaterial + module.PyCoolscanError = FakePyCoolscanError + module.DeviceNotFound = FakeDeviceNotFound + module.ManualReviewRequired = FakeManualReviewRequired + module.SafeStopRequested = FakeSafeStopRequested + + monkeypatch.setitem(sys.modules, "coolscanpy", module) + + return types.SimpleNamespace( + module=module, + state=state, + Device=FakeDevice, + Roll=FakeRoll, + Thumbnail=FakeThumbnail, + Frame=FakeFrame, + Receipt=FakeReceipt, + ) diff --git a/tests/scanners/test_coolscanpy_roll.py b/tests/scanners/test_coolscanpy_roll.py new file mode 100644 index 00000000..74009d7d --- /dev/null +++ b/tests/scanners/test_coolscanpy_roll.py @@ -0,0 +1,166 @@ +"""Tests for the optional coolscanpy-backed roll-scanning adapter. + +No real coolscanpy install is required or assumed: `fake_coolscanpy` (see +tests/scanners/conftest.py) injects a minimal stand-in module, exercising +the same lazy `import coolscanpy` path the production code takes. +""" + +from __future__ import annotations + +import importlib.util +import sys + +import numpy as np +import pytest + +from negpy.infrastructure.scanners import coolscanpy_roll + + +class TestAvailable: + def test_false_when_not_importable(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delitem(sys.modules, "coolscanpy", raising=False) + # This environment genuinely doesn't have coolscanpy installed (it's + # an optional dependency-group); if that ever changes, find_spec + # itself would need monkeypatching instead of just sys.modules. + assert importlib.util.find_spec("coolscanpy") is None + assert coolscanpy_roll.available() is False + + def test_true_when_fake_module_present(self, fake_coolscanpy) -> None: + assert coolscanpy_roll.available() is True + + +class TestListDevices: + def test_passthrough(self, fake_coolscanpy) -> None: + fake_coolscanpy.state["devices"] = ["device-a", "device-b"] + assert coolscanpy_roll.list_devices() == ["device-a", "device-b"] + + +class TestOpenRoll: + def test_opens_named_device_with_default_material(self, fake_coolscanpy) -> None: + roll = fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + + handle = coolscanpy_roll.open_roll("ls5000-usb-001") + + assert device.roll_called_with == fake_coolscanpy.module.Material.COLOR_NEGATIVE + handle.close() + assert roll.closed is True + assert device.closed is True + + def test_explicit_material_forwarded(self, fake_coolscanpy) -> None: + roll = fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + + coolscanpy_roll.open_roll(material=fake_coolscanpy.module.Material.BLACK_AND_WHITE_NEGATIVE) + + assert device.roll_called_with == fake_coolscanpy.module.Material.BLACK_AND_WHITE_NEGATIVE + + def test_device_not_found_translated_to_runtime_error(self, fake_coolscanpy) -> None: + fake_coolscanpy.state["open_error"] = fake_coolscanpy.module.DeviceNotFound("no Coolscan LS-5000 unit is attached") + + with pytest.raises(RuntimeError, match="no Coolscan LS-5000 unit is attached"): + coolscanpy_roll.open_roll() + + def test_roll_open_failure_closes_device_and_translates(self, fake_coolscanpy) -> None: + class ExplodingDevice(fake_coolscanpy.Device): + def roll(self, *, material=None): + raise ValueError("no roll adapter is attached/detected on this device") + + device = ExplodingDevice(None) + fake_coolscanpy.state["open_device"] = device + + with pytest.raises(RuntimeError, match="no roll adapter is attached"): + coolscanpy_roll.open_roll() + assert device.closed is True + + +class TestRollHandle: + @staticmethod + def _handle(fake_coolscanpy, roll=None): + roll = roll if roll is not None else fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + return coolscanpy_roll.open_roll(), roll, device + + def test_preview_passthrough(self, fake_coolscanpy) -> None: + thumb = fake_coolscanpy.Thumbnail(slot=1, image=np.zeros((4, 4, 3)), boundary_rows=(0, 4), spacing_offset=0, needs_approval=False) + handle, _roll, _device = self._handle(fake_coolscanpy, fake_coolscanpy.Roll(thumbnails=[thumb])) + + assert handle.preview() == [thumb] + + def test_preview_filters_by_requested_slots(self, fake_coolscanpy) -> None: + thumbs = [ + fake_coolscanpy.Thumbnail(slot=i, image=np.zeros((2, 2, 3)), boundary_rows=(0, 2), spacing_offset=0, needs_approval=False) + for i in (1, 2, 3) + ] + handle, _roll, _device = self._handle(fake_coolscanpy, fake_coolscanpy.Roll(thumbnails=thumbs)) + + result = handle.preview([2]) + + assert [t.slot for t in result] == [2] + + def test_preview_exception_translated(self, fake_coolscanpy) -> None: + handle, _roll, _device = self._handle( + fake_coolscanpy, + fake_coolscanpy.Roll(raise_on={"preview": fake_coolscanpy.module.PyCoolscanError("fingerprint mismatch")}), + ) + with pytest.raises(RuntimeError, match="fingerprint mismatch"): + handle.preview() + + def test_approve_records_slot(self, fake_coolscanpy) -> None: + handle, roll, _device = self._handle(fake_coolscanpy) + handle.approve(3) + assert roll.approved == [3] + + def test_manual_review_required_translated(self, fake_coolscanpy) -> None: + handle, _roll, _device = self._handle( + fake_coolscanpy, + fake_coolscanpy.Roll(raise_on={"approve": fake_coolscanpy.module.ManualReviewRequired("slot 3 needs review", slot=3)}), + ) + with pytest.raises(RuntimeError, match="slot 3 needs review"): + handle.approve(3) + + def test_set_spacing_offset_records_value(self, fake_coolscanpy) -> None: + handle, roll, _device = self._handle(fake_coolscanpy) + handle.set_spacing_offset(5, -12) + assert roll.spacing_offsets[5] == -12 + + def test_scan_many_yields_frames(self, fake_coolscanpy) -> None: + frame = fake_coolscanpy.Frame(slot=2, rgb=np.zeros((4, 4, 3), dtype=np.uint16), ir=None, ir_validity=None, receipt=None) + handle, _roll, _device = self._handle(fake_coolscanpy, fake_coolscanpy.Roll(frames=[frame])) + + assert list(handle.scan_many([2])) == [frame] + + def test_scan_many_exception_translated(self, fake_coolscanpy) -> None: + error = fake_coolscanpy.module.SafeStopRequested("safe stop requested; 0 of 1 requested frames completed") + handle, _roll, _device = self._handle(fake_coolscanpy, fake_coolscanpy.Roll(raise_on={"scan_many_slots": {5: error}})) + + with pytest.raises(RuntimeError, match="safe stop requested") as excinfo: + list(handle.scan_many([5])) + # __cause__ preserves the original typed exception for a caller that + # wants to distinguish a deliberate stop from a genuine failure. + assert isinstance(excinfo.value.__cause__, fake_coolscanpy.module.SafeStopRequested) + + def test_safe_stop_forwarded(self, fake_coolscanpy) -> None: + handle, roll, _device = self._handle(fake_coolscanpy) + handle.safe_stop() + assert roll.safe_stop_called is True + + def test_close_releases_roll_then_device(self, fake_coolscanpy) -> None: + handle, roll, device = self._handle(fake_coolscanpy) + handle.close() + assert roll.closed is True + assert device.closed is True + + def test_context_manager_closes_on_exit(self, fake_coolscanpy) -> None: + roll = fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + + with coolscanpy_roll.open_roll() as handle: + assert handle is not None + assert roll.closed is False + + assert roll.closed is True From 7b94deb74b92606c0938798adac35621c02857c2 Mon Sep 17 00:00:00 2001 From: Rohan Pandula Date: Sat, 18 Jul 2026 22:32:38 -0700 Subject: [PATCH 3/5] Add roll-scanning service and output writer RollScanningService orchestrates the coolscanpy_roll adapter: open/ close a roll (one at a time, matching coolscanpy's own reservation lock), preview, approve, spacing correction, batch scan_many, and safe_stop. write_frame() writes each scanned Frame to disk in NegPy's existing scan-output shape: a 16-bit TIFF master, an "_IR" TIFF sidecar when the frame carries an infrared plane (matching writer.write_tiff_16bit's own suffix), and an "_receipt.json" sidecar with the full Receipt. The filename pattern is the same Jinja2 template the plain Scan panel uses, but seq is seeded from the frame's physical slot number rather than probed for a free name, so redoing one bad slot overwrites that slot's files instead of accumulating a second copy. ir_validity is intentionally not persisted; nothing downstream reads it yet. --- negpy/services/scanning/roll_service.py | 184 ++++++++++++++++++++++++ tests/scanners/test_roll_service.py | 163 +++++++++++++++++++++ 2 files changed, 347 insertions(+) create mode 100644 negpy/services/scanning/roll_service.py create mode 100644 tests/scanners/test_roll_service.py diff --git a/negpy/services/scanning/roll_service.py b/negpy/services/scanning/roll_service.py new file mode 100644 index 00000000..5855a8f8 --- /dev/null +++ b/negpy/services/scanning/roll_service.py @@ -0,0 +1,184 @@ +"""Roll-scanning service backed by the optional `coolscanpy` package. + +Mirrors `negpy.services.scanning.service.ScannerService`: one class that +orchestrates device discovery, the hardware workflow, and writing results to +disk in NegPy's conventional layout. Where `ScannerService` wraps a single +ad-hoc `Device.scan()`, this wraps coolscanpy's whole-roll workflow +(preview -> approve -> batch fine-scan) exposed by +`negpy.infrastructure.scanners.coolscanpy_roll`. + +Entirely inert if `coolscanpy` is not installed: `available()` is a cheap +presence check re-exported from that module, and every other method only +reaches coolscanpy (transitively, through `coolscanpy_roll`) once actually +called. +""" + +from __future__ import annotations + +import dataclasses +import json +import os +import tempfile +from datetime import date as _date +from typing import TYPE_CHECKING, Iterable, Iterator + +import numpy as np +import tifffile + +from negpy.infrastructure.scanners import coolscanpy_roll +from negpy.kernel.system.logging import get_logger +from negpy.services.scanning.templating import render_scan_filename + +if TYPE_CHECKING: + import coolscanpy + +logger = get_logger(__name__) + +# Re-exported so callers only need this module: `from negpy.services.scanning +# .roll_service import available`, matching how the plain Scan sidebar checks +# `_sane_available()` before showing its device combo. +available = coolscanpy_roll.available + + +@dataclasses.dataclass(frozen=True) +class RollFrameOutput: + """Where one scanned frame's files landed on disk.""" + + slot: int + rgb_path: str + ir_path: str | None + receipt_path: str + + +class RollScanningService: + """Orchestrates coolscanpy device/roll lifecycle and output writing. + + One roll open at a time, matching `coolscanpy.Device.roll()`'s own + single-reservation lock -- call `close()` (or use this as a context + manager) before opening another. + """ + + def __init__(self) -> None: + self._roll: coolscanpy_roll.RollHandle | None = None + + # -- device / roll lifecycle ----------------------------------------- + + def list_devices(self) -> "list[coolscanpy.DeviceInfo]": + return coolscanpy_roll.list_devices() + + def open_roll(self, device_id: str | None = None, *, material: "coolscanpy.Material | None" = None) -> None: + """Open a device and its roll extension. Call `close()` when done.""" + if self._roll is not None: + raise RuntimeError("a roll is already open on this service; call close() first") + self._roll = coolscanpy_roll.open_roll(device_id, material=material) + + def close(self) -> None: + """Idempotent. Ends the roll reservation and releases the device.""" + if self._roll is not None: + self._roll.close() + self._roll = None + + def __enter__(self) -> "RollScanningService": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + # -- preview / approval ------------------------------------------------ + + def preview(self, slots: Iterable[int] | None = None, *, on_progress=None) -> list: + return self._require_roll().preview(slots, on_progress=on_progress) + + def set_spacing_offset(self, slot: int, offset_rows: int) -> None: + self._require_roll().set_spacing_offset(slot, offset_rows) + + def approve(self, slot: int) -> None: + self._require_roll().approve(slot) + + def needs_approval(self, slot: int) -> bool: + return self._require_roll().needs_approval(slot) + + # -- scanning ---------------------------------------------------------- + + def scan_many(self, slots: Iterable[int], *, on_progress=None) -> Iterator["coolscanpy.Frame"]: + yield from self._require_roll().scan_many(slots, on_progress=on_progress) + + def safe_stop(self) -> None: + if self._roll is not None: + self._roll.safe_stop() + + # -- writing ------------------------------------------------------------- + + def write_frame(self, frame: "coolscanpy.Frame", output_folder: str, filename_pattern: str) -> RollFrameOutput: + """Write one scanned `Frame` to disk: a 16-bit TIFF master, an `_IR` + TIFF sidecar when the frame carries an infrared plane (matching + `writer.write_tiff_16bit`'s own `_IR.tif` convention), and + a `_receipt.json` sidecar holding the full `Receipt`. + + `filename_pattern` is the same Jinja2 template the plain scan path + uses (variables: `date`, `seq`), but `seq` is seeded from the frame's + physical slot number rather than probed for the next free name: a + roll slot is already a stable identity, so re-scanning slot 5 + replaces slot 5's old files instead of piling up a `..._002` beside + them. + + `frame.ir_validity` (the per-pixel IR confidence mask) is not + persisted here -- it isn't part of NegPy's existing scan-output + convention and nothing downstream reads it yet. + """ + os.makedirs(output_folder, exist_ok=True) + date_str = _date.today().strftime("%Y%m%d") + basename = render_scan_filename(filename_pattern, date_str, frame.slot) + base_path = os.path.join(output_folder, basename) + + rgb_path = _atomic_write_tiff(base_path + ".tif", frame.rgb, photometric="rgb") + + ir_path = None + if frame.ir is not None: + ir_path = _atomic_write_tiff(base_path + "_IR.tif", frame.ir, photometric="minisblack") + + receipt_path = base_path + "_receipt.json" + _atomic_write_json(receipt_path, dataclasses.asdict(frame.receipt)) + + return RollFrameOutput(slot=frame.slot, rgb_path=rgb_path, ir_path=ir_path, receipt_path=receipt_path) + + # -- internals ----------------------------------------------------------- + + def _require_roll(self) -> coolscanpy_roll.RollHandle: + if self._roll is None: + raise RuntimeError("no roll is open; call open_roll() first") + return self._roll + + +def _ensure_uint16(array: np.ndarray) -> np.ndarray: + return array if array.dtype == np.uint16 else array.astype(np.uint16) + + +def _atomic_write_tiff(path: str, array: np.ndarray, *, photometric: str) -> str: + """Write `array` to `path` via a temp file + rename, matching the + atomic-write convention `negpy.services.scanning.writer` already uses + for the plain scan path.""" + fd, tmp_path = tempfile.mkstemp(suffix=".tif", dir=os.path.dirname(path) or ".") + os.close(fd) + try: + tifffile.imwrite(tmp_path, _ensure_uint16(array), photometric=photometric, compression="lzw") + os.replace(tmp_path, path) + except Exception: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + raise + return path + + +def _atomic_write_json(path: str, payload: dict) -> str: + tmp_path = None + try: + with tempfile.NamedTemporaryFile("w", dir=os.path.dirname(path) or ".", delete=False, suffix=".part", encoding="utf-8") as tmp: + tmp_path = tmp.name + json.dump(payload, tmp, default=str, indent=2) + os.replace(tmp_path, path) + except Exception: + if tmp_path is not None and os.path.exists(tmp_path): + os.unlink(tmp_path) + raise + return path diff --git a/tests/scanners/test_roll_service.py b/tests/scanners/test_roll_service.py new file mode 100644 index 00000000..2ac621ba --- /dev/null +++ b/tests/scanners/test_roll_service.py @@ -0,0 +1,163 @@ +"""Tests for RollScanningService: lifecycle orchestration and output writing. + +Lifecycle tests use `fake_coolscanpy` (see tests/scanners/conftest.py) the +same way test_coolscanpy_roll.py does. write_frame() tests construct a fake +Frame/Receipt directly -- writing to disk never touches coolscanpy itself, +so no module injection is needed for those. +""" + +from __future__ import annotations + +import json +import os + +import numpy as np +import pytest +import tifffile + +from negpy.services.scanning import roll_service +from negpy.services.scanning.roll_service import RollScanningService + + +class TestAvailable: + def test_reexports_backend_availability(self, fake_coolscanpy) -> None: + assert roll_service.available() is True + + +class TestRollLifecycle: + def _open_service(self, fake_coolscanpy, roll=None): + roll = roll if roll is not None else fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + service = RollScanningService() + service.open_roll("ls5000-usb-001") + return service, roll, device + + def test_open_then_close(self, fake_coolscanpy) -> None: + service, roll, device = self._open_service(fake_coolscanpy) + service.close() + assert roll.closed is True + assert device.closed is True + + def test_double_open_raises(self, fake_coolscanpy) -> None: + service, _roll, _device = self._open_service(fake_coolscanpy) + with pytest.raises(RuntimeError, match="already open"): + service.open_roll() + + def test_methods_before_open_raise(self) -> None: + service = RollScanningService() + with pytest.raises(RuntimeError, match="no roll is open"): + service.preview() + + def test_close_without_open_is_a_no_op(self) -> None: + RollScanningService().close() # must not raise + + def test_preview_and_approve_delegate_to_handle(self, fake_coolscanpy) -> None: + thumb = fake_coolscanpy.Thumbnail(slot=1, image=np.zeros((2, 2, 3)), boundary_rows=(0, 2), spacing_offset=0, needs_approval=True) + service, roll, _device = self._open_service(fake_coolscanpy, fake_coolscanpy.Roll(thumbnails=[thumb])) + + assert service.preview() == [thumb] + service.approve(1) + assert roll.approved == [1] + + def test_scan_many_delegates_to_handle(self, fake_coolscanpy) -> None: + frame = fake_coolscanpy.Frame(slot=1, rgb=np.zeros((2, 2, 3), dtype=np.uint16), ir=None, ir_validity=None, receipt=None) + service, _roll, _device = self._open_service(fake_coolscanpy, fake_coolscanpy.Roll(frames=[frame])) + + assert list(service.scan_many([1])) == [frame] + + def test_safe_stop_before_open_is_a_no_op(self) -> None: + RollScanningService().safe_stop() # must not raise + + def test_context_manager_closes(self, fake_coolscanpy) -> None: + roll = fake_coolscanpy.Roll() + device = fake_coolscanpy.Device(roll) + fake_coolscanpy.state["open_device"] = device + + with RollScanningService() as service: + service.open_roll() + assert roll.closed is True + + +class TestWriteFrame: + def _frame(self, fake_coolscanpy, *, slot=7, ir=None): + rgb = np.random.randint(0, 65535, (40, 60, 3), dtype=np.uint16) + receipt = fake_coolscanpy.Receipt(version=1, slot=slot, dpi=4000, depth=16, device_id="usb:1:2", transport_smear_verdict="clean") + return fake_coolscanpy.Frame(slot=slot, rgb=rgb, ir=ir, ir_validity=None, receipt=receipt) + + def test_writes_rgb_tiff(self, fake_coolscanpy, tmp_path) -> None: + frame = self._frame(fake_coolscanpy) + service = RollScanningService() + + output = service.write_frame(frame, str(tmp_path), '{{ date }}_slot{{ "%02d" % seq }}') + + assert os.path.exists(output.rgb_path) + assert output.rgb_path.endswith(".tif") + readback = tifffile.imread(output.rgb_path) + assert readback.shape == (40, 60, 3) + assert readback.dtype == np.uint16 + np.testing.assert_array_equal(readback, frame.rgb) + + def test_seq_seeded_from_slot_number(self, fake_coolscanpy, tmp_path) -> None: + frame = self._frame(fake_coolscanpy, slot=23) + service = RollScanningService() + + output = service.write_frame(frame, str(tmp_path), '{{ "%03d" % seq }}') + + assert "023" in os.path.basename(output.rgb_path) + + def test_writes_ir_sidecar_when_present(self, fake_coolscanpy, tmp_path) -> None: + ir = np.random.randint(0, 65535, (40, 60), dtype=np.uint16) + frame = self._frame(fake_coolscanpy, ir=ir) + service = RollScanningService() + + output = service.write_frame(frame, str(tmp_path), '{{ "%03d" % seq }}') + + assert output.ir_path is not None + assert output.ir_path.endswith("_IR.tif") + assert os.path.exists(output.ir_path) + readback = tifffile.imread(output.ir_path) + np.testing.assert_array_equal(readback, ir) + + def test_no_ir_sidecar_when_absent(self, fake_coolscanpy, tmp_path) -> None: + frame = self._frame(fake_coolscanpy, ir=None) + service = RollScanningService() + + output = service.write_frame(frame, str(tmp_path), '{{ "%03d" % seq }}') + + assert output.ir_path is None + assert not any(name.endswith("_IR.tif") for name in os.listdir(tmp_path)) + + def test_writes_receipt_json_sidecar(self, fake_coolscanpy, tmp_path) -> None: + frame = self._frame(fake_coolscanpy, slot=9) + service = RollScanningService() + + output = service.write_frame(frame, str(tmp_path), '{{ "%03d" % seq }}') + + assert output.receipt_path.endswith("_receipt.json") + with open(output.receipt_path) as fh: + payload = json.load(fh) + assert payload["slot"] == 9 + assert payload["dpi"] == 4000 + assert payload["transport_smear_verdict"] == "clean" + + def test_rescanning_same_slot_overwrites(self, fake_coolscanpy, tmp_path) -> None: + service = RollScanningService() + pattern = '{{ "%03d" % seq }}' + + first = service.write_frame(self._frame(fake_coolscanpy, slot=4), str(tmp_path), pattern) + second_frame = self._frame(fake_coolscanpy, slot=4) + second = service.write_frame(second_frame, str(tmp_path), pattern) + + assert first.rgb_path == second.rgb_path + readback = tifffile.imread(second.rgb_path) + np.testing.assert_array_equal(readback, second_frame.rgb) + + def test_creates_output_folder(self, fake_coolscanpy, tmp_path) -> None: + nested = tmp_path / "does" / "not" / "exist" / "yet" + frame = self._frame(fake_coolscanpy) + service = RollScanningService() + + output = service.write_frame(frame, str(nested), '{{ "%03d" % seq }}') + + assert os.path.exists(output.rgb_path) From dbe32d00e650708e6bd871d8412a93e6b5457118 Mon Sep 17 00:00:00 2001 From: Rohan Pandula Date: Sat, 18 Jul 2026 22:32:43 -0700 Subject: [PATCH 4/5] Document coolscanpy roll-scanning integration Explains what the integration adds, install steps, the LS-5000/SA-30 hardware caveat, the on-disk output convention, error handling, and where the single integration point lives for a future re-point onto the maintainer's own generic SANE-based coolscan route. Also documents the follow-up: no contact-sheet GUI panel yet, and where one would plug in. --- docs/COOLSCANPY_ROLL_SCANNING.md | 158 +++++++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 docs/COOLSCANPY_ROLL_SCANNING.md diff --git a/docs/COOLSCANPY_ROLL_SCANNING.md b/docs/COOLSCANPY_ROLL_SCANNING.md new file mode 100644 index 00000000..56cda623 --- /dev/null +++ b/docs/COOLSCANPY_ROLL_SCANNING.md @@ -0,0 +1,158 @@ +# Roll scanning with coolscanpy + +This page describes NegPy's optional integration with +[coolscanpy](https://github.com/rohanpandula/coolscanpy), a standalone +Python library for whole-roll scanning on a Nikon Coolscan LS-5000 with a +roll feeder. coolscanpy talks to the scanner directly over USB and does not +need SANE installed. NegPy consumes it the same way it consumes +python-sane or gphoto2: as an optional dependency that the rest of the +application never requires. + +## What it adds + +The plain Scan panel in NegPy scans one frame at a time through SANE. A +roll feeder changes the shape of the problem: you preview the whole roll in +one transport pass, check or nudge each frame's spacing, and then run a +batch fine scan across a list of slots without reloading the film. That +whole-roll workflow, preview, spacing correction, approval of any frame the +transport could not place automatically, and batch scanning with per-frame +receipts, is what coolscanpy implements and what this integration exposes +inside NegPy. + +Coverage is narrower than the plain Scan panel. Only color negative film +scans through coolscanpy's roll engine end to end today. Black and white +negative previews and approves normally, but batch fine scanning it raises +`NotImplementedError` inside coolscanpy itself; that limitation is +documented in the coolscanpy README and is unchanged by this integration. + +## Install + +Nothing here is required to run NegPy. The feature is entirely absent +until coolscanpy is importable, and every entry point checks for that +before doing anything else. + +If you use uv, sync the new dependency group the same way you already +sync the `scanner` group for python-sane: + +``` +uv sync --group coolscan-roll +``` + +If you manage your own environment instead, install the package directly: + +``` +pip install coolscanpy +``` + +coolscanpy has no SANE dependency for the roll workflow. See the +coolscanpy README if you also want its plain single-frame `scan()` path, +which does need SANE. + +## Hardware support + +coolscanpy, and by extension this integration, has been validated against +one specific setup: a Nikon Super Coolscan 5000 ED (LS-5000) running +firmware 1.03, with an SA-21 roll feeder wired for SA-30 compatibility. +Live hardware validation on 2026-07-18 confirmed USB enumeration, device +open, and a full roll preview with correct spacing and manual-review +flagging. Fine scanning and infrared capture through the roll engine have +not yet been re-run live since coolscanpy was extracted into its own +package; that is the remaining validation step on the coolscanpy side, not +something specific to this NegPy integration. + +Every other Coolscan model and every roll feeder other than the SA-21/SA-30 +combination above is untested. The transport protocol is not assumed to be +LS-5000-only where it does not have to be, but nothing beyond that one +combination has scanned real film. Windows has never run the roll engine +against hardware. + +## What this integration writes + +A batch scan produces, per slot, three files in the configured output +folder: + +- a 16-bit TIFF holding the scanner-linear RGB image +- a 16-bit TIFF holding the infrared plane, named with an `_IR` suffix + next to the RGB file, matching the suffix NegPy's plain scan writer + already uses for infrared +- a JSON file with an `_receipt` suffix, holding the full scan receipt + coolscanpy returns for that frame: exposure, clipping and focus + telemetry, transport-smear assessment, and the fingerprints the frame was + checked against + +The filename pattern is the same Jinja2 template the plain Scan panel +uses, with `date` and `seq` variables. For roll scanning, `seq` is the +frame's physical slot number rather than an incrementing counter. Slot +numbers are already a stable identity for a given roll, so re-scanning one +bad slot overwrites that slot's old files instead of accumulating a second +copy beside them. + +The infrared confidence mask coolscanpy also returns per frame is not +written to disk by this integration. Nothing downstream currently reads +it, and adding a file format for it before there is a consumer would be +guessing at a convention rather than following one. + +## Error handling + +coolscanpy raises a typed exception hierarchy rooted at +`PyCoolscanError`, covering things like a roll that no longer matches its +last preview, a slot that needs manual approval before it can be scanned, +and a requested safe stop. NegPy's existing scanner code has no typed +exception vocabulary of its own; `SaneBackend` reports every failure as a +plain `RuntimeError` with a human-readable message, and that message is +what ends up in the Scan panel's status label. This integration follows +the same convention: every coolscanpy exception is flattened to a +`RuntimeError` at the boundary, with the original exception preserved as +`__cause__` for any caller that wants to branch on it. + +One case is worth calling out for whoever builds the interactive panel +described below. `Roll.safe_stop()` lets the frame already in flight +finish and only refuses the next one, by raising `SafeStopRequested`. The +existing plain-scan worker treats a cancellation the same way, checking +its own cancel flag and returning quietly instead of reporting an error. +A roll-scanning worker should do the same for a `RuntimeError` whose +`__cause__` is `SafeStopRequested`: that is an expected stop, not a +failure. + +## The integration point + +Everything that touches coolscanpy itself lives in one file, +`negpy/infrastructure/scanners/coolscanpy_roll.py`. `open_roll()` in that +file is the single place a device handle gets resolved for the roll +workflow; it currently calls `coolscanpy.open()` directly. Everything else, +including `negpy/services/scanning/roll_service.py`, only ever talks to the +`RollHandle` that function returns. + +That module is marked in a comment as the intended target for a future +change. NegPy's maintainer has a generic SANE-based coolscan route planned +upstream, which is expected to add a real backend-selection seam to +`ScannerService` (today `ScannerService._get_backend()` simply hardcodes +`SaneBackend()`, with no selection mechanism at all). When that seam +lands, re-pointing this roll integration at it should only require changing +how `open_roll()` resolves a device. The roll workflow built on top of it, +the exception translation in the same file, and the whole of +`roll_service.py`, should not need to change. + +## What is not built yet + +This integration ships the backend and service layer: device discovery, +roll preview, spacing correction, approval, batch scanning, and file +writing, all covered by hardware-free tests. It does not ship a contact +sheet panel with rendered thumbnail images in the desktop GUI. Building +that panel from scratch, before the maintainer's own SANE-based route +defines how a Coolscan device is meant to surface in the UI, risked adding +exactly the kind of large, hard-to-review, foreign-feeling change this +integration is meant to avoid. + +A future panel can follow the existing Scan panel closely. Look at +`negpy/desktop/view/sidebar/scan.py` for the sidebar structure and its +`_sane_available()` availability gate, `negpy/desktop/workers/scan_worker.py` +for the background-thread pattern (a `QObject` moved to its own `QThread`, +progress and result reported back over Qt signals), and +`negpy/desktop/controller.py` for how that worker's signals get wired into +the rest of the application. A roll panel would follow the same shape, +calling `RollScanningService` instead of `ScannerService`, gated behind +`coolscanpy_roll.available()` the same way the plain panel gates itself +behind SANE. The simplest first version would not need thumbnail images at +all: a plain list of slots with their approval state, backed by +`Roll.preview()`, is enough to let someone pick which slots to scan. From 6a1be25fe2204be5e070c221c72cbaef29d82229 Mon Sep 17 00:00:00 2001 From: Rohan Pandula Date: Sat, 18 Jul 2026 23:17:17 -0700 Subject: [PATCH 5/5] Move roll scanning into its own infrastructure/services package The camera capture route lives beside scanners/ and scanning/ as its own infrastructure/capture + services/capture pair, gated on an optional dependency, rather than inside the single-frame ScannerBackend protocol. Give roll scanning the same shape: infrastructure/roll/coolscanpy_roll.py and services/roll/service.py, instead of sitting inside the plain-SANE scanners/scanning packages where it doesn't implement that protocol and never will. Also adds a RollScanningError for this service's own lifecycle misuse (mirrors CaptureError), an is_safe_stop() helper next to the existing exception translation so a future worker can tell a deliberate safe_stop() outcome from a real failure, and precise on_progress/return typing sourced from coolscanpy's actual Progress/Thumbnail/Frame types. No behavior change to the preview/approve/scan/write workflow itself. --- negpy/infrastructure/roll/__init__.py | 8 +++ .../{scanners => roll}/coolscanpy_roll.py | 50 +++++++++++++------ negpy/infrastructure/roll/settings.py | 26 ++++++++++ negpy/services/roll/__init__.py | 15 ++++++ .../roll_service.py => roll/service.py} | 36 ++++++++----- tests/roll/__init__.py | 0 tests/{scanners => roll}/conftest.py | 12 ++--- .../test_coolscanpy_roll.py | 2 +- .../test_service.py} | 10 ++-- 9 files changed, 120 insertions(+), 39 deletions(-) create mode 100644 negpy/infrastructure/roll/__init__.py rename negpy/infrastructure/{scanners => roll}/coolscanpy_roll.py (77%) create mode 100644 negpy/infrastructure/roll/settings.py create mode 100644 negpy/services/roll/__init__.py rename negpy/services/{scanning/roll_service.py => roll/service.py} (82%) create mode 100644 tests/roll/__init__.py rename tests/{scanners => roll}/conftest.py (92%) rename tests/{scanners => roll}/test_coolscanpy_roll.py (99%) rename tests/{scanners/test_roll_service.py => roll/test_service.py} (94%) diff --git a/negpy/infrastructure/roll/__init__.py b/negpy/infrastructure/roll/__init__.py new file mode 100644 index 00000000..c390fe95 --- /dev/null +++ b/negpy/infrastructure/roll/__init__.py @@ -0,0 +1,8 @@ +"""Coolscan roll-feeder capture layer -- no Qt, no NegPy file model. + +Drives a Nikon Coolscan LS-5000 plus SA-21/SA-30 roll feeder through the +optional `coolscanpy` package: whole-roll preview, per-slot spacing +correction and approval, and batch fine-scanning with receipts. Entirely +absent when `coolscanpy` is not installed -- see +`coolscanpy_roll.available()`. +""" diff --git a/negpy/infrastructure/scanners/coolscanpy_roll.py b/negpy/infrastructure/roll/coolscanpy_roll.py similarity index 77% rename from negpy/infrastructure/scanners/coolscanpy_roll.py rename to negpy/infrastructure/roll/coolscanpy_roll.py index bd466550..a011f0f0 100644 --- a/negpy/infrastructure/scanners/coolscanpy_roll.py +++ b/negpy/infrastructure/roll/coolscanpy_roll.py @@ -9,9 +9,10 @@ This module owns every place coolscanpy itself gets imported or driven for the roll workflow: opening a device, opening its roll extension, and -translating coolscanpy's exceptions. `negpy.services.scanning.roll_service` -builds the file-writing workflow on top of it but never imports coolscanpy -directly. +translating coolscanpy's exceptions. `negpy.services.roll.service` builds +the file-writing workflow on top of it but never imports coolscanpy +directly, matching how `negpy.services.capture.service` never imports +`gphoto2` directly either. -------------------------------------------------------------------------- INTEGRATION POINT for a future re-point @@ -25,9 +26,9 @@ adapter at it should only mean changing how `open_roll()` resolves a device -- everything built on top of the returned `RollHandle` (this module's exception translation, and all of -`negpy/services/scanning/roll_service.py`) is unaffected, since it never -talks to coolscanpy except through this one function and the `RollHandle` -it returns. +`negpy/services/roll/service.py`) is unaffected, since it never talks to +coolscanpy except through this one function and the `RollHandle` it +returns. -------------------------------------------------------------------------- """ @@ -70,7 +71,9 @@ def __init__(self, device: "coolscanpy.Device", roll: "coolscanpy.Roll") -> None self._device = device self._roll = roll - def preview(self, slots: Iterable[int] | None = None, *, on_progress=None) -> list: + def preview( + self, slots: Iterable[int] | None = None, *, on_progress: "coolscanpy.ProgressCallback | None" = None + ) -> "list[coolscanpy.Thumbnail]": """One whole-roll transport read. See `coolscanpy.Roll.preview`.""" try: return self._roll.preview(slots, on_progress=on_progress) @@ -95,15 +98,15 @@ def needs_approval(self, slot: int) -> bool: except Exception as error: raise _translate(error) from error - def scan_many(self, slots: Iterable[int], *, on_progress=None) -> Iterator["coolscanpy.Frame"]: + def scan_many( + self, slots: Iterable[int], *, on_progress: "coolscanpy.ProgressCallback | None" = None + ) -> Iterator["coolscanpy.Frame"]: """Batch fine-scan `slots` in one transport reservation. - Note for a future caller that wires this to a cancel affordance - (mirroring how `ScanWorker.run_scan` swallows a cancelled plain scan - rather than reporting it as an error): a `RuntimeError` raised here - whose `__cause__` is a `coolscanpy.SafeStopRequested` means - `safe_stop()` was already called deliberately and should be treated - the same way, not surfaced as a failure. + A `RuntimeError` raised here whose `__cause__` is a + `coolscanpy.SafeStopRequested` means `safe_stop()` was already + called deliberately and should be treated as a clean stop, not + surfaced as a failure -- see `is_safe_stop()`. """ try: yield from self._roll.scan_many(slots, on_progress=on_progress) @@ -170,6 +173,23 @@ def _translate(error: BaseException) -> RuntimeError: `FingerprintRefused`, `ManualReviewRequired`, `SafeStopRequested`, ...) and the handful of plain `ValueError`/`RuntimeError` coolscanpy itself raises (e.g. "no roll adapter is attached"), so nothing coolscanpy-shaped - ever crosses out of this module. + ever crosses out of this module. `raise _translate(error) from error` + always preserves the original as `__cause__` -- see `is_safe_stop()`. """ return RuntimeError(str(error)) + + +def is_safe_stop(error: BaseException) -> bool: + """True if `error` is this module's translation of a deliberate + `RollHandle.safe_stop()` outcome, not a genuine failure. + + `error` is normally the `RuntimeError` a caller of `scan_many()` just + caught; every translation in this module chains `from error`, so the + original `coolscanpy.SafeStopRequested` (when that is what happened) + is always sitting in `__cause__`. Imports coolscanpy lazily like the + rest of this module -- safe here because an error this module raised + already means coolscanpy was reachable. + """ + import coolscanpy + + return isinstance(getattr(error, "__cause__", None), coolscanpy.SafeStopRequested) diff --git a/negpy/infrastructure/roll/settings.py b/negpy/infrastructure/roll/settings.py new file mode 100644 index 00000000..e27799dc --- /dev/null +++ b/negpy/infrastructure/roll/settings.py @@ -0,0 +1,26 @@ +"""Persisted Roll Scanning panel settings (stored as a global setting dict).""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class RollScanSettings: + """Sticky settings for the Roll Scanning sidebar. + + Persisted via the session repo under the `roll_scan_settings` key, + mirroring `ScannerSettings` and `ScanlightSettings`. `last_device_id` + empty means no device remembered yet; per-slot spacing offsets and + approvals are not persisted here at all -- they live on the open + coolscanpy `Roll` for as long as it stays open, and `preview()` resets + them on its own the moment it re-reads the transport. + """ + + last_device_id: str = "" + output_folder: str = "" + filename_pattern: str = '{{ date }}_{{ "%03d" % seq }}' + + @classmethod + def defaults(cls) -> "RollScanSettings": + return cls() diff --git a/negpy/services/roll/__init__.py b/negpy/services/roll/__init__.py new file mode 100644 index 00000000..810cf5b8 --- /dev/null +++ b/negpy/services/roll/__init__.py @@ -0,0 +1,15 @@ +"""Roll-scanning orchestration -- drives a coolscanpy Device + Roll extension +through preview, spacing correction, approval and batch fine-scanning. + +No Qt dependencies (mirrors `negpy.services.capture` and +`negpy.services.scanning`). +""" + +from negpy.services.roll.service import RollFrameOutput, RollScanningError, RollScanningService, available + +__all__ = [ + "RollFrameOutput", + "RollScanningError", + "RollScanningService", + "available", +] diff --git a/negpy/services/scanning/roll_service.py b/negpy/services/roll/service.py similarity index 82% rename from negpy/services/scanning/roll_service.py rename to negpy/services/roll/service.py index 5855a8f8..13d64a5c 100644 --- a/negpy/services/scanning/roll_service.py +++ b/negpy/services/roll/service.py @@ -1,11 +1,13 @@ """Roll-scanning service backed by the optional `coolscanpy` package. -Mirrors `negpy.services.scanning.service.ScannerService`: one class that -orchestrates device discovery, the hardware workflow, and writing results to -disk in NegPy's conventional layout. Where `ScannerService` wraps a single +Sibling of `negpy.services.capture.service.CaptureService`: one class that +orchestrates the hardware workflow and writes results to disk in NegPy's +conventional layout, on top of an infrastructure-layer adapter for an +optional dependency (`negpy.infrastructure.roll.coolscanpy_roll`, mirroring +`negpy.infrastructure.capture.gphoto`). It also mirrors the older, simpler +`negpy.services.scanning.service.ScannerService`: where that wraps a single ad-hoc `Device.scan()`, this wraps coolscanpy's whole-roll workflow -(preview -> approve -> batch fine-scan) exposed by -`negpy.infrastructure.scanners.coolscanpy_roll`. +(preview -> approve -> batch fine-scan). Entirely inert if `coolscanpy` is not installed: `available()` is a cheap presence check re-exported from that module, and every other method only @@ -25,7 +27,7 @@ import numpy as np import tifffile -from negpy.infrastructure.scanners import coolscanpy_roll +from negpy.infrastructure.roll import coolscanpy_roll from negpy.kernel.system.logging import get_logger from negpy.services.scanning.templating import render_scan_filename @@ -34,12 +36,18 @@ logger = get_logger(__name__) -# Re-exported so callers only need this module: `from negpy.services.scanning -# .roll_service import available`, matching how the plain Scan sidebar checks +# Re-exported so callers only need this module: `from negpy.services.roll +# .service import available`, matching how the plain Scan sidebar checks # `_sane_available()` before showing its device combo. available = coolscanpy_roll.available +class RollScanningError(RuntimeError): + """Raised for a roll-scanning failure that originates in this service's + own orchestration (lifecycle misuse), as opposed to one translated from + coolscanpy itself. Mirrors `negpy.services.capture.service.CaptureError`.""" + + @dataclasses.dataclass(frozen=True) class RollFrameOutput: """Where one scanned frame's files landed on disk.""" @@ -69,7 +77,7 @@ def list_devices(self) -> "list[coolscanpy.DeviceInfo]": def open_roll(self, device_id: str | None = None, *, material: "coolscanpy.Material | None" = None) -> None: """Open a device and its roll extension. Call `close()` when done.""" if self._roll is not None: - raise RuntimeError("a roll is already open on this service; call close() first") + raise RollScanningError("a roll is already open on this service; call close() first") self._roll = coolscanpy_roll.open_roll(device_id, material=material) def close(self) -> None: @@ -86,7 +94,9 @@ def __exit__(self, *exc: object) -> None: # -- preview / approval ------------------------------------------------ - def preview(self, slots: Iterable[int] | None = None, *, on_progress=None) -> list: + def preview( + self, slots: Iterable[int] | None = None, *, on_progress: "coolscanpy.ProgressCallback | None" = None + ) -> "list[coolscanpy.Thumbnail]": return self._require_roll().preview(slots, on_progress=on_progress) def set_spacing_offset(self, slot: int, offset_rows: int) -> None: @@ -100,7 +110,9 @@ def needs_approval(self, slot: int) -> bool: # -- scanning ---------------------------------------------------------- - def scan_many(self, slots: Iterable[int], *, on_progress=None) -> Iterator["coolscanpy.Frame"]: + def scan_many( + self, slots: Iterable[int], *, on_progress: "coolscanpy.ProgressCallback | None" = None + ) -> Iterator["coolscanpy.Frame"]: yield from self._require_roll().scan_many(slots, on_progress=on_progress) def safe_stop(self) -> None: @@ -146,7 +158,7 @@ def write_frame(self, frame: "coolscanpy.Frame", output_folder: str, filename_pa def _require_roll(self) -> coolscanpy_roll.RollHandle: if self._roll is None: - raise RuntimeError("no roll is open; call open_roll() first") + raise RollScanningError("no roll is open; call open_roll() first") return self._roll diff --git a/tests/roll/__init__.py b/tests/roll/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/scanners/conftest.py b/tests/roll/conftest.py similarity index 92% rename from tests/scanners/conftest.py rename to tests/roll/conftest.py index 5890566a..3cfc79ce 100644 --- a/tests/scanners/conftest.py +++ b/tests/roll/conftest.py @@ -1,10 +1,10 @@ -"""Shared fixtures for tests/scanners/: a fake `coolscanpy` module. +"""Shared fixtures for tests/roll/: a fake `coolscanpy` module. -`coolscanpy_roll.py` / `roll_service.py` treat coolscanpy as optional and -import it lazily, so these tests never need the real package installed -- -`fake_coolscanpy` injects a minimal stand-in module into `sys.modules` -instead, exercising the same `import coolscanpy` path the production code -takes. +`negpy.infrastructure.roll.coolscanpy_roll` and `negpy.services.roll.service` +treat coolscanpy as optional and import it lazily, so these tests never need +the real package installed -- `fake_coolscanpy` injects a minimal stand-in +module into `sys.modules` instead, exercising the same `import coolscanpy` +path the production code takes. """ from __future__ import annotations diff --git a/tests/scanners/test_coolscanpy_roll.py b/tests/roll/test_coolscanpy_roll.py similarity index 99% rename from tests/scanners/test_coolscanpy_roll.py rename to tests/roll/test_coolscanpy_roll.py index 74009d7d..b90cccda 100644 --- a/tests/scanners/test_coolscanpy_roll.py +++ b/tests/roll/test_coolscanpy_roll.py @@ -13,7 +13,7 @@ import numpy as np import pytest -from negpy.infrastructure.scanners import coolscanpy_roll +from negpy.infrastructure.roll import coolscanpy_roll class TestAvailable: diff --git a/tests/scanners/test_roll_service.py b/tests/roll/test_service.py similarity index 94% rename from tests/scanners/test_roll_service.py rename to tests/roll/test_service.py index 2ac621ba..df07168b 100644 --- a/tests/scanners/test_roll_service.py +++ b/tests/roll/test_service.py @@ -1,6 +1,6 @@ """Tests for RollScanningService: lifecycle orchestration and output writing. -Lifecycle tests use `fake_coolscanpy` (see tests/scanners/conftest.py) the +Lifecycle tests use `fake_coolscanpy` (see tests/roll/conftest.py) the same way test_coolscanpy_roll.py does. write_frame() tests construct a fake Frame/Receipt directly -- writing to disk never touches coolscanpy itself, so no module injection is needed for those. @@ -15,8 +15,8 @@ import pytest import tifffile -from negpy.services.scanning import roll_service -from negpy.services.scanning.roll_service import RollScanningService +from negpy.services.roll import service as roll_service +from negpy.services.roll.service import RollScanningError, RollScanningService class TestAvailable: @@ -41,12 +41,12 @@ def test_open_then_close(self, fake_coolscanpy) -> None: def test_double_open_raises(self, fake_coolscanpy) -> None: service, _roll, _device = self._open_service(fake_coolscanpy) - with pytest.raises(RuntimeError, match="already open"): + with pytest.raises(RollScanningError, match="already open"): service.open_roll() def test_methods_before_open_raise(self) -> None: service = RollScanningService() - with pytest.raises(RuntimeError, match="no roll is open"): + with pytest.raises(RollScanningError, match="no roll is open"): service.preview() def test_close_without_open_is_a_no_op(self) -> None: