From 0acd8dc9a134f1f2ad33807287fde65c1274cf56 Mon Sep 17 00:00:00 2001 From: thetalkingdrum <254868117+thetalkingdrum@users.noreply.github.com> Date: Sun, 11 Oct 2026 17:36:51 +0200 Subject: [PATCH 1/2] fix(scan): keep Windows awake during a scan batch A scan batch runs for many minutes without keyboard or mouse input, so Windows idle sleep can suspend the PC in the middle of a pass. The pass then ends at the wake with the unread part of the frame left as zeros, and the IR validity mask still says the whole frame is valid. Four frames of one session were cut off this way: each was saved within two seconds of a System Idle wake, at 0.7%, 26%, 58% and 66% of the frame. power_assertion.py held a real assertion on macOS only. On Windows the assertion is now the calling thread's execution state with ES_SYSTEM_REQUIRED, so the display may still turn off. run_batch acquires and releases it on one thread, as the state requires. Linux still gets an inert assertion. Co-Authored-By: Claude Sonnet 5.5 --- docs/USER_GUIDE.md | 1 + negpy/desktop/power_assertion.py | 57 ++++++++++++++++++++++++--- tests/test_power_assertion.py | 66 ++++++++++++++++++++++++++++++-- 3 files changed, 116 insertions(+), 8 deletions(-) diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index f72123be8..14967fcb1 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -1053,6 +1053,7 @@ Capture film directly into NegPy. The **Scanner** card picks the scanner; the ta * **Device**: the scanner; the arrows refresh the list. The eject menu holds **Eject Now** and, on a strip feeder, **Eject When Done** (on by default), which returns the strip after a batch. Off keeps it loaded with its picks and previews. A strip the scanner returns by itself counts as an eject: insert it again, then **Detect frames**. After 9 minutes with no scanner activity NegPy measures the strip again. * **Debug log** (nkscan; Off, Debug, Trace): writes `nkscan.log` in the NegPy folder. For a bug report, set Trace, reproduce the problem and attach the file. +* **Sleep**: during a scan batch, NegPy keeps macOS and Windows from going to sleep. On Linux, turn off automatic suspend before a long roll. **pyOpticfilm (Plustek)**: the **OpticFilm 8200i SE** (`07b3:1825`) and **8100 V2** (`07b3:1824`) scan; for other models try **SANE** on Linux and macOS. **IR** comes in the same pass. On Windows, bind the device to **WinUSB** with Zadig first ([PLUSTEK_WINDOWS.md](PLUSTEK_WINDOWS.md)). From source, install with `uv sync --group plustek` or `pip install negpy[plustek]`. diff --git a/negpy/desktop/power_assertion.py b/negpy/desktop/power_assertion.py index 1eaf3425d..fbc5951c7 100644 --- a/negpy/desktop/power_assertion.py +++ b/negpy/desktop/power_assertion.py @@ -1,8 +1,9 @@ """Best-effort idle-sleep protection for unattended desktop operations. The scanner process remains responsible for its own USB lifecycle. This module -only asks macOS to keep the computer awake while a user-requested operation is -running. Other platforms deliberately receive an inert assertion. +only asks the operating system to keep the computer awake while a user-requested +operation is running. macOS and Windows receive a real assertion; other platforms +receive an inert one. """ from __future__ import annotations @@ -19,6 +20,9 @@ logger = get_logger(__name__) +_ES_CONTINUOUS = 0x80000000 +_ES_SYSTEM_REQUIRED = 0x00000001 + class UnattendedPowerAssertion(Protocol): """One idempotently releasable operating-system power assertion.""" @@ -59,6 +63,23 @@ def release(self) -> None: logger.warning("could not end the macOS NSProcessInfo activity", exc_info=True) +class _WindowsPowerAssertion: + """Windows keeps the execution state per thread: release on the thread that acquired it.""" + + def __init__(self, kernel32: Any) -> None: + self._kernel32: Any | None = kernel32 + + def release(self) -> None: + kernel32 = self._kernel32 + if kernel32 is None: + return + self._kernel32 = None + try: + kernel32.SetThreadExecutionState(_ES_CONTINUOUS) + except Exception: + logger.warning("could not clear the Windows execution state", exc_info=True) + + class _CaffeinatePowerAssertion: _WAIT_SECONDS = 2.0 @@ -91,20 +112,44 @@ def _load_foundation() -> Any: return importlib.import_module("Foundation") +def _load_kernel32() -> Any: + import ctypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.SetThreadExecutionState.argtypes = [ctypes.c_uint32] + kernel32.SetThreadExecutionState.restype = ctypes.c_uint32 + return kernel32 + + +def _acquire_windows(kernel32_loader: Callable[[], Any] | None) -> UnattendedPowerAssertion: + """Block idle system sleep for the calling thread; the display may still turn off.""" + + try: + kernel32 = (_load_kernel32 if kernel32_loader is None else kernel32_loader)() + if not kernel32.SetThreadExecutionState(_ES_CONTINUOUS | _ES_SYSTEM_REQUIRED): + raise OSError("SetThreadExecutionState refused the request") + return _WindowsPowerAssertion(kernel32) + except Exception: + logger.warning("could not create a Windows idle-sleep assertion", exc_info=True) + return _NoOpPowerAssertion() + + def acquire_unattended_power_assertion( reason: str, *, _platform: str | None = None, _foundation_loader: Callable[[], Any] | None = None, + _kernel32_loader: Callable[[], Any] | None = None, _process_factory: Callable[..., _CaffeinateProcess] | None = None, _parent_pid: int | None = None, ) -> UnattendedPowerAssertion: - """Prevent macOS idle sleep until the returned assertion is released. + """Prevent idle sleep until the returned assertion is released. - ``NSProcessInfo`` is preferred because its activity is owned by this - process. PyObjC is optional, so packaged builds fall back to + On macOS, ``NSProcessInfo`` is preferred because its activity is owned by + this process. PyObjC is optional, so packaged builds fall back to ``/usr/bin/caffeinate -i -w ``. The ``-w`` binding makes the helper release its assertion even if NegPy exits without normal cleanup. + On Windows the assertion belongs to the calling thread and ends with it. Backend failures are logged and degrade to a no-op instead of blocking a scan. """ @@ -113,6 +158,8 @@ def acquire_unattended_power_assertion( if not reason: raise ValueError("power assertion reason must not be empty") platform = sys.platform if _platform is None else _platform + if platform == "win32": + return _acquire_windows(_kernel32_loader) if platform != "darwin": return _NoOpPowerAssertion() diff --git a/tests/test_power_assertion.py b/tests/test_power_assertion.py index e57ee6ee8..89d31933d 100644 --- a/tests/test_power_assertion.py +++ b/tests/test_power_assertion.py @@ -54,6 +54,19 @@ def wait(self, timeout: float | None = None) -> int: return self.returncode +class _Kernel32: + def __init__(self, *, result: int = 0x80000000, clear_fails: bool = False) -> None: + self.calls: list[int] = [] + self.result = result + self.clear_fails = clear_fails + + def SetThreadExecutionState(self, flags: int) -> int: # noqa: N802 + self.calls.append(flags) + if self.clear_fails and flags == 0x80000000: + raise OSError("clear failed") + return self.result + + def _foundation(process_info: _ProcessInfo) -> Any: return SimpleNamespace( NSActivityUserInitiated=0x01, @@ -62,20 +75,67 @@ def _foundation(process_info: _ProcessInfo) -> Any: ) -def test_non_macos_assertion_is_an_idempotent_noop() -> None: +def test_unsupported_platform_assertion_is_an_idempotent_noop() -> None: def fail_loader() -> Any: - raise AssertionError("non-macOS must not load Foundation") + raise AssertionError("an unsupported platform must not load a backend") assertion = acquire_unattended_power_assertion( "test scan", _platform="linux", _foundation_loader=fail_loader, - _process_factory=lambda *_args, **_kwargs: pytest.fail("non-macOS must not run caffeinate"), + _kernel32_loader=fail_loader, + _process_factory=lambda *_args, **_kwargs: pytest.fail("an unsupported platform must not run a helper"), + ) + + assertion.release() + assertion.release() + + +def test_windows_blocks_idle_system_sleep_and_clears_it_once() -> None: + kernel32 = _Kernel32() + + assertion = acquire_unattended_power_assertion( + "NegPy scan batch", + _platform="win32", + _foundation_loader=lambda: pytest.fail("Windows must not load Foundation"), + _kernel32_loader=lambda: kernel32, + _process_factory=lambda *_args, **_kwargs: pytest.fail("Windows must not run caffeinate"), + ) + + assert kernel32.calls == [0x80000001] + assertion.release() + assertion.release() + assert kernel32.calls == [0x80000001, 0x80000000] + + +def test_windows_refused_request_degrades_to_noop() -> None: + kernel32 = _Kernel32(result=0) + + assertion = acquire_unattended_power_assertion("NegPy scan batch", _platform="win32", _kernel32_loader=lambda: kernel32) + + assertion.release() + assert kernel32.calls == [0x80000001] + + +def test_windows_backend_failures_degrade_to_noop() -> None: + assertion = acquire_unattended_power_assertion( + "NegPy scan batch", + _platform="win32", + _kernel32_loader=lambda: (_ for _ in ()).throw(OSError("no kernel32")), ) assertion.release() + + +def test_windows_release_failure_is_swallowed() -> None: + kernel32 = _Kernel32(clear_fails=True) + assertion = acquire_unattended_power_assertion("NegPy scan batch", _platform="win32", _kernel32_loader=lambda: kernel32) + + assertion.release() assertion.release() + assert kernel32.calls == [0x80000001, 0x80000000] + def test_macos_prefers_nsprocessinfo_and_releases_once() -> None: process_info = _ProcessInfo() From c34c7050c11f42d94518aed51d17e29974b8e7ec Mon Sep 17 00:00:00 2001 From: thetalkingdrum <254868117+thetalkingdrum@users.noreply.github.com> Date: Sun, 11 Oct 2026 17:38:08 +0200 Subject: [PATCH 2/2] fix(scan): hold the power assertion for a single-frame scan too A single-frame scan runs for minutes without input, so idle sleep can cut it off the same way as a batch. run_scan now acquires the assertion like run_batch and releases it in its finally block. Tests cover the single scan, its error path and the batch, which had none. Co-Authored-By: Claude Sonnet 5.5 --- docs/USER_GUIDE.md | 2 +- negpy/desktop/workers/scan_worker.py | 2 + tests/test_scan_worker.py | 61 ++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+), 1 deletion(-) diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 14967fcb1..62341813d 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -1053,7 +1053,7 @@ Capture film directly into NegPy. The **Scanner** card picks the scanner; the ta * **Device**: the scanner; the arrows refresh the list. The eject menu holds **Eject Now** and, on a strip feeder, **Eject When Done** (on by default), which returns the strip after a batch. Off keeps it loaded with its picks and previews. A strip the scanner returns by itself counts as an eject: insert it again, then **Detect frames**. After 9 minutes with no scanner activity NegPy measures the strip again. * **Debug log** (nkscan; Off, Debug, Trace): writes `nkscan.log` in the NegPy folder. For a bug report, set Trace, reproduce the problem and attach the file. -* **Sleep**: during a scan batch, NegPy keeps macOS and Windows from going to sleep. On Linux, turn off automatic suspend before a long roll. +* **Sleep**: while a scan runs, NegPy keeps macOS and Windows from going to sleep. On Linux, turn off automatic suspend before a long roll. **pyOpticfilm (Plustek)**: the **OpticFilm 8200i SE** (`07b3:1825`) and **8100 V2** (`07b3:1824`) scan; for other models try **SANE** on Linux and macOS. **IR** comes in the same pass. On Windows, bind the device to **WinUSB** with Zadig first ([PLUSTEK_WINDOWS.md](PLUSTEK_WINDOWS.md)). From source, install with `uv sync --group plustek` or `pip install negpy[plustek]`. diff --git a/negpy/desktop/workers/scan_worker.py b/negpy/desktop/workers/scan_worker.py index 2f0f0bd2a..fbae1cc1f 100644 --- a/negpy/desktop/workers/scan_worker.py +++ b/negpy/desktop/workers/scan_worker.py @@ -163,6 +163,7 @@ def run_scan(self, req: ScanRequest) -> None: self._request_prepared = False self._scanning = True + assertion = acquire_unattended_power_assertion("NegPy film scan") outcome: tuple[str, str | None] | None = None returned: StripReturned | None = None try: @@ -208,6 +209,7 @@ def run_scan(self, req: ScanRequest) -> None: logger.exception("Could not initialize scanner service") outcome = ("error", str(error)) finally: + assertion.release() with self._state_lock: self._scanning = False diff --git a/tests/test_scan_worker.py b/tests/test_scan_worker.py index a3b74256b..43d795fac 100644 --- a/tests/test_scan_worker.py +++ b/tests/test_scan_worker.py @@ -2,6 +2,7 @@ import dataclasses import threading +from typing import Any import pytest @@ -341,6 +342,66 @@ def test_cancel_during_write_does_not_hide_write_failure() -> None: assert worker._scanning is False +class _PowerAssertion: + def __init__(self, events: list[str]) -> None: + self._events = events + + def release(self) -> None: + self._events.append("release") + + +def _record_power_assertions(monkeypatch: pytest.MonkeyPatch, service: Any) -> list[str]: + """Events in order: each acquire, each scan the service runs, each release.""" + events: list[str] = [] + + def acquire(reason: str) -> _PowerAssertion: + events.append(f"acquire:{reason}") + return _PowerAssertion(events) + + monkeypatch.setattr("negpy.desktop.workers.scan_worker.acquire_unattended_power_assertion", acquire) + run_scan = service.run_scan + + def spy(*args: Any, **kwargs: Any) -> Any: + events.append("scan") + return run_scan(*args, **kwargs) + + service.run_scan = spy + return events + + +def test_run_scan_holds_a_power_assertion_for_the_whole_scan(monkeypatch: pytest.MonkeyPatch) -> None: + worker = ScanWorker() + service = _ScanService() + worker._service = service # type: ignore[assignment] + events = _record_power_assertions(monkeypatch, service) + + worker.run_scan(_scan_request()) + + assert events == ["acquire:NegPy film scan", "scan", "release"] + + +def test_run_scan_releases_the_power_assertion_after_an_error(monkeypatch: pytest.MonkeyPatch) -> None: + worker = ScanWorker() + service = _ScanService(write_error=OSError("scratch disk full")) + worker._service = service # type: ignore[assignment] + events = _record_power_assertions(monkeypatch, service) + + worker.run_scan(_scan_request()) + + assert events == ["acquire:NegPy film scan", "scan", "release"] + + +def test_run_batch_holds_a_power_assertion_for_the_whole_batch(monkeypatch: pytest.MonkeyPatch) -> None: + worker = ScanWorker() + service = _BatchService() + worker._service = service # type: ignore[assignment] + events = _record_power_assertions(monkeypatch, service) + + worker.run_batch(_batch_request((2, 3))) + + assert events == ["acquire:NegPy film scan batch", "scan", "scan", "release"] + + def test_run_batch_scans_each_frame_and_reports_paths() -> None: worker = ScanWorker() service = _BatchService()