orisys-app/shared/tests/test_worker_frame_rate.py

"""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: ~100 frames per second.
FRAME_PERIOD_S = 0.01
#: 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)
        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_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)


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)