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. 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/roll/coolscanpy_roll.py b/negpy/infrastructure/roll/coolscanpy_roll.py new file mode 100644 index 00000000..a011f0f0 --- /dev/null +++ b/negpy/infrastructure/roll/coolscanpy_roll.py @@ -0,0 +1,195 @@ +"""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.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 +-------------------------------------------------------------------------- +`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/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: "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) + 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: "coolscanpy.ProgressCallback | None" = None + ) -> Iterator["coolscanpy.Frame"]: + """Batch fine-scan `slots` in one transport reservation. + + 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) + 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. `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/roll/service.py b/negpy/services/roll/service.py new file mode 100644 index 00000000..13d64a5c --- /dev/null +++ b/negpy/services/roll/service.py @@ -0,0 +1,196 @@ +"""Roll-scanning service backed by the optional `coolscanpy` package. + +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). + +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.roll 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.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.""" + + 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 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: + """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: "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: + 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: "coolscanpy.ProgressCallback | None" = 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 RollScanningError("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/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/tests/roll/__init__.py b/tests/roll/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/roll/conftest.py b/tests/roll/conftest.py new file mode 100644 index 00000000..3cfc79ce --- /dev/null +++ b/tests/roll/conftest.py @@ -0,0 +1,175 @@ +"""Shared fixtures for tests/roll/: a fake `coolscanpy` module. + +`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 + +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/roll/test_coolscanpy_roll.py b/tests/roll/test_coolscanpy_roll.py new file mode 100644 index 00000000..b90cccda --- /dev/null +++ b/tests/roll/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.roll 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 diff --git a/tests/roll/test_service.py b/tests/roll/test_service.py new file mode 100644 index 00000000..df07168b --- /dev/null +++ b/tests/roll/test_service.py @@ -0,0 +1,163 @@ +"""Tests for RollScanningService: lifecycle orchestration and output writing. + +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. +""" + +from __future__ import annotations + +import json +import os + +import numpy as np +import pytest +import tifffile + +from negpy.services.roll import service as roll_service +from negpy.services.roll.service import RollScanningError, 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(RollScanningError, match="already open"): + service.open_roll() + + def test_methods_before_open_raise(self) -> None: + service = RollScanningService() + with pytest.raises(RollScanningError, 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) 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"