From e299775c4416b7afb4cbe856ebd76a9f33831e36 Mon Sep 17 00:00:00 2001 From: x01dc Date: Mon, 27 Jul 2026 10:35:38 +0200 Subject: [PATCH] feat(LamNI): measure camera live-mode FPS and smear-sweep capture FPS Neither rate was previously measured anywhere, making it hard to tell a real acquisition bottleneck from a display/overhead one. - IDSCamera.get_live_fps() (new USER_ACCESS entry): rolling-buffer measurement of the live-mode push rate in _live_mode_loop(), readable as dev.cam_xeye.get_live_fps(). SimIDSCamera inherits it unchanged, so it works against the simulated deployment too. - XrayEyeAlign._smear_sweep() now tracks timestamps of the first and most recent distinct frame it captures, storing the computed rate as self.last_smear_fps and printing it in the sweep-done summary line. Co-Authored-By: Claude Sonnet 5 --- .../plugins/LamNI/x_ray_eye_align.py | 20 +++++++- csaxs_bec/devices/ids_cameras/ids_camera.py | 17 +++++++ .../test_x_ray_eye_align.py | 47 +++++++++++++++++++ tests/tests_devices/test_ids_camera.py | 31 ++++++++++++ 4 files changed, 114 insertions(+), 1 deletion(-) diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py index 9abe225..b589805 100644 --- a/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py +++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/x_ray_eye_align.py @@ -83,6 +83,9 @@ class XrayEyeAlign: # k=9 : sample at 315° self.alignment_values: dict[int, list[float]] = {} self.alignment_images = [] + # Measured capture rate of the most recent _smear_sweep() call, in Hz. + # None until a smear sweep has run. + self.last_smear_fps: float | None = None # ------------------------------------------------------------------ # GUI shortcut @@ -745,11 +748,17 @@ class XrayEyeAlign: last_frame = None frames_seen = 0 last_push = time.time() + # Timestamps of the first captured frame and the most recent distinct + # one, used to compute the actual smear capture rate below -- bounded + # above by the camera's own live-mode push rate (see get_live_fps()). + first_frame_time = None + last_distinct_time = None try: while report.status != "COMPLETED": frame = dev.cam_xeye.get_last_image() if composite is None: composite = frame.copy() + first_frame_time = time.time() elif frame.shape != composite.shape: print( f"[rotation-center][smear] skipping frame with mismatched shape " @@ -758,6 +767,7 @@ class XrayEyeAlign: else: if last_frame is None or not np.array_equal(frame, last_frame): frames_seen += 1 + last_distinct_time = time.time() composite = np.maximum(composite, frame) last_frame = frame if time.time() - last_push > display_update_interval_s: @@ -775,9 +785,17 @@ class XrayEyeAlign: self.gui.set_smear_active(False) if not keep_shutter_open: dev.fsh.fshclose() + elapsed = ( + last_distinct_time - first_frame_time + if frames_seen >= 1 and last_distinct_time is not None + else 0.0 + ) + self.last_smear_fps = frames_seen / elapsed if elapsed > 0 else None + fps_str = f"{self.last_smear_fps:.2f}" if self.last_smear_fps is not None else "n/a" print( f"[rotation-center][smear] sweep done -- {frames_seen} distinct frames over " - f"{sweep_deg:.1f} deg, now at {self.get_tomo_angle():.1f} deg" + f"{sweep_deg:.1f} deg, now at {self.get_tomo_angle():.1f} deg " + f"(capture rate: {fps_str} Hz)" ) # Note: the GUI stays on the "smear_preview" channel (showing this # composite) until the caller has collected the operator's click -- diff --git a/csaxs_bec/devices/ids_cameras/ids_camera.py b/csaxs_bec/devices/ids_cameras/ids_camera.py index 381e959..2ea5bdc 100644 --- a/csaxs_bec/devices/ids_cameras/ids_camera.py +++ b/csaxs_bec/devices/ids_cameras/ids_camera.py @@ -3,6 +3,8 @@ from __future__ import annotations import threading +import time +from collections import deque from typing import TYPE_CHECKING, Literal import numpy as np @@ -70,6 +72,7 @@ class IDSCamera(PSIDeviceBase): "get_last_image", "push_preview_image", "push_smear_preview", + "get_live_fps", ] def __init__( @@ -101,6 +104,9 @@ class IDSCamera(PSIDeviceBase): 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() + # 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, m_n_colormode=m_n_colormode, @@ -199,9 +205,20 @@ class IDSCamera(PSIDeviceBase): except Exception as e: logger.error(f"Error in live mode loop: {e}") break + self._live_frame_times.append(time.time()) stop_event.wait(0.2) # 5 Hz 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_bec_ipython_client/test_x_ray_eye_align.py b/tests/tests_bec_ipython_client/test_x_ray_eye_align.py index 4aeace9..3bbc25a 100644 --- a/tests/tests_bec_ipython_client/test_x_ray_eye_align.py +++ b/tests/tests_bec_ipython_client/test_x_ray_eye_align.py @@ -424,6 +424,53 @@ def test_smear_sweep_max_projection(bec_client_mock): assert [c.args[0] for c in align.gui.set_smear_active.call_args_list] == [True, False] +def test_smear_sweep_computes_capture_fps(bec_client_mock): + client = bec_client_mock + align, dev_mock = _make_calibration_align(client) + + frame_a = np.array([[1, 5], [3, 2]], dtype=np.uint8) + frame_b = np.array([[4, 2], [3, 6]], dtype=np.uint8) + frame_c = np.array([[0, 1], [9, 0]], dtype=np.uint8) + dev_mock.cam_xeye.get_last_image.side_effect = [frame_a, frame_b, frame_c] + report = _report_with_status_sequence("RUNNING", "RUNNING", "RUNNING", "COMPLETED") + + # Deterministic clock: each time.time() call advances by 0.05s. With + # frame_a (first frame, uncounted per the existing off-by-one) then + # frame_b and frame_c as the 2 distinct frames that follow, the elapsed + # time between the first frame and the last distinct one spans exactly + # 4 of these ticks (0.20s) -- see the call trace in _smear_sweep(). + clock = {"t": 0.0} + + def fake_time(): + clock["t"] += 0.05 + return clock["t"] + + with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock): + with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report): + with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"): + with mock.patch(f"{XRAY_EYE_ALIGN}.time.time", side_effect=fake_time): + align._smear_sweep() + + assert align.last_smear_fps == pytest.approx(10.0) + + +def test_smear_sweep_fps_none_with_fewer_than_two_distinct_frames(bec_client_mock): + client = bec_client_mock + align, dev_mock = _make_calibration_align(client) + + frame_a = np.array([[1, 5], [3, 2]], dtype=np.uint8) + dev_mock.cam_xeye.get_last_image.return_value = frame_a + report = _report_with_status_sequence("RUNNING", "COMPLETED") + + with mock.patch(f"{XRAY_EYE_ALIGN}.dev", dev_mock): + with mock.patch(f"{XRAY_EYE_ALIGN}.mv", return_value=report): + with mock.patch(f"{XRAY_EYE_ALIGN}.time.sleep"): + align._smear_sweep() + + # only the first (uncounted) frame was seen -- frames_seen stays 0 + assert align.last_smear_fps is None + + def test_smear_sweep_issues_single_continuous_move(bec_client_mock): client = bec_client_mock align, dev_mock = _make_calibration_align(client) diff --git a/tests/tests_devices/test_ids_camera.py b/tests/tests_devices/test_ids_camera.py index e6897d0..1c639b4 100644 --- a/tests/tests_devices/test_ids_camera.py +++ b/tests/tests_devices/test_ids_camera.py @@ -1,5 +1,6 @@ """Unit tests for the IDS Camera device.""" +import threading from unittest import mock import numpy as np @@ -81,6 +82,36 @@ def test_get_last_image(ids_camera): assert np.array_equal(result, test_image) +def test_get_live_fps_none_before_frames(ids_camera): + """No frames pushed yet -> get_live_fps() returns None.""" + assert ids_camera.get_live_fps() is None + + +def test_get_live_fps_computes_rate(ids_camera): + """get_live_fps() computes rate from the recorded push timestamps.""" + ids_camera._live_frame_times.extend([0.0, 0.2, 0.4, 0.6, 0.8]) + assert ids_camera.get_live_fps() == pytest.approx(5.0) + + +def test_live_mode_loop_records_frame_times(ids_camera): + """_live_mode_loop() appends one timestamp per pushed frame.""" + ids_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 + ids_camera._live_mode_loop(stop_event) + + assert len(ids_camera._live_frame_times) == 3 + assert ids_camera.get_live_fps() is not None + + def test_on_destroy(ids_camera): """Test the on_destroy method to ensure it cleans up resources.""" ids_camera.cam.on_disconnect = mock.Mock()