diff --git a/shared/src/orisys_app_shared/acquisition/finger_worker.py b/shared/src/orisys_app_shared/acquisition/finger_worker.py index f19949b..5391e95 100644 --- a/shared/src/orisys_app_shared/acquisition/finger_worker.py +++ b/shared/src/orisys_app_shared/acquisition/finger_worker.py @@ -75,7 +75,7 @@ class _FingerProducer: """The ``StreamPipeline`` produce callable for a ``FingerRuntime``. The finger pipeline's producer is ``runtime.grab_and_process`` — read the - source and UV-unroll it — so this wrapper adds the two things that have to + source and UV-unroll it — so this wrapper adds the things that have to happen on the producer thread: * **One published frame per delivered frame.** ``CaptureSource.read_latest`` @@ -88,6 +88,10 @@ class _FingerProducer: frame to the tactile sensor and drops the source frame; the record channel's pre-roll history wants the source frame, so it is copied here while it is still this frame's, and handed on by sequence. + * **The camera's own rate.** The consumer reads ``capture_fps`` from this + producer, the same way the box path reads it from ``_FrameProducer``: the + source's delivered-frame counter, anchored to the wall clock of the wait + that observed it. """ CACHE = 8 @@ -104,6 +108,11 @@ class _FingerProducer: self._cache: deque = deque(maxlen=self.CACHE) self._times: deque = deque() self._last_read_at: float | None = None + #: ``(clock, seconds, delivered frames)`` samples behind ``capture_fps``. + self._rate_samples: deque = deque() + self._capture_mark: tuple[float, int] | None = None + #: The wait's own anchor into the counter, consumed by ``capture_fps``. + self._repeat_check: tuple[float, int] | None = None def __call__(self): self._wait_for_a_new_frame() @@ -134,6 +143,10 @@ class _FingerProducer: def _wait_for_a_new_frame(self) -> None: if self.is_file or self.capture_seq is None: return + #: When the wait started, and what the counter said then: the two together + #: are the only honest basis for the camera's own rate when this loop is + #: slower than the camera and skips a frame entirely. + self._repeat_check = (time.perf_counter(), int(self.capture_seq)) deadline = time.perf_counter() + CAMERA_DISCONNECT_TIMEOUT_S while True: current = _capture_sequence(self.capture_source) @@ -168,6 +181,63 @@ class _FingerProducer: while self._times and self._times[0][1] < cutoff: self._times.popleft() + def capture_window(self, previous: tuple[float, int] | None) -> tuple[float, int]: + """``(seconds, delivered frames)`` over the interval since ``previous``. + + Both anchors have to be read together, and on this thread: the source's + counter advances on the capture thread whether or not a frame is taken, so + pairing a stale time with the current counter would measure the interval + between two *publishes* while crediting every frame delivered since an + earlier read — a rate that comes out at roughly double the camera's. + """ + if previous is None or self.capture_seq is None: + return 0.0, 0 + t0, delivered0 = previous + span = time.perf_counter() - t0 + delivered = int(self.capture_seq) - int(delivered0) + return (span, delivered) if span > 0.0 else (0.0, 0) + + def capture_fps(self) -> float: + """The capture source's own published rate, as a rate over a window. + + Counted from the source's delivered-frame counter between two of this + producer's reads, against the wall-clock between them, and summed over the + last second so one scheduler-jittered interval does not become the reported + rate. Only a source with that counter can answer — a live camera. A video + file decodes a new frame per call, so it has no "the camera had it and we did + not" to measure. + """ + if self.capture_seq is None: + return 0.0 + now = time.perf_counter() + span, delivered = self.capture_window(self._capture_mark) + self._capture_mark = (now, int(self.capture_seq)) + if span <= 0.0 or delivered <= 0: + # Credit what the repeat wait already observed: when this loop cannot + # keep up, no frame is read at all until the counter moves again, and a + # read-anchored window would then credit a whole interval's frames to a + # fraction of it. + span, delivered = self._repeat_window() + if span > 0.0 and delivered > 0: + self._rate_samples.append((now, span, delivered)) + cutoff = now - FPS_WINDOW_S + while len(self._rate_samples) > 1 and self._rate_samples[0][0] < cutoff: + self._rate_samples.popleft() + total_span = sum(sample[1] for sample in self._rate_samples) + total_frames = sum(sample[2] for sample in self._rate_samples) + return max(0.0, total_frames / total_span) if total_span > 0.0 else 0.0 + + def _repeat_window(self) -> tuple[float, int]: + """``(seconds, frames)`` the repeat wait saw, and then forget it.""" + previous = self._repeat_check + self._repeat_check = None + if previous is None or self._capture_mark is None or self.capture_seq is None: + return 0.0, 0 + t0, seq0 = previous + span = self._capture_mark[0] - t0 + frames = int(self.capture_seq) - int(seq0) + return (span, frames) if span > 0.0 and frames > 0 else (0.0, 0) + def resolve_finger_export_dir(config_ref: str | Path) -> Path | None: """Return local ``assets/30fps-0703`` when present and complete. @@ -441,22 +511,34 @@ class FingerAcquisitionWorker(RecordSinkMixin): raise RuntimeError("Finger runtime is not running") self._pending_reset = True - def _consume_pending_reset(self, runtime, frame_id: int, mono_t: float) -> None: - """Apply a queued reset and report the frame it actually took effect at.""" + def _consume_pending_reset( + self, runtime, frame_id: int, mono_t: float, *, pipeline=None + ) -> None: + """Apply a queued reset and report the frame it actually took effect at. + + A running stream pipeline owns the tracker's frame queue on its flow + thread. Resetting from this thread empties that queue while the flow + stage is indexing it (``list index out of range``) and the session ends. + ``pipeline.reset`` queues the same ``Sensor.reset`` for the flow stage, + which applies it before the next frame. + """ with self._lock: pending = self._pending_reset self._pending_reset = False session = self._session_id if not pending or runtime is None: return - tactile = getattr(runtime, "tactile", None) - target = tactile if callable(getattr(tactile, "reset", None)) else None - if target is None: - sensor = getattr(tactile, "raw_sensor", None) - target = sensor if callable(getattr(sensor, "reset", None)) else None - if target is None: - raise RuntimeError("runtime does not support reset") - target.reset() + if pipeline is not None: + pipeline.reset() + else: + tactile = getattr(runtime, "tactile", None) + target = tactile if callable(getattr(tactile, "reset", None)) else None + if target is None: + sensor = getattr(tactile, "raw_sensor", None) + target = sensor if callable(getattr(sensor, "reset", None)) else None + if target is None: + raise RuntimeError("runtime does not support reset") + target.reset() if self._on_reset_applied is not None: self._on_reset_applied(int(frame_id), float(mono_t), int(session)) @@ -754,9 +836,10 @@ class FingerAcquisitionWorker(RecordSinkMixin): # it was built: see the box worker. A paused worker publishes # nothing, including frames already in flight. continue - # A zeroing and new coefficients are applied by this thread at a - # frame boundary, then reported with the frame and time they took - # effect at — the same confirmation rule as the serial loop. + # Coefficients are applied by this thread at a frame boundary. + # A zeroing is queued onto the pipeline's flow thread, which + # owns the tracker, and reported with the frame and time it + # takes effect — the same confirmation rule as the serial loop. try: self._consume_pending_calibration( tactile, frame_id + 1, float(packet.mono_t or 0.0) @@ -771,7 +854,10 @@ class FingerAcquisitionWorker(RecordSinkMixin): self._on_error(errors[-1]) try: self._consume_pending_reset( - runtime, frame_id + 1, float(packet.mono_t or 0.0) + runtime, + frame_id + 1, + float(packet.mono_t or 0.0), + pipeline=pipe, ) except Exception as exc: # noqa: BLE001 message = f"reset failed: {type(exc).__name__}: {exc}" diff --git a/shared/src/orisys_app_shared/acquisition/sdk_worker.py b/shared/src/orisys_app_shared/acquisition/sdk_worker.py index 5042e43..833984d 100644 --- a/shared/src/orisys_app_shared/acquisition/sdk_worker.py +++ b/shared/src/orisys_app_shared/acquisition/sdk_worker.py @@ -959,18 +959,31 @@ class SdkAcquisitionWorker(RecordSinkMixin): raise RuntimeError("Sensor is not running") self._pending_reset = True - def _consume_pending_reset(self, sensor, frame_id: int, mono_t: float) -> None: - """Apply a queued reset and report the frame it actually took effect at.""" + def _consume_pending_reset( + self, sensor, frame_id: int, mono_t: float, *, pipeline=None + ) -> None: + """Apply a queued reset and report the frame it actually took effect at. + + A running stream pipeline owns the tracker's frame queue on its flow + thread. ``Sensor.reset`` from this thread empties that queue while the + flow stage is indexing it (``list index out of range``) and the session + ends. ``pipeline.reset`` queues the reset for the flow stage, which + applies it before the next frame. The serial loop has no other thread + in the tracker, so it still resets inline. + """ with self._lock: pending = self._pending_reset self._pending_reset = False session = self._session_id if not pending or sensor is None: return - reset = getattr(sensor, "reset", None) - if not callable(reset): - raise RuntimeError("sensor does not support reset") - reset() + if pipeline is not None: + pipeline.reset() + else: + reset = getattr(sensor, "reset", None) + if not callable(reset): + raise RuntimeError("sensor does not support reset") + reset() if self._on_reset_applied is not None: self._on_reset_applied(int(frame_id), float(mono_t), int(session)) @@ -1277,9 +1290,10 @@ class SdkAcquisitionWorker(RecordSinkMixin): # includes not handing the operator a packet they paused # before, so the in-flight frame is dropped here. continue - # A zeroing and new coefficients are applied by this thread at a - # frame boundary, then reported with the frame and time they took - # effect at — the same confirmation rule as the serial loop. + # Coefficients are applied by this thread at a frame boundary. + # A zeroing is queued onto the pipeline's flow thread, which + # owns the tracker, and reported with the frame and time it + # takes effect — the same confirmation rule as the serial loop. try: self._consume_pending_calibration( sensor, frame_id + 1, float(packet.mono_t or 0.0) @@ -1296,7 +1310,10 @@ class SdkAcquisitionWorker(RecordSinkMixin): self._on_error(errors[-1]) try: self._consume_pending_reset( - sensor, frame_id + 1, float(packet.mono_t or 0.0) + sensor, + frame_id + 1, + float(packet.mono_t or 0.0), + pipeline=pipe, ) except Exception as exc: # noqa: BLE001 message = f"reset failed: {type(exc).__name__}: {exc}" diff --git a/shared/tests/test_calibration_identity.py b/shared/tests/test_calibration_identity.py index 5657cfd..33cfcc7 100644 --- a/shared/tests/test_calibration_identity.py +++ b/shared/tests/test_calibration_identity.py @@ -129,6 +129,61 @@ def test_a_zeroing_is_applied_by_the_worker_and_reported_with_its_frame(): assert events == [(42, 1.4, 2)] +def test_a_live_zeroing_is_queued_on_the_pipeline(): + """The stream path must not reset the tracker on the consumer thread. + + The flow stage is indexing the tracker's frame queue at the same time. + Clearing it here is the ``list index out of range`` that ends the session. + """ + events: list[tuple] = [] + worker = _worker(on_reset_applied=lambda *args: events.append(args)) + sensor = _Sensor() + queued: list[int] = [] + + class _Pipeline: + def reset(self) -> None: + queued.append(sensor.resets) + + worker._session_id = 2 + worker.reset() + worker._consume_pending_reset( + sensor, frame_id=42, mono_t=1.4, pipeline=_Pipeline() + ) + assert sensor.resets == 0, "the live zeroing ran on the consumer thread" + assert queued == [0] + assert events == [(42, 1.4, 2)] + + +def test_a_live_finger_zeroing_is_queued_on_the_pipeline(): + from orisys_app_shared.acquisition.finger_worker import FingerAcquisitionWorker + + events: list[tuple] = [] + worker = FingerAcquisitionWorker(on_reset_applied=lambda *args: events.append(args)) + tactile_resets: list[int] = [] + + class _Tactile: + def reset(self) -> None: + tactile_resets.append(1) + + class _Runtime: + tactile = _Tactile() + + queued: list[int] = [] + + class _Pipeline: + def reset(self) -> None: + queued.append(len(tactile_resets)) + + worker._session_id = 3 + worker.reset() + worker._consume_pending_reset( + _Runtime(), frame_id=7, mono_t=0.2, pipeline=_Pipeline() + ) + assert tactile_resets == [], "the finger zeroing ran on the consumer thread" + assert queued == [0] + assert events == [(7, 0.2, 3)] + + def test_a_started_session_drops_a_stale_zeroing_request(): root = _root() events: list[tuple] = [] diff --git a/shared/tests/test_stream_pipeline_acquisition.py b/shared/tests/test_stream_pipeline_acquisition.py index c360cc0..4c6ef93 100644 --- a/shared/tests/test_stream_pipeline_acquisition.py +++ b/shared/tests/test_stream_pipeline_acquisition.py @@ -416,3 +416,32 @@ def test_a_failed_chain_marks_the_frame_rather_than_claiming_its_forces(): assert packet.contact.centroid_uv_xy is None finally: _restore(saved) + + +def test_the_finger_producer_reports_the_camera_rate(monkeypatch): + """The first live frame asks the producer for ``capture_fps``. + + That call is what ended a finger session with + ``'_FingerProducer' object has no attribute 'capture_fps'``. A file has no + delivered-frame counter, so it reports nothing; a live source reports the + frames its counter advanced over the wall-clock between two reads. + """ + from orisys_app_shared.acquisition import finger_worker + + clock = {"t": 1000.0} + monkeypatch.setattr(finger_worker.time, "perf_counter", lambda: clock["t"]) + + file_source = types.SimpleNamespace(running=False, status=types.SimpleNamespace(last_seq=0)) + file_runtime = types.SimpleNamespace(capture_source=file_source) + file_producer = finger_worker._FingerProducer(file_runtime, is_file=True) + assert file_producer.capture_fps() == 0.0 + + status = types.SimpleNamespace(last_seq=10, dropped_frames=0) + live = types.SimpleNamespace(running=True, status=status) + producer = finger_worker._FingerProducer( + types.SimpleNamespace(capture_source=live), is_file=False + ) + assert producer.capture_fps() == 0.0 + clock["t"] = 1001.0 + producer.capture_seq = 40 + assert producer.capture_fps() == pytest.approx(30.0)