diff --git a/shared/src/orisys_app_shared/acquisition/finger_worker.py b/shared/src/orisys_app_shared/acquisition/finger_worker.py index 017ebde..52740b5 100644 --- a/shared/src/orisys_app_shared/acquisition/finger_worker.py +++ b/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 - #: so a change can be checked against it frame-for-frame on the same input - #: (``False`` selects the reference). + #: File analysis can pipeline every frame; live capture uses the serial + #: chain so old inputs and completed results cannot accumulate in FIFOs. 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: - # 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. + if self._use_stream_pipeline and is_file: 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. - - ``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. - """ + """Run whole frames sequentially; live capture always takes the newest input.""" 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, - depth=depth, - valid_mask=runtime.valid_mask, + flow=None if flow is None else np.array(flow, copy=True), + depth=None if depth is None else np.array(depth, copy=True), + 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, diff --git a/shared/tests/test_finger_realtime.py b/shared/tests/test_finger_realtime.py new file mode 100644 index 0000000..dcbb552 --- /dev/null +++ b/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 diff --git a/shared/tests/test_stream_pipeline_acquisition.py b/shared/tests/test_stream_pipeline_acquisition.py index b3be318..0f263e9 100644 --- a/shared/tests/test_stream_pipeline_acquisition.py +++ b/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"]) -def test_a_stopped_session_leaves_no_pipeline_threads(kind): +@pytest.mark.parametrize( + "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():