diff --git a/negpy/desktop/controller.py b/negpy/desktop/controller.py index f7189b4c..3dfbb8e9 100644 --- a/negpy/desktop/controller.py +++ b/negpy/desktop/controller.py @@ -206,7 +206,7 @@ class AppController(QObject): scan_backend_requested = pyqtSignal(str) scan_requested = pyqtSignal(ScanRequest) scan_devices_ready = pyqtSignal(list) - scan_progress = pyqtSignal(float) + scan_progress = pyqtSignal(float, str) # progress, phase name scan_finished = pyqtSignal(str) scan_error = pyqtSignal(str) scan_started = pyqtSignal() diff --git a/negpy/desktop/view/sidebar/right_panel.py b/negpy/desktop/view/sidebar/right_panel.py index a3807c35..8f65eb18 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 @@ -90,12 +89,9 @@ 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) + 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..66c2be12 100644 --- a/negpy/desktop/view/sidebar/scan.py +++ b/negpy/desktop/view/sidebar/scan.py @@ -355,6 +355,8 @@ def _update_device_caps(self) -> None: 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") + self.autofocus_check.setChecked(caps.autofocus) + self.autofocus_check.setVisible(caps.autofocus) # If no film sources, show banner if not caps.sources: @@ -596,9 +598,10 @@ def _on_scan(self) -> None: self.set_scanning(False) self.status_label.setText(f"Scanner busy: {e}") - @pyqtSlot(float) - def _on_scan_progress(self, progress: float) -> None: + @pyqtSlot(float, str) + def _on_scan_progress(self, progress: float, phase_name: str = 'Scanning') -> None: self.progress_bar.setVisible(True) + self.progress_bar.setFormat(f"{phase_name}… %p%") self.progress_bar.setValue(int(progress * 100)) @pyqtSlot(str) diff --git a/negpy/desktop/workers/scan_worker.py b/negpy/desktop/workers/scan_worker.py index 7a60cbee..7840cc03 100644 --- a/negpy/desktop/workers/scan_worker.py +++ b/negpy/desktop/workers/scan_worker.py @@ -53,7 +53,7 @@ class ScanWorker(QObject): """Background worker for scanner operations. Mirrors RenderWorker pattern.""" devices_ready = pyqtSignal(list) # list[ScannerDevice] - progress = pyqtSignal(float) # 0.0..1.0 + progress = pyqtSignal(float, str) # 0.0..1.0, phase name finished = pyqtSignal(str) # output rgb file path frame_done = pyqtSignal(int, str) # batch: frame number, rgb file path batch_finished = pyqtSignal(list) # batch: all written rgb paths (also on stop/error) @@ -128,7 +128,9 @@ def run_scan(self, req: ScanRequest) -> None: result = service.run_scan( device_id=req.device_id, params=req.params, - progress=self.progress.emit, + # A one-phase backend calls progress(fraction), which a + # two-argument signal's emit rejects on its own. + progress=lambda fraction, phase="Scanning": self.progress.emit(fraction, phase), cancel=self._cancel_event, ) except Exception as error: @@ -204,8 +206,8 @@ def run_batch(self, req: BatchRequest) -> None: 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) + def _progress(fraction: float, phase: str = "Scanning", _base: float = base) -> None: + self.progress.emit(_base + min(1.0, max(0.0, fraction)) / total, phase) try: result = service.run_scan(req.device_id, frame_params, _progress, self._cancel_event) diff --git a/negpy/infrastructure/scanners/base.py b/negpy/infrastructure/scanners/base.py index fbea4051..8ab2a172 100644 --- a/negpy/infrastructure/scanners/base.py +++ b/negpy/infrastructure/scanners/base.py @@ -28,6 +28,7 @@ class ScannerCapabilities: supported_depths: tuple[int, ...] sources: tuple[ScanMode, ...] max_area_mm: tuple[float, float] # (width, height) + autofocus: bool = True auto_exposure: bool = False adapter_frame_capacity: int | None = None # transport capacity bound, not an exposure count adapter_frame_control: bool = False @@ -55,7 +56,7 @@ class ScannerSession(Protocol): def scan( self, params: ScanParams, - progress: Callable[[float], None], + progress: Callable[[float, str], None], cancel: threading.Event, ) -> ScanResult: ... def eject(self) -> bool: ... @@ -73,6 +74,11 @@ class ScannerBackend(Protocol): scanner with no selectable source must still populate it or it never appears. - `scan` raises `TransientScanError` for retryable transport failures and a plain exception for everything else — that choice is the backend's alone. + - `scan` reports progress as `progress(fraction)`, or `progress(fraction, phase)` + when it has more than one phase to distinguish. The fraction is relative to + the phase, not the scan, so a backend reporting several rewinds to 0.0 at each + one and the label is what makes that legible. A caller supplying `progress` + must therefore accept the phase as optional, defaulting it to "Scanning". - `eject` returns False for a device with no eject action; it raises only when a present eject genuinely fails. - The constructor raises `ScannerUnavailable` when the driver is missing, with an @@ -88,7 +94,7 @@ def scan( self, device_id: str, params: ScanParams, - progress: Callable[[float], None], + progress: Callable[[float, str], None], cancel: threading.Event, ) -> ScanResult: ... def open_session(self, device_id: str) -> ScannerSession: ... diff --git a/negpy/infrastructure/scanners/per_frame_roll.py b/negpy/infrastructure/scanners/per_frame_roll.py index 16c76530..4298ef0a 100644 --- a/negpy/infrastructure/scanners/per_frame_roll.py +++ b/negpy/infrastructure/scanners/per_frame_roll.py @@ -57,7 +57,7 @@ def preview(self, slots: Iterable[int], *, cancel: threading.Event) -> Iterator[ frame=slot, ) try: - result = self._backend.scan(self._device.id, params, lambda _fraction: None, cancel) + result = self._backend.scan(self._device.id, params, lambda _fraction, _phase="": None, cancel) except Exception as error: if cancel.is_set(): return diff --git a/negpy/infrastructure/scanners/pieusb_backend.py b/negpy/infrastructure/scanners/pieusb_backend.py new file mode 100644 index 00000000..805164c3 --- /dev/null +++ b/negpy/infrastructure/scanners/pieusb_backend.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + + +from negpy.infrastructure.scanners.base import ( + ScannerCapabilities, + ScannerDevice, + ScannerSession, + ScannerUnavailable, + TransientScanError, +) +from negpy.infrastructure.scanners.params import ScanParams, ScanMode +from negpy.infrastructure.scanners.result import ScanResult + +if TYPE_CHECKING: + from pieusb.types import DeviceInfo + +from collections.abc import Callable + +import errno +import threading + +# libusb failures worth another attempt: a fresh `with Scanner(dev)` re-claims the +# interface, which clears a stall or a busy handle. pyusb maps these from +# LIBUSB_ERROR_BUSY/TIMEOUT/PIPE (usb.backend.libusb1._libusb_errno). +_RETRYABLE_USB_ERRNOS = frozenset({errno.EBUSY, errno.ETIMEDOUT, errno.EPIPE}) + +# ...and the ones a retry cannot help, because _devices_map still holds the +# vanished device and every attempt would reuse a dead handle. +_GONE_USB_ERRNOS = frozenset({errno.ENODEV, errno.ENOENT}) + + +def _require_pieusb() -> None: + try: + # An actual import, not find_spec: a resolvable spec still fails to load + # if pyusb or libusb_package's bundled library is missing or ABI-broken, + # which is the failure this is here to catch. + import pieusb # noqa: F401 + except ImportError: + raise ScannerUnavailable("pieusb not importable. Install: uv sync --group pieusb") from None + + +def _as_scan_error(exc: BaseException, params: ScanParams) -> Exception: + """Re-type a failed scan so ScannerService can decide on type alone. + + `TransientScanError` earns a retry (ScannerService gives it three attempts + with a settle delay); anything else fails fast with a message the sidebar + shows verbatim, so each one has to say what the user can do about it. + + pieusb reports failures by exception type, unlike SANE's message markers in + `sane_backend._as_scan_error` — with one exception, noted below. Note that + pyusb's USBError is NOT wrapped by pieusb's transport layer, so it arrives + raw and must be handled here. + """ + import usb.core + from pieusb.exceptions import ( + CheckCondition, + DeviceNotReady, + PieusbError, + Timeout, + TransportError, + WarmingUp, + ) + + # --- retryable --------------------------------------------------------- + if isinstance(exc, (TransportError, Timeout)): + # TransportError already reset the transport before raising, so the + # device is deliberately left in a state a fresh open can use. + return TransientScanError(f"Scanner transport failure: {exc}") + if isinstance(exc, WarmingUp): + # Retryable by nature, but pieusb has already waited out its own budget + # (~150s at START SCAN), so each further attempt costs that again. The + # WARMING_UP phase reaches the progress bar, so the wait is at least visible. + return TransientScanError(f"Scanner lamp is still warming up: {exc}") + if isinstance(exc, usb.core.USBError) and getattr(exc, "errno", None) in _RETRYABLE_USB_ERRNOS: + return TransientScanError(f"USB error during the scan: {exc}") + + # --- fatal, most specific first ---------------------------------------- + if isinstance(exc, usb.core.USBError): + if getattr(exc, "errno", None) in _GONE_USB_ERRNOS: + return RuntimeError(f"Lost contact with the scanner mid-scan; refresh the device list and try again ({exc})") + # No errno, or one not worth a claim either way: say what happened and + # stop, rather than guessing at a cause the message cannot support. + return RuntimeError(f"USB error during the scan: {exc}") + if isinstance(exc, MemoryError): + return RuntimeError( + f"Not enough memory to hold a {params.dpi} dpi {params.depth}-bit scan" + f"{' with infrared' if params.capture_ir else ''}. Lower the resolution, " + f"scan 8-bit, or select a smaller area." + ) + if isinstance(exc, DeviceNotReady): + # WarmingUp, its one retryable subclass, was taken above. + return RuntimeError(f"The scanner is not ready to scan — check the holder and the cover ({exc})") + if isinstance(exc, CheckCondition): + # str() carries the sense key/code/qualifier, which is the only thing that + # makes an unrecognised refusal diagnosable after the fact. + return RuntimeError(f"The scanner refused a command: {exc}") + if isinstance(exc, PieusbError): + # The one message match, because pieusb has no distinct type for it and it + # is the likeliest failure with a user action attached. Worth an upstream + # exception type; until then, matching beats showing raw geometry. + if "empty image" in str(exc): + return RuntimeError("The scanner reported an empty image — check that film is loaded and the area is in range") + return RuntimeError(f"Scan failed: {exc}") + return RuntimeError(f"Scan failed: {exc}") + + +class PieusbSession: + device_id: str + + + def __init__(self, backend: PieusbBackend) -> None: + raise NotImplementedError('PieusbSession not yet implemented') + + def scan( + self, + params: ScanParams, + progress: Callable[[float, str], None], + cancel: threading.Event, + ) -> ScanResult: + raise NotImplementedError("PieusbSession is a stub; use PieusbBackend.scan") + + def eject(self) -> bool: + return False + + def close(self) -> None: + self.dev.close() + + def __enter__(self) -> "ScannerSession": + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + +class PieusbBackend: + def __init__(self) -> None: + _require_pieusb() + + self._devices_cache: list[ScannerDevice] | None = None + self._devices_map: dict[str, DeviceInfo] = {} + + def list_devices(self) -> list[ScannerDevice]: + if self._devices_cache is not None: + return self._devices_cache + + return self.refresh_devices() + + def refresh_devices(self) -> list[ScannerDevice]: + from pieusb import get_devices + from pieusb.types import Filter + + self._devices_cache = None + devices = get_devices() + self._devices_map = {} + self._devices_cache = [] + for dev in devices: + native_res = dev.inquiry.max_resolution_x + max_w = dev.inquiry.max_scan_w / native_res * 25.4 + max_h = dev.inquiry.max_scan_h / native_res * 25.4 + supported_dpi = tuple([int(native_res / d) for d in [1, 2, 4, 5, 8, 10, 20]]) + supported_depths = tuple([d for d in [8, 16] if d in dev.inquiry.color_depths]) + caps = ScannerCapabilities( + ir_channel=Filter.INFRARED in dev.inquiry.filters, + supported_dpi=supported_dpi, + supported_depths=supported_depths, + sources=(ScanMode.POSITIVE,), + max_area_mm=(max_w, max_h), + auto_exposure=True, + autofocus=False, + ) + device_str = f"pieusb:{dev.dev.bus}:{dev.dev.address}" + self._devices_map[device_str] = dev + self._devices_cache.append( + ScannerDevice(id=device_str, vendor=dev.inquiry.vendor, model=dev.inquiry.model_str, capabilities=caps) + ) + + return self._devices_cache + + def scan( + self, + device_id: str, + params: ScanParams, + progress: Callable[[float, str], None], + cancel: threading.Event, + ) -> ScanResult: + from pieusb.scanner import Scanner + + if cancel.is_set(): + raise Exception("Scan was cancelled") + + dev = self._devices_map[device_id] + + with Scanner(dev) as s: + if params.capture_ir: + s.mode = "rgbi" + else: + s.mode = "rgb" + + s.color_depth = params.depth + s.resolution = params.dpi + s.auto_exp = params.auto_exposure + + if params.window is not None: + tl_x, tl_y, br_x, br_y = params.window + tl_x *= dev.inquiry.max_scan_w + br_x *= dev.inquiry.max_scan_w + tl_y *= dev.inquiry.max_scan_h + br_y *= dev.inquiry.max_scan_h + s.tl_x = int(tl_x) + s.tl_y = int(tl_y) + s.br_x = int(br_x) + s.br_y = int(br_y) + + result = None + scan_error = None + scan_cancelled = False + + def on_update(update): + progress(update.progress, update.phase) + + def on_complete(scan_result): + nonlocal result, scan_error, scan_cancelled + scan_error = scan_result.error + scan_cancelled = scan_result.cancelled + result = ScanResult( + rgb=scan_result.rgb, + ir=scan_result.ir, + dpi=params.dpi, + device_model=dev.inquiry.model_str + ) + + s.scan(on_update, on_complete) + + done = False + while not done: + if cancel.is_set(): + s.cancel() + raise Exception("Scan was cancelled") + done = s.wait(0.2) + + # A pieusb scan reports exactly one outcome, and the three below are + # not interchangeable: only `error` may be retried, and a cancelled + # scan is not a failure at all. + if scan_cancelled: + # The worker noticed the cancel at a chunk boundary and stopped the + # device itself, so the poll above saw it finish rather than cancel. + raise Exception("Scan was cancelled") + if scan_error is not None: + raise _as_scan_error(scan_error, params) from scan_error + if result is None: + # on_complete never ran: the worker thread died on something that + # is not an Exception, so the device state is unknown, not failed. + raise RuntimeError("The scan worker did not report an outcome") + if result.rgb is None: + raise RuntimeError("The scan completed but returned no image data") + + return result + + def open_session(self, device_id: str) -> ScannerSession: + raise NotImplementedError("open_session not yet implemented in PieusbBackend") + + def eject(self, device_id: str) -> bool: + return False diff --git a/negpy/infrastructure/scanners/registry.py b/negpy/infrastructure/scanners/registry.py index d06d44fd..609fbaf1 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_pieusb() -> ScannerBackend: + from negpy.infrastructure.scanners.pieusb_backend import PieusbBackend + + return PieusbBackend() + + 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), + "pieusb": ("PIEUSB", _make_pieusb), } diff --git a/negpy/infrastructure/scanners/sane_backend.py b/negpy/infrastructure/scanners/sane_backend.py index c631f6f0..4eaa3c4d 100644 --- a/negpy/infrastructure/scanners/sane_backend.py +++ b/negpy/infrastructure/scanners/sane_backend.py @@ -606,10 +606,16 @@ def __exit__(self, *_exc: object) -> None: def scan( self, params: ScanParams, - progress: Callable[[float], None], + progress: Callable[..., None], cancel: threading.Event, ) -> ScanResult: - """Scan one frame on the held handle. Blocks until complete or cancelled.""" + """Scan one frame on the held handle. Blocks until complete or cancelled. + + `progress` is `...` rather than `[[float], None]` because SANE reports one + phase and so calls `progress(fraction)`, while ScannerSession declares the + `(fraction, phase)` form callers must accept; only `...` is assignable to + both. See ScannerBackend's progress obligation. + """ if self.closed: 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) @@ -811,7 +817,7 @@ def scan( self, device_id: str, params: ScanParams, - progress: Callable[[float], None], + progress: Callable[..., None], cancel: threading.Event, ) -> ScanResult: """Execute a one-shot scan via SANE (open, scan, close). Blocks until complete or cancelled.""" @@ -842,7 +848,7 @@ def _scan_on_device( dev, device_id: str, params: ScanParams, - progress: Callable[[float], None], + progress: Callable[..., None], cancel: threading.Event, ) -> ScanResult: """Scan one frame on an already-open handle. sane_cancel()s the frame when diff --git a/pyproject.toml b/pyproject.toml index 368661ee..3e72d569 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'", ] +pieusb = [ + "pieusb>=0.3.2", +] [project.urls] Homepage = "https://github.com/marcinz606/NegPy" diff --git a/tests/scanners/test_pieusb_autoexposure.py b/tests/scanners/test_pieusb_autoexposure.py new file mode 100644 index 00000000..86d8f30b --- /dev/null +++ b/tests/scanners/test_pieusb_autoexposure.py @@ -0,0 +1,392 @@ +"""Parity tests for pieusb's auto-exposure against the SANE backend it ports. + +The reference is sane-backends' pieusb: getGain/getGainSetting +(pieusb_specific.c:2420, 2440), updateGain2 (2528), the dg derivation in +sanei_pieusb_set_gain_offset's "from preview" branch (1912) and the 1%/99% loop +in sanei_pieusb_analyze_preview (2394). The expected numbers below are the C's, +not this implementation's. +""" + +import numpy +import pytest + +pieusb_calibration = pytest.importorskip("pieusb.calibration") + +GAINS = pieusb_calibration.GAINS +gain_increase = pieusb_calibration.gain_increase +get_gain = pieusb_calibration.get_gain +get_gain_setting = pieusb_calibration.get_gain_setting +percentile_bounds = pieusb_calibration.percentile_bounds +update_gain = pieusb_calibration.update_gain + + +def test_gain_table_matches_the_firmware_values(): + assert len(GAINS) == 13 + assert GAINS[0] == 1.000 + assert GAINS[12] == 4.627 + + +@pytest.mark.parametrize( + "setting,expected", + [ + (0, 1.000), + (5, 1.075), + (19, 1.3398), # the default gain + (30, 1.653), + (59, 4.4292), # last setting before the extrapolated branch + (60, 4.627), + (63, 5.2204), # extrapolated past the table + ], +) +def test_get_gain_interpolates_the_table(setting, expected): + assert get_gain(setting) == pytest.approx(expected, abs=1e-4) + + +def test_get_gain_clamps_below_zero(): + assert get_gain(-5) == GAINS[0] + + +def test_get_gain_setting_inverts_get_gain(): + # 60-62 are excluded: getGain extrapolates from (setting - 55) while + # getGainSetting's matching branch starts at 60, so the C's own pair does not + # round-trip there. Reproducing that skew is deliberate. + for setting in list(range(60)) + [63]: + assert get_gain_setting(get_gain(setting)) == setting + + +def test_get_gain_setting_is_bounded(): + assert get_gain_setting(0.5) == 0 + assert get_gain_setting(1.0) == 0 + assert get_gain_setting(1000.0) == 63 + + +def test_update_gain_splits_the_boost_between_gain_and_exposure(): + # Defaults (gain 19, exposure 2937) with the maximum dg the C allows. + assert update_gain(19, 2937, 3.0) == (43, 5087) + + +@pytest.mark.parametrize("dg", [0.5, 1.0, 1.5, 2.0, 2.5, 3.0]) +def test_update_gain_delivers_exactly_dg(dg): + """Gain takes sqrt(dg); exposure takes whatever quantisation left over.""" + setting, exposure = update_gain(19, 2937, dg) + achieved = (get_gain(setting) / get_gain(19)) * (exposure / 2937) + assert achieved == pytest.approx(dg, rel=2e-3) + + +def test_update_gain_can_reduce(): + setting, exposure = update_gain(40, 2937, 0.5) + assert setting < 40 + assert exposure < 2937 + + +def test_gain_increase_takes_the_smallest_channel_ratio(): + # Green saturates lowest, so green sets the ceiling for all three. + dg = gain_increase((100, 200, 100), (58981, 52428, 58981)) + expected = (52428 / 65536) / (200 / 256) + assert dg == pytest.approx(expected) + + +def test_gain_increase_is_capped_at_three(): + assert gain_increase((5, 5, 5), (58981, 52428, 58981)) == 3.00 + + +def test_gain_increase_can_ask_to_pull_back(): + # A preview already brighter than the saturation reference wants dg < 1. + assert gain_increase((250, 250, 250), (32768, 32768, 32768)) < 1.0 + + +def test_gain_increase_skips_empty_channels(): + dg = gain_increase((0, 200, 0), (58981, 52428, 58981)) + assert dg == pytest.approx((52428 / 65536) / (200 / 256)) + + +def test_gain_increase_is_a_noop_when_nothing_was_measured(): + assert gain_increase((0, 0, 0), (58981, 52428, 58981)) == 1.0 + + +def test_percentile_bounds_shifts_16_bit_into_256_bins(): + # 40000 >> 8 = 156, and the bound is the last bin still under 99% (see + # test_percentile_bounds_is_the_last_bin_below_the_threshold), so 155. + plane = numpy.full((10, 10), 40000, dtype="> 8) - 1 + + +def test_percentile_bounds_bins_8_bit_directly(): + # Divergence from the C, which would shift 8-bit samples into bin 0. + plane = numpy.full((10, 10), 200, dtype="u1") + _lower, upper = percentile_bounds(plane) + assert upper == 199 + + +def test_percentile_bounds_ignores_the_top_one_percent(): + # 0.5% specular highlights at full scale must not drag the bound up to 255. + plane = numpy.zeros(1000, dtype="= 0.99 at bin 10, so the C's loop stops before it: bound 9, not 10. + plane = numpy.zeros((100, 100), dtype=" None: + assert isinstance(_as_scan_error(exc, _PARAMS), TransientScanError) + + +@pytest.mark.parametrize( + "exc", + [ + pytest.param(_pieusb_exc("DeviceNotReady", "not ready"), id="not-ready"), + pytest.param(_pieusb_exc("CheckCondition", 0x05, 0x20, 0x00), id="check-condition"), + pytest.param(_pieusb_exc("PieusbError", "GET SHADING PARMS returned no entries"), id="protocol"), + pytest.param(_pieusb_exc("ParamError", "sharpen and fast_infrared are exclusive"), id="bad-options"), + pytest.param(MemoryError(), id="out-of-memory"), + pytest.param(ValueError("Invalid value provided to option 'resolution'"), id="unexpected"), + ], +) +def test_real_errors_fail_fast(exc: BaseException) -> None: + """A retry would cost another full scan and end the same way.""" + reported = _as_scan_error(exc, _PARAMS) + assert isinstance(reported, Exception) + assert not isinstance(reported, TransientScanError) + + +def test_a_stalled_interface_is_transient_but_a_vanished_device_is_not() -> None: + """Both are USBError, which pieusb's transport does not wrap — errno decides. + + A retry re-opens the device, which clears a stall; it cannot help once the + device is gone, because _devices_map still holds the dead handle. + """ + import errno + + import usb.core + + # pyusb's third argument is the errno; the second is libusb's own code. + stalled = usb.core.USBError("Resource busy", -6, errno.EBUSY) + vanished = usb.core.USBError("No such device", -4, errno.ENODEV) + + assert isinstance(_as_scan_error(stalled, _PARAMS), TransientScanError) + vanished_report = _as_scan_error(vanished, _PARAMS) + assert not isinstance(vanished_report, TransientScanError) + assert "refresh" in str(vanished_report) + + +def test_a_usb_error_without_an_errno_makes_no_claim_about_the_cause() -> None: + """pieusb raises some USBErrors with no errno; the message must not overreach.""" + import usb.core + + reported = _as_scan_error(usb.core.USBError("something went wrong"), _PARAMS) + + assert not isinstance(reported, TransientScanError) + assert "refresh" not in str(reported) + assert "something went wrong" in str(reported) + + +def test_the_sense_data_survives_into_the_message() -> None: + """An unrecognised refusal is only diagnosable from its sense triple.""" + reported = _as_scan_error(_pieusb_exc("CheckCondition", 0x05, 0x20, 0x00), _PARAMS) + assert "0x05" in str(reported) and "0x20" in str(reported) + + +def test_out_of_memory_names_what_to_change() -> None: + """MemoryError alone tells the user nothing they can act on.""" + reported = _as_scan_error(MemoryError(), _PARAMS) + assert "5000" in str(reported) and "16" in str(reported) + + +def test_an_empty_image_is_reported_as_missing_film() -> None: + exc = _pieusb_exc("PieusbError", "GET PARAMETERS reported an empty image (0x0)") + assert "film" in str(_as_scan_error(exc, _PARAMS)).lower() + + +# ── outcome triage ──────────────────────────────────────────────────────── + + +class _FakeScanner: + """Enough Scanner for PieusbBackend.scan: options are plain attributes.""" + + def __init__(self, outcome) -> None: + self._outcome = outcome + + def __enter__(self) -> "_FakeScanner": + return self + + def __exit__(self, *exc: object) -> None: + return None + + def scan(self, on_update, on_complete) -> None: + from pieusb.types import ScanPhase, UpdateData + + on_update(UpdateData(phase=ScanPhase.SCANNING, progress=0.5)) + if self._outcome is not None: + on_complete(self._outcome) + + def wait(self, timeout=None) -> bool: + return True + + def cancel(self) -> None: + return None + + +class _FakeInquiry: + max_scan_w = 10000 + max_scan_h = 10000 + model_str = "ProScan 10T" + + +class _FakeInfo: + inquiry = _FakeInquiry() + + +def _backend_with(monkeypatch: pytest.MonkeyPatch, outcome) -> PieusbBackend: + monkeypatch.setattr("pieusb.scanner.Scanner", lambda info: _FakeScanner(outcome)) + monkeypatch.setattr(pieusb_backend, "_require_pieusb", lambda: None) + backend = PieusbBackend() + backend._devices_map = {"pieusb:1:2": _FakeInfo()} # type: ignore[dict-item] + return backend + + +def _scan(backend: PieusbBackend, progress=None): + return backend.scan("pieusb:1:2", _PARAMS, progress or (lambda *_: None), threading.Event()) + + +def _pieusb_result(**kwargs): + from pieusb.types import ScanResult + + return ScanResult(**kwargs) + + +def test_a_clean_scan_returns_the_image(monkeypatch: pytest.MonkeyPatch) -> None: + rgb = np.ones((4, 3, 3), dtype=np.uint16) + backend = _backend_with(monkeypatch, _pieusb_result(rgb=rgb, width=3, height=4)) + seen: list[tuple[float, str]] = [] + + result = _scan(backend, progress=lambda fraction, phase: seen.append((fraction, phase))) + + assert result.rgb.shape == (4, 3, 3) + assert result.dpi == _PARAMS.dpi + assert result.device_model == "ProScan 10T" + assert seen == [(0.5, "Scanning")] + + +def test_a_transient_failure_reaches_the_caller_typed(monkeypatch: pytest.MonkeyPatch) -> None: + """Without this the service cannot retry: the type is its only signal.""" + error = _pieusb_exc("TransportError", "USB status 0x08") + backend = _backend_with(monkeypatch, _pieusb_result(rgb=None, error=error)) + + with pytest.raises(TransientScanError) as excinfo: + _scan(backend) + assert excinfo.value.__cause__ is error # the traceback survives for the log + + +def test_a_real_failure_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None: + error = _pieusb_exc("CheckCondition", 0x05, 0x20, 0x00) + backend = _backend_with(monkeypatch, _pieusb_result(rgb=None, error=error)) + + with pytest.raises(Exception) as excinfo: + _scan(backend) + assert not isinstance(excinfo.value, TransientScanError) + assert excinfo.value.__cause__ is error + + +def test_a_scan_cancelled_by_the_worker_reports_cancellation(monkeypatch: pytest.MonkeyPatch) -> None: + """pieusb stops at a chunk boundary, so the poll sees it finish, not cancel.""" + backend = _backend_with(monkeypatch, _pieusb_result(rgb=None, cancelled=True)) + + with pytest.raises(Exception, match="[Cc]ancel") as excinfo: + _scan(backend) + assert not isinstance(excinfo.value, TransientScanError) + + +def test_a_missing_outcome_is_not_disguised_as_a_failed_scan(monkeypatch: pytest.MonkeyPatch) -> None: + """on_complete never ran, so the device state is unknown rather than failed.""" + backend = _backend_with(monkeypatch, None) + + with pytest.raises(RuntimeError, match="did not report an outcome"): + _scan(backend) + + +def test_an_empty_result_without_an_error_is_still_a_failure(monkeypatch: pytest.MonkeyPatch) -> None: + backend = _backend_with(monkeypatch, _pieusb_result(rgb=None)) + + with pytest.raises(RuntimeError, match="no image data"): + _scan(backend) diff --git a/tests/test_scan_sidebar.py b/tests/test_scan_sidebar.py index 949f6b94..3f1fcb85 100644 --- a/tests/test_scan_sidebar.py +++ b/tests/test_scan_sidebar.py @@ -81,7 +81,7 @@ def save_global_setting(self, key: str, value) -> None: class _FakeController(QObject): scan_devices_ready = pyqtSignal(list) - scan_progress = pyqtSignal(float) + scan_progress = pyqtSignal(float, str) # progress, phase name scan_finished = pyqtSignal(str) scan_error = pyqtSignal(str) scan_cancelled = pyqtSignal() diff --git a/tests/test_scan_worker.py b/tests/test_scan_worker.py index 4c9dea5f..d30621c1 100644 --- a/tests/test_scan_worker.py +++ b/tests/test_scan_worker.py @@ -210,6 +210,29 @@ def test_scan_worker_reports_eject_failure() -> None: assert errors == ["transport refused"] +def test_progress_without_a_phase_is_reported_as_scanning() -> None: + """A backend with one phase calls progress(fraction); the signal needs both. + + SANE does exactly this, so passing `self.progress.emit` straight to the + backend would raise on every update. + """ + + class _ProgressService(_ScanService): + def run_scan(self, *, device_id, params, progress, cancel): + progress(0.25) # no phase — the SANE shape + progress(0.5, "Calibrating") # a backend that names its phases + return object() + + worker = ScanWorker() + worker._service = _ProgressService() # type: ignore[assignment] + seen: list[tuple[float, str]] = [] + worker.progress.connect(lambda fraction, phase: seen.append((fraction, phase))) + + worker.run_scan(_scan_request()) + + assert seen == [(0.25, "Scanning"), (0.5, "Calibrating")] + + def test_scan_worker_emits_cancelled_when_acquisition_returns_after_cancel() -> None: worker = ScanWorker() service = _ScanService(cancel_during_acquisition=True) diff --git a/uv.lock b/uv.lock index cfbb8304..6d50b871 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.13" [[package]] @@ -185,6 +185,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/49/fa/391e437a34e55095173dca5f24070d89cbc233ff85bf1c29c93248c6588d/imageio-2.37.3-py3-none-any.whl", hash = "sha256:46f5bb8522cd421c0f5ae104d8268f569d856b29eb1a13b92829d1970f32c9f0", size = 317646, upload-time = "2026-03-09T11:31:10.771Z" }, ] +[[package]] +name = "importlib-resources" +version = "7.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e4/06/b56dfa750b44e86157093bc8fca0ab81dccbf5260510de4eaf1cb69b5b99/importlib_resources-7.1.0.tar.gz", hash = "sha256:0722d4c6212489c530f2a145a34c0a7a3b4721bc96a15fada5930e2a0b760708", size = 44985, upload-time = "2026-04-12T16:36:09.232Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8a/db/55a262f3606bebcae07cc14095338471ad7c0bbcaa37707e6f0ee49725b7/importlib_resources-7.1.0-py3-none-any.whl", hash = "sha256:1bd7b48b4088eddb2cd16382150bb515af0bd2c70128194392725f82ad2c96a1", size = 37232, upload-time = "2026-04-12T16:36:08.219Z" }, +] + [[package]] name = "iniconfig" version = "2.3.0" @@ -206,6 +215,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" }, ] +[[package]] +name = "libusb-package" +version = "1.0.30.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "importlib-resources" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/42/a8/8b3d5dae7340880d556f9f866874b9674b2e3a22fee1e2b96f1ab0feae36/libusb_package-1.0.30.0-py3-none-macosx_10_9_x86_64.whl", hash = "sha256:2b98784bac3bedda7e95bb5039dd0eded84b02f36110e4d5d607375366c61090", size = 69516, upload-time = "2026-06-30T07:41:02.451Z" }, + { url = "https://files.pythonhosted.org/packages/52/6f/26de4e9f858ab50e87931f0be268f3c1bbfce33e8584add60da857632142/libusb_package-1.0.30.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:4ad25f8d254bbbdd234446d5580b24d62d59bb0e690d8090457dac347f044437", size = 65700, upload-time = "2026-06-30T07:41:03.557Z" }, + { url = "https://files.pythonhosted.org/packages/11/12/9ba8fa91dc95b1cbfa4a68207d4048b08b900fd3686fa72b99846608de01/libusb_package-1.0.30.0-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:c71b5f8c65b286425c02c7d624eee1fd7f08bfb1dfa492ddc20e27f06fee3829", size = 76166, upload-time = "2026-06-30T07:41:04.462Z" }, + { url = "https://files.pythonhosted.org/packages/6f/e1/a7c83535749332825f02d6693868f8ee9ae99c4104e60a483773bb652c0e/libusb_package-1.0.30.0-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f502ad5a0527b8c0431de817662325c88a1bba2cc334173665b04ad168d7b6d3", size = 76159, upload-time = "2026-06-30T07:41:05.543Z" }, + { url = "https://files.pythonhosted.org/packages/de/56/a1fb1726fc96a8ee3bd0e04e5e505500f022dd310dd348ecebf5cdae7a60/libusb_package-1.0.30.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:7a2a1c82f6ef85d9920cbe2898aa6927d4c487744ef9f20b0d7d6d95705095bb", size = 77162, upload-time = "2026-06-30T07:41:06.581Z" }, + { url = "https://files.pythonhosted.org/packages/7b/03/264cefc51275ecb047194c4973bc40b3b278566a64bbe6e74f86bf1e0afc/libusb_package-1.0.30.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:2ee5ef8ac6f08402b4e8334f7404850316bdd26037c9a97f2a7456ee9a5d1550", size = 76676, upload-time = "2026-06-30T07:41:07.505Z" }, + { url = "https://files.pythonhosted.org/packages/e5/95/d166eeefe0d9dd5833d5724a97b023ec380c3a2d364040ec0485fd57703c/libusb_package-1.0.30.0-py3-none-win32.whl", hash = "sha256:79728146e1f01e525786900b2ccd8298a8eca53cd94b37fb0885e56aa6f9d60a", size = 78769, upload-time = "2026-06-30T07:41:08.432Z" }, + { url = "https://files.pythonhosted.org/packages/60/4a/ff49bd77f33af05ca26fee29d601beb1e19ca791bb86efece82a3833885c/libusb_package-1.0.30.0-py3-none-win_amd64.whl", hash = "sha256:90808da724c8939a333d931d0c7226372ca5fefd98e2f2f3ef79fd918aa37522", size = 90711, upload-time = "2026-06-30T07:41:09.318Z" }, +] + [[package]] name = "llvmlite" version = "0.47.0" @@ -322,6 +349,9 @@ dev = [ { name = "ruff" }, { name = "ty" }, ] +pieusb = [ + { name = "pieusb" }, +] scanner = [ { name = "python-sane" }, ] @@ -354,6 +384,7 @@ dev = [ { name = "ruff", specifier = "==0.14.10" }, { name = "ty", specifier = ">=0.0.26" }, ] +pieusb = [{ name = "pieusb", specifier = ">=0.3.2" }] scanner = [{ name = "python-sane", specifier = ">=2.9" }] [[package]] @@ -466,6 +497,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/16/12b82f791c7f50ddec566873d5bdd245baa1491bac11d15ffb98aecc8f8b/pefile-2024.8.26-py3-none-any.whl", hash = "sha256:76f8b485dcd3b1bb8166f1128d395fa3d87af26360c2358fb75b80019b957c6f", size = 74766, upload-time = "2024-08-26T21:01:02.632Z" }, ] +[[package]] +name = "pieusb" +version = "0.3.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "libusb-package" }, + { name = "numpy" }, + { name = "pyusb" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/d5/1e/70da3b080b141201deabf3353716ea4c8e645444248d230bbcc1127066f3/pieusb-0.3.2.tar.gz", hash = "sha256:aa71bc9b04cb28537e8881086fae35e6fad070621e261d1a59c10a985257ab0d", size = 60581, upload-time = "2026-08-03T12:37:46.188Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/15/c0/7196dee7f3337595b37f45a2788c69681d98fd39e12aef07a4aaf3810e24/pieusb-0.3.2-py3-none-any.whl", hash = "sha256:74c1fba821177e69d8900a3a67379bd597cbb98b7368955f494863f632012511", size = 49761, upload-time = "2026-08-03T12:37:45.106Z" }, +] + [[package]] name = "piexif" version = "1.1.3" @@ -725,6 +770,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"