Merge branch 'main' of ssh://892887cvmk08.aicp.io:24928/orisys-tech/orisys-app

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

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

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] = []

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)