"""Which frame the acquisition loop counts, and at what rate. ``CaptureSource.read_latest`` returns the newest frame from its background reader and, if that reader has not produced another one yet, the *same* frame again. The loop must not treat the repeat as a new sample: doing so duplicated every measurement (a 30 fps device recorded 63 rows per second) and the rail read the loop's own polling rate as the device's frame rate. """ from __future__ import annotations import sys import threading import time import types from pathlib import Path APP_ROOT = Path(__file__).resolve().parents[2] BOX_CONFIG = APP_ROOT / "apps" / "studio" / "config" / "sensors" / "box" / "sensor.json" #: The fake device's own frame period: 30 frames per second, the rate the studio #: sensor configs declare. FRAME_PERIOD_S = 1.0 / 30.0 #: How many times the SDK hands the same frame back before the next one arrives. REPEATS_PER_FRAME = 3 class _Status: """Stands in for ``orisys.capture.config.CaptureStatus``.""" def __init__(self) -> None: self.last_seq = 0 self.dropped_frames = 0 class _CaptureSource: """A background reader that publishes a new frame on its own thread. The real one is a thread too: the sequence advances whether or not the consumer asks for a frame, which is exactly why an unchanged sequence means "no new frame" rather than "the consumer has not called yet". """ def __init__(self, period_s: float = FRAME_PERIOD_S) -> None: self.status = _Status() self.running = True self._period = float(period_s) self._stop = threading.Event() self._thread = threading.Thread(target=self._publish, daemon=True) self._thread.start() def _publish(self) -> None: while not self._stop.is_set(): if self._stop.wait(timeout=self._period): return self.status.last_seq += 1 def close(self) -> None: self.running = False self._stop.set() self._thread.join(timeout=1.0) class _Info: """The channel names the worker reads off ``sensor.info``.""" FNORMAL = "FNORMAL" FSHEARX = "FSHEARX" FSHEARY = "FSHEARY" FNORMAL_RAW = "FNORMAL_RAW" FSHEARX_RAW = "FSHEARX_RAW" FSHEARY_RAW = "FSHEARY_RAW" #: Every sensor the fake SDK handed out, so a test can read its frame counter. CREATED: list = [] class _Sensor: def __init__(self, *args, **kwargs) -> None: import numpy as np self.configured: dict = {} self.connected = True self.capture_source = _CaptureSource() self.frame = np.zeros((8, 8, 3), dtype=np.uint8) self.info = _Info() self.force_calibrated = True CREATED.append(self) def configure(self, **kwargs): self.configured = kwargs def read_info(self, *names): """The SDK's readout: one value per requested channel, in order.""" values = { _Info.FNORMAL: 1.5, _Info.FSHEARX: 0.25, _Info.FSHEARY: -0.25, _Info.FNORMAL_RAW: 15.0, _Info.FSHEARX_RAW: 2.5, _Info.FSHEARY_RAW: -2.5, } return tuple(values.get(str(name), 0.0) for name in names) def get_img(self): # ``read_latest``: hand back the newest frame immediately, the same one # again when nothing new has arrived. return self.frame def disconnect(self): self.connected = False self.capture_source.close() def _install() -> object: saved = {name: sys.modules.get(name) for name in ("orisys", "orisys.finger")} module = types.ModuleType("orisys") module.__version__ = "fake" module.Sensor = _Sensor # type: ignore[attr-defined] finger_module = types.ModuleType("orisys.finger") finger_module.FingerRuntime = _Sensor # type: ignore[attr-defined] module.finger = finger_module # type: ignore[attr-defined] sys.modules["orisys"] = module sys.modules["orisys.finger"] = finger_module return saved def _restore(saved) -> None: for name, original in saved.items(): if original is None: sys.modules.pop(name, None) else: sys.modules[name] = original def _collect(duration_s: float = 1.2): """Run the real worker loop against the fake device. Returns ``(packets, delivered)`` — what the loop recorded, and how many frames the fake capture source actually delivered in the same period. """ from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker CREATED.clear() saved = _install() worker = SdkAcquisitionWorker() packets: list = [] lock = threading.Lock() def sink(packet) -> None: with lock: packets.append(packet) try: worker.add_sink(sink) # Camera id 0, not "0": a string source is a file, and a file runs on the # video's timeline rather than the wall clock (see the file test below). worker.start(0, config_ref=BOX_CONFIG) time.sleep(duration_s) worker.stop(wait=True) delivered = CREATED[-1].capture_source.status.last_seq if CREATED else 0 finally: _restore(saved) return packets, int(delivered) def _rate(packets) -> float: span = float(packets[-1].mono_t) - float(packets[0].mono_t) return (len(packets) - 1) / span if span > 0 else 0.0 def test_the_same_frame_is_not_counted_twice(): """One packet per delivered frame, whatever the loop's polling rate is.""" packets, delivered = _collect() assert packets, "the loop emitted nothing at all" assert delivered > 10, f"the fake device only delivered {delivered} frames" assert len(packets) <= delivered * 1.5, ( f"{len(packets)} packets for {delivered} delivered frames: the repeats " "are being recorded as samples" ) assert len(packets) >= delivered * 0.7, ( f"only {len(packets)} packets for {delivered} delivered frames" ) def test_a_30_fps_device_is_consumed_at_30_fps(): """The loop must not add a timer tick of latency per frame. ``time.sleep(0.002)`` costs a full 15.6 ms on Windows, so a loop that sleeps before looking for the next frame spends ~47 ms per frame on a 30 fps device and records 16.7 fps with half the frames dropped. """ packets, delivered = _collect(duration_s=1.5) assert packets, "the loop emitted nothing at all" rate = _rate(packets) assert 27.0 <= rate <= 31.5, ( f"a 30 fps device was consumed at {rate:.1f} fps " f"({len(packets)} packets for {delivered} delivered frames)" ) def test_the_reported_rate_is_the_delivered_rate(): """``packet.fps`` describes the frames, not how often the loop asked.""" packets, delivered = _collect() reported = [float(packet.fps) for packet in packets if packet.fps] assert reported, "no packet carried a frame rate" measured = _rate(packets) settled = reported[-5:] assert all(value > 10.0 for value in settled), settled assert abs(settled[-1] - measured) < 0.25 * measured, (settled[-1], measured) def test_a_file_source_runs_on_the_video_s_own_timeline(): """A pass over a video is stamped with the video's time, not the wall clock.""" from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker CREATED.clear() saved = _install() worker = SdkAcquisitionWorker() packets: list = [] lock = threading.Lock() def sink(packet) -> None: with lock: packets.append(packet) try: worker.add_sink(sink) worker.start("recorded.mp4", config_ref=BOX_CONFIG) time.sleep(0.4) worker.stop(wait=True) finally: _restore(saved) assert len(packets) > 3, f"the pass produced {len(packets)} packets" steps = [ round(packets[index + 1].mono_t - packets[index].mono_t, 6) for index in range(min(5, len(packets) - 1)) ] assert packets[0].mono_t == 0.0, packets[0].mono_t assert all(abs(step - 1.0 / 30.0) < 1e-6 for step in steps), steps assert packets[-1].mono_t < 0.4 + 1.0, ( f"the pass ran on the wall clock: {packets[-1].mono_t} s of video in " f"{0.4} s of real time" ) if __name__ == "__main__": # pragma: no cover - manual runner failures = 0 for name, function in sorted(globals().items()): if name.startswith("test_") and callable(function): try: function() print(f" ok {name}") except Exception as exc: # noqa: BLE001 failures += 1 print(f" FAIL {name}: {type(exc).__name__}: {exc}") print(f"failures: {failures}") raise SystemExit(1 if failures else 0)