diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 44ff7ed9..34308b51 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -466,11 +466,20 @@ When you set capture gear, it's written to standard EXIF and the digitizing rig ## 12. Scan tab -Capture film directly into NegPy (Linux and macOS; unavailable on Windows). Two collapsible sections: +Capture film directly into NegPy. Two collapsible sections: -* **Scanner (SANE)**: drive a supported flatbed/film scanner over SANE. +* **Scanner**: drive a scanner over one of two backends, picked from the **Backend** dropdown: + * **SANE**: any supported flatbed/film scanner via SANE. Linux and macOS only. + * **nkscan**: a Nikon Coolscan directly over USB or SCSI (FireWire). Linux, macOS and Windows (macOS is USB-only for now). Needs the optional `nkscan` dependency (`uv sync --group nkscan`). * **Camera Scanning**: DSLR/mirrorless copy-stand capture. Auto-connects the camera over USB (PC-Remote mode). With a NegPy **Scanlight** connected it captures narrowband R/G/B triplets from saved film-stock presets; without one it does a single white-light exposure. A **Live View** window helps you frame and focus; captured frames land in the hot folder and flow straight into RGB-Scan mode. +The Scanner panel's controls are mostly backend-agnostic (DPI, bit depth, IR, autofocus, auto-exposure, frame range), but a few only appear for devices that support them: + +* **Quality** (multisample / single-line): nkscan-specific. Multisample averages multiple passes to trade scan time for lower noise; single-line reads the sensor one line at a time, which is slower but can reduce banding. Hidden on devices that report neither. +* **Lock white balance** / **Lock exposure across batch**: nkscan-specific, shown once a frame range exists, and independent of each other. **Lock white balance** holds white balance during every frame's own metering, so the film's cast isn't scanned out. **Lock exposure across batch** meters only the first frame, then holds that exact exposure for every remaining frame. Combine both for roll analysis (consistent color and brightness across the whole strip); either can be used alone. Both off by default. + +A batch scan holds the scanner open for the whole frame range rather than reopening per frame, regardless of backend. + Camera scanning needs the optional `python-gphoto2` dependency (`pip install gphoto2`; no Windows build). See [CAMERA_SCANNING.md](CAMERA_SCANNING.md). --- diff --git a/negpy/desktop/view/sidebar/right_panel.py b/negpy/desktop/view/sidebar/right_panel.py index 6492c894..db81dcff 100644 --- a/negpy/desktop/view/sidebar/right_panel.py +++ b/negpy/desktop/view/sidebar/right_panel.py @@ -1,4 +1,3 @@ -import sys from typing import Any, Dict import qtawesome as qta @@ -94,12 +93,12 @@ def wrap_scroll(widget: QWidget) -> QScrollArea: self.metadata_sidebar = MetadataSidebar(self.controller) self.history_panel = HistoryPanel(self.controller) - from negpy.desktop.view.sidebar.scan import ScanSidebar, _ScanUnsupportedPlaceholder + from negpy.desktop.view.sidebar.scan import ScanSidebar - if sys.platform == "win32": - self.scan_sidebar = _ScanUnsupportedPlaceholder() - else: - self.scan_sidebar = ScanSidebar(self.controller) + # nkscan supports Windows for some devices, so always build the real sidebar; + # a backend that isn't available on this platform surfaces through + # ScannerUnavailable's install hint instead. + self.scan_sidebar = ScanSidebar(self.controller) from negpy.desktop.view.sidebar.scanlight import ScanlightSidebar diff --git a/negpy/desktop/view/sidebar/scan.py b/negpy/desktop/view/sidebar/scan.py index 4d533d80..46781b8f 100644 --- a/negpy/desktop/view/sidebar/scan.py +++ b/negpy/desktop/view/sidebar/scan.py @@ -1,5 +1,5 @@ import qtawesome as qta -from PyQt6.QtCore import Qt, pyqtSlot +from PyQt6.QtCore import pyqtSlot from PyQt6.QtWidgets import ( QCheckBox, QComboBox, @@ -128,6 +128,23 @@ def _init_ui(self) -> None: depth_row.addWidget(self.ir_check) self.form.addRow("Depth", depth_row) + # Multisample averaging / single-line CCD (nkscan-specific — hidden entirely + # on devices that don't report either capability). + quality_row = QHBoxLayout() + quality_row.setContentsMargins(0, 0, 0, 0) + self.multisample_combo = QComboBox() + self.multisample_combo.setToolTip("Multisample averaging (trades scan time for noise)") + self.single_line_check = QCheckBox("Single-line") + self.single_line_check.setToolTip("Single-line CCD mode (slower, may reduce banding)") + quality_row.addWidget(self.multisample_combo, 1) + quality_row.addWidget(self.single_line_check) + self.quality_row_widget = QWidget() + self.quality_row_widget.setLayout(quality_row) + self.quality_row_label = QLabel("Quality") + self.form.addRow(self.quality_row_label, self.quality_row_widget) + self.quality_row_label.setVisible(False) + self.quality_row_widget.setVisible(False) + # Spanning rows (no label column) so the checkboxes sit at the left edge. self.autofocus_check = QCheckBox("Autofocus") self.autofocus_check.setChecked(True) @@ -138,6 +155,19 @@ def _init_ui(self) -> None: self.ae_check.setToolTip("Meter exposure in hardware before the scan") self.form.addRow(self.ae_check) + # nkscan-specific, independent controls — only meaningful once a frame + # range exists. Lock WB affects every frame's own metering; lock exposure + # freezes the numeric gain frame 1 settles on for the rest of the batch. + self.lock_white_balance_check = QCheckBox("Lock white balance") + self.lock_white_balance_check.setToolTip("Hold white balance during metering, so film keeps its cast instead of scanning neutral") + self.form.addRow(self.lock_white_balance_check) + self.lock_white_balance_check.setVisible(False) + + self.lock_exposure_check = QCheckBox("Lock exposure across batch") + self.lock_exposure_check.setToolTip("Meter the first frame, then hold that exact exposure for the rest of the batch") + self.form.addRow(self.lock_exposure_check) + self.lock_exposure_check.setVisible(False) + # Frame range (roll/strip feeders only — shown when a live capacity is known). self.frame_range_widget = QWidget() frame_row = QHBoxLayout(self.frame_range_widget) @@ -225,6 +255,9 @@ def _init_ui(self) -> None: self.pattern_edit.setText(self._settings.filename_pattern) self.autofocus_check.setChecked(self._settings.autofocus) self.ae_check.setChecked(self._settings.auto_exposure) + self.single_line_check.setChecked(self._settings.single_line) + self.lock_white_balance_check.setChecked(self._settings.lock_white_balance) + self.lock_exposure_check.setChecked(self._settings.lock_exposure) def _connect_signals(self) -> None: self.refresh_btn.clicked.connect(self._on_refresh) @@ -241,6 +274,10 @@ def _connect_signals(self) -> None: self.ir_check.toggled.connect(lambda: self._update_settings_from_ui()) self.autofocus_check.toggled.connect(lambda: self._update_settings_from_ui()) self.ae_check.toggled.connect(lambda: self._update_settings_from_ui()) + self.multisample_combo.currentTextChanged.connect(lambda: self._update_settings_from_ui()) + self.single_line_check.toggled.connect(lambda: self._update_settings_from_ui()) + self.lock_white_balance_check.toggled.connect(lambda: self._update_settings_from_ui()) + self.lock_exposure_check.toggled.connect(lambda: self._update_settings_from_ui()) self.frame_from_spin.valueChanged.connect(self._on_frame_from_changed) self.frame_to_spin.valueChanged.connect(self._on_frame_to_changed) self.scan_window_btn.clicked.connect(self._on_set_scan_window) @@ -343,6 +380,10 @@ def _update_device_caps(self) -> None: self.dpi_combo.setEnabled(False) self.depth_combo.setEnabled(False) self.ir_check.setEnabled(False) + self.quality_row_label.setVisible(False) + self.quality_row_widget.setVisible(False) + self.lock_white_balance_check.setVisible(False) + self.lock_exposure_check.setVisible(False) self.eject_btn.setVisible(False) self.frame_range_label.setVisible(False) self.frame_range_widget.setVisible(False) @@ -352,6 +393,8 @@ def _update_device_caps(self) -> None: self.dpi_combo.setEnabled(True) self.depth_combo.setEnabled(True) self.ir_check.setEnabled(True) + self.multisample_combo.setEnabled(True) + self.single_line_check.setEnabled(True) self.eject_btn.setVisible(caps.can_eject) self.eject_btn.setEnabled(caps.can_eject and not self._scanning) self.frame_label.setText(f"Frame: {caps.max_area_mm[0]:.0f} × {caps.max_area_mm[1]:.0f} mm") @@ -371,6 +414,10 @@ def _populate_form(self, caps: ScannerCapabilities) -> None: self.depth_combo.blockSignals(True) self.ir_check.blockSignals(True) self.ae_check.blockSignals(True) + self.multisample_combo.blockSignals(True) + self.single_line_check.blockSignals(True) + self.lock_white_balance_check.blockSignals(True) + self.lock_exposure_check.blockSignals(True) self.frame_from_spin.blockSignals(True) self.frame_to_spin.blockSignals(True) @@ -417,6 +464,25 @@ def _populate_form(self, caps: ScannerCapabilities) -> None: self.ae_check.setChecked(False) self.ae_check.setToolTip("Auto-exposure not supported by this device") + # Multisample / single-line (nkscan-specific quality knobs; hidden entirely + # on devices that report neither). + has_multisample = len(caps.multisample) > 1 + self.quality_row_label.setVisible(has_multisample or caps.single_line) + self.quality_row_widget.setVisible(has_multisample or caps.single_line) + + self.multisample_combo.clear() + self.multisample_combo.setVisible(has_multisample) + self.multisample_combo.setEnabled(has_multisample) + if has_multisample: + for m in caps.multisample: + self.multisample_combo.addItem(f"{m}×", m) + idx = self.multisample_combo.findData(self._settings.multisample) + self.multisample_combo.setCurrentIndex(idx if idx >= 0 else 0) + + self.single_line_check.setVisible(caps.single_line) + self.single_line_check.setEnabled(caps.single_line) + self.single_line_check.setChecked(caps.single_line and self._settings.single_line) + # Frame range — only a roll/strip feeder reporting a live capacity capacity = caps.adapter_frame_capacity has_frames = capacity is not None @@ -433,6 +499,17 @@ def _populate_form(self, caps: ScannerCapabilities) -> None: self.frame_from_spin.setValue(frm) self.frame_to_spin.setValue(to) + # Lock white balance / lock exposure across batch (nkscan-specific, + # independent controls; only meaningful once there's a batch to hold + # exposure across). + show_lock = caps.supports_exposure_lock and has_frames + self.lock_white_balance_check.setVisible(show_lock) + self.lock_white_balance_check.setEnabled(show_lock) + self.lock_white_balance_check.setChecked(show_lock and self._settings.lock_white_balance) + self.lock_exposure_check.setVisible(show_lock) + self.lock_exposure_check.setEnabled(show_lock) + self.lock_exposure_check.setChecked(show_lock and self._settings.lock_exposure) + self.scan_window_row_label.setVisible(has_frames) self.scan_window_widget.setVisible(has_frames) self.scan_window_status.setVisible(has_frames) @@ -445,6 +522,10 @@ def _populate_form(self, caps: ScannerCapabilities) -> None: self.depth_combo.blockSignals(False) self.ir_check.blockSignals(False) self.ae_check.blockSignals(False) + self.multisample_combo.blockSignals(False) + self.single_line_check.blockSignals(False) + self.lock_white_balance_check.blockSignals(False) + self.lock_exposure_check.blockSignals(False) self.frame_from_spin.blockSignals(False) self.frame_to_spin.blockSignals(False) @@ -548,6 +629,10 @@ def _on_scan(self) -> None: capture_ir = self.ir_check.isEnabled() and self.ir_check.isChecked() autofocus = self.autofocus_check.isChecked() auto_exposure = self.ae_check.isEnabled() and self.ae_check.isChecked() + multisample = int(self.multisample_combo.currentData() or 1) if self.multisample_combo.isEnabled() else 1 + single_line = self.single_line_check.isEnabled() and self.single_line_check.isChecked() + lock_white_balance = self.lock_white_balance_check.isEnabled() and self.lock_white_balance_check.isChecked() + lock_exposure = self.lock_exposure_check.isEnabled() and self.lock_exposure_check.isChecked() pattern = self.pattern_edit.text().strip() or '{{ date }}_{{ "%03d" % seq }}' fmt = self.fmt_combo.currentText() @@ -562,6 +647,8 @@ def _on_scan(self) -> None: auto_exposure=auto_exposure, window=base_window, frame_offset_mm=self._settings.frame_offset_mm, + multisample=multisample, + single_line=single_line, ) self._update_settings_from_ui() @@ -580,6 +667,8 @@ def _on_scan(self) -> None: frames=frames, frame_windows=frame_windows, frame_offset_modifier_mm=self._settings.frame_offset_modifier_mm, + lock_white_balance=lock_white_balance, + lock_exposure=lock_exposure, ) ) else: @@ -685,27 +774,13 @@ def _update_settings_from_ui(self) -> None: capture_ir=self.ir_check.isChecked() and self.ir_check.isEnabled(), autofocus=self.autofocus_check.isChecked(), auto_exposure=self.ae_check.isChecked() and self.ae_check.isEnabled(), + multisample=int(self.multisample_combo.currentData() or 1) if self.multisample_combo.isEnabled() else 1, + single_line=self.single_line_check.isChecked() and self.single_line_check.isEnabled(), + lock_white_balance=self.lock_white_balance_check.isChecked() and self.lock_white_balance_check.isEnabled(), + lock_exposure=self.lock_exposure_check.isChecked() and self.lock_exposure_check.isEnabled(), frame_from=self.frame_from_spin.value(), frame_to=self.frame_to_spin.value(), output_folder=self.folder_edit.text().strip(), output_format=self.fmt_combo.currentText(), filename_pattern=self.pattern_edit.text().strip() or '{{ date }}_{{ "%03d" % seq }}', ) - - -class _ScanUnsupportedPlaceholder(QWidget): - """Shown on Windows where SANE is not available.""" - - def __init__(self, parent: QWidget | None = None) -> None: - super().__init__(parent) - # No layout alignment and no QSS padding: either one breaks the wrapped - # QLabel's height-for-width negotiation and clips the text — the label - # must be stretched to full width so it can report its wrapped height. - layout = QVBoxLayout(self) - layout.setContentsMargins(20, 20, 20, 20) - - label = QLabel("Scanner support not yet available on Windows.") - label.setAlignment(Qt.AlignmentFlag.AlignCenter) - label.setWordWrap(True) - label.setStyleSheet(f"color: {THEME.text_muted}; font-size: {THEME.font_size_base}px;") - layout.addWidget(label) diff --git a/negpy/desktop/workers/scan_worker.py b/negpy/desktop/workers/scan_worker.py index 7a60cbee..f0d3b7b6 100644 --- a/negpy/desktop/workers/scan_worker.py +++ b/negpy/desktop/workers/scan_worker.py @@ -35,7 +35,7 @@ class RollPreviewRequest: @dataclass(frozen=True) class BatchRequest: - """Scan an explicit set of frames, one SANE session each, frame-numbered output.""" + """Scan an explicit set of frames under one held session, frame-numbered output.""" device_id: str params: ScanParams # base; frame + window + offset overridden per iteration @@ -47,6 +47,12 @@ class BatchRequest: # Feed-axis drift (mm/frame): frame N scans at # frame_offset_mm + (N-1) * modifier, floored at 0. frame_offset_modifier_mm: float = 0.0 + # Independent controls, both no-ops on a backend with nothing to hold/freeze: + # hold white balance during every frame's own metering (session open-time intent). + lock_white_balance: bool = False + # Meter the first frame, then freeze that exact gain for the rest of the batch + # (session.lock_exposure(), called after frame 1 completes). + lock_exposure: bool = False class ScanWorker(QObject): @@ -175,7 +181,11 @@ def run_scan(self, req: ScanRequest) -> None: @pyqtSlot(BatchRequest) def run_batch(self, req: BatchRequest) -> None: - """Scan a frame range, one SANE session per frame, frame-numbered output. + """Scan a frame range under one held session, frame-numbered output. + + The device stays open for the whole range (see ScannerSession) rather than + reopening per frame — required for nkscan's held-exposure workflow, and + avoids the Coolscan feeder auto-parking between frames. Holds an idle-sleep assertion for the whole run — a 40-frame SA-30 batch at 4000 dpi is a long unattended operation. `batch_finished` always fires @@ -195,44 +205,53 @@ def run_batch(self, req: BatchRequest) -> None: outcome: tuple[str, str | None] = ("finished", None) try: service = self._ensure_service() - for index, frame in enumerate(frames): - if self._cancel_event.is_set(): - outcome = ("cancelled", None) - break - window = req.frame_windows.get(frame, req.params.window) - offset = max(0.0, req.params.frame_offset_mm + (frame - 1) * req.frame_offset_modifier_mm) - frame_params = dataclasses.replace(req.params, frame=frame, window=window, frame_offset_mm=offset) - base = index / total - - def _progress(fraction: float, _base: float = base) -> None: - self.progress.emit(_base + min(1.0, max(0.0, fraction)) / total) - - try: - result = service.run_scan(req.device_id, frame_params, _progress, self._cancel_event) - except Exception as error: + session = service.open_session(req.device_id, lock_white_balance=req.lock_white_balance) + try: + for index, frame in enumerate(frames): if self._cancel_event.is_set(): outcome = ("cancelled", None) - else: - logger.exception("Batch frame %s scan failed", frame) + break + window = req.frame_windows.get(frame, req.params.window) + offset = max(0.0, req.params.frame_offset_mm + (frame - 1) * req.frame_offset_modifier_mm) + frame_params = dataclasses.replace(req.params, frame=frame, window=window, frame_offset_mm=offset) + base = index / total + + def _progress(fraction: float, _base: float = base) -> None: + self.progress.emit(_base + min(1.0, max(0.0, fraction)) / total) + + try: + result = service.run_session_scan(session, frame_params, _progress, self._cancel_event) + except Exception as error: + if self._cancel_event.is_set(): + outcome = ("cancelled", None) + else: + logger.exception("Batch frame %s scan failed", frame) + outcome = ("error", str(error)) + break + if self._cancel_event.is_set(): + outcome = ("cancelled", None) + break + if index == 0 and req.lock_exposure: + try: + session.lock_exposure() + except Exception: + logger.exception("Could not lock exposure after frame %s", frame) + try: + path = service.write_result( + result=result, + output_folder=req.output_folder, + filename_pattern=req.filename_pattern, + output_format=req.output_format, + seq=frame, + ) + except Exception as error: + logger.exception("Could not write batch frame %s", frame) outcome = ("error", str(error)) - break - if self._cancel_event.is_set(): - outcome = ("cancelled", None) - break - try: - path = service.write_result( - result=result, - output_folder=req.output_folder, - filename_pattern=req.filename_pattern, - output_format=req.output_format, - seq=frame, - ) - except Exception as error: - logger.exception("Could not write batch frame %s", frame) - outcome = ("error", str(error)) - break - paths.append(path) - self.frame_done.emit(frame, path) + break + paths.append(path) + self.frame_done.emit(frame, path) + finally: + session.close() except Exception as error: logger.exception("Could not run scan batch") outcome = ("error", str(error)) diff --git a/negpy/infrastructure/scanners/base.py b/negpy/infrastructure/scanners/base.py index fbea4051..198cdb6b 100644 --- a/negpy/infrastructure/scanners/base.py +++ b/negpy/infrastructure/scanners/base.py @@ -33,6 +33,9 @@ class ScannerCapabilities: adapter_frame_control: bool = False can_eject: bool = False frame_pitch_mm: float = 0.0 # feed-axis distance between frame positions; 0.0 = unknown + multisample: tuple[int, ...] = (1,) # supported multisample factors; (1,) = no averaging option + single_line: bool = False # device offers a single-line CCD scan mode + supports_exposure_lock: bool = False # session.lock_exposure() actually freezes something @dataclass(frozen=True) @@ -58,6 +61,13 @@ def scan( progress: Callable[[float], None], cancel: threading.Event, ) -> ScanResult: ... + def lock_exposure(self) -> None: + """Freeze whatever exposure the last scan settled on for the rest of this session. + + No-op on a transport with nothing to freeze. + """ + ... + def eject(self) -> bool: ... def close(self) -> None: ... def __enter__(self) -> "ScannerSession": ... @@ -91,5 +101,11 @@ def scan( progress: Callable[[float], None], cancel: threading.Event, ) -> ScanResult: ... - def open_session(self, device_id: str) -> ScannerSession: ... + def open_session(self, device_id: str, *, lock_white_balance: bool = False) -> ScannerSession: + """Open an exclusive session. `lock_white_balance` is the caller's intent to hold + white balance during this session's own metering, independent of the later + lock_exposure() gain freeze — a backend with no cast-preserving metering option + may ignore it.""" + ... + def eject(self, device_id: str) -> bool: ... diff --git a/negpy/infrastructure/scanners/nkscan_backend.py b/negpy/infrastructure/scanners/nkscan_backend.py new file mode 100644 index 00000000..f1256959 --- /dev/null +++ b/negpy/infrastructure/scanners/nkscan_backend.py @@ -0,0 +1,218 @@ +import threading +from typing import Callable + +import numpy as np + +from negpy.infrastructure.scanners.base import ( + ScanMode, + ScannerCapabilities, + ScannerDevice, + ScannerUnavailable, + TransientScanError, +) +from negpy.infrastructure.scanners.params import ScanParams +from negpy.infrastructure.scanners.result import ScanResult +from negpy.kernel.system.logging import get_logger + +logger = get_logger(__name__) + +_PREPARE_WAIT_FOR_MEDIA_S = 300.0 + + +def _caps_from_nkscan(caps) -> ScannerCapabilities: + """Build ScannerCapabilities from an nkscan.Capabilities. Pure — no `nkscan` import.""" + return ScannerCapabilities( + ir_channel=bool(caps.ir_channel), + supported_dpi=tuple(sorted(int(d) for d in caps.dpi)), + supported_depths=tuple(sorted(int(d) for d in caps.depths)) or (16,), + # nkscan devices are dedicated film scanners with no source selector at all. + sources=(ScanMode.NEGATIVE, ScanMode.POSITIVE, ScanMode.TRANSPARENCY), + max_area_mm=(float(caps.max_area_mm[0]), float(caps.max_area_mm[1])), + auto_exposure=bool(caps.auto_exposure), + can_eject=bool(caps.can_eject), + multisample=tuple(sorted(int(m) for m in caps.multisample)) or (1,), + single_line=bool(caps.single_line), + supports_exposure_lock=True, # every nkscan session can lock_gain() + ) + + +def _as_scan_error(exc: Exception) -> Exception: + """Re-type an nkscan transient failure so the service can retry without reading messages.""" + import nkscan + + if isinstance(exc, nkscan.TransientError): + return TransientScanError(str(exc)) + return exc + + +def _progress_adapter( + progress: Callable[[float], None] | None, + cancel: threading.Event, +) -> Callable[[int, int], bool]: + """Adapt nkscan's `(read, total) -> bool | None` mid-read callback (returning False + aborts the pass) to NegPy's fractional `progress(float)` callable plus the shared + cancel Event, giving real mid-scan cancellation.""" + + def _cb(read: int, total: int) -> bool: + if progress is not None and total: + try: + progress(min(1.0, max(0.0, read / total))) + except Exception: + pass + return not cancel.is_set() + + return _cb + + +class NkscanSession: + """Wraps one nkscan.Session: lazily prepares on first scan(), then reuses the + placed session for subsequent frames without re-preparing.""" + + def __init__(self, session, *, lock_white_balance: bool = False) -> None: + self._session = session + self._prepared = False + self.device_id: str = session.device_id + # Held during frame 1's own auto-exposure metering — independent of + # lock_exposure() (which freezes the *gain* a metered frame settled on). + # WB lock without a gain freeze still re-meters brightness each frame, + # just without neutralizing the film's orange base while doing it. + self._lock_white_balance = lock_white_balance + + def _ensure_prepared(self, params: ScanParams) -> None: + if self._prepared: + return + try: + self._session.prepare( + frames=None, # hardware-sensed placement; the caller never declares a frame count + pitch_mm=None, + offset_mm=params.frame_offset_mm, + gain=None, # auto-exposure; a caller freezes it afterwards via lock_exposure() + lock_white_balance=self._lock_white_balance, + wait_for_media_s=_PREPARE_WAIT_FOR_MEDIA_S, + ) + except Exception as e: + raise _as_scan_error(e) from e + self._prepared = True + + def scan( + self, + params: ScanParams, + progress: Callable[[float], None], + cancel: threading.Event, + ) -> ScanResult: + if cancel.is_set(): + # A cancel set before this call must not pay for prepare()'s hardware + # wait (up to 300s for media) or a real scan. + raise RuntimeError("Scan cancelled before start") + self._ensure_prepared(params) + # NegPy's ScanParams.frame is 1-indexed (batch ranges are `range(frame_from, frame_to+1)`); + # nkscan's scan(index=...) is explicitly 0-indexed. Deliberate translation, not an oversight. + index = (params.frame - 1) if params.frame is not None else 0 + try: + result = self._session.scan( + index, + dpi=params.dpi, + ir=params.capture_ir, + # nkscan has no way to report back or accept a prior autofocus setpoint, so + # ScanParams.autofocus=False cannot be honoured — always autofocus fresh. + focus="auto", + multisample=params.multisample, + single_line=params.single_line, + window=params.window, + progress=_progress_adapter(progress, cancel), + ) + except Exception as e: + raise _as_scan_error(e) from e + # nkscan's IR plane always covers the full frame. + ir_valid_mask = np.ones(result.ir.shape[:2], dtype=np.bool_) if result.ir is not None else None + return ScanResult( + rgb=result.rgb, + ir=result.ir, + dpi=result.dpi, + device_model=result.device_model, + ir_valid_mask=ir_valid_mask, + ) + + def lock_exposure(self) -> None: + self._session.lock_gain() + + def eject(self) -> bool: + # Capability-gated no-op for a device with no eject action, matching + # SaneBackend.eject()'s contract — nkscan's own eject() has no such gate. + if not bool(self._session.capabilities.can_eject): + self.close() + return False + try: + return bool(self._session.eject()) + except Exception as e: + raise _as_scan_error(e) from e + finally: + self.close() + + def close(self) -> None: + self._session.close() + + def __enter__(self) -> "NkscanSession": + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + +class NkscanBackend: + """nkscan implementation of ScannerBackend. Only module that imports `nkscan`.""" + + def __init__(self) -> None: + try: + import nkscan # noqa: F401 + except ImportError: + raise ScannerUnavailable("nkscan not importable. Install: uv sync --group nkscan") from None + self._nkscan = nkscan + self._devices_cache: list[ScannerDevice] | None = None + + def list_devices(self) -> list[ScannerDevice]: + if self._devices_cache is not None: + return self._devices_cache + try: + raw_devices = self._nkscan.list_devices() + except Exception as e: + logger.error(f"nkscan device listing failed: {e}") + return [] + devices = [ + ScannerDevice( + id=d.id, + vendor=d.vendor, + model=d.model, + capabilities=_caps_from_nkscan(d.capabilities), + ) + for d in raw_devices + ] + self._devices_cache = devices + return devices + + def refresh_devices(self) -> list[ScannerDevice]: + self._devices_cache = None + return self.list_devices() + + def open_session(self, device_id: str, *, lock_white_balance: bool = False) -> NkscanSession: + try: + session = self._nkscan.Session(device_id) + except Exception as e: + raise _as_scan_error(e) from e + return NkscanSession(session, lock_white_balance=lock_white_balance) + + def scan( + self, + device_id: str, + params: ScanParams, + progress: Callable[[float], None], + cancel: threading.Event, + ) -> ScanResult: + with self.open_session(device_id) as session: + return session.scan(params, progress, cancel) + + def eject(self, device_id: str) -> bool: + # NkscanSession.eject() already terminates the session itself; no `with` + # here, or it would close a second time (harmless — Session.close() is + # idempotent — but pointless). + return self.open_session(device_id).eject() diff --git a/negpy/infrastructure/scanners/params.py b/negpy/infrastructure/scanners/params.py index bb0de0e6..b5907835 100644 --- a/negpy/infrastructure/scanners/params.py +++ b/negpy/infrastructure/scanners/params.py @@ -25,6 +25,10 @@ class ScanParams: # Hardware auto-exposure (SANE `ae`), distinct from NegPy's rendering # auto-exposure. An explicit request fails if the option is unavailable. auto_exposure: bool = False + # Multisample averaging factor and single-line CCD mode. nkscan-specific; + # ignored by backends that don't support them. + multisample: int = 1 + single_line: bool = False MIN_FRAME_EXTENT_MM = 1.0 # below this a capped scan is a useless sliver diff --git a/negpy/infrastructure/scanners/registry.py b/negpy/infrastructure/scanners/registry.py index d06d44fd..7a0fb138 100644 --- a/negpy/infrastructure/scanners/registry.py +++ b/negpy/infrastructure/scanners/registry.py @@ -9,12 +9,19 @@ def _make_sane() -> ScannerBackend: return SaneBackend() +def _make_nkscan() -> ScannerBackend: + from negpy.infrastructure.scanners.nkscan_backend import NkscanBackend + + return NkscanBackend() + + DEFAULT_BACKEND_ID = "sane" # id -> (display label, factory). Insertion order drives the sidebar dropdown. # Adding a backend is one entry here plus its implementation module. BACKENDS: dict[str, tuple[str, Callable[[], ScannerBackend]]] = { "sane": ("SANE", _make_sane), + "nkscan": ("nkscan (Coolscan)", _make_nkscan), } diff --git a/negpy/infrastructure/scanners/sane_backend.py b/negpy/infrastructure/scanners/sane_backend.py index c631f6f0..38eebbc3 100644 --- a/negpy/infrastructure/scanners/sane_backend.py +++ b/negpy/infrastructure/scanners/sane_backend.py @@ -614,6 +614,9 @@ def scan( raise RuntimeError(f"Scanner session for {self.device_id} is closed") return self._backend._scan_on_device(self._dev, self.device_id, params, progress, cancel) + def lock_exposure(self) -> None: + """No-op: SANE's `ae` re-meters fresh on every scan, there's nothing to freeze.""" + def eject(self) -> bool: """Press the vendor eject action, if any. Always ends the session. @@ -749,11 +752,15 @@ def _open_device(self, device_id: str): logger.info(f"Scanner {device_id} re-enumerated; remapped to {fresh_id}") return dev, fresh_id - def open_session(self, device_id: str) -> SaneSession: + def open_session(self, device_id: str, *, lock_white_balance: bool = False) -> SaneSession: """Open an exclusive scanning session — the batch/roll handover seam. The device is opened once (self-healing via _open_device) and stays open until SaneSession.close()/eject(). One session per device. + + `lock_white_balance` is accepted for ScannerBackend conformance and ignored: + `ae` re-meters fresh on every scan, so there's no cast-preserving metering + mode to hold (see SaneSession.lock_exposure()). """ try: self._ensure_initialized() diff --git a/negpy/infrastructure/scanners/settings.py b/negpy/infrastructure/scanners/settings.py index 93eb6840..577e1595 100644 --- a/negpy/infrastructure/scanners/settings.py +++ b/negpy/infrastructure/scanners/settings.py @@ -14,6 +14,10 @@ class ScannerSettings: capture_ir: bool = False autofocus: bool = True auto_exposure: bool = False + multisample: int = 1 + single_line: bool = False + lock_white_balance: bool = False + lock_exposure: bool = False frame_from: int = 1 frame_to: int = 1 output_folder: str = "" diff --git a/negpy/services/scanning/service.py b/negpy/services/scanning/service.py index 6d62dafb..a70811ae 100644 --- a/negpy/services/scanning/service.py +++ b/negpy/services/scanning/service.py @@ -46,13 +46,15 @@ def list_devices(self) -> list[ScannerDevice]: def refresh_devices(self) -> list[ScannerDevice]: return self._get_backend().refresh_devices() - def open_session(self, device_id: str) -> ScannerSession: + def open_session(self, device_id: str, *, lock_white_balance: bool = False) -> ScannerSession: """Open an exclusive device session for batch/roll workflows. The session owns the scanner until closed: one continuous open, per-frame - scan() calls, one release (close/eject) at the end. + scan() calls, one release (close/eject) at the end. `lock_white_balance` is + the caller's intent to hold white balance during this session's own + metering — see ScannerBackend.open_session. """ - return self._get_backend().open_session(device_id) + return self._get_backend().open_session(device_id, lock_white_balance=lock_white_balance) def eject(self, device_id: str) -> bool: """Trigger the device's eject action. @@ -80,11 +82,46 @@ def run_scan( *, retry_delay: float = _SCAN_IO_RETRY_DELAY_S, ) -> ScanResult: - """Scan, retrying once on a transient USB I/O glitch (fresh open each try).""" + """One-shot scan (backend opens, scans, closes), retrying once on a + transient USB I/O glitch (fresh open each try).""" backend = self._get_backend() + return self._retry_transient( + lambda: backend.scan(device_id, params, progress, cancel), + device_id=device_id, + cancel=cancel, + retry_delay=retry_delay, + ) + + def run_session_scan( + self, + session: ScannerSession, + params: ScanParams, + progress: Callable[[float], None], + cancel: threading.Event, + *, + retry_delay: float = _SCAN_IO_RETRY_DELAY_S, + ) -> ScanResult: + """Scan one frame on an already-open session (batch/roll workflows that hold + the device across many frames), with the same transient-retry behaviour as + run_scan.""" + return self._retry_transient( + lambda: session.scan(params, progress, cancel), + device_id=session.device_id, + cancel=cancel, + retry_delay=retry_delay, + ) + + @staticmethod + def _retry_transient( + scan_fn: Callable[[], ScanResult], + *, + device_id: str, + cancel: threading.Event, + retry_delay: float, + ) -> ScanResult: for attempt in range(1, _SCAN_IO_RETRY_ATTEMPTS + 1): try: - return backend.scan(device_id, params, progress, cancel) + return scan_fn() except TransientScanError as e: if attempt >= _SCAN_IO_RETRY_ATTEMPTS or cancel.is_set(): raise diff --git a/pyproject.toml b/pyproject.toml index 368661ee..2b886200 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,10 +3,7 @@ name = "negpy" description = "A tool for processing film negatives." readme = "README.md" license = "GPL-3.0" -authors = [ - { name = "marcinz606" }, - { name = "NegPy Contributors" } -] +authors = [{ name = "marcinz606" }, { name = "NegPy Contributors" }] dynamic = ["version"] requires-python = ">=3.13" dependencies = [ @@ -36,7 +33,7 @@ classifiers = [ "Programming Language :: Python :: 3.13", "Topic :: Multimedia :: Graphics :: Graphics Conversion", "Topic :: Multimedia :: Graphics :: Viewers", - "Topic :: Scientific/Engineering :: Image Processing" + "Topic :: Scientific/Engineering :: Image Processing", ] [dependency-groups] @@ -47,14 +44,16 @@ dev = [ "ruff==0.14.10", "ty>=0.0.26", ] -scanner = [ - "python-sane>=2.9", -] +scanner = ["python-sane>=2.9"] camera = [ # Tethered camera scanning. libgphoto2 has no Windows build, so the wheels — and the # Camera Scanning tab with them — are macOS and Linux only. "gphoto2>=2.5 ; sys_platform != 'win32'", ] +nkscan = [ + # Rust-based USB/SCSI Nikon Coolscan driver. + "nkscan>=0.1.0", +] [project.urls] Homepage = "https://github.com/marcinz606/NegPy" diff --git a/tests/scanners/test_backend_contract.py b/tests/scanners/test_backend_contract.py index 49fd9012..bf498b8a 100644 --- a/tests/scanners/test_backend_contract.py +++ b/tests/scanners/test_backend_contract.py @@ -41,6 +41,13 @@ def __call__( """ ... + def transient_error(self) -> Exception: + """A sample exception this backend's own scan() treats as transient. + + What counts as "transient" is backend-specific — not a shared literal. + """ + ... + # ── the SANE entry ──────────────────────────────────────────────────────── @@ -71,21 +78,75 @@ def get_devices(self) -> list[tuple[str, str, str]]: return [(self.device_id, "Nikon", "LS-50")] -def _sane_backend( - *, - scan_error: Exception | None = None, - with_eject: bool = False, - film: bool = True, - progress_steps: int = 0, -) -> tuple[Any, str]: - # A non-film transport is spelled as a flatbed id with no `source` option, so - # nothing infers film sources for it. - device_id = _DEV_ID if film else "epson2:libusb:001:002" - dev = _ContractDev(_opt_map(eject=with_eject), scan_error, progress_steps) - return _make_backend(_ModuleWithDevice(dev, device_id)), device_id +class _SaneBackendFactory: + def __call__( + self, + *, + scan_error: Exception | None = None, + with_eject: bool = False, + film: bool = True, + progress_steps: int = 0, + ) -> tuple[Any, str]: + # A non-film transport is spelled as a flatbed id with no `source` option, so + # nothing infers film sources for it. + device_id = _DEV_ID if film else "epson2:libusb:001:002" + dev = _ContractDev(_opt_map(eject=with_eject), scan_error, progress_steps) + return _make_backend(_ModuleWithDevice(dev, device_id)), device_id + + def transient_error(self) -> Exception: + # A string match against SANE's own stable status text (see + # sane_backend._TRANSIENT_IO_MARKERS), not a shared literal. + return RuntimeError("Error during device I/O") + + +_sane_backend = _SaneBackendFactory() + + +# ── the nkscan entry ────────────────────────────────────────────────────── -BACKENDS: list[tuple[str, _Factory]] = [("sane", _sane_backend)] +class _NkscanBackendFactory: + def __call__( + self, + *, + scan_error: Exception | None = None, + with_eject: bool = False, + film: bool = True, + progress_steps: int = 0, + ) -> tuple[Any, str]: + import sys + + from negpy.infrastructure.scanners.nkscan_backend import NkscanBackend + from tests.scanners.test_nkscan_backend import _DEV_ID as _NKSCAN_DEV_ID + from tests.scanners.test_nkscan_backend import FakeCapabilities, FakeDevice, FakeNkscanModule, FakeSession + + # nkscan devices are always dedicated film scanners; a "no film sources" + # transport is spelled as the module reporting no devices at all. + caps = FakeCapabilities(can_eject=with_eject) + devices = [FakeDevice(_NKSCAN_DEV_ID, "Nikon", "LS-50", caps)] if film else [] + session = FakeSession( + _NKSCAN_DEV_ID, + scan_error=scan_error, + capabilities=caps, + progress_steps=max(progress_steps, 1), + ) + module = FakeNkscanModule(devices=devices, session=session) + # Process-global and deliberately left in place (not monkeypatch-scoped): every + # caller either goes through this factory again or test_nkscan_backend.py's own + # fixture, both of which reassign sys.modules["nkscan"] fresh before use. + sys.modules["nkscan"] = module # type: ignore[assignment] + return NkscanBackend(), _NKSCAN_DEV_ID + + def transient_error(self) -> Exception: + from tests.scanners.test_nkscan_backend import _TransientError + + return _TransientError("USB glitch") + + +_nkscan_backend = _NkscanBackendFactory() + + +BACKENDS: list[tuple[str, _Factory]] = [("sane", _sane_backend), ("nkscan", _nkscan_backend)] pytestmark = pytest.mark.parametrize("name,make_backend", BACKENDS) @@ -146,8 +207,12 @@ def test_progress_stays_within_the_unit_range(name: str, make_backend: _Factory) def test_transport_glitches_are_typed_transient(name: str, make_backend: _Factory) -> None: - """The service retries on type alone — it must not have to read messages.""" - backend, device_id = make_backend(scan_error=RuntimeError("Error during device I/O")) + """The service retries on type alone — it must not have to read messages. + + What counts as "transient" is each backend's own call — `.transient_error()` + is a sample of whatever that backend's factory attaches, not a shared literal. + """ + backend, device_id = make_backend(scan_error=make_backend.transient_error()) with pytest.raises(TransientScanError): backend.scan(device_id, _PARAMS, lambda _: None, threading.Event()) diff --git a/tests/scanners/test_backend_registry.py b/tests/scanners/test_backend_registry.py index 7a23f67d..cf956f63 100644 --- a/tests/scanners/test_backend_registry.py +++ b/tests/scanners/test_backend_registry.py @@ -5,6 +5,10 @@ def test_backend_choices_includes_sane(): assert ("sane", "SANE") in registry.backend_choices() +def test_backend_choices_includes_nkscan(): + assert ("nkscan", "nkscan (Coolscan)") in registry.backend_choices() + + def test_create_backend_resolves_and_falls_back(monkeypatch): sentinel = object() monkeypatch.setattr(registry, "BACKENDS", {"sane": ("SANE", lambda: sentinel)}) diff --git a/tests/scanners/test_nkscan_backend.py b/tests/scanners/test_nkscan_backend.py new file mode 100644 index 00000000..bf7312df --- /dev/null +++ b/tests/scanners/test_nkscan_backend.py @@ -0,0 +1,372 @@ +"""NkscanBackend/NkscanSession: capability mapping, lazy prepare, frame index +translation (1-indexed ScanParams.frame -> nkscan's 0-indexed scan(index=)), +the progress/cancel adapter, and error translation. + +A fake `nkscan` module is installed into sys.modules for the duration of each +test — NkscanBackend.__init__ and _as_scan_error both do their own `import +nkscan`, so the fake must be importable, not just attached to an instance. +""" + +from __future__ import annotations + +import sys +import threading +from dataclasses import dataclass, field +from typing import Any, Callable + +import numpy as np +import pytest + +from negpy.infrastructure.scanners.base import ScanMode, ScannerUnavailable, TransientScanError +from negpy.infrastructure.scanners.nkscan_backend import NkscanBackend, _progress_adapter +from negpy.infrastructure.scanners.params import ScanParams + +_DEV_ID = "usb:001:002" +_PARAMS = ScanParams(dpi=1000, depth=16, capture_ir=False) + + +class _ScannerError(RuntimeError): + pass + + +class _TransientError(_ScannerError): + pass + + +class _TransportError(_TransientError): + pass + + +class _DeviceBusy(_TransientError): + pass + + +class _MediaError(_ScannerError): + pass + + +class _ScanCancelled(_ScannerError): + pass + + +class _UnsupportedError(_ScannerError): + pass + + +class _DeviceNotFound(_ScannerError): + pass + + +@dataclass +class FakeCapabilities: + dpi: list[int] = field(default_factory=lambda: [4000, 1000]) + depths: list[int] = field(default_factory=lambda: [16]) + multisample: list[int] = field(default_factory=lambda: [1, 2, 4]) + ir_channel: bool = True + max_area_mm: tuple[float, float] = (36.0, 24.0) + auto_exposure: bool = True + frame_control: bool = True + detects_frames: bool = False + senses_frames: bool = True + single_line: bool = True + can_eject: bool = True + + +@dataclass +class FakeDevice: + id: str + vendor: str + model: str + capabilities: FakeCapabilities + + +@dataclass +class FakeScanResult: + rgb: np.ndarray + ir: np.ndarray | None + dpi: int + device_model: str + frame: int + + +class FakeSession: + """Stand-in for nkscan.Session: records prepare()/scan()/lock_gain() calls.""" + + def __init__( + self, + device_id: str, + *, + scan_error: Exception | None = None, + rgb_shape: tuple[int, int, int] = (6, 5, 3), + capabilities: FakeCapabilities | None = None, + progress_steps: int = 2, + ) -> None: + self.device_id = device_id + self.model = "LS-50" + self.capabilities = capabilities or FakeCapabilities() + self.prepare_calls: list[dict[str, Any]] = [] + self.scan_calls: list[dict[str, Any]] = [] + self.progress_returns: list[bool] = [] + self.lock_gain_calls = 0 + self.eject_calls = 0 + self.close_calls = 0 + self._scan_error = scan_error + self._rgb_shape = rgb_shape + self._progress_steps = progress_steps + + def prepare(self, **kwargs: Any) -> int: + self.prepare_calls.append(kwargs) + return 1 + + def scan( + self, + index: int, + *, + dpi: int, + ir: bool, + focus: str, + multisample: int, + single_line: bool, + window: tuple[float, float, float, float] | None, + progress: Callable[[int, int], bool] | None = None, + ) -> FakeScanResult: + if progress is not None: + for i in range(1, self._progress_steps + 1): + self.progress_returns.append(progress(i, self._progress_steps)) + self.scan_calls.append( + dict(index=index, dpi=dpi, ir=ir, focus=focus, multisample=multisample, single_line=single_line, window=window) + ) + if self._scan_error is not None: + raise self._scan_error + h, w, c = self._rgb_shape + rgb = np.zeros((h, w, c), dtype=np.uint16) + ir_arr = np.zeros((h, w), dtype=np.uint16) if ir else None + return FakeScanResult(rgb=rgb, ir=ir_arr, dpi=dpi, device_model=self.model, frame=index) + + def lock_gain(self) -> None: + self.lock_gain_calls += 1 + + def eject(self) -> bool: + self.eject_calls += 1 + return True + + def close(self) -> None: + self.close_calls += 1 + + +class FakeNkscanModule: + """Stand-in for the compiled `nkscan` extension module.""" + + def __init__(self, *, devices: list[FakeDevice] | None = None, session: FakeSession | None = None) -> None: + self.devices = devices if devices is not None else [FakeDevice(_DEV_ID, "Nikon", "LS-50", FakeCapabilities())] + self._session = session + self.opened: list[str] = [] + self.TransientError = _TransientError + self.TransportError = _TransportError + self.DeviceBusy = _DeviceBusy + self.ScannerError = _ScannerError + self.MediaError = _MediaError + self.ScanCancelled = _ScanCancelled + self.UnsupportedError = _UnsupportedError + self.DeviceNotFound = _DeviceNotFound + + def list_devices(self) -> list[FakeDevice]: + return self.devices + + def Session(self, device_id: str) -> FakeSession: + self.opened.append(device_id) + return self._session or FakeSession(device_id) + + +@pytest.fixture +def fake_nkscan(monkeypatch: pytest.MonkeyPatch) -> FakeNkscanModule: + module = FakeNkscanModule() + monkeypatch.setitem(sys.modules, "nkscan", module) + return module + + +def test_backend_unavailable_when_nkscan_is_not_importable(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setitem(sys.modules, "nkscan", None) # forces ImportError on `import nkscan` + + with pytest.raises(ScannerUnavailable, match="uv sync --group nkscan"): + NkscanBackend() + + +def test_list_devices_maps_capabilities(fake_nkscan: FakeNkscanModule) -> None: + backend = NkscanBackend() + devices = backend.list_devices() + + assert len(devices) == 1 + device = devices[0] + assert device.id == _DEV_ID + caps = device.capabilities + # Dedicated film scanner with no source selector: all three modes, unconditionally. + assert set(caps.sources) == {ScanMode.NEGATIVE, ScanMode.POSITIVE, ScanMode.TRANSPARENCY} + assert caps.supported_dpi == (1000, 4000) # sorted + assert caps.supported_depths == (16,) + assert caps.multisample == (1, 2, 4) + assert caps.single_line is True + assert caps.ir_channel is True + assert caps.can_eject is True + assert caps.supports_exposure_lock is True # every nkscan session can lock_gain() + + +def test_scan_prepares_lazily_and_only_once_per_session(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + with backend.open_session(_DEV_ID) as session: + session.scan(_PARAMS, lambda _: None, threading.Event()) + session.scan(_PARAMS, lambda _: None, threading.Event()) + + assert len(session_impl.prepare_calls) == 1 + assert len(session_impl.scan_calls) == 2 + assert session_impl.close_calls == 1 + + +def test_open_session_with_lock_white_balance_holds_it_during_prepare(fake_nkscan: FakeNkscanModule) -> None: + """Independent of lock_exposure()/lock_gain() (which freezes the *gain* a + metered frame settled on): this affects frame 1's own metering, so the film's + orange base isn't neutralized away while auto-exposure is computing gains.""" + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + with backend.open_session(_DEV_ID, lock_white_balance=True) as session: + session.scan(_PARAMS, lambda _: None, threading.Event()) + + assert session_impl.prepare_calls[0]["lock_white_balance"] is True + + +def test_open_session_without_lock_white_balance_leaves_it_unlocked(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + with backend.open_session(_DEV_ID) as session: + session.scan(_PARAMS, lambda _: None, threading.Event()) + + assert session_impl.prepare_calls[0]["lock_white_balance"] is False + + +def test_scan_frame_index_is_translated_from_1_indexed_to_0_indexed(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + with backend.open_session(_DEV_ID) as session: + session.scan(ScanParams(dpi=1000, depth=16, capture_ir=False, frame=3), lambda _: None, threading.Event()) + session.scan(ScanParams(dpi=1000, depth=16, capture_ir=False, frame=None), lambda _: None, threading.Event()) + + assert [c["index"] for c in session_impl.scan_calls] == [2, 0] + + +def test_one_shot_scan_returns_a_well_formed_result(fake_nkscan: FakeNkscanModule) -> None: + backend = NkscanBackend() + + result = backend.scan(_DEV_ID, _PARAMS, lambda _: None, threading.Event()) + + assert result.rgb.shape == (6, 5, 3) + assert result.dpi == _PARAMS.dpi + assert result.device_model == "LS-50" + assert result.ir is None + assert result.ir_valid_mask is None + + +def test_ir_valid_mask_is_set_when_ir_is_captured(fake_nkscan: FakeNkscanModule) -> None: + backend = NkscanBackend() + params = ScanParams(dpi=1000, depth=16, capture_ir=True) + + result = backend.scan(_DEV_ID, params, lambda _: None, threading.Event()) + + assert result.ir is not None + assert result.ir_valid_mask is not None + assert result.ir_valid_mask.shape == result.ir.shape + assert result.ir_valid_mask.all() + + +def test_progress_adapter_reports_fraction_and_honours_cancel() -> None: + """Unit-tested directly: a scan() call with cancel already set never reaches + the adapter at all (see test_scan_honours_a_precancelled_call below) — this + is what nkscan's mid-read callback sees once a read is actually underway.""" + seen: list[float] = [] + cancel = threading.Event() + adapter = _progress_adapter(seen.append, cancel) + + assert adapter(1, 2) is True + assert adapter(2, 2) is True + assert seen == pytest.approx([0.5, 1.0]) + + cancel.set() + assert adapter(1, 2) is False # mid-read cancel: tell nkscan to stop + + +def test_scan_honours_a_precancelled_call(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + cancel = threading.Event() + cancel.set() + + with pytest.raises(RuntimeError, match="[Cc]ancel"): + backend.scan(_DEV_ID, _PARAMS, lambda _: None, cancel) + + assert session_impl.scan_calls == [] # never reached the hardware + + +def test_transient_error_is_translated(fake_nkscan: FakeNkscanModule) -> None: + fake_nkscan._session = FakeSession(_DEV_ID, scan_error=fake_nkscan.TransientError("USB glitch")) + backend = NkscanBackend() + + with pytest.raises(TransientScanError): + backend.scan(_DEV_ID, _PARAMS, lambda _: None, threading.Event()) + + +def test_media_error_is_not_translated_to_transient(fake_nkscan: FakeNkscanModule) -> None: + fake_nkscan._session = FakeSession(_DEV_ID, scan_error=fake_nkscan.MediaError("no film")) + backend = NkscanBackend() + + with pytest.raises(Exception) as excinfo: + backend.scan(_DEV_ID, _PARAMS, lambda _: None, threading.Event()) + assert not isinstance(excinfo.value, TransientScanError) + + +def test_scan_cancelled_propagates_with_a_cancel_shaped_message(fake_nkscan: FakeNkscanModule) -> None: + fake_nkscan._session = FakeSession(_DEV_ID, scan_error=fake_nkscan.ScanCancelled("the pass was cancelled")) + backend = NkscanBackend() + + with pytest.raises(Exception, match="[Cc]ancel"): + backend.scan(_DEV_ID, _PARAMS, lambda _: None, threading.Event()) + + +def test_lock_exposure_calls_lock_gain(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + with backend.open_session(_DEV_ID) as session: + session.scan(_PARAMS, lambda _: None, threading.Event()) + session.lock_exposure() + + assert session_impl.lock_gain_calls == 1 + + +def test_eject_closes_the_session(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + assert backend.eject(_DEV_ID) is True + assert session_impl.eject_calls == 1 + assert session_impl.close_calls == 1 + + +def test_eject_is_capability_gated_but_still_releases(fake_nkscan: FakeNkscanModule) -> None: + session_impl = FakeSession(_DEV_ID, capabilities=FakeCapabilities(can_eject=False)) + fake_nkscan._session = session_impl + backend = NkscanBackend() + + assert backend.eject(_DEV_ID) is False + assert session_impl.eject_calls == 0 + assert session_impl.close_calls == 1 diff --git a/tests/scanners/test_sane_session.py b/tests/scanners/test_sane_session.py index ba4e5a2d..7a058845 100644 --- a/tests/scanners/test_sane_session.py +++ b/tests/scanners/test_sane_session.py @@ -288,7 +288,7 @@ class FakeBackend: def __init__(self) -> None: self.calls: list[str] = [] - def open_session(self, device_id: str): + def open_session(self, device_id: str, *, lock_white_balance: bool = False): self.calls.append(device_id) return sentinel diff --git a/tests/test_scan_worker.py b/tests/test_scan_worker.py index 4c9dea5f..2d483cca 100644 --- a/tests/test_scan_worker.py +++ b/tests/test_scan_worker.py @@ -62,7 +62,12 @@ def write_result(self, **_kwargs) -> str: class _BatchService: - """Records per-frame scans; returns a frame-numbered path from write_result.""" + """Records per-frame scans; returns a frame-numbered path from write_result. + + Stands in for ScannerService: `open_session` hands back itself as the held + session (one open for the whole batch, not one per frame — see run_batch), + and `run_session_scan`/`lock_exposure`/`close` record how the worker used it. + """ def __init__(self, *, fail_on: int | None = None, cancel_before: int | None = None) -> None: self.fail_on = fail_on @@ -72,12 +77,28 @@ def __init__(self, *, fail_on: int | None = None, cancel_before: int | None = No self.windows: list = [] self.offsets: list[float] = [] self.eject_calls: list[str] = [] + self.session_open_calls: list[str] = [] + self.session_open_lock_white_balance: list[bool] = [] + self.session_close_calls = 0 + self.lock_exposure_calls = 0 def eject(self, device_id: str) -> bool: self.eject_calls.append(device_id) return True - def run_scan(self, device_id, params, progress, cancel): + def open_session(self, device_id: str, *, lock_white_balance: bool = False) -> "_BatchService": + self.session_open_calls.append(device_id) + self.session_open_lock_white_balance.append(lock_white_balance) + return self + + def close(self) -> None: + self.session_close_calls += 1 + + def lock_exposure(self) -> None: + self.lock_exposure_calls += 1 + + def run_session_scan(self, session, params, progress, cancel): + assert session is self if self.cancel_before is not None and params.frame == self.cancel_before: cancel.set() self.frames.append(params.frame) @@ -104,7 +125,7 @@ def _scan_request() -> ScanRequest: ) -def _batch_request(frames=(2, 3, 4), frame_windows=None) -> BatchRequest: +def _batch_request(frames=(2, 3, 4), frame_windows=None, lock_white_balance=False, lock_exposure=False) -> BatchRequest: return BatchRequest( device_id="coolscan3:test", params=ScanParams(dpi=4_000, depth=16, capture_ir=False), @@ -113,6 +134,8 @@ def _batch_request(frames=(2, 3, 4), frame_windows=None) -> BatchRequest: output_format="TIFF", frames=tuple(frames), frame_windows=frame_windows or {}, + lock_white_balance=lock_white_balance, + lock_exposure=lock_exposure, ) @@ -384,6 +407,62 @@ def test_run_batch_scans_an_explicit_frame_subset() -> None: assert service.written_seqs == [1, 2, 4, 6] +def test_run_batch_opens_one_session_for_the_whole_range() -> None: + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + + worker.run_batch(_batch_request((2, 3, 4))) + + assert service.session_open_calls == ["coolscan3:test"] # once, not per frame + assert service.session_close_calls == 1 + + +def test_run_batch_locks_exposure_once_after_the_first_frame_when_requested() -> None: + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + + worker.run_batch(_batch_request((2, 3, 4), lock_exposure=True)) + + assert service.lock_exposure_calls == 1 + + +def test_run_batch_passes_lock_white_balance_intent_to_open_session() -> None: + """The session must know up front, not get it after the fact — a backend + (nkscan) uses this to hold white balance during frame 1's own metering, + which lock_exposure()'s later gain freeze cannot retroactively fix.""" + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + + worker.run_batch(_batch_request((2, 3, 4), lock_white_balance=True)) + + assert service.session_open_lock_white_balance == [True] + + +def test_run_batch_lock_white_balance_and_lock_exposure_are_independent() -> None: + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + + worker.run_batch(_batch_request((2, 3, 4), lock_white_balance=True, lock_exposure=False)) + + assert service.session_open_lock_white_balance == [True] + assert service.lock_exposure_calls == 0 + + +def test_run_batch_does_not_lock_anything_by_default() -> None: + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + + worker.run_batch(_batch_request((2, 3, 4))) + + assert service.lock_exposure_calls == 0 + assert service.session_open_lock_white_balance == [False] + + def test_run_batch_applies_per_frame_windows_falling_back_to_base() -> None: worker = ScanWorker() service = _BatchService() diff --git a/uv.lock b/uv.lock index cfbb8304..fd3a00ec 100644 --- a/uv.lock +++ b/uv.lock @@ -322,6 +322,9 @@ dev = [ { name = "ruff" }, { name = "ty" }, ] +nkscan = [ + { name = "nkscan" }, +] scanner = [ { name = "python-sane" }, ] @@ -354,8 +357,22 @@ dev = [ { name = "ruff", specifier = "==0.14.10" }, { name = "ty", specifier = ">=0.0.26" }, ] +nkscan = [{ name = "nkscan", specifier = ">=0.1.0" }] scanner = [{ name = "python-sane", specifier = ">=2.9" }] +[[package]] +name = "nkscan" +version = "0.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/10/75/be7781b46868cda43314ff98522a200a9d19b5ab20bb09e2155639042a2c/nkscan-0.1.0.tar.gz", hash = "sha256:66d84dc1910b2d87ac270dcaf33ca249a44287cb691a77c5a789cbd8ee5cbee3", size = 201179, upload-time = "2026-07-30T17:04:53.778Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/96/fa/d7e5bec233f36746e687f1ce7444c159167aae4b7aea2435fc884cdf8906/nkscan-0.1.0-cp313-abi3-macosx_10_12_x86_64.whl", hash = "sha256:f39fd98a02b6fb5d293c06febb56e8aa7ac94b0afb0e1b37e5846d5c31d99198", size = 549065, upload-time = "2026-07-30T17:04:46.278Z" }, + { url = "https://files.pythonhosted.org/packages/f9/5e/8addcd41d8c60a10dfd43d5e0fab7b7266ce4564c56272997074f139ae94/nkscan-0.1.0-cp313-abi3-macosx_11_0_arm64.whl", hash = "sha256:285350c5efee51e7fc56c6933ece863d83d1f1aa6f653052ddaae13e18f085d8", size = 532609, upload-time = "2026-07-30T17:04:47.843Z" }, + { url = "https://files.pythonhosted.org/packages/48/c6/33df56f628486f5c338375f806eb6efe2380a7e024e2aa7ad1ef8e27d006/nkscan-0.1.0-cp313-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:4a7253339c4e9f9b5f6040b8b66bb50ae9d2a1786d7679a2ce71fc3dfeebfb44", size = 615385, upload-time = "2026-07-30T17:04:49.425Z" }, + { url = "https://files.pythonhosted.org/packages/df/ee/a7aa7bc9208d18e3637590927974e5a74a0f9d2df35b24ebb4b97d4e2aa5/nkscan-0.1.0-cp313-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2fec70749cc876b665b3fac0d727fbbc36a7680c8bae6462919900ba1bbdbc92", size = 663715, upload-time = "2026-07-30T17:04:50.825Z" }, + { url = "https://files.pythonhosted.org/packages/b6/a7/2f5a8dfef2b78e668b22de42ed115eb627f71d54cc1106b4b34d49a3b5b4/nkscan-0.1.0-cp313-abi3-win_amd64.whl", hash = "sha256:ec63b85d2b6aafe43e59a7e1ab460cd8b145f486335cd5d475b58f536d9e3389", size = 401021, upload-time = "2026-07-30T17:04:52.37Z" }, +] + [[package]] name = "numba" version = "0.65.1"