feat(flomni, lamni): end-of-scan statistics with the time split: scan time (kickoff -> end in the RT loop), exposure (points x requested exp_time x frames, passed with the kickoff), overhead per point

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QnSJvqxh9oDbshx8qXgeiX

csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py

@@ -13,6 +13,7 @@ from ophyd_devices.utils.socket import SocketIO, raise_if_disconnected
 from prettytable import PrettyTable
 
 from csaxs_bec.devices.omny.rt.rt_ophyd import (
+    format_scan_timing,
     RtCommunicationError,
     RtError,
     RtReadbackSignal,
@@ -816,6 +817,11 @@ class RtFlomniFlyer(Device):
         )
         self.average_stdeviations_x_st_fzp = 0
         self.average_stdeviations_y_st_fzp = 0
+        # end-of-scan statistics: kickoff parameters from the scan, start / end of the scan
+        self._exp_time = None
+        self._frames_per_trigger = 1
+        self._t_kickoff = None
+        self._t_scan_end = None
         self.readout_thread = None
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
@@ -856,6 +862,8 @@ class RtFlomniFlyer(Device):
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
 
+        self._t_scan_end = time.monotonic()  # scan finished in the RT loop (statistics)
+
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
             # if we are here, the shutdown_event is set. We can exit the readout loop.
@@ -890,6 +898,8 @@ class RtFlomniFlyer(Device):
             f" ({read_counter} points, sample vs FZP): x"
             f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f} nm, y"
             f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f} nm"
+            + format_scan_timing(read_counter, self._t_kickoff, self._t_scan_end, self._exp_time,
+                                 self._frames_per_trigger)
         )
         logger.info(msg)
         try:
@@ -932,12 +942,22 @@ class RtFlomniFlyer(Device):
         )
         self.readout_thread.start()
 
+    def configure(self, d: dict) -> tuple:
+        """The device server calls this with the kickoff parameters of the scan before kickoff():
+        exp_time / frames_per_trigger for the end-of-scan statistics (not passed: no exposure part)."""
+        d = dict(d or {})
+        self._exp_time = d.pop("exp_time", None)
+        self._frames_per_trigger = d.pop("frames_per_trigger", 1) or 1
+        return super().configure(d) if d else ({}, {})
+
     def kickoff(self) -> DeviceStatus:
         self.shutdown_event.clear()
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
         self.controller.start_scan()
+        self._t_kickoff = time.monotonic()
+        self._t_scan_end = None
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
         status.set_finished()

csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py

@@ -13,7 +13,7 @@ from ophyd_devices.utils.controller import Controller, retry_once, threadlocked
 from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
 from prettytable import PrettyTable
 
-from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtError
+from csaxs_bec.devices.omny.rt.rt_ophyd import RtCommunicationError, RtError, format_scan_timing
 
 logger = bec_logger.logger
 
@@ -752,6 +752,11 @@ class RtLamniFlyer(Device):
         self.average_stdeviations_x_st_fzp = 0
         self.average_stdeviations_y_st_fzp = 0
         self.average_lamni_angle = 0
+        # end-of-scan statistics: kickoff parameters from the scan, start / end of the scan
+        self._exp_time = None
+        self._frames_per_trigger = 1
+        self._t_kickoff = None
+        self._t_scan_end = None
         self.readout_thread = None
         self.scan_done_event = threading.Event()
         self.scan_done_event.set()
@@ -767,12 +772,22 @@ class RtLamniFlyer(Device):
         )
         self.readout_thread.start()
 
+    def configure(self, d: dict) -> tuple:
+        """The device server calls this with the kickoff parameters of the scan before kickoff():
+        exp_time / frames_per_trigger for the end-of-scan statistics (not passed: no exposure part)."""
+        d = dict(d or {})
+        self._exp_time = d.pop("exp_time", None)
+        self._frames_per_trigger = d.pop("frames_per_trigger", 1) or 1
+        return super().configure(d) if d else ({}, {})
+
     def kickoff(self) -> DeviceStatus:
         self.shutdown_event.clear()
         self.scan_done_event.clear()
         while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001):
             ...
         self.controller.start_scan()
+        self._t_kickoff = time.monotonic()
+        self._t_scan_end = None
         self.shutdown_event.wait(0.1)
         status = DeviceStatus(self)
         status.set_finished()
@@ -834,6 +849,8 @@ class RtLamniFlyer(Device):
                     signals = self._get_signals_from_table(return_table)
                     self.data.set(signals)
 
+        self._t_scan_end = time.monotonic()  # scan finished in the RT loop (statistics)
+
         if self.shutdown_event.wait(0.05):
             logger.info("Shutdown event set, stopping readout.")
             # if we are here, the shutdown_event is set. We can exit the readout loop.
@@ -870,6 +887,8 @@ class RtLamniFlyer(Device):
             f" {self.average_stdeviations_x_st_fzp/read_counter*1000:.1f} nm, y"
             f" {self.average_stdeviations_y_st_fzp/read_counter*1000:.1f} nm;"
             f" average angle {self.average_lamni_angle/read_counter:.4f}"
+            + format_scan_timing(read_counter, self._t_kickoff, self._t_scan_end, self._exp_time,
+                                 self._frames_per_trigger)
         )
         logger.info(msg)
         try:

csaxs_bec/devices/omny/rt/rt_ophyd.py

@@ -14,6 +14,26 @@ from ophyd_devices.utils.socket import SocketSignal
 logger = bec_logger.logger
 
 
+def format_scan_timing(n_points: int, t_start, t_end, exp_time=None, frames_per_trigger=1) -> str:
+    """Time split of a flyer scan for the end-of-scan statistics (Mirko 2026-10-26, BEC side only): scan time from
+    the kickoff to the end of the scan in the RT loop (wall clock of the device server), exposure = points x
+    requested exp_time x frames_per_trigger (not measured: the detector may expose longer), overhead = the rest
+    (positioning, trigger, detector readout). '' when the times are missing."""
+    if not n_points or t_start is None or t_end is None or t_end <= t_start:
+        return ""
+    total = t_end - t_start
+    txt = f"; scan time {total:.1f} s ({total / n_points * 1000:.0f} ms/point)"
+    if exp_time:
+        expo = n_points * float(exp_time) * int(frames_per_trigger or 1)
+        over = total - expo
+        txt += (
+            f": exposure {expo:.1f} s ({expo / total * 100:.0f} %, requested {float(exp_time) * 1000:g} ms"
+            f" x {int(frames_per_trigger or 1)}), overhead {over:.1f} s ({over / n_points * 1000:.0f} ms/point:"
+            " positioning, trigger, detector)"
+        )
+    return txt
+
+
 class RtCommunicationError(ControllerCommunicationError):
     pass
 

csaxs_bec/scans/flomni_fermat_scan.py

@@ -238,7 +238,11 @@ class FlomniFermatScan(ScanBase):
         """
 
         # send off the flyer
-        self.actions.kickoff(device=self.dev.rt_positions)
+        self.actions.kickoff(
+            device=self.dev.rt_positions,
+            # for the end-of-scan statistics of the flyer (exposure vs overhead)
+            parameters={"exp_time": self.exp_time, "frames_per_trigger": self.frames_per_trigger},
+        )
 
         # start the readout loop of the flyer
         status = self.actions.complete(device=self.dev.rt_positions, wait=False)

csaxs_bec/scans/lamni_fermat_scan.py

@@ -218,7 +218,11 @@ class LamniFermatScan(ScanBase):
         This is where the main scan logic should be implemented.
         """
 
-        self.actions.kickoff(device="rt_positions")
+        self.actions.kickoff(
+            device="rt_positions",
+            # for the end-of-scan statistics of the flyer (exposure vs overhead)
+            parameters={"exp_time": self.exp_time, "frames_per_trigger": self.frames_per_trigger},
+        )
 
         status = self.actions.complete(device="rt_positions", wait=False)
 

tests/tests_devices/test_rt_flomni.py

@@ -107,11 +107,23 @@ def test_flyer_reports_average_stdev_to_the_client(rt_flomni):
     for row in (["0", "0", "1", "1", "0.002", "2", "2", "0.001", "0", "0"],
                 ["1", "0", "1", "1", "0.004", "2", "2", "0.003", "0", "0"]):
         flyer._get_signals_from_table(row)
+    flyer.configure({"exp_time": 0.1, "frames_per_trigger": 2})        # kickoff parameters of the scan
+    flyer._t_kickoff, flyer._t_scan_end = 100.0, 101.0
     flyer._report_statistics(2)
     send = rt_flomni.device_manager.connector.send_client_info
     msg = send.call_args.args[0]
     assert "x 3.0 nm, y 2.0 nm" in msg and "2 points" in msg
+    assert "scan time 1.0 s (500 ms/point)" in msg and "exposure 0.4 s (40 %" in msg and "overhead 0.6 s (300 ms/point" in msg
     assert send.call_args.kwargs["show_asap"] is True
     send.reset_mock()
     flyer._report_statistics(0)                    # no points: nothing sent, no ZeroDivisionError
     send.assert_not_called()
+
+
+def test_format_scan_timing():
+    from csaxs_bec.devices.omny.rt.rt_ophyd import format_scan_timing
+
+    assert format_scan_timing(0, 0, 1) == "" and format_scan_timing(5, None, 1) == ""
+    assert format_scan_timing(10, 0, 2.0) == "; scan time 2.0 s (200 ms/point)"     # no exp_time: no split
+    txt = format_scan_timing(10, 0, 2.0, 0.1, 1)
+    assert "exposure 1.0 s (50 %" in txt and "overhead 1.0 s (100 ms/point" in txt

tests/tests_devices/test_rt_lamni_statistics.py

@@ -13,10 +13,13 @@ def test_lamni_flyer_reports_average_stdev_to_the_client():
         row = ["0"] * 21
         row[0], row[5], row[8], row[19] = str(i), str(sx), str(sy), "0.5"
         flyer._get_signals_from_table(row)
+    flyer.configure({"exp_time": 0.05})                                  # frames_per_trigger default 1
+    flyer._t_kickoff, flyer._t_scan_end = 10.0, 10.5
     flyer._report_statistics(2)
     send = flyer.controller.device_manager.connector.send_client_info
     msg = send.call_args.args[0]
     assert "x 3.0 nm, y 2.0 nm" in msg and "2 points" in msg and "average angle 0.5000" in msg
+    assert "scan time 0.5 s (250 ms/point)" in msg and "exposure 0.1 s (20 %" in msg
     send.reset_mock()
     flyer._report_statistics(0)                    # no points: nothing sent, no ZeroDivisionError
     send.assert_not_called()