ykcheungab
371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""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
|
|
#: What the driver reported back for the negotiated stream.
|
|
self.width = 1920
|
|
self.height = 1080
|
|
self.fps = 30.0
|
|
self.fourcc = "MJPG"
|
|
self.backend_name = "MSMF"
|
|
|
|
|
|
class _CaptureConfig:
|
|
"""The requested stream, as ``CaptureConfig`` carries it."""
|
|
|
|
def __init__(self) -> None:
|
|
self.width = 1920
|
|
self.height = 1080
|
|
self.fps = 30.0
|
|
self.fourcc = "MJPG"
|
|
self.source = 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". It also counts
|
|
a drop whenever it replaces a frame nobody took, like
|
|
``orisys.capture.source.CaptureSource._background_loop`` does.
|
|
"""
|
|
|
|
def __init__(self, period_s: float = FRAME_PERIOD_S) -> None:
|
|
self.status = _Status()
|
|
self.config = _CaptureConfig()
|
|
self.running = True
|
|
self._period = float(period_s)
|
|
self._stop = threading.Event()
|
|
self._taken_seq = 0
|
|
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
|
|
# Replacing a frame that was never consumed is a drop on our side.
|
|
if self.status.last_seq > self._taken_seq:
|
|
self.status.dropped_frames += 1
|
|
self.status.last_seq += 1
|
|
|
|
def take(self) -> None:
|
|
self._taken_seq = self.status.last_seq
|
|
|
|
def close(self) -> None:
|
|
self.running = False
|
|
self._stop.set()
|
|
self._thread.join(timeout=1.0)
|
|
|
|
|
|
#: Per-frame processing cost the fake device charges, so a test can make the
|
|
#: pipeline slower than the camera (the real box chain costs ~10 ms at 480p).
|
|
PROCESS_COST_S = 0.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. A test can make this cost time, so
|
|
# the loop can be put under a pipeline slower than the camera.
|
|
if PROCESS_COST_S > 0:
|
|
time.sleep(PROCESS_COST_S)
|
|
self.capture_source.take()
|
|
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_the_camera_s_own_rate_is_reported_next_to_ours(monkeypatch):
|
|
"""A low frame rate must be attributable, not just visible.
|
|
|
|
With a pipeline that fits the frame period the two rates agree. With one that
|
|
does not, the *camera's* rate stays at 30 while ours falls and the source
|
|
counts the frames nobody took — which is the whole difference between "the
|
|
camera is slow" and "we are slow".
|
|
"""
|
|
packets, _ = _collect(duration_s=1.5)
|
|
assert packets, "the loop emitted nothing at all"
|
|
camera = [float(packet.capture_fps) for packet in packets if packet.capture_fps]
|
|
assert camera, "no packet carried the capture source's own rate"
|
|
assert 28.0 <= camera[-1] <= 31.0, f"the fake camera publishes at 30 fps, got {camera[-1]}"
|
|
assert packets[-1].capture_dropped == 0, (
|
|
f"the loop dropped {packets[-1].capture_dropped} frames while keeping up"
|
|
)
|
|
|
|
# Now make the pipeline three times slower than the camera's own period.
|
|
monkeypatch.setattr(sys.modules[__name__], "PROCESS_COST_S", 0.06)
|
|
slow, _ = _collect(duration_s=2.0)
|
|
try:
|
|
assert slow, "the loop emitted nothing at all"
|
|
ours = _rate(slow)
|
|
assert ours < 22.0, f"a 60 ms pipeline still kept up at {ours:.1f} fps"
|
|
assert 28.0 <= float(slow[-1].capture_fps) <= 31.0, (
|
|
"the camera's rate was reported as slow too: "
|
|
f"{slow[-1].capture_fps}"
|
|
)
|
|
assert slow[-1].capture_dropped > 5, (
|
|
"the loop fell behind without recording a single dropped frame: "
|
|
f"{slow[-1].capture_dropped}"
|
|
)
|
|
finally:
|
|
monkeypatch.setattr(sys.modules[__name__], "PROCESS_COST_S", 0.0)
|
|
|
|
|
|
def test_a_low_camera_rate_is_recorded_as_a_negotiation(monkeypatch):
|
|
"""A camera that cannot do the configured rate must be visible as such.
|
|
|
|
The pipeline cannot turn 16.7 fps into 30, so the only honest thing left is to
|
|
record what was asked for and what the driver gave, and to have a one-line
|
|
description ready for the operator.
|
|
"""
|
|
from orisys_app_shared.acquisition.sdk_worker import (
|
|
SdkAcquisitionWorker,
|
|
short_capture_fallacy,
|
|
)
|
|
|
|
CREATED.clear()
|
|
saved = _install()
|
|
worker = SdkAcquisitionWorker()
|
|
try:
|
|
worker.start(0, config_ref=BOX_CONFIG)
|
|
deadline = time.perf_counter() + 2.0
|
|
while time.perf_counter() < deadline and not CREATED:
|
|
time.sleep(0.01)
|
|
assert CREATED, "the fake sensor was never constructed"
|
|
source = CREATED[-1].capture_source
|
|
source.status.fps = 16.67 # what the driver actually negotiated
|
|
status = worker.capture_status()
|
|
assert status["requested"]["fps"] == 30.0
|
|
assert status["requested"]["width"] == 1920
|
|
assert status["negotiated"]["fps"] == 16.67, status
|
|
assert status["negotiated"]["fourcc"] == "MJPG"
|
|
assert status["negotiated"]["backend"] == "MSMF"
|
|
detail = short_capture_fallacy(status)
|
|
assert "1920x1080" in detail and "16.67" in detail and "30" in detail, detail
|
|
# A driver that gives what was asked for says nothing at all.
|
|
source.status.fps = 30.0
|
|
assert short_capture_fallacy(worker.capture_status()) == ""
|
|
finally:
|
|
worker.stop(wait=True)
|
|
_restore(saved)
|
|
|
|
|
|
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)
|