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 --

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:

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)

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()