diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 2b070f72..109c201a 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -3,6 +3,7 @@ ## 0.36.0 +- New: **Nikon Coolscan RGB+IR capture** — NegPy can read 1x inline RGBI frames from SANE's `coolscan3` backend, including scans made through `saned`. It recovers python-sane's misgrouped fourth channel, drops an incomplete final scanner row when necessary, and fails instead of silently saving RGB alone. This path is verified on an LS-5000 ED and requires a patched SANE backend with its infrared option enabled. - New: **Camera Scanning** — a new tab captures negatives with a tethered camera and feeds the RGB Scan merge. Two auto-selected modes: **Narrowband RGB** (jackw01's open-source Scanlight flashes red, then green, then blue while the camera captures each exposure) and **normal** (a single exposure under any light, imported as an ordinary RAW). Includes per-channel ETTR auto-calibration metered on the clear film base, film-stock presets, and a live view with the camera's hardware focus magnifier and live ISO/shutter controls. Cameras are detected on the USB bus automatically — no address, login or pairing. macOS and Linux only (verified on Sony bodies, other brands need testing). @light-sntchr - New: **Split grade** — **Shadows Grade / Highlights Grade** sliders on the Tone page trim each zone's contrast in ISO-R points, like a split-grade darkroom exposure: harder shadows without blowing the highlights, or softer highlights without flattening the shadows. Mid-sparing and bounded by the paper's black and white, like the zone density sliders, and scoped per colour layer through the Global/R/G/B selector like the main Grade. - New: **Per-layer trims (crossover correction)** — a **Global / Red / Green / Blue** selector on the Tone and Process pages scopes the curve controls (Grade, Toe, Shoulder, Widths, Snap) and the White/Black Point sliders to a single colour layer. Filtration can only *shift* a layer; these trims re-shape it, fixing casts that differ between shadows, mids and highlights. The H&D chart tracks the per-layer curves live. diff --git a/negpy/infrastructure/scanners/sane_backend.py b/negpy/infrastructure/scanners/sane_backend.py index e34745ec..c6906eb9 100644 --- a/negpy/infrastructure/scanners/sane_backend.py +++ b/negpy/infrastructure/scanners/sane_backend.py @@ -29,11 +29,29 @@ # SANE option py_names (underscored) that expose a dedicated infrared channel/scan. _IR_OPTION_NAMES = ("ir", "preview_ir") +_COOLSCAN3_IR_OPTION_NAMES = ("infrared",) _PIEUSB_PREFIX = "pieusb:" +_COOLSCAN3_PREFIX = "coolscan3:" # Backends for dedicated film scanners that expose no `source` option. -_FILM_BACKEND_PREFIXES = (_PIEUSB_PREFIX,) +_FILM_BACKEND_PREFIXES = (_PIEUSB_PREFIX, _COOLSCAN3_PREFIX) + + +def _strip_net_prefix(device_id: str) -> str: + """Drop a leading `net::` so backend-prefix checks work over saned. + + Handles both `net:scanner.example:coolscan3:...` and bracketed IPv6 hosts + (`net:[2001:db8::1]:coolscan3:...`). + """ + if not device_id.startswith("net:"): + return device_id + rest = device_id[len("net:") :] + if rest.startswith("["): # bracketed IPv6 host + close = rest.find("]:") + return rest[close + 2 :] if close > 0 else device_id + host, sep, backend_part = rest.partition(":") + return backend_part if sep and host else device_id def _mode_has_rgbi(opt) -> bool: @@ -59,7 +77,7 @@ def _infer_film_scanner(opt, device_id: str) -> bool: desc = str(getattr(invert, "desc", "") or "").lower() if "negative" in desc and "film" in desc: return True - return device_id.startswith(_FILM_BACKEND_PREFIXES) + return _strip_net_prefix(device_id).startswith(_FILM_BACKEND_PREFIXES) def _resolve_install_hint() -> str: @@ -147,10 +165,71 @@ def _upper(name: str) -> float: return (36.0, 25.0) # default 35mm frame -def _detect_ir(opt) -> bool: +def _find_ir_option(opt) -> str | None: + """Return a legacy IR option name without changing its old semantics.""" + for key in opt: + if str(key).lower().replace("-", "_").strip("_") in _IR_OPTION_NAMES: + return str(key) + return None + + +def _find_coolscan3_ir_option(opt) -> str | None: + """Return coolscan3's inline-IR boolean option, if present.""" + for key in opt: + if str(key).lower().replace("-", "_").strip("_") in _COOLSCAN3_IR_OPTION_NAMES: + return str(key) + return None + + +def _option_is_usable(opt, option_name: str) -> bool: + """Return whether an option can be selected without failing device probe.""" + if option_name not in opt: + return False + option = opt[option_name] + for method_name in ("is_active", "is_settable"): + method = getattr(option, method_name, None) + if not callable(method): + continue + try: + if not bool(method()): + return False + except Exception: + return False + return True + + +def _require_active_option(opt, option_name: str, absent_message: str) -> None: + """Fail before scanner configuration when a requested option is unusable.""" + if option_name not in opt: + raise RuntimeError(absent_message) + option = opt[option_name] + is_active = getattr(option, "is_active", None) + if callable(is_active): + try: + active = bool(is_active()) + except Exception as e: + raise RuntimeError(f"Could not determine whether SANE option {option_name!r} is active: {e}") from e + if not active: + raise RuntimeError(f"Requested SANE option {option_name!r} is inactive") + is_settable = getattr(option, "is_settable", None) + if callable(is_settable): + try: + settable = bool(is_settable()) + except Exception as e: + raise RuntimeError(f"Could not determine whether SANE option {option_name!r} is settable: {e}") from e + if not settable: + raise RuntimeError(f"Requested SANE option {option_name!r} is not settable") + + +def _detect_ir(opt, device_id: str = "") -> bool: if _mode_has_rgbi(opt): return True - return any(str(key).lower().replace("-", "_").strip("_") in _IR_OPTION_NAMES for key in opt) + if _find_ir_option(opt) is not None: + return True + if _strip_net_prefix(device_id).startswith(_COOLSCAN3_PREFIX): + option_name = _find_coolscan3_ir_option(opt) + return option_name is not None and _option_is_usable(opt, option_name) + return False def _caps_from_options(opt, device_id: str = "") -> ScannerCapabilities: @@ -161,7 +240,7 @@ def _caps_from_options(opt, device_id: str = "") -> ScannerCapabilities: # detection/UI gate — never applied to the device — so populate it to unblock scanning. sources = (ScanMode.NEGATIVE, ScanMode.POSITIVE, ScanMode.TRANSPARENCY) return ScannerCapabilities( - ir_channel=_detect_ir(opt), + ir_channel=_detect_ir(opt, device_id), supported_dpi=_detect_dpi(opt), supported_depths=_detect_depths(opt), sources=sources, @@ -174,6 +253,62 @@ def _split_rgbi(arr: np.ndarray) -> tuple[np.ndarray, np.ndarray]: return arr[:, :, :3], arr[:, :, 3] +def _reinterpret_channels(arr: np.ndarray, width: int, lines: int) -> np.ndarray: + """Recover the true channel count of a frame python-sane misread. + + python-sane's C reader hardcodes 3 samples/pixel for every non-gray frame + and reads the stream in `3 * width`-sample chunks, so a 4-sample + inline-RGBI stream (pieusb/coolscan3 convention: SANE_FRAME_RGB with + bytes_per_line = 4 x pixels_per_line x sample size) arrives misshaped. + Worse, a partial final chunk is *discarded* at EOF: when `4 * lines` is + not a multiple of 3, the stream's trailing `(4 * lines mod 3) * width` + samples are lost (both 1489- and 5959-line LS-5000 frames hit this). + The loss is confined to the last row, so return the complete prefix as a + view and drop that incomplete edge row rather than copy the full frame. + """ + if width <= 0 or lines <= 0: + return arr + total = int(arr.size) + expected = 4 * width * lines + missing = expected - total + if missing in (width, 2 * width): + flat = arr.reshape(-1) + complete = (lines - 1) * width * 4 + return flat[:complete].reshape(lines - 1, width, 4) + if total % (width * lines): + return arr + nch = total // (width * lines) + if nch not in (1, 3, 4): + return arr + if arr.ndim == 3 and arr.shape == (lines, width, nch): + return arr + return arr.reshape(lines, width, nch) + + +def _validate_coolscan3_rgbi_parameters( + frame_format, + last_frame, + width: int, + lines: int, + depth: int, + bytes_per_line: int, + requested_depth: int, +) -> None: + """Validate the four-sample coolscan3 contract before reshaping data.""" + if not bool(last_frame): + raise RuntimeError("Inline RGBI scan did not report a last frame") + if str(frame_format).strip().lower() != "color": + raise RuntimeError(f"Inline RGBI scan reported unexpected frame format {frame_format!r}") + if width <= 0 or lines <= 0: + raise RuntimeError(f"Inline RGBI scan reported invalid geometry {width}x{lines}") + if int(depth) != requested_depth: + raise RuntimeError(f"Inline RGBI scan reported depth {depth}, expected {requested_depth}") + sample_bytes = (int(depth) + 7) // 8 + expected_bpl = width * 4 * sample_bytes + if int(bytes_per_line) != expected_bpl: + raise RuntimeError(f"Inline RGBI scan reported {bytes_per_line} bytes per line, expected {expected_bpl}") + + class SaneBackend: """python-sane implementation of ScannerBackend. Only module that imports `sane`.""" @@ -249,12 +384,37 @@ def scan( try: # IR capture strategy decides the scan mode (RGBI yields a 4th channel inline). ir_strategy = self._ir_strategy(dev, device_id) if params.capture_ir else None - - # Configure SANE parameters - dev.mode = "RGBI" if ir_strategy == "rgbi" else "Color" + backend_id = _strip_net_prefix(device_id) + + # Coolscan3 IR is an explicit, fail-loud contract. Other backends + # keep their existing fallback behavior when no strategy is found. + option_map = dev.opt if hasattr(dev, "opt") else {} + if params.capture_ir and ir_strategy is None and backend_id.startswith(_COOLSCAN3_PREFIX): + raise RuntimeError("IR capture requested but the device exposes no usable infrared mechanism") + ir_opt = _find_coolscan3_ir_option(option_map) if ir_strategy == "option" else None + if ir_strategy == "option": + if ir_opt is None: + raise RuntimeError("IR capture requested but the device exposes no usable infrared option") + _require_active_option(option_map, ir_opt, "IR capture requested but the infrared option is unavailable") + + # Configure SANE parameters. Not every backend has a `mode` option + # (coolscan3 exposes none) — only touch options the device has. + if hasattr(dev, "opt") and "mode" in dev.opt: + dev.mode = "RGBI" if ir_strategy == "rgbi" else "Color" dev.depth = params.depth dev.resolution = params.dpi + # Inline IR via a boolean option (coolscan3 `infrared`): the 4th + # sample rides in the same frame, no mode/source change involved. + # IR was explicitly requested — fail loud rather than silently + # returning a scan without the channel. + if ir_strategy == "option": + assert ir_opt is not None # established by the preflight above + try: + setattr(dev, ir_opt, True) + except Exception as e: + raise RuntimeError(f"IR capture requested but enabling option {ir_opt!r} failed: {e}") from e + # Apply hardware-specific optimizations if device_id.startswith(_PIEUSB_PREFIX): self._set_pieusb_flags(dev, params.capture_ir) @@ -287,6 +447,30 @@ def scan( rgb_array = None ir_array = None + # Frame geometry truth, for channel reinterpretation below + # (python-sane assumes 3 samples/pixel; see _reinterpret_channels). + try: + frame_format, last_frame, (px_per_line, n_lines), frame_depth, bytes_per_line = dev.get_parameters() + except Exception as e: + if ir_strategy == "option": + dev.cancel() + raise RuntimeError(f"Could not read inline RGBI frame parameters: {e}") from e + px_per_line = n_lines = -1 + if ir_strategy == "option": + try: + _validate_coolscan3_rgbi_parameters( + frame_format, + last_frame, + px_per_line, + n_lines, + frame_depth, + bytes_per_line, + params.depth, + ) + except RuntimeError: + dev.cancel() + raise + # Read RGB frame. Use arr_snap() (numpy path) — snap() goes via PIL # which is 8-bit only and silently truncates 16-bit RGB buffers. try: @@ -295,8 +479,19 @@ def scan( dev.cancel() raise RuntimeError(f"RGB scan failed: {e}") from e - # RGBI mode returns infrared as the 4th channel inline — split it off. - if ir_strategy == "rgbi" and rgb_array.ndim == 3 and rgb_array.shape[2] == 4: + # coolscan3 inline IR carries a 4th sample in an RGB frame, which + # python-sane misgroups. Keep other RGBI-mode backends on their + # pre-existing path; their row and frame contracts may differ. + if ir_strategy == "option": + rgb_array = _reinterpret_channels(rgb_array, px_per_line, n_lines) + if rgb_array.ndim == 3 and rgb_array.shape[2] == 4: + rgb_array, ir_array = _split_rgbi(rgb_array) + else: + # IR was explicitly requested; a long archival scan must + # not quietly complete without its defect channel. + dev.cancel() + raise RuntimeError(f"IR strategy '{ir_strategy}' yielded no 4th channel (shape={rgb_array.shape})") + elif ir_strategy == "rgbi" and rgb_array.ndim == 3 and rgb_array.shape[2] == 4: rgb_array, ir_array = _split_rgbi(rgb_array) expected_dtype = np.uint16 if params.depth == 16 else np.uint8 @@ -377,13 +572,20 @@ def _detect_caps(self, dev, device_id: str = "") -> ScannerCapabilities: @staticmethod def _ir_strategy(dev, device_id) -> str | None: - """How to capture IR for this device: 'rgbi' (4th channel), 'source' (Plustek + """How to capture IR for this device: 'rgbi' (RGBI scan mode), 'option' + (boolean IR option, 4th channel inline — coolscan3), 'source' (Plustek second scan), 'internal' (applied by the Backend/Scanner itself) or None.""" opt = dev.opt if hasattr(dev, "opt") else {} + backend_id = _strip_net_prefix(device_id) if device_id.startswith(_PIEUSB_PREFIX): return "internal" if _mode_has_rgbi(opt): return "rgbi" + # Inline-boolean IR is a coolscan3 contract (4th sample in the same + # frame). coolscan2 exposes the same option name but delivers IR as a + # separate later frame — do not claim it here. + if backend_id.startswith(_COOLSCAN3_PREFIX) and _find_coolscan3_ir_option(opt) is not None: + return "option" if SaneBackend._get_ir_source(dev): return "source" return None diff --git a/tests/scanners/test_coolscan_ir.py b/tests/scanners/test_coolscan_ir.py new file mode 100644 index 00000000..0e347e9f --- /dev/null +++ b/tests/scanners/test_coolscan_ir.py @@ -0,0 +1,479 @@ +"""Tests for coolscan3-style inline IR: option strategy, net IDs, channel repair. + +The tested Nikon Coolscan LS-5000 exposes IR as a boolean `infrared` +option; the frame then carries 4 samples/pixel while reporting +SANE_FRAME_RGB (pieusb convention). python-sane's C reader hardcodes +3 samples/pixel for non-gray frames, so the array arrives byte-intact but +misshaped — `_reinterpret_channels` recovers it from the frame geometry. +""" + +import threading +from dataclasses import dataclass, field +from typing import Any + +import numpy as np +import pytest + +from negpy.infrastructure.scanners.params import ScanParams +from negpy.infrastructure.scanners.sane_backend import ( + SaneBackend, + _caps_from_options, + _detect_ir, + _find_coolscan3_ir_option, + _find_ir_option, + _reinterpret_channels, + _strip_net_prefix, +) + + +@dataclass +class FakeOption: + """Stand-in for python-sane's Option (only the fields the module reads).""" + + constraint: Any = None + desc: str = "" + active: bool = True + settable: bool = True + + def is_active(self) -> bool: + return self.active + + def is_settable(self) -> bool: + return self.settable + + +class SettableOnlyOption: + """Option shim for bindings that expose writability but not activity.""" + + constraint = None + desc = "" + + def is_settable(self) -> bool: + return False + + +COOLSCAN3_OPT = { + "infrared": FakeOption(), + "depth": FakeOption(constraint=[8, 16]), + "resolution": FakeOption(constraint=[4000, 2000, 1000]), + "frame": FakeOption(constraint=(1, 40, 1)), + # NB: no "mode", no "source" — like the real backend. +} + + +class TestNetPrefix: + def test_strips_saned_prefix(self) -> None: + assert _strip_net_prefix("net:scanner.example:coolscan3:usb:libusb:001:007") == "coolscan3:usb:libusb:001:007" + + def test_plain_id_unchanged(self) -> None: + assert _strip_net_prefix("coolscan3:usb:libusb:001:007") == "coolscan3:usb:libusb:001:007" + + def test_pieusb_over_net(self) -> None: + assert _strip_net_prefix("net:host:pieusb:libusb:001:004") == "pieusb:libusb:001:004" + + def test_bracketed_ipv6_host(self) -> None: + assert _strip_net_prefix("net:[2001:db8::1]:coolscan3:usb:test") == "coolscan3:usb:test" + + +class TestIrOptionDetection: + def test_finds_coolscan3_infrared(self) -> None: + assert _find_coolscan3_ir_option(COOLSCAN3_OPT) == "infrared" + + def test_legacy_finder_does_not_claim_coolscan3_infrared(self) -> None: + assert _find_ir_option(COOLSCAN3_OPT) is None + + def test_detect_ir_true(self) -> None: + assert _detect_ir(COOLSCAN3_OPT, "coolscan3:usb:test") is True + + def test_inactive_ir_option_is_not_advertised(self) -> None: + assert _detect_ir({"infrared": FakeOption(active=False)}, "coolscan3:usb:test") is False + + def test_read_only_ir_option_is_not_advertised(self) -> None: + assert _detect_ir({"infrared": FakeOption(settable=False)}, "coolscan3:usb:test") is False + + def test_coolscan2_infrared_option_is_not_advertised(self) -> None: + assert _detect_ir({"infrared": FakeOption()}, "coolscan2:usb:test") is False + + def test_legacy_ir_option_keeps_presence_based_capability(self) -> None: + assert _detect_ir({"ir": FakeOption(active=False)}, "legacy:usb:test") is True + + def test_no_ir_on_flatbed(self) -> None: + assert _find_ir_option({"mode": FakeOption(constraint=["Color", "Gray"])}) is None + + +class TestCoolscanCapabilities: + def test_film_scanner_inferred_without_source(self) -> None: + caps = _caps_from_options(COOLSCAN3_OPT, "coolscan3:usb:libusb:001:007") + assert caps.ir_channel is True + assert caps.sources # inferred, not skipped + assert caps.supported_depths == (8, 16) + + def test_film_scanner_inferred_over_net(self) -> None: + caps = _caps_from_options(COOLSCAN3_OPT, "net:scanner.example:coolscan3:usb:libusb:001:007") + assert caps.sources + + def test_unknown_backend_ir_option_does_not_imply_film_scanner(self) -> None: + caps = _caps_from_options({"infrared": FakeOption()}, "mystery:001") + assert caps.sources == () + assert caps.ir_channel is False + + def test_inactive_ir_still_identifies_film_scanner_without_advertising_ir(self) -> None: + caps = _caps_from_options({"infrared": FakeOption(active=False)}, "coolscan3:usb:test") + assert caps.sources + assert caps.ir_channel is False + + def test_coolscan2_does_not_advertise_unsupported_inline_ir(self) -> None: + caps = _caps_from_options({"infrared": FakeOption()}, "coolscan2:usb:test") + assert caps.ir_channel is False + + +def _emulate_python_sane_read(true_frame: np.ndarray) -> np.ndarray: + """Reproduce python-sane's C reader on a 4-sample frame. + + It assumes 3 samples/pixel, reads `3 * width`-sample chunks, and + DISCARDS a partial final chunk at EOF (_sane.c snap loop) — so when + `4 * lines` is not a multiple of 3, trailing samples are lost. + """ + h, w, c = true_frame.shape + flat = true_frame.reshape(-1) + chunk = 3 * w + n_full = flat.size // chunk + return flat[: n_full * chunk].reshape(n_full, w, 3) + + +class TestReinterpretChannels: + def _rgbi(self, h: int, w: int) -> np.ndarray: + rng = np.random.default_rng(42) + return rng.integers(0, 65535, size=(h, w, 4), dtype=np.uint16) + + def test_recovers_misread_rgbi_divisible(self) -> None: + h, w = 6, 5 # 4h % 3 == 0: no truncation, full recovery + true = self._rgbi(h, w) + fixed = _reinterpret_channels(_emulate_python_sane_read(true), w, h) + assert fixed.shape == (h, w, 4) + assert np.array_equal(fixed, true) + + def test_recovers_truncated_rgbi_mod1(self) -> None: + h, w = 7, 5 # 4h % 3 == 1 — the real LS-5000 case (1489, 5959 lines) + true = self._rgbi(h, w) + encoded = _emulate_python_sane_read(true) + fixed = _reinterpret_channels(encoded, w, h) + assert fixed.shape == (h - 1, w, 4) # padded edge row dropped + assert np.array_equal(fixed, true[: h - 1]) + assert np.shares_memory(fixed, encoded) + + def test_recovers_truncated_rgbi_mod2(self) -> None: + h, w = 5, 4 # 4h % 3 == 2 + true = self._rgbi(h, w) + fixed = _reinterpret_channels(_emulate_python_sane_read(true), w, h) + assert fixed.shape == (h - 1, w, 4) + assert np.array_equal(fixed, true[: h - 1]) + + def test_correct_rgb_untouched(self) -> None: + arr = np.zeros((10, 20, 3), dtype=np.uint16) + assert _reinterpret_channels(arr, 20, 10) is arr + + def test_unknown_geometry_untouched(self) -> None: + arr = np.zeros((8, 5, 3), dtype=np.uint16) + assert _reinterpret_channels(arr, -1, -1) is arr + + def test_indivisible_untouched(self) -> None: + arr = np.zeros((7, 5, 3), dtype=np.uint16) + assert _reinterpret_channels(arr, 5, 6) is arr + + +class FakeSaneDev: + """Mimics python-sane SaneDev for a coolscan3-like device. + + Setting an attribute that is not an internal field and not a known SANE + option raises AttributeError, like python-sane does. arr_snap() returns + the misshaped array the real C module produces for 4-sample frames. + """ + + _INTERNAL = ( + "recorded", + "true_frame", + "cancelled", + "closed", + "opt_map", + "started", + "parameter_format", + "parameter_last", + "parameter_depth", + "parameter_bpl", + ) + + def __init__( + self, + true_frame: np.ndarray, + opt_map: dict | None = None, + *, + parameter_format: str = "color", + parameter_last: bool = True, + parameter_depth: int = 16, + parameter_bpl: int | None = None, + ) -> None: + object.__setattr__(self, "opt_map", COOLSCAN3_OPT if opt_map is None else opt_map) + object.__setattr__(self, "recorded", {}) + object.__setattr__(self, "true_frame", true_frame) + object.__setattr__(self, "cancelled", False) + object.__setattr__(self, "closed", False) + object.__setattr__(self, "started", False) + object.__setattr__(self, "parameter_format", parameter_format) + object.__setattr__(self, "parameter_last", parameter_last) + object.__setattr__(self, "parameter_depth", parameter_depth) + object.__setattr__(self, "parameter_bpl", parameter_bpl) + + @property + def opt(self): + return self.opt_map + + def __setattr__(self, name: str, value: Any) -> None: + if name in self._INTERNAL: + object.__setattr__(self, name, value) + return + if name not in self.opt_map: + raise AttributeError(f"No such SANE option: {name}") + self.recorded[name] = value + + def start(self) -> None: + object.__setattr__(self, "started", True) + + def get_parameters(self): + h, w, _ = self.true_frame.shape + bytes_per_line = self.parameter_bpl + if bytes_per_line is None: + bytes_per_line = w * 4 * ((self.parameter_depth + 7) // 8) + return (self.parameter_format, self.parameter_last, (w, h), self.parameter_depth, bytes_per_line) + + def arr_snap(self) -> np.ndarray: + if self.true_frame.shape[2] == 3: + return self.true_frame # 3-channel frames come back correctly + return _emulate_python_sane_read(self.true_frame) + + def cancel(self) -> None: + object.__setattr__(self, "cancelled", True) + + def close(self) -> None: + object.__setattr__(self, "closed", True) + + +@dataclass +class FakeSaneModule: + dev: FakeSaneDev + opened: list = field(default_factory=list) + + def open(self, device_id: str) -> FakeSaneDev: + self.opened.append(device_id) + return self.dev + + +def _make_backend(dev: FakeSaneDev) -> SaneBackend: + backend = SaneBackend.__new__(SaneBackend) + backend._sane = FakeSaneModule(dev) + backend._sane_initialized = True + backend._devices_cache = None + return backend + + +class TestScanWithOptionStrategy: + def _run(self, device_id: str, h: int = 6) -> tuple: + rng = np.random.default_rng(7) + true = rng.integers(0, 65535, size=(h, 5, 4), dtype=np.uint16) + dev = FakeSaneDev(true) + backend = _make_backend(dev) + params = ScanParams(dpi=1000, depth=16, capture_ir=True) + result = backend.scan(device_id, params, None, threading.Event()) + return true, dev, result + + def test_ir_split_and_shape_repair(self) -> None: + true, dev, result = self._run("coolscan3:usb:libusb:001:007") + assert result.rgb.shape == (6, 5, 3) + assert result.ir is not None and result.ir.shape == (6, 5) + assert np.array_equal(result.rgb, true[:, :, :3]) + assert np.array_equal(result.ir, true[:, :, 3]) + + def test_ir_split_with_python_sane_truncation(self) -> None: + # 4h % 3 == 1: python-sane drops the stream tail (real LS-5000 case); + # scan() must still deliver IR, minus the one padded edge row. + true, dev, result = self._run("coolscan3:usb:libusb:001:007", h=7) + assert result.rgb.shape == (6, 5, 3) + assert result.ir is not None and result.ir.shape == (6, 5) + assert np.array_equal(result.rgb, true[:6, :, :3]) + assert np.array_equal(result.ir, true[:6, :, 3]) + + def test_infrared_option_enabled_and_mode_untouched(self) -> None: + _, dev, _ = self._run("coolscan3:usb:libusb:001:007") + assert dev.recorded.get("infrared") is True + assert "mode" not in dev.recorded # device has no mode option; must not be set + + def test_works_over_net_device_id(self) -> None: + _, dev, result = self._run("net:scanner.example:coolscan3:usb:libusb:001:007") + assert result.ir is not None + assert dev.recorded.get("infrared") is True + + def test_no_ir_requested_scans_plain(self) -> None: + rng = np.random.default_rng(9) + true = rng.integers(0, 65535, size=(6, 5, 3), dtype=np.uint16) + dev = FakeSaneDev(true) + backend = _make_backend(dev) + params = ScanParams(dpi=1000, depth=16, capture_ir=False) + result = backend.scan("coolscan3:usb:libusb:001:007", params, None, threading.Event()) + assert "infrared" not in dev.recorded + assert result.ir is None + assert result.rgb.shape == (6, 5, 3) + assert np.array_equal(result.rgb, true) + + def test_inactive_infrared_option_fails_before_scan_start(self) -> None: + rng = np.random.default_rng(10) + opt = dict(COOLSCAN3_OPT) + opt["infrared"] = FakeOption(active=False) + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16), opt_map=opt) + + with pytest.raises(RuntimeError, match="infrared.*inactive"): + self._scan_device(dev, ScanParams(dpi=1000, depth=16, capture_ir=True)) + + assert dev.started is False + assert "infrared" not in dev.recorded + + def test_read_only_infrared_option_fails_before_scanner_configuration(self) -> None: + rng = np.random.default_rng(100) + opt = dict(COOLSCAN3_OPT) + opt["infrared"] = FakeOption(settable=False) + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16), opt_map=opt) + + with pytest.raises(RuntimeError, match="infrared.*not settable"): + self._scan_device(dev, ScanParams(dpi=1000, depth=16, capture_ir=True)) + + assert dev.started is False + assert dev.recorded == {} + + def test_settable_only_read_only_option_still_fails_preflight(self) -> None: + rng = np.random.default_rng(1001) + opt = dict(COOLSCAN3_OPT) + opt["infrared"] = SettableOnlyOption() + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16), opt_map=opt) + + with pytest.raises(RuntimeError, match="infrared.*not settable"): + self._scan_device(dev, ScanParams(dpi=1000, depth=16, capture_ir=True)) + + assert dev.started is False + assert dev.recorded == {} + + @pytest.mark.parametrize( + ("device_kwargs", "message"), + ( + ({"parameter_last": False}, "last frame"), + ({"parameter_format": "gray"}, "frame format"), + ({"parameter_depth": 8}, "depth"), + ({"parameter_bpl": 5 * 4 * 2 + 2}, "bytes per line"), + ), + ) + def test_invalid_inline_rgbi_parameters_fail_loud(self, device_kwargs: dict, message: str) -> None: + rng = np.random.default_rng(1002) + dev = FakeSaneDev( + rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16), + **device_kwargs, + ) + + with pytest.raises(RuntimeError, match=message): + self._scan_device(dev, ScanParams(dpi=1000, depth=16, capture_ir=True)) + + assert dev.cancelled + + def test_requested_ir_without_a_usable_mechanism_fails_before_start(self) -> None: + rng = np.random.default_rng(101) + opt = {name: value for name, value in COOLSCAN3_OPT.items() if name != "infrared"} + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 3), dtype=np.uint16), opt_map=opt) + + with pytest.raises(RuntimeError, match="no usable infrared mechanism"): + self._scan_device(dev, ScanParams(dpi=1000, depth=16, capture_ir=True)) + + assert dev.started is False + + @staticmethod + def _scan_device(dev: FakeSaneDev, params: ScanParams): + backend = _make_backend(dev) + return backend.scan("coolscan3:usb:libusb:001:007", params, None, threading.Event()) + + def test_option_strategy_not_claimed_for_coolscan2(self) -> None: + # coolscan2 exposes the same option name but delivers IR as a separate + # later frame — the inline-option contract must not be applied to it. + rng = np.random.default_rng(11) + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16)) + assert SaneBackend._ir_strategy(dev, "coolscan3:usb:libusb:001:007") == "option" + assert SaneBackend._ir_strategy(dev, "coolscan2:usb:libusb:001:007") is None + assert SaneBackend._ir_strategy(dev, "net:scanner.example:coolscan2:usb:x") is None + + def test_net_pieusb_keeps_preexisting_rgbi_strategy(self) -> None: + rng = np.random.default_rng(12) + opt = {"mode": FakeOption(constraint=["Color", "RGBI"])} + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 4), dtype=np.uint16), opt_map=opt) + assert SaneBackend._ir_strategy(dev, "pieusb:libusb:001:004") == "internal" + assert SaneBackend._ir_strategy(dev, "net:scanner.example:pieusb:libusb:001:004") == "rgbi" + + def test_generic_rgbi_mode_keeps_preexisting_rgb_fallback(self) -> None: + rng = np.random.default_rng(13) + true = rng.integers(0, 65535, size=(6, 5, 3), dtype=np.uint16) + opt = { + "mode": FakeOption(constraint=["Color", "RGBI"]), + "depth": FakeOption(constraint=[8, 16]), + "resolution": FakeOption(constraint=[1000]), + } + dev = FakeSaneDev(true, opt_map=opt) + backend = _make_backend(dev) + + result = backend.scan( + "generic:usb:test", + ScanParams(dpi=1000, depth=16, capture_ir=True), + None, + threading.Event(), + ) + + assert np.array_equal(result.rgb, true) + assert result.ir is None + + @pytest.mark.parametrize( + ("device_id", "opt"), + ( + ("legacy:usb:test", {"ir": FakeOption()}), + ("coolscan2:usb:test", {"infrared": FakeOption()}), + ), + ) + def test_non_coolscan3_option_keeps_preexisting_rgb_fallback(self, device_id: str, opt: dict) -> None: + rng = np.random.default_rng(14) + true = rng.integers(0, 65535, size=(6, 5, 3), dtype=np.uint16) + opt.update( + { + "depth": FakeOption(constraint=[8, 16]), + "resolution": FakeOption(constraint=[1000]), + } + ) + dev = FakeSaneDev(true, opt_map=opt) + backend = _make_backend(dev) + + result = backend.scan( + device_id, + ScanParams(dpi=1000, depth=16, capture_ir=True), + None, + threading.Event(), + ) + + assert np.array_equal(result.rgb, true) + assert result.ir is None + + +def test_requested_ir_without_channel_raises() -> None: + rng = np.random.default_rng(17) + # Device frame is plain 3-channel: inline-IR strategy cannot deliver + # a 4th channel and the scan must fail loud, not return ir=None. + dev = FakeSaneDev(rng.integers(0, 65535, size=(6, 5, 3), dtype=np.uint16)) + backend = _make_backend(dev) + with pytest.raises(RuntimeError, match="no 4th channel"): + backend.scan( + "coolscan3:usb:libusb:001:007", + ScanParams(dpi=1000, depth=16, capture_ir=True), + None, + threading.Event(), + ) + assert dev.cancelled