diff --git a/sensors/finger/assets/30fps-0703/unroll_map_uv_s0.42.npz b/sensors/finger/assets/30fps-0703/unroll_map_uv_s0.42.npz new file mode 100644 index 0000000..f4884a1 Binary files /dev/null and b/sensors/finger/assets/30fps-0703/unroll_map_uv_s0.42.npz differ diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index 17bdb29..811be66 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -11,7 +11,7 @@ __author__ = "Orisys" from . import calibration, capture, configs, finger, signal_processing, util from .export import StreamRecorder, mosaic_videos from .frame_processors import FFCFrameProcessor -from .pipeline import FrameSnapshot, PipelineStats, StreamPipeline +from .pipeline import FrameSnapshot, PauseLevel, PipelineStats, StreamPipeline from .sensor import Sensor CompoundEyeSensor = Sensor @@ -45,6 +45,7 @@ __all__ = [ 'StreamPipeline', 'FrameSnapshot', 'PipelineStats', + 'PauseLevel', # offline recording of arrow overlays 'StreamRecorder', diff --git a/src/orisys/pipeline.py b/src/orisys/pipeline.py index c3bcde4..7df075d 100644 --- a/src/orisys/pipeline.py +++ b/src/orisys/pipeline.py @@ -68,17 +68,42 @@ import threading import time from collections.abc import Callable, Iterable, Sequence from dataclasses import dataclass, field +from enum import Enum from queue import Empty, Full, Queue from typing import Any import numpy as np -__all__ = ["FrameSnapshot", "PipelineStats", "StreamPipeline"] +__all__ = ["FrameSnapshot", "PipelineStats", "PauseLevel", "StreamPipeline"] #: Array fields a snapshot can carry. ``image`` is the working grey frame; the #: rest come from ``Sensor.read_info`` (already copied by the SDK). _ARRAY_FIELDS = ("image", "flow", "vnormal", "vshear", "depth", "centroid", "contour") +#: How many periods the stages wait on the pause event before re-checking for a +#: stop. Small enough to resume promptly, large enough not to spin a core. +_PAUSE_POLL_S = 0.05 + + +class PauseLevel(str, Enum): + """What a pause holds still. + + ``BOTH`` stops the pipeline read and the compute stages: nothing is captured + and nothing is computed, so the sensor reports nothing until it resumes. + + ``CAPTURE_ONLY`` keeps reading frames and skips the flow and HHD stages. The + read is still issued, so a live camera keeps streaming: its exposure, + white balance and frame rate stay where the operator set them, and resuming + does not have to re-converge. What stops is the work - the three-stage + pipeline drops to one stage of cost, which is what makes a long hold cheap. + + Both levels rebase the tracker on resume, so the hold does not turn into a + phantom deformation reading. + """ + + BOTH = "both" + CAPTURE_ONLY = "capture_only" + @dataclass(frozen=True) class FrameSnapshot: @@ -187,6 +212,13 @@ class StreamPipeline: #: Set by :meth:`reset` and consumed by the flow stage, so that resetting sensor state #: never races the stage that owns that state. self._reset_pending = threading.Event() + #: Pause state. ``_pause_level`` is what a pause would hold when armed; + #: ``_currently_capturing`` records whether the producer is still + #: reading, which decides whether a resume has to rebase the tracker. + self._pause = threading.Event() + self._pause_level: PauseLevel = PauseLevel.BOTH + self._currently_capturing = True + self._sensor_source: Any = None self._lock = threading.Lock() self._latest: FrameSnapshot | None = None @@ -278,6 +310,107 @@ class StreamPipeline: """ self._reset_pending.set() + @property + def paused(self) -> bool: + return self._pause.is_set() + + @property + def pause_level(self) -> PauseLevel: + """The level in effect while paused (the last level requested).""" + return self._pause_level + + def pause(self, level: PauseLevel | str = PauseLevel.CAPTURE_ONLY) -> None: + """Hold the pipeline still until :meth:`resume`. + + ``level`` selects what stays running: + + * :attr:`PauseLevel.BOTH` - stop reading *and* computing. The sensor + reports nothing at all while held. + * :attr:`PauseLevel.CAPTURE_ONLY` (default) - keep reading so a live + camera stays in its converged state (exposure, white balance, frame + rate), but skip the flow and HHD stages. Cost drops to the read alone. + + No snapshots are published while paused, so :meth:`wait` returns + ``None``. The tracker is rebased on resume, because a held reference + frame would otherwise report the hold itself as a large deformation - + use ``reset=False`` to resume without that rebase. + + Safe to call from the consumer thread, and idempotent; re-pausing with a + different level takes effect immediately. + """ + if not isinstance(level, PauseLevel): + level = PauseLevel(level) + if self._done.is_set(): + return + self._set_reading(level is PauseLevel.CAPTURE_ONLY) + self._pause_level = level + self._pause.set() + + def resume(self, *, reset: bool = True) -> None: + """Release a pause and continue processing. + + With ``reset`` (the default) the tracker is rebased on the next frame, + which is what keeps the hold out of the force reading. Pass + ``reset=False`` when the caller wants the pre-pause reference frame kept + - for example when the held period is known to be motion-free. + """ + if not self._pause.is_set(): + return + if reset and self._currently_capturing: + # Frames read during the pause were never computed, so the sensor's + # reference frame is stale by however long the hold lasted. Ask for a + # rebase; the flow stage applies it before its next frame. + self._reset_pending.set() + # Arm capture *before* lifting the hold. The producer stops as soon as + # the pause flag is set and only returns once it is clear, so a level-1 + # resume must have `_currently_capturing` true by then or the producer + # settles with capture mode off. + self._currently_capturing = True + self._pause.clear() + self._set_reading(True) + + def _set_reading(self, reading: bool) -> None: + """Turn the producer's capture on or off (level 1 versus level 2).""" + self._currently_capturing = bool(reading) + source = self._resolve_sensor_source() + if source is None: + return + try: + if reading: + source.start_background() + else: + source.stop_background() + except Exception: # noqa: BLE001 - a video file may have no such control + if self._verbose: + print("[pipeline] pause: could not toggle background capture", flush=True) + + def _resolve_sensor_source(self) -> Any: + """The sensor's capture source, or ``None`` for file/derived sources.""" + if self._sensor_source is not None: + return self._sensor_source + source = getattr(self.sensor, "_capture_source", None) + if source is None: + return None + # File playback cannot be held this way, and stopping a fake source only + # breaks the read, so leave those alone. + try: + if not hasattr(source, "start_background") or not source.is_open(): + return None + except Exception: # noqa: BLE001 + return None + self._sensor_source = source + return source + + def _skip_while_paused(self) -> bool: + """True while a pause discards this stage's work.""" + return self._pause.is_set() and not self._wait_out_pause() + + def _wait_out_pause(self) -> bool: + """Block while paused; ``False`` means the pipeline is stopping.""" + while self._pause.is_set() and not self._done.is_set(): + self._pause.wait(_PAUSE_POLL_S) + return not self._done.is_set() + def wait(self, timeout: float | None = None) -> FrameSnapshot | None: """Block until a snapshot newer than the last returned one exists. @@ -285,11 +418,14 @@ class StreamPipeline: frames are gone, which is what a live GUI/control loop wants. Use ``on_frame`` when every frame must be handled exactly once. - Returns ``None`` on timeout, or once the pipeline has ended without a - newer frame. + Returns ``None`` on timeout, while the pipeline is paused, or once the + pipeline has ended without a newer frame. ``running`` distinguishes a + hold from the end of input. """ - with self._lock: - seen = self._published + if self._pause.is_set(): + # A hold publishes nothing. Returning the pre-pause snapshot here + # would look like live data to a control loop. + return None deadline = None if timeout is None else time.perf_counter() + timeout while True: with self._lock: @@ -351,8 +487,32 @@ class StreamPipeline: except Empty: return None + def _keep_camera_fed(self) -> None: + """Read and discard one frame, without touching the shared sensor state. + + ``get_img`` would write ``Sensor.img`` while the flow stage may still be + draining, so the capture source is read directly. The read is what keeps + a live camera's exposure and auto-white-balance converged for however + long the pause lasts; nothing downstream sees the frame. + """ + source = self._sensor_source + if source is None: + time.sleep(_PAUSE_POLL_S) + return + try: + source.read_latest(copy=False, wait_s=0.0) + except Exception: # noqa: BLE001 - a dropped frame is not a failure + time.sleep(_PAUSE_POLL_S) + def _producer_loop(self) -> None: while not self._done.is_set(): + if self._pause.is_set() and not self._currently_capturing: + # Level 1 hold: capture itself is stopped, so there is nothing to + # read. Wait it out, then drop the stale backlog. + if not self._wait_out_pause(): + break + self._drain(self._q1) + continue t0 = time.perf_counter() frame = self._produce() if frame is None: @@ -371,6 +531,15 @@ class StreamPipeline: item = self._get(self._q1) if item is None: break + if self._pause.is_set(): + # Level 2 holds the frame in camera only: skip the flow work and + # let the queue drop this frame. A motion-free hold adds nothing + # to the track, and the reference frame is rebased on resume. + if not self._wait_out_pause(): + break + self._drain(self._q1) + self._drain(self._q2) + continue # Applied here, on the thread that owns the sensor's flow state, before this frame is # computed: see reset() for why this must not run on the caller's thread. if self._reset_pending.is_set(): @@ -394,6 +563,13 @@ class StreamPipeline: item = self._get(self._q2) if item is None: break + if self._pause.is_set(): + # Discard rather than compute: this is the stage the level-2 hold + # is meant to avoid paying for. + self._drain(self._q2) + if not self._wait_out_pause(): + break + continue frame, flow, t_cap, produce_s, flow_s = item t0 = time.perf_counter() # Explicit flow: compute_hhd reads self.flow, which the flow stage diff --git a/tests/test_pipeline_pause.py b/tests/test_pipeline_pause.py new file mode 100644 index 0000000..f0cbb24 --- /dev/null +++ b/tests/test_pipeline_pause.py @@ -0,0 +1,300 @@ +"""Tests for StreamPipeline pause/resume. + +A fake sensor stands in for the real one: the pause contract is about which +stages run, not about camera hardware, and a fake makes that observable without +a device. The sensor exposes the same surface the stages use - ``get_img``, +``compute_flow``, ``compute_hhd``, ``compute_contact``, ``reset``, ``read_info`` +and ``_force_raw`` - so the pipeline runs its real loops unchanged. +""" + +from __future__ import annotations + +import threading +import time + +import numpy as np +import pytest + +import orisys +from orisys import PauseLevel, StreamPipeline + + +class FakeInfo: + FNORMAL = "fnormal" + FNORMAL_RAW = "fnormal_raw" + FSHEARX = "fshearx" + FSHEARY = "fsheary" + VNORMAL = "vnormal" + VSHEAR = "vshear" + DEPTH = "depth" + CENTROID = "centroid" + CONTOUR = "contour" + + +class FakeSensor: + """Minimal sensor surface, with counters for what each stage did.""" + + def __init__(self, *, source=None) -> None: + self.info = FakeInfo() + self.force_unit = "raw" + self.fnormal = 1.0 + self.fshearx = 2.0 + self.fsheary = 3.0 + self.ftor = 0.0 + self._force_raw = (10.0, 20.0, 30.0) + self.img_size_raw = (8, 8) + self.flow = None + self.img = None + self._force_filtered_raw = (10.0, 20.0, 30.0) + self._force_raw_unfiltered = (10.0, 20.0, 30.0) + self._force_unfiltered = (1.0, 2.0, 3.0) + self._lock = threading.Lock() + self.produced = 0 + self.flow_calls = 0 + self.hhd_calls = 0 + self.reset_calls = 0 + if source is not None: + self._capture_source = source + + # ---- producer stage + def get_img(self): + with self._lock: + self.produced += 1 + # A distinct frame each time, so the stages can be told apart. + return np.full((4, 4), float(self.produced), dtype=np.float32) + + # ---- flow stage + def compute_flow(self, frame=None): + with self._lock: + self.flow_calls += 1 + shape = (4, 4, 2) + self.flow = np.zeros(shape, dtype=np.float32) + + # ---- hhd stage + def compute_hhd(self, flow=None): + with self._lock: + self.hhd_calls += 1 + + def compute_contact(self): + return None + + def reset(self): + with self._lock: + self.reset_calls += 1 + + def read_info(self, *which): + values = [] + for key in which: + if key in (FakeInfo.VNORMAL, FakeInfo.VSHEAR): + values.append(np.zeros((4, 4, 2), dtype=np.float32)) + elif key == FakeInfo.DEPTH: + values.append(np.zeros((4, 4), dtype=np.float32)) + elif key == FakeInfo.CENTROID: + values.append(np.zeros(2, dtype=np.float32)) + elif key == FakeInfo.CONTOUR: + values.append(np.zeros(0, dtype=np.float32)) + else: + values.append(1.0) + return tuple(values) + + +class FakeCaptureSource: + """Records whether the camera was told to keep streaming.""" + + def __init__(self) -> None: + self.started = 0 + self.stopped = 0 + self.reads = 0 + self._running = True + + def is_open(self): + return True + + def start_background(self): + self.started += 1 + self._running = True + + def stop_background(self, timeout: float = 1.0): + self.stopped += 1 + self._running = False + + def read_latest(self, *, copy: bool = True, wait_s: float = 0.0): + self.reads += 1 + return True, np.zeros((4, 4), dtype=np.float32) + + +def _wait_for(predicate, timeout=2.0, interval=0.01): + deadline = time.perf_counter() + timeout + while time.perf_counter() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +def test_pause_level_is_exported_from_the_package(): + assert orisys.PauseLevel is PauseLevel + assert PauseLevel.CAPTURE_ONLY.value == "capture_only" + assert PauseLevel.BOTH.value == "both" + + +def test_capture_only_keeps_reading_but_stops_computing(): + source = FakeCaptureSource() + sensor = FakeSensor(source=source) + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 3) + assert pipe.paused is False + + pipe.pause(PauseLevel.CAPTURE_ONLY) + assert pipe.paused is True + assert pipe.pause_level is PauseLevel.CAPTURE_ONLY + + flow_at_pause = sensor.flow_calls + hhd_at_pause = sensor.hhd_calls + produced_at_pause = sensor.produced + time.sleep(0.4) + + # The read keeps going, so a live camera stays converged... + assert sensor.produced > produced_at_pause, "capture-only must keep reading" + assert source.stopped == 0, "the camera must not be stopped" + # ...and the work stops: this is the part the hold is bought for. + assert sensor.flow_calls == flow_at_pause, "flow must not run while held" + assert sensor.hhd_calls == hhd_at_pause, "no HHD work may happen while held" + + published_at_pause = pipe.stats.published + pipe.resume() + assert pipe.paused is False + assert _wait_for(lambda: pipe.stats.published > published_at_pause), ( + "resume must publish frames again" + ) + assert sensor.hhd_calls > hhd_at_pause + + +def test_pause_both_stops_capture_as_well_as_compute(): + source = FakeCaptureSource() + sensor = FakeSensor(source=source) + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 3) + + pipe.pause(PauseLevel.BOTH) + assert _wait_for(lambda: source.stopped >= 1), "capture must be stopped" + produced_at_pause = sensor.produced + hhd_at_pause = sensor.hhd_calls + published_at_pause = pipe.stats.published + time.sleep(0.4) + + assert sensor.produced == produced_at_pause, "no capture while fully held" + assert sensor.hhd_calls == hhd_at_pause + assert pipe.stats.published == published_at_pause + + pipe.resume() + assert _wait_for(lambda: source.started >= 1), "capture must restart" + assert _wait_for(lambda: pipe.stats.published > published_at_pause) + + +def test_resume_rebases_the_tracker_by_default(): + sensor = FakeSensor() + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 2) + resets_before = sensor.reset_calls + pipe.pause(PauseLevel.CAPTURE_ONLY) + time.sleep(0.1) + pipe.resume() + assert _wait_for(lambda: sensor.reset_calls > resets_before), ( + "a held reference frame would otherwise read as deformation" + ) + + +def test_resume_can_keep_the_reference_frame(): + sensor = FakeSensor() + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 2) + resets_before = sensor.reset_calls + pipe.pause(PauseLevel.CAPTURE_ONLY) + time.sleep(0.1) + pipe.resume(reset=False) + assert _wait_for(lambda: pipe.stats.published > 0) + time.sleep(0.2) + assert sensor.reset_calls == resets_before + + +def test_wait_returns_none_while_paused_and_frames_after_resume(): + sensor = FakeSensor() + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert pipe.wait(timeout=2.0) is not None + pipe.pause(PauseLevel.CAPTURE_ONLY) + assert pipe.wait(timeout=0.1) is None, "a hold publishes nothing" + # A paused pipeline is not a finished pipeline: None here must not be + # read as end of input. + assert pipe.running is True + pipe.resume() + assert pipe.wait(timeout=2.0) is not None + + +def test_pause_is_idempotent_and_level_changes_apply(): + source = FakeCaptureSource() + sensor = FakeSensor(source=source) + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 2) + pipe.pause(PauseLevel.CAPTURE_ONLY) + pipe.pause(PauseLevel.CAPTURE_ONLY) + assert pipe.paused is True + pipe.pause(PauseLevel.BOTH) + assert pipe.pause_level is PauseLevel.BOTH + assert _wait_for(lambda: source.stopped >= 1) + pipe.resume() + pipe.resume() # second resume is a no-op + assert pipe.paused is False + + +def test_pause_accepts_a_plain_string_level(): + sensor = FakeSensor() + with StreamPipeline(sensor, sensor.get_img, include=("image",)) as pipe: + assert _wait_for(lambda: pipe.stats.published >= 1) + pipe.pause("capture_only") + assert pipe.pause_level is PauseLevel.CAPTURE_ONLY + pipe.resume() + + +def test_stop_while_paused_terminates_promptly(): + sensor = FakeSensor() + pipe = StreamPipeline(sensor, sensor.get_img, include=("image",)) + pipe.start() + try: + assert _wait_for(lambda: pipe.stats.published >= 2) + pipe.pause(PauseLevel.BOTH) + time.sleep(0.1) + started = time.perf_counter() + pipe.stop(timeout=3.0) + assert time.perf_counter() - started < 2.0, "stop must not wait out the pause" + assert pipe.running is False + finally: + if pipe.running: + pipe.stop() + + +def test_end_of_input_is_still_reported_as_none(): + """The pause path must not swallow the end-of-stream sentinel. + + Counted through ``on_frame``, which delivers every frame exactly once: with + a fast producer and a 5-frame source, ``wait`` is documented latest-wins and + legitimately skips frames, so it cannot be used to count them. + """ + frames = {"produced": 0} + delivered: list[int] = [] + + def produce(): + frames["produced"] += 1 + if frames["produced"] > 5: + return None + return np.zeros((4, 4), dtype=np.float32) + + sensor = FakeSensor() + with StreamPipeline( + sensor, produce, include=("image",), on_frame=lambda snap: delivered.append(snap.seq) + ) as pipe: + assert _wait_for(lambda: not pipe.running, timeout=3.0), "producer never finished" + assert len(delivered) == 5, delivered + # Once the input has ended, wait reports the end rather than hanging. + assert pipe.wait(timeout=0.5) is None + assert pipe.running is False diff --git a/tools/lint_gate.py b/tools/lint_gate.py index c40c8c4..3f49306 100644 --- a/tools/lint_gate.py +++ b/tools/lint_gate.py @@ -131,16 +131,24 @@ def run_doc_gate(force_all: bool) -> dict: """ script = REPO.parent / "scripts" / "doc_lint.py" if not script.is_file(): - return {"status": "unavailable", "findings": [], "baselined": []} + print(f"lint_gate: required documentation checker missing: {script}", file=sys.stderr) + return {"status": "error", "findings": [], "baselined": []} cmd = [sys.executable, str(script), "--json"] if force_all: cmd.append("--all") result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", check=False) + if result.returncode not in (0, 1): + print(f"lint_gate: documentation checker failed ({result.returncode}): {result.stderr.strip()}", file=sys.stderr) + return {"status": "error", "findings": [], "baselined": []} if not result.stdout.strip(): print(f"lint_gate: doc gate produced no output: {result.stderr.strip()[:200]}", file=sys.stderr) return {"status": "error", "findings": [], "baselined": []} try: - return json.loads(result.stdout) + payload = json.loads(result.stdout) + expected = "clean" if result.returncode == 0 else "findings" + if not isinstance(payload, dict) or payload.get("status") != expected: + return {"status": "error", "findings": [], "baselined": []} + return payload except json.JSONDecodeError: print("lint_gate: could not parse doc gate output", file=sys.stderr) return {"status": "error", "findings": [], "baselined": []} @@ -161,6 +169,10 @@ def main(argv: list[str] | None = None) -> int: docs: dict = {"status": "skipped", "findings": [], "baselined": []} if not args.no_docs and args.files is None: docs = run_doc_gate(force_all=args.all) + if docs.get("status") not in ("clean", "findings"): + if args.json: + print(json.dumps({"status": "error", "docs": docs, "findings": []})) + return 2 # Report the documentation verdict even when Ruff is unavailable: a missing linter must not # hide the docs result, and the exit code still signals the environment problem.