feat(devices): add live-mode FPS measurement to AlliedVisionAravisCamera
CI for csaxs_bec / test (push) Failing after 1m42s
CI for csaxs_bec / test (pull_request) Failing after 1m38s

Port get_live_fps()/live_mode_poll_interval_s from IDSCamera (added on
main): a rolling deque of the last 10 live-mode push timestamps, exposed
as a measured frame rate rather than assuming the nominal poll interval
is actually achieved -- verified against real hardware, where the
Alvium G1-507m's actual acquisition/readout time brings the achievable
rate to ~3 fps despite the 0.2s (5 Hz) nominal poll interval.

Mirrors the same three unit tests added for IDSCamera's version.
This commit is contained in:
x01dc
2026-07-28 10:26:40 +02:00
parent 13326c0985
commit 285691b629
2 changed files with 57 additions and 1 deletions
@@ -3,6 +3,8 @@
from __future__ import annotations
import threading
import time
from collections import deque
from typing import TYPE_CHECKING
import numpy as np
@@ -62,6 +64,7 @@ class AlliedVisionAravisCamera(PSIDeviceBase):
"set_exposure_time",
"get_gain",
"set_gain",
"get_live_fps",
]
def __init__(
@@ -74,6 +77,7 @@ class AlliedVisionAravisCamera(PSIDeviceBase):
live_mode: bool = False,
num_rotation_90: int = 0,
transpose: bool = False,
live_mode_poll_interval_s: float = 0.2,
**kwargs,
):
"""Initialize the Allied Vision Camera.
@@ -85,10 +89,18 @@ class AlliedVisionAravisCamera(PSIDeviceBase):
prefix (str): Prefix for the device.
scan_info (ScanInfo | None): Scan information for the device.
live_mode (bool): Whether to enable live mode for the camera.
live_mode_poll_interval_s (float): Delay between frame grabs in
the live-mode loop. Lower this for cameras/use cases that
need a higher live-mode push rate; the achievable rate is
still bounded by the camera's own exposure/acquisition time.
"""
super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
self._live_mode_thread: threading.Thread | None = None
self._stop_live_mode_event: threading.Event = threading.Event()
self._live_mode_poll_interval_s = live_mode_poll_interval_s
# Rolling buffer of push timestamps from _live_mode_loop, used to
# measure the actual live-mode frame rate (see get_live_fps()).
self._live_frame_times: deque[float] = deque(maxlen=10)
self.cam = Camera(camera_id=camera_id, connect=False)
self._inputs = {"live_mode": live_mode}
self._mask = np.zeros((1, 1), dtype=np.uint8)
@@ -181,9 +193,20 @@ class AlliedVisionAravisCamera(PSIDeviceBase):
except Exception as e:
logger.error(f"Error in live mode loop: {e}")
break
stop_event.wait(0.2) # 5 Hz
self._live_frame_times.append(time.time())
stop_event.wait(self._live_mode_poll_interval_s)
self.cam.set_camera_rate_limiting(False)
def get_live_fps(self) -> float | None:
"""Measured live-mode frame rate, from the timestamps of the last
pushed frames. Returns None if live mode hasn't pushed at least two
frames yet (e.g. not running, or just started).
"""
times = self._live_frame_times
if len(times) < 2:
return None
return (len(times) - 1) / (times[-1] - times[0])
def process_data(self, image: np.ndarray | None):
"""Process the image data before sending it to the preview signal."""
if image is None:
@@ -1,5 +1,6 @@
"""Unit tests for the AlliedVisionAravisCamera device."""
import threading
from unittest import mock
import numpy as np
@@ -80,6 +81,38 @@ def test_get_last_image(allied_vision_camera):
assert np.array_equal(result, test_image)
def test_get_live_fps_none_before_frames(allied_vision_camera):
"""No frames pushed yet -> get_live_fps() returns None."""
assert allied_vision_camera.get_live_fps() is None
def test_get_live_fps_computes_rate(allied_vision_camera):
"""get_live_fps() computes rate from the recorded push timestamps."""
allied_vision_camera._live_frame_times.extend([0.0, 0.2, 0.4, 0.6, 0.8])
assert allied_vision_camera.get_live_fps() == pytest.approx(5.0)
def test_live_mode_loop_records_frame_times(allied_vision_camera):
"""_live_mode_loop() appends one timestamp per pushed frame."""
allied_vision_camera.cam.get_image_data = mock.Mock(
return_value=np.zeros((2, 2), dtype=np.uint8)
)
stop_event = threading.Event()
calls = {"n": 0}
def fake_wait(timeout):
calls["n"] += 1
if calls["n"] >= 3:
stop_event.set()
return False
stop_event.wait = fake_wait
allied_vision_camera._live_mode_loop(stop_event)
assert len(allied_vision_camera._live_frame_times) == 3
assert allied_vision_camera.get_live_fps() is not None
def test_on_destroy(allied_vision_camera):
"""Test the on_destroy method to ensure it cleans up resources."""
allied_vision_camera.cam.on_disconnect = mock.Mock()