fix(scan): keep Windows awake while a scan runs

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**: 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]`.
 

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 <parent pid>``.  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()
 

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
 

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()

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()