fix(studio): 消除 Finger 实时采集的帧排队

实时处理相机最新帧，并修复重复采样、数组快照和暂停恢复。通过共享模块及相关 Studio 测试。

shared/src/orisys_app_shared/acquisition/finger_worker.py

@@ -294,9 +294,8 @@ class FingerAcquisitionWorker(RecordSinkMixin):
         use_stream_pipeline: bool = True,
     ):
         self._init_sinks()
-        #: ``runtime.pipeline()`` is the acquisition path; the serial chain is kept
+        #: File analysis can pipeline every frame; live capture uses the serial
-        #: so a change can be checked against it frame-for-frame on the same input
+        #: chain so old inputs and completed results cannot accumulate in FIFOs.
-        #: (``False`` selects the reference).
         self._use_stream_pipeline = bool(use_stream_pipeline)
         #: The running session's ``StreamPipeline``, when there is one.
         self._pipe = None
@@ -500,8 +499,11 @@ class FingerAcquisitionWorker(RecordSinkMixin):
         never computed: keeping the pre-pause reference frame would report the hold
         itself as a large deformation.
         """
-        self._pause.clear()
         pipe = self._pipe
+        if pipe is None and self._pause.is_set():
+            with self._lock:
+                self._pending_reset = True
+        self._pause.clear()
         if pipe is not None:
             try:
                 pipe.resume(reset=True)
@@ -642,12 +644,7 @@ class FingerAcquisitionWorker(RecordSinkMixin):
         if opened is None:
             return
         mode, source, runtime, is_file = opened
-        if self._use_stream_pipeline:
+        if self._use_stream_pipeline and is_file:
-            # The recommended path: the SDK runs produce (capture + UV unroll) ->
-            # flow -> HHD on three threads, so throughput is bounded by the slowest
-            # stage rather than by their sum. The app's own layer — session and
-            # frame bookkeeping, the record sink, calibration reload, contact
-            # provenance — stays on top of the frames the pipeline finishes.
             self._run_stream_pipeline(session, mode, source, runtime, is_file)
             return
         self._run_serial_loop(session, mode, source, runtime, is_file)
@@ -1033,19 +1030,15 @@ class FingerAcquisitionWorker(RecordSinkMixin):
     def _run_serial_loop(
         self, session: int, mode: FixedCameraMode, source, runtime, is_file: bool
     ) -> None:
-        """The single-threaded chain, kept whole for the packet-diff check.
+        """Run whole frames sequentially; live capture always takes the newest input."""
-
-        ``use_stream_pipeline=False`` selects it: read, unroll, decompose and read
-        the forces back to back on this one thread, which bounds throughput by the
-        sum of the stages. It is the reference the concurrent path is measured
-        against, and it is not the path a session uses.
-        """
         frame_times: list[float] = []
         #: (collection time, published sequence), so the camera's own rate
         #: can be told apart from the rate this loop consumes (see the box
         #: worker: the two are the same measurement and must mean the same).
         source_times: list[tuple[float, int]] = []
         last_source_seq = None
+        last_processed_seq = None
+        capture = _RuntimeCapture(runtime)
         capture_fps = 0.0
         saw_frame = False
         last_frame_mono = 0.0
@@ -1107,6 +1100,12 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                     time.sleep(PAUSE_POLL_S)
                     continue
 
+                if not is_file and source_seq is not None:
+                    if source_seq == last_processed_seq:
+                        time.sleep(0.001)
+                        continue
+                    last_processed_seq = source_seq
+
                 unrolled = runtime.process_frame(source_frame, color=False)
                 if unrolled is None:
                     if saw_frame and is_file and none_streak_start is None:
@@ -1116,6 +1115,7 @@ class FingerAcquisitionWorker(RecordSinkMixin):
 
                 none_streak_start = None
                 tactile = runtime.tactile
+                sensor = getattr(tactile, "raw_sensor", tactile)
                 compute_error = ""
                 # The frame's acquisition time, so an event that happens on this
                 # frame is dated when it happened rather than when it was queued.
@@ -1135,9 +1135,11 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                     if self._on_error is not None and session == self._session_id:
                         self._on_error(message)
                 # The same chain the offline depth replay runs on the tactile.
-                chain_depth, chain_error = process_chain(tactile)
+                chain_depth, chain_error = process_chain(sensor)
                 compute_error = compute_error or chain_error
                 t1 = time.perf_counter()
+                if self._pause.is_set() or self._stop.is_set():
+                    continue
                 frame_times.append(t0)
                 cutoff = t1 - FPS_WINDOW_S
                 frame_times = [t for t in frame_times if t >= cutoff]
@@ -1147,7 +1149,7 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                 span = frame_times[-1] - frame_times[0] if len(frame_times) > 1 else 0.0
                 fps = (len(frame_times) - 1) / span if span > 0.0 else 0.0
                 # The camera's own rate, so this view can attribute a low rate too.
-                delivered = _capture_sequence(tactile)
+                delivered = source_seq if not is_file else None
                 if delivered is not None and delivered != last_source_seq:
                     last_source_seq = delivered
                     source_times.append((t0, delivered))
@@ -1163,11 +1165,11 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                         with self._lock:
                             self._measured_capture_fps = capture_fps
                 forces = read_forces_from_sensor(
-                    tactile,
+                    sensor,
                     config_ref=mode.config_ref,
                 )
                 invalid_reason = forces.get("invalid_reason") or compute_error
-                raw_frame = _ensure_bgr(source_frame)
+                raw_frame = _ensure_bgr(source_frame).copy()
                 # The frame the tactile chain consumed, kept for depth replay.
                 self.note_preroll(raw_frame)
                 # Provenance for anything that colours a derived array (the 3D pane):
@@ -1187,13 +1189,13 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                 if depth_valid:
                     last_depth_frame_id = frame_id
                 contact = _contact_from_sensor(
-                    tactile,
+                    sensor,
                     frame_id=frame_id,
                     compute_error=compute_error,
                 )
                 packet = SdkPacket(
                     raw_frame=raw_frame,
-                    processed_frame=_ensure_bgr(unrolled),
+                    processed_frame=_ensure_bgr(unrolled).copy(),
                     timestamp=utc_ns / 1e9,
                     wall_t=utc_ns / 1e9,
                     mono_t=frame_mono,
@@ -1206,6 +1208,8 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                     media_time_ms=media_ms,
                     fps=fps,
                     process_ms=(t1 - t0) * 1000.0,
+                    capture_fps=capture_fps,
+                    capture_dropped=int(_capture_dropped(capture) or 0),
                     normal_force=forces["normal_force"],
                     shear_x=forces["shear_x"],
                     shear_y=forces["shear_y"],
@@ -1224,9 +1228,12 @@ class FingerAcquisitionWorker(RecordSinkMixin):
                     # place; the app must not record them as this frame's data.
                     stale_forces=bool(compute_error),
                     raw_available=bool(forces.get("raw_available", True)),
-                    flow=flow,
+                    flow=None if flow is None else np.array(flow, copy=True),
-                    depth=depth,
+                    depth=None if depth is None else np.array(depth, copy=True),
-                    valid_mask=runtime.valid_mask,
+                    valid_mask=(
+                        np.array(runtime.valid_mask, copy=True)
+                        if runtime.valid_mask is not None else None
+                    ),
                     flow_valid=flow_valid,
                     flow_frame_id=frame_id if flow_valid else last_flow_frame_id,
                     depth_valid=depth_valid,

shared/tests/test_finger_realtime.py

@@ -0,0 +1,135 @@
+"""Live Finger consumes current frames without a FIFO of pending work."""
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from orisys_app_shared.acquisition import finger_worker as fw
+
+
+@pytest.mark.parametrize("is_file, expected", [(False, "serial"), (True, "stream")])
+def test_default_path_keeps_only_offline_processing_queued(
+    monkeypatch, is_file, expected
+):
+    worker = fw.FingerAcquisitionWorker()
+    paths = []
+    monkeypatch.setattr(worker, "_open_runtime", lambda *_: (None, None, None, is_file))
+    monkeypatch.setattr(worker, "_run_serial_loop", lambda *_: paths.append("serial"))
+    monkeypatch.setattr(
+        worker, "_run_stream_pipeline", lambda *_: paths.append("stream")
+    )
+    worker._run_session(1, None)
+    assert paths == [expected]
+
+
+def _live_loop(monkeypatch, *, sequences=None, compute_s=0.0, sink_s=0.0, pause=False):
+    clock = [10.0]
+    worker = fw.FingerAcquisitionWorker()
+    worker._session_id = 1
+    worker._mono_t0 = clock[0]
+    packets = []
+    resets = []
+    source = SimpleNamespace(
+        running=True, status=SimpleNamespace(last_seq=0, dropped_frames=7)
+    )
+    flow = np.zeros((8, 8, 2), np.float32)
+    depth = np.zeros((8, 8), np.float32)
+    tactile = SimpleNamespace(reset=lambda: resets.append(clock[0]))
+    runtime = SimpleNamespace(
+        tactile=tactile, capture_source=source, valid_mask=np.ones((8, 8), bool)
+    )
+    reads = [0]
+    paused_once = [False]
+
+    def read():
+        reads[0] += 1
+        if reads[0] > 40:
+            worker._stop.set()
+            return None
+        if sequences is not None:
+            seq = sequences[min(reads[0] - 1, len(sequences) - 1)]
+        else:
+            seq = 1 + int((clock[0] - 10.0) * 30 + 1e-6)
+        source.status.last_seq = seq
+        return SimpleNamespace(
+            image=np.full((8, 8, 3), seq, np.uint8),
+            sequence=seq,
+            host_mono_ns=int(clock[0] * 1e9),
+            host_utc_ns=int(clock[0] * 1e9),
+            media_time_ms=None,
+        )
+
+    def process(frame, **_):
+        flow.fill(frame[0, 0, 0])
+        depth.fill(frame[0, 0, 0])
+        return frame[:, :, 0]
+
+    def chain(_):
+        clock[0] += compute_s
+        if pause and not paused_once[0]:
+            paused_once[0] = True
+            worker.pause()
+        return depth, ""
+
+    def sleep(seconds):
+        clock[0] += max(seconds, 0.001)
+        if worker.paused:
+            worker.resume()
+
+    def sink(packet):
+        packets.append(packet)
+        clock[0] += sink_s
+        if len(packets) >= 3:
+            worker._stop.set()
+
+    runtime.read_source_stamped = read
+    runtime.process_frame = process
+    monkeypatch.setattr(
+        fw, "time", SimpleNamespace(perf_counter=lambda: clock[0], sleep=sleep)
+    )
+    monkeypatch.setattr(fw, "process_chain", chain)
+    monkeypatch.setattr(fw, "_contact_from_sensor", lambda *_, **__: None)
+    monkeypatch.setattr(
+        fw,
+        "read_forces_from_sensor",
+        lambda *_, **__: {
+            "normal_force": 1.0,
+            "shear_x": 2.0,
+            "shear_y": 3.0,
+            "normal_force_raw": 1.0,
+            "shear_x_raw": 2.0,
+            "shear_y_raw": 3.0,
+            "force_unit": "raw",
+            "force_calibrated": False,
+            "flow": flow,
+            "depth": depth,
+        },
+    )
+    worker.add_sink(sink)
+    worker._run_serial_loop(1, SimpleNamespace(config_ref=None), 0, runtime, False)
+    return packets, worker, resets
+
+
+def test_repeated_camera_frames_are_not_processed_or_recorded(monkeypatch):
+    packets, _, _ = _live_loop(monkeypatch, sequences=[1, 1, 2, 2, 3])
+    assert [p.source_frame_id for p in packets] == [1, 2, 3]
+    assert [p.frame_id for p in packets] == [0, 1, 2]
+
+
+def test_slow_compute_and_sink_take_current_input_without_a_backlog(monkeypatch):
+    packets, _, _ = _live_loop(monkeypatch, compute_s=0.07, sink_s=0.10)
+    assert [p.source_frame_id for p in packets] == [1, 6, 11]
+    for packet in packets:
+        assert packet.flow_frame_id == packet.depth_frame_id == packet.frame_id
+        assert np.all(packet.flow == packet.source_frame_id)
+        assert np.all(packet.depth == packet.source_frame_id)
+        assert packet.capture_dropped == 7
+
+
+def test_pause_during_compute_does_not_publish_and_resume_rebases(monkeypatch):
+    packets, _, resets = _live_loop(monkeypatch, compute_s=0.07, pause=True)
+    assert len(packets) == 3
+    assert packets[0].source_frame_id > 1
+    assert packets[0].frame_id == 0
+    assert len(resets) == 1

shared/tests/test_stream_pipeline_acquisition.py

@@ -283,6 +283,8 @@ def test_finger_video_and_camera_use_the_same_packet_path(source, monkeypatch):
         assert opened == [source]
         assert packets
         packet = packets[0]
+        assert packet.valid
+        assert packet.force_calibrated
         assert packet.raw_frame.shape == (8, 8, 3)
         assert packet.processed_frame.shape == (8, 8, 3)
         assert packet.session_id == 1
@@ -300,20 +302,25 @@ def test_finger_video_and_camera_use_the_same_packet_path(source, monkeypatch):
         _restore(saved)
 
 
-@pytest.mark.parametrize("kind", ["box", "finger"])
+@pytest.mark.parametrize(
-def test_a_stopped_session_leaves_no_pipeline_threads(kind):
+    "kind, source", [("box", "0"), ("finger", "recording.mp4"), ("finger", 0)]
+)
+def test_a_stopped_session_leaves_no_pipeline_threads(kind, source):
     """``stop`` must join the three stages: they own the sensor's tracker state."""
     packets: list = []
     saved = _install()
     worker = _worker(kind)
     try:
         worker.add_sink(packets.append)
-        worker.start("0", config_ref=_config(kind))
+        worker.start(source, config_ref=_config(kind))
         deadline = time.perf_counter() + 5.0
         while time.perf_counter() < deadline and not packets:
             time.sleep(0.02)
         assert packets, "the session published nothing to stop"
-        assert _pipeline_threads(), "the fake session did not start the pipeline"
+        if kind == "finger" and source == 0:
+            assert _pipeline_threads() == []
+        else:
+            assert _pipeline_threads(), "the fake session did not start the pipeline"
         worker.stop(wait=True)
         deadline = time.perf_counter() + 5.0
         while time.perf_counter() < deadline and _pipeline_threads():