diff --git a/csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py b/csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py index ce3ecdb..5e41faf 100644 --- a/csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py +++ b/csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py @@ -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: diff --git a/tests/tests_devices/test_allied_vision_aravis_camera.py b/tests/tests_devices/test_allied_vision_aravis_camera.py index 587da56..b3134d0 100644 --- a/tests/tests_devices/test_allied_vision_aravis_camera.py +++ b/tests/tests_devices/test_allied_vision_aravis_camera.py @@ -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()