diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index 811be66..8466988 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -5,13 +5,19 @@ A Python library for processing and analyzing optical tactile sensor data, including image stitching, optical flow tracking, and deformation decomposition. """ -__version__ = "0.5.3" +__version__ = "0.5.4" __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, PauseLevel, PipelineStats, StreamPipeline +from .pipeline import ( + FrameSnapshot, + PauseLevel, + PipelineStats, + ProducedFrame, + StreamPipeline, +) from .sensor import Sensor CompoundEyeSensor = Sensor @@ -44,6 +50,7 @@ __all__ = [ # pipelined multi-stream processing 'StreamPipeline', 'FrameSnapshot', + 'ProducedFrame', 'PipelineStats', 'PauseLevel', diff --git a/src/orisys/capture/__init__.py b/src/orisys/capture/__init__.py index 98d6505..7643050 100644 --- a/src/orisys/capture/__init__.py +++ b/src/orisys/capture/__init__.py @@ -16,6 +16,7 @@ from .discovery import ( pick_default_camera, ) from .source import ( + CapturedFrame, CaptureSource, CaptureWorker, is_live_camera, @@ -31,6 +32,7 @@ __all__ = [ "CameraFilter", "CameraInfo", "CaptureConfig", + "CapturedFrame", "CaptureHealth", "CaptureSource", "CaptureStatus", diff --git a/src/orisys/capture/source.py b/src/orisys/capture/source.py index d758c4c..8ef939e 100644 --- a/src/orisys/capture/source.py +++ b/src/orisys/capture/source.py @@ -7,6 +7,8 @@ import os import sys import threading import time +from dataclasses import dataclass +from math import isfinite from typing import Any, Protocol, runtime_checkable import cv2 @@ -23,6 +25,21 @@ from .config import ( logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class CapturedFrame: + """One image and the host clocks sampled when its capture read completed. + + ``sequence`` belongs to this source instance. ``media_time_ms`` is advisory + file metadata and is never substituted for the host acquisition clocks. + """ + + image: np.ndarray + sequence: int + host_mono_ns: int + host_utc_ns: int + media_time_ms: float | None = None + + @runtime_checkable class VideoCaptureLike(Protocol): """Minimal protocol so tests can inject fake captures.""" @@ -223,7 +240,10 @@ class CaptureSource: self._frame_lock = threading.Lock() self._frame_cond = threading.Condition(self._frame_lock) self._latest_frame: np.ndarray | None = None + self._latest_captured: CapturedFrame | None = None self._latest_seq = 0 + self._read_seq = 0 + self._last_delivered_seq = 0 self._thread: threading.Thread | None = None self._background_running = False @@ -338,6 +358,7 @@ class CaptureSource: self._cap = None with self._frame_cond: self._latest_frame = None + self._latest_captured = None self._latest_seq = 0 def _refresh_stream_meta_unlocked(self) -> None: @@ -363,28 +384,80 @@ class CaptureSource: for _ in range(max(0, int(count))): self.read_raw() - def read_raw(self) -> tuple[bool, np.ndarray | None]: - """Read one frame directly from the underlying capture (no reconnect).""" + def read_raw_stamped(self) -> CapturedFrame | None: + """Read a frame and bind its identity and clocks before another read.""" cap = self._cap if cap is None: - return False, None + return None try: ok, frame = cap.read() except Exception as exc: with self._lock: self._status.consecutive_failures += 1 self._status.last_error = str(exc) - return False, None + return None if not ok or frame is None: with self._lock: self._status.consecutive_failures += 1 - return False, None + return None + host_mono_ns = time.perf_counter_ns() + host_utc_ns = time.time_ns() + media_time_ms = None + if not is_live_camera(self._status.source): + try: + value = float(cap.get(cv2.CAP_PROP_POS_MSEC)) + if isfinite(value) and value >= 0: + media_time_ms = value + except Exception as exc: # noqa: BLE001 - metadata must not stop capture + logger.debug("capture: media timestamp unavailable: %s", exc) with self._lock: + self._read_seq += 1 + sequence = self._read_seq self._status.consecutive_failures = 0 - self._status.last_frame_time = time.perf_counter() + self._status.last_frame_time = host_mono_ns / 1e9 if self._status.health != CaptureHealth.CONNECTED: self._status.health = CaptureHealth.CONNECTED - return True, frame + return CapturedFrame(frame, sequence, host_mono_ns, host_utc_ns, media_time_ms) + + def read_raw(self) -> tuple[bool, np.ndarray | None]: + """Compatibility read without capture metadata.""" + captured = self.read_raw_stamped() + return (captured is not None, None if captured is None else captured.image) + + def read_latest_stamped( + self, *, copy: bool = True, wait_s: float = 0.0 + ) -> CapturedFrame | None: + """Return a frame and its own stamp atomically from the background slot.""" + if not self._background_running: + captured = self.read_raw_stamped() + if captured is not None and copy: + return CapturedFrame( + captured.image.copy(), + captured.sequence, + captured.host_mono_ns, + captured.host_utc_ns, + captured.media_time_ms, + ) + return captured + + deadline = time.perf_counter() + max(0.0, wait_s) + with self._frame_cond: + while self._latest_captured is None: + remaining = deadline - time.perf_counter() + if remaining <= 0: + return None + self._frame_cond.wait(timeout=remaining) + captured = self._latest_captured + self._last_delivered_seq = captured.sequence + if copy: + return CapturedFrame( + captured.image.copy(), + captured.sequence, + captured.host_mono_ns, + captured.host_utc_ns, + captured.media_time_ms, + ) + return captured def read_latest( self, @@ -393,21 +466,8 @@ class CaptureSource: wait_s: float = 0.0, ) -> tuple[bool, np.ndarray | None]: """Return the newest frame from the background thread, or read sync.""" - if not self._background_running: - ok, frame = self.read_raw() - if ok and copy and frame is not None: - return ok, frame.copy() - return ok, frame - - deadline = time.perf_counter() + max(0.0, wait_s) - with self._frame_cond: - while self._latest_frame is None: - remaining = deadline - time.perf_counter() - if remaining <= 0: - return False, None - self._frame_cond.wait(timeout=remaining) - frame = self._latest_frame.copy() if copy else self._latest_frame - return True, frame + captured = self.read_latest_stamped(copy=copy, wait_s=wait_s) + return (captured is not None, None if captured is None else captured.image) def read(self, *, copy: bool = True, wait_s: float = 1.0) -> tuple[bool, np.ndarray | None]: """Compat alias used by finger runtime and older worker callers.""" @@ -441,6 +501,7 @@ class CaptureSource: self._thread = None with self._frame_cond: self._latest_frame = None + self._latest_captured = None def stop(self, timeout: float = 1.0) -> None: """Compat alias for ``stop_background``.""" @@ -455,19 +516,23 @@ class CaptureSource: time.sleep(0.2) continue - ok, frame = self.read_raw() - if not ok or frame is None: + captured = self.read_raw_stamped() + if captured is None: if self.should_reconnect(): continue time.sleep(0.001) continue with self._frame_cond: - if self._latest_frame is not None: + if ( + self._latest_captured is not None + and self._latest_captured.sequence > self._last_delivered_seq + ): with self._lock: self._status.dropped_frames += 1 - self._latest_seq += 1 - self._latest_frame = frame + self._latest_seq = captured.sequence + self._latest_frame = captured.image + self._latest_captured = captured with self._lock: self._status.last_seq = self._latest_seq self._frame_cond.notify_all() diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py index fd2dc8f..7b60ed0 100644 --- a/src/orisys/finger/runtime.py +++ b/src/orisys/finger/runtime.py @@ -358,6 +358,14 @@ class FingerRuntime: return self.capture_source.read_raw() return self.capture.read() + def read_source_stamped(self): + """Read the source image with its capture-thread identity and host clocks.""" + if self.capture_source is None: + return None + if self.capture_source.running: + return self.capture_source.read_latest_stamped(copy=True, wait_s=1.0) + return self.capture_source.read_raw_stamped() + def process_frame(self, frame, *, color=False): """Prepare, unroll, and feed one source frame into the tactile processor.""" if color: diff --git a/src/orisys/pipeline.py b/src/orisys/pipeline.py index a682611..1654ca0 100644 --- a/src/orisys/pipeline.py +++ b/src/orisys/pipeline.py @@ -74,7 +74,13 @@ from typing import Any import numpy as np -__all__ = ["FrameSnapshot", "PipelineStats", "PauseLevel", "StreamPipeline"] +__all__ = [ + "FrameSnapshot", + "ProducedFrame", + "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). @@ -105,6 +111,14 @@ class PauseLevel(str, Enum): CAPTURE_ONLY = "capture_only" +@dataclass(frozen=True) +class ProducedFrame: + """A producer image with opaque metadata carried to its finished snapshot.""" + + image: np.ndarray + metadata: Any = None + + @dataclass(frozen=True) class FrameSnapshot: """An immutable view of one finished frame. @@ -139,6 +153,7 @@ class FrameSnapshot: #: when the producer did not report one. Read from ``Sensor.img_size_raw`` because callers #: compare it with ``image.shape`` to see how much the frame was reduced. image_size_raw: tuple[int, int] | None = None + source_metadata: Any = None @dataclass @@ -171,8 +186,9 @@ class PipelineStats: class StreamPipeline: """Run ``produce -> flow -> hhd`` as three concurrent stages. - ``produce`` is a zero-argument callable returning either the working frame - (``(H, W)`` grey or ``(H, W, 3)`` BGR) or ``None`` to signal end of input. + ``produce`` is a zero-argument callable returning the working frame + (``(H, W)`` grey or ``(H, W, 3)`` BGR), a ``ProducedFrame`` carrying source + metadata, or ``None`` to signal end of input. For a ``Sensor`` that is ``sensor.get_img``; for a ``FingerRuntime`` use :meth:`for_finger_runtime`. """ @@ -180,7 +196,7 @@ class StreamPipeline: def __init__( self, sensor: Any, - produce: Callable[[], np.ndarray | None], + produce: Callable[[], np.ndarray | ProducedFrame | None], *, contact: bool = False, depth: int = 2, @@ -519,10 +535,14 @@ class StreamPipeline: frame = self._produce() if frame is None: break + source_metadata = None + if isinstance(frame, ProducedFrame): + source_metadata = frame.metadata + frame = frame.image t1 = time.perf_counter() # Private copy: the SDK buffer is rewritten in place by the next # produce() call while the flow stage is still reading this frame. - payload = (frame.copy(), t0, t1 - t0) + payload = (frame.copy(), t0, t1 - t0, source_metadata) if not self._put(self._q1, payload): break self.stats.produced += 1 @@ -547,7 +567,7 @@ class StreamPipeline: if self._reset_pending.is_set(): self._reset_pending.clear() self.sensor.reset() - frame, t_cap, produce_s = item + frame, t_cap, produce_s, source_metadata = item t0 = time.perf_counter() # The frame is passed explicitly: the SDK's self.img slot is shared # with the producer, so it must not be the transport. @@ -557,7 +577,10 @@ class StreamPipeline: # Private copy: with the flow EMA enabled Sensor.flow is a buffer the # next frame rewrites. flow_copy = flow.copy() if isinstance(flow, np.ndarray) else None - self._put(self._q2, (frame, flow_copy, t_cap, produce_s, flow_s)) + self._put( + self._q2, + (frame, flow_copy, t_cap, produce_s, flow_s, source_metadata), + ) self._put(self._q2, None) def _hhd_loop(self) -> None: @@ -572,7 +595,7 @@ class StreamPipeline: if not self._wait_out_pause(): break continue - frame, flow, t_cap, produce_s, flow_s = item + frame, flow, t_cap, produce_s, flow_s, source_metadata = item t0 = time.perf_counter() # Explicit flow: compute_hhd reads self.flow, which the flow stage # overwrites for the next frame, so it cannot be the transport. @@ -580,7 +603,9 @@ class StreamPipeline: if self._contact: self.sensor.compute_contact() hhd_s = time.perf_counter() - t0 - snap = self._build(frame, flow, produce_s, flow_s, hhd_s, t_cap) + snap = self._build( + frame, flow, produce_s, flow_s, hhd_s, t_cap, source_metadata + ) self._publish(snap) if self._on_frame is not None: # Runs on this stage's thread: a slow consumer exerts real @@ -600,7 +625,8 @@ class StreamPipeline: return value.copy() def _build(self, frame: np.ndarray, flow: np.ndarray | None, produce_s: float, - flow_s: float, hhd_s: float, t_cap: float) -> FrameSnapshot: + flow_s: float, hhd_s: float, t_cap: float, + source_metadata: Any = None) -> FrameSnapshot: sensor = self.sensor info = sensor.info include = self._include @@ -641,6 +667,7 @@ class StreamPipeline: flow_ms=flow_s * 1000.0, hhd_ms=hhd_s * 1000.0, image_size_raw=image_size_raw, + source_metadata=source_metadata, ) def _fps(self) -> float: diff --git a/tests/test_capture.py b/tests/test_capture.py index 1084529..0bfd5da 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -199,6 +199,89 @@ def test_capture_source_read_and_status(): assert cap.release_count == 1 +def test_file_read_binds_frame_sequence_and_host_clocks(): + cap = FakeCap([_frame(7), _frame(8)]) + src = CaptureSource( + CaptureConfig(source="sample.mp4", reconnect=ReconnectPolicy(enabled=False)), + factory=lambda cfg: (cap, "FAKE"), + ) + src.open() + before = time.perf_counter_ns() + first = src.read_raw_stamped() + second = src.read_raw_stamped() + after = time.perf_counter_ns() + assert first is not None and second is not None + assert (first.sequence, second.sequence) == (1, 2) + assert (int(first.image[0, 0, 0]), int(second.image[0, 0, 0])) == (7, 8) + assert before <= first.host_mono_ns <= second.host_mono_ns <= after + assert first.host_utc_ns > 0 + assert first.media_time_ms == 0.0 + src.close() + + +def test_background_read_keeps_its_own_stamp_when_newer_frame_arrives(): + cap = FakeCap([_frame(i) for i in range(10)]) + src = CaptureSource( + CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False)), + factory=lambda cfg: (cap, "FAKE"), + ) + src.open() + src.start_background() + try: + first = src.read_latest_stamped(wait_s=1.0) + assert first is not None + deadline = time.perf_counter() + 1.0 + second = first + while second.sequence == first.sequence and time.perf_counter() < deadline: + second = src.read_latest_stamped(wait_s=0.01) + assert second is not None and second.sequence > first.sequence + assert int(first.image[0, 0, 0]) == min(first.sequence - 1, 9) + assert int(second.image[0, 0, 0]) == min(second.sequence - 1, 9) + assert first.host_mono_ns <= second.host_mono_ns + assert src.status.last_seq >= 2 + finally: + src.close() + + +def test_background_drop_count_only_increases_for_unread_replacements(): + class StepCap(FakeCap): + def __init__(self): + super().__init__([_frame(i) for i in range(4)]) + self.allow = threading.Semaphore(0) + + def read(self): + if not self.allow.acquire(timeout=0.1): + return False, None + return super().read() + + cap = StepCap() + src = CaptureSource( + CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False)), + factory=lambda cfg: (cap, "FAKE"), + ) + src.open() + src.start_background() + + def wait_seq(seq): + deadline = time.perf_counter() + 2.0 + while src.status.last_seq < seq and time.perf_counter() < deadline: + time.sleep(0.001) + assert src.status.last_seq >= seq + + try: + cap.allow.release() + wait_seq(1) + assert src.read_latest_stamped().sequence == 1 + cap.allow.release() + wait_seq(2) + assert src.status.dropped_frames == 0 + cap.allow.release() + wait_seq(3) + assert src.status.dropped_frames == 1 + finally: + src.close() + + def test_auto_reconnect_after_failed_reads(): first = FakeCap([_frame(1)], fail_after=1) second = FakeCap([_frame(9)]) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 5c97497..3c82a59 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -8,7 +8,7 @@ import time import numpy as np import pytest -from orisys import FrameSnapshot, StreamPipeline +from orisys import FrameSnapshot, ProducedFrame, StreamPipeline class FakeSensor: @@ -149,6 +149,30 @@ def test_on_frame_receives_every_frame_in_order(): assert pipe.stats.published == 12 +def test_producer_metadata_stays_with_its_image_through_the_queues(): + sensor = FakeSensor(flow_delay=0.002) + frames = [] + produced = 0 + + def produce(): + nonlocal produced + if produced == 12: + return None + value = produced + produced += 1 + return ProducedFrame(sensor.make_frame(value), {"source_seq": value}) + + pipe = StreamPipeline(sensor, produce, on_frame=frames.append) + pipe.start() + deadline = time.perf_counter() + 10.0 + while len(frames) < 12 and time.perf_counter() < deadline: + time.sleep(0.001) + pipe.stop() + assert len(frames) == 12 + for snap in frames: + assert snap.source_metadata["source_seq"] == int(snap.image[0, 0]) + + def test_forces_come_from_the_matching_frame(): """Each snapshot must carry the forces computed from its own frame's flow.""" sensor = FakeSensor()