为采集帧绑定主机时间与来源信息

采集时间随图像穿过处理队列，并修正未读帧的丢帧计数。通过采集与流水线针对性测试。

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',
 

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",

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]:
+    def read_raw_stamped(self) -> CapturedFrame | None:
-        """Read one frame directly from the underlying capture (no reconnect)."""
+        """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:
+        captured = self.read_latest_stamped(copy=copy, wait_s=wait_s)
-            ok, frame = self.read_raw()
+        return (captured is not None, None if captured is None else captured.image)
-            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
 
     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()
+            captured = self.read_raw_stamped()
-            if not ok or frame is None:
+            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_seq = captured.sequence
-                self._latest_frame = frame
+                self._latest_frame = captured.image
+                self._latest_captured = captured
                 with self._lock:
                     self._status.last_seq = self._latest_seq
                 self._frame_cond.notify_all()

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:

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
+    ``produce`` is a zero-argument callable returning the working frame
-    (``(H, W)`` grey or ``(H, W, 3)`` BGR) or ``None`` to signal end of input.
+    (``(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:

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

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