From 845822cc34e835ce14be643ed50f85438c50e720 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Sat, 19 Sep 2026 09:40:52 +0200 Subject: [PATCH] feat(rt): add data monitor --- csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py | 66 +++++- csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py | 65 +++++- csaxs_bec/devices/omny/rt/rt_omny_ophyd.py | 68 +++++- csaxs_bec/devices/omny/rt/rt_ophyd.py | 82 ++++++- csaxs_bec/scans/flomni_fermat_scan.py | 1 + csaxs_bec/scans/lamni_fermat_scan.py | 1 + csaxs_bec/scans/omny_fermat_scan.py | 1 + tests/tests_devices/test_rt_data_monitor.py | 214 +++++++++++++++++++ tests/tests_scans/test_rt_scan_timeout.py | 44 ++++ 9 files changed, 531 insertions(+), 11 deletions(-) create mode 100644 tests/tests_devices/test_rt_data_monitor.py create mode 100644 tests/tests_scans/test_rt_scan_timeout.py diff --git a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py index 0cf66f39..97c2e903 100644 --- a/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_flomni_ophyd.py @@ -14,6 +14,7 @@ from prettytable import PrettyTable from csaxs_bec.devices.omny.rt.rt_ophyd import ( RtCommunicationError, + RtDataMonitor, RtError, RtReadbackSignal, RtSetpointSignal, @@ -520,6 +521,7 @@ class RtFlomniController(Controller): raise RtError("Cannot start scan because no target positions are planned.") # hier exception # start a point-by-point scan (for cont scan in flomni it would be "sa") + logger.info("RtFlomniController: Starting scan.") self.socket_put_and_receive("sd") @retry_once @@ -809,7 +811,7 @@ class RtFlomniMotor(Device, PositionerBase): class RtFlomniFlyer(Device): - USER_ACCESS = ["controller"] + USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"] data = Cpt( AsyncMultiSignal, name="data", @@ -846,9 +848,12 @@ class RtFlomniFlyer(Device): port=2222, socket_cls=SocketIO, device_manager=None, + data_timeout=5.0, + scan_timeout=60.0, **kwargs, ): super().__init__(prefix=prefix, name=name, parent=parent, **kwargs) + self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout) self.shutdown_event = threading.Event() self.controller = RtFlomniController( socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager @@ -859,6 +864,49 @@ class RtFlomniFlyer(Device): self.scan_done_event = threading.Event() self.scan_done_event.set() + @property + def data_timeout(self) -> float: + """Seconds without received data before logging diagnostics (default: 5).""" + return self._data_monitor.data_timeout + + @data_timeout.setter + def data_timeout(self, value: float) -> None: + self._data_monitor.data_timeout = value + + @property + def scan_timeout(self) -> float: + """Seconds without received points before failing the scan (default: 60). + + Includes initial buffering: during an active scan, samples are only read + once the reported scan position exceeds five. Allow enough time for this + first read; each received sample then resets the timeout. + """ + return self._data_monitor.scan_timeout + + @scan_timeout.setter + def scan_timeout(self, value: float) -> None: + self._data_monitor.scan_timeout = value + + def fetch_diagnostics(self) -> dict: + """Return scan status, feedback state, and recent controller commands.""" + diagnostics = {} + try: + mode, planned, current = self.controller.get_scan_status() + diagnostics["scan_status"] = { + "mode": mode, + "planned_positions": planned, + "current_position": current, + } + except Exception as exc: + diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}" + try: + diagnostics["feedback_running"] = self.controller.feedback_is_running() + except Exception as exc: + diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}" + with self.controller._lock: + diagnostics["command_history"] = list(self.controller.command_history) + return diagnostics + def read_positions_from_sampler(self, status: DeviceStatus): """ Read the positions from the sampler and update the data signal. @@ -894,6 +942,13 @@ class RtFlomniFlyer(Device): read_counter = read_counter + 1 signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() + + if mode > 0 and not self.shutdown_event.is_set(): + exception = self._data_monitor.check(self.name, self.fetch_diagnostics) + if exception is not None: + status.set_exception(exception) + return if self.shutdown_event.wait(0.05): logger.info("Shutdown event set, stopping readout.") @@ -910,6 +965,7 @@ class RtFlomniFlyer(Device): signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() # NOTE: No need to set the status to failed if the shutdown_event is set. # The stop() method will take care of that. @@ -959,6 +1015,7 @@ class RtFlomniFlyer(Device): self.scan_done_event.clear() while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001): ... + self._data_monitor.reset() self.controller.start_scan() self.shutdown_event.wait(0.1) status = DeviceStatus(self) @@ -980,7 +1037,10 @@ class RtFlomniFlyer(Device): def stop(self, *, success=False): self.shutdown_event.set() self.scan_done_event.set() - if self.readout_thread is not None: + if ( + self.readout_thread is not None + and self.readout_thread is not threading.current_thread() + ): self.readout_thread.join() return super().stop(success=success) @@ -990,4 +1050,4 @@ if __name__ == "__main__": socket_cls=SocketIO, socket_host="mpc2844.psi.ch", socket_port=2222, device_manager=None ) rtcontroller.on() - rtcontroller.laser_tracker_on() \ No newline at end of file + rtcontroller.laser_tracker_on() diff --git a/csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py b/csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py index c9c70f87..e9dc899c 100644 --- a/csaxs_bec/devices/omny/rt/rt_lamni_ophyd.py +++ b/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, RtDataMonitor, RtError logger = bec_logger.logger @@ -217,6 +217,7 @@ class RtLamniController(Controller): raise RtError("Cannot start scan because no target positions are planned.") # hier exception # start a point-by-point scan (for cont scan in flomni it would be "sa") + logger.info("RtLamniController: Starting scan.") self.socket_put_and_receive("sd") def feedback_is_running(self) -> bool: @@ -695,7 +696,7 @@ class RtLamniMotor(Device, PositionerBase): class RtLamniFlyer(Device): - USER_ACCESS = ["controller"] + USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"] data = Cpt( AsyncMultiSignal, name="data", @@ -741,9 +742,12 @@ class RtLamniFlyer(Device): port=3333, socket_cls=SocketIO, device_manager=None, + data_timeout=5.0, + scan_timeout=60.0, **kwargs, ): super().__init__(prefix=prefix, name=name, parent=parent, **kwargs) + self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout) self.shutdown_event = threading.Event() self.controller = RtLamniController( socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager @@ -755,6 +759,49 @@ class RtLamniFlyer(Device): self.scan_done_event = threading.Event() self.scan_done_event.set() + @property + def data_timeout(self) -> float: + """Seconds without received data before logging diagnostics (default: 5).""" + return self._data_monitor.data_timeout + + @data_timeout.setter + def data_timeout(self, value: float) -> None: + self._data_monitor.data_timeout = value + + @property + def scan_timeout(self) -> float: + """Seconds without received points before failing the scan (default: 60). + + Includes initial buffering: during an active scan, samples are only read + once the reported scan position exceeds five. Allow enough time for this + first read; each received sample then resets the timeout. + """ + return self._data_monitor.scan_timeout + + @scan_timeout.setter + def scan_timeout(self, value: float) -> None: + self._data_monitor.scan_timeout = value + + def fetch_diagnostics(self) -> dict: + """Return scan status, feedback state, and recent controller commands.""" + diagnostics = {} + try: + mode, planned, current = self.controller.get_scan_status() + diagnostics["scan_status"] = { + "mode": mode, + "planned_positions": planned, + "current_position": current, + } + except Exception as exc: + diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}" + try: + diagnostics["feedback_running"] = self.controller.feedback_is_running() + except Exception as exc: + diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}" + with self.controller._lock: + diagnostics["command_history"] = list(self.controller.command_history) + return diagnostics + def stage(self): self.shutdown_event.clear() self.scan_done_event.set() @@ -771,6 +818,7 @@ class RtLamniFlyer(Device): self.scan_done_event.clear() while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001): ... + self._data_monitor.reset() self.controller.start_scan() self.shutdown_event.wait(0.1) status = DeviceStatus(self) @@ -792,7 +840,10 @@ class RtLamniFlyer(Device): def stop(self, *, success=False): self.shutdown_event.set() self.scan_done_event.set() - if self.readout_thread is not None: + if ( + self.readout_thread is not None + and self.readout_thread is not threading.current_thread() + ): self.readout_thread.join() return super().stop(success=success) @@ -832,6 +883,13 @@ class RtLamniFlyer(Device): read_counter = read_counter + 1 signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() + + if mode > 0 and not self.shutdown_event.is_set(): + exception = self._data_monitor.check(self.name, self.fetch_diagnostics) + if exception is not None: + status.set_exception(exception) + return if self.shutdown_event.wait(0.05): logger.info("Shutdown event set, stopping readout.") @@ -848,6 +906,7 @@ class RtLamniFlyer(Device): signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() # NOTE: No need to set the status to failed if the shutdown_event is set. # The stop() method will take care of that. diff --git a/csaxs_bec/devices/omny/rt/rt_omny_ophyd.py b/csaxs_bec/devices/omny/rt/rt_omny_ophyd.py index f2566f25..f99bb10c 100644 --- a/csaxs_bec/devices/omny/rt/rt_omny_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_omny_ophyd.py @@ -23,6 +23,7 @@ if builtins.__dict__.get("bec") is not None: from csaxs_bec.devices.omny.rt.rt_ophyd import ( RtCommunicationError, + RtDataMonitor, RtError, RtReadbackSignal, RtSetpointSignal, @@ -859,6 +860,7 @@ class RtOMNYController(Controller): raise RtError("Cannot start scan because no target positions are planned.") # hier exception # start a point-by-point scan (for cont scan in flomni it would be "sa") + logger.info("RtOMNYController: Starting scan.") self.socket_put_and_receive("sd") @retry_once @@ -1140,7 +1142,7 @@ class RtOMNYMotor(Device, PositionerBase): class RtOMNYFlyer(Device): - USER_ACCESS = ["controller"] + USER_ACCESS = ["controller", "data_timeout", "scan_timeout", "fetch_diagnostics"] data = Cpt( AsyncMultiSignal, @@ -1184,9 +1186,12 @@ class RtOMNYFlyer(Device): port=2222, socket_cls=SocketIO, device_manager=None, + data_timeout=5.0, + scan_timeout=60.0, **kwargs, ): super().__init__(prefix=prefix, name=name, parent=parent, **kwargs) + self._data_monitor = RtDataMonitor(data_timeout=data_timeout, scan_timeout=scan_timeout) self.shutdown_event = threading.Event() self.controller = RtOMNYController( socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager @@ -1197,6 +1202,49 @@ class RtOMNYFlyer(Device): self.scan_done_event = threading.Event() self.scan_done_event.set() + @property + def data_timeout(self) -> float: + """Seconds without received data before logging diagnostics (default: 5).""" + return self._data_monitor.data_timeout + + @data_timeout.setter + def data_timeout(self, value: float) -> None: + self._data_monitor.data_timeout = value + + @property + def scan_timeout(self) -> float: + """Seconds without received points before failing the scan (default: 60). + + Includes initial buffering: during an active scan, samples are only read + once the reported scan position exceeds five. Allow enough time for this + first read; each received sample then resets the timeout. + """ + return self._data_monitor.scan_timeout + + @scan_timeout.setter + def scan_timeout(self, value: float) -> None: + self._data_monitor.scan_timeout = value + + def fetch_diagnostics(self) -> dict: + """Return scan status, feedback state, and recent controller commands.""" + diagnostics = {} + try: + mode, planned, current = self.controller.get_scan_status() + diagnostics["scan_status"] = { + "mode": mode, + "planned_positions": planned, + "current_position": current, + } + except Exception as exc: + diagnostics["scan_status_error"] = f"{type(exc).__name__}: {exc}" + try: + diagnostics["feedback_running"] = self.controller.feedback_is_running() + except Exception as exc: + diagnostics["feedback_error"] = f"{type(exc).__name__}: {exc}" + with self.controller._lock: + diagnostics["command_history"] = list(self.controller.command_history) + return diagnostics + def stage(self): self.shutdown_event.clear() self.scan_done_event.set() @@ -1213,6 +1261,7 @@ class RtOMNYFlyer(Device): self.scan_done_event.clear() while not self.controller._min_scan_buffer_reached and not self.shutdown_event.wait(0.001): ... + self._data_monitor.reset() self.controller.start_scan() self.shutdown_event.wait(0.1) status = DeviceStatus(self) @@ -1234,7 +1283,10 @@ class RtOMNYFlyer(Device): def stop(self, *, success=False): self.shutdown_event.set() self.scan_done_event.set() - if self.readout_thread is not None: + if ( + self.readout_thread is not None + and self.readout_thread is not threading.current_thread() + ): self.readout_thread.join() return super().stop(success=success) @@ -1251,7 +1303,9 @@ class RtOMNYFlyer(Device): self.average_stdeviations_x_st_fzp = 0 self.average_stdeviations_y_st_fzp = 0 - mode, number_of_positions_planned, current_position_in_scan = self.get_scan_status() + mode, number_of_positions_planned, current_position_in_scan = ( + self.controller.get_scan_status() + ) while mode > 0 and not self.shutdown_event.wait(0.01): # logger.info(f"Current scan position {current_position_in_scan} out of {number_of_positions_planned}") mode, number_of_positions_planned, current_position_in_scan = ( @@ -1269,6 +1323,13 @@ class RtOMNYFlyer(Device): read_counter = read_counter + 1 signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() + + if mode > 0 and not self.shutdown_event.is_set(): + exception = self._data_monitor.check(self.name, self.fetch_diagnostics) + if exception is not None: + status.set_exception(exception) + return if self.shutdown_event.wait(0.05): logger.info("Shutdown event set, stopping readout.") @@ -1285,6 +1346,7 @@ class RtOMNYFlyer(Device): signals = self._get_signals_from_table(return_table) self.data.set(signals) + self._data_monitor.reset() # NOTE: No need to set the status to failed if the shutdown_event is set. # The stop() method will take care of that. diff --git a/csaxs_bec/devices/omny/rt/rt_ophyd.py b/csaxs_bec/devices/omny/rt/rt_ophyd.py index f456729c..45deaeb8 100644 --- a/csaxs_bec/devices/omny/rt/rt_ophyd.py +++ b/csaxs_bec/devices/omny/rt/rt_ophyd.py @@ -1,10 +1,12 @@ """ -This module contains base signals for RT devices. Controller and motors are implemented in the -bespoke modules such as `rt_flomni_ophyd.py` or `rt_lamni_ophyd.py`. +This module contains base signals and data monitoring for RT devices. Controllers and motors are +implemented in bespoke modules such as `rt_flomni_ophyd.py` or `rt_lamni_ophyd.py`. """ import functools +import math import time +from collections.abc import Callable from bec_lib import bec_logger from ophyd.utils import ReadOnlyError @@ -26,6 +28,82 @@ class BECConfigError(Exception): pass +class RtDataMonitor: + """Track gaps in received scan samples for an owning RT flyer. + + ``data_timeout`` warns once with diagnostics; ``scan_timeout`` fails the scan. + The readout loop checks both after draining available samples. Status replies do + not count as data, and receiving a sample resets both deadlines. This does not + monitor blocked socket I/O. + + Both intervals start at kickoff and include initial sampler buffering. While + the scan is active, RT flyers only fetch samples once the reported scan + position exceeds five. Controller progress alone does not reset either timer, + so slow scans may warn or time out before this first batch is read. Configure + ``scan_timeout`` to allow enough time for the first sampler read. It limits + gaps between received samples, not the total scan duration. + """ + + def __init__(self, data_timeout=5.0, scan_timeout=60.0): + self.data_timeout = data_timeout + self.scan_timeout = scan_timeout + self._last_data_time = None + self._data_timeout_logged = False + + @staticmethod + def _validate_timeout(value, name): + value = float(value) + if not math.isfinite(value) or value <= 0: + raise ValueError(f"{name} must be a finite, positive number of seconds.") + return value + + @property + def data_timeout(self) -> float: + """Seconds without a received sample before logging a warning; must be positive.""" + return self._data_timeout + + @data_timeout.setter + def data_timeout(self, value: float) -> None: + self._data_timeout = self._validate_timeout(value, "data_timeout") + + @property + def scan_timeout(self) -> float: + """Seconds without a received sample before failing the scan.""" + return self._scan_timeout + + @scan_timeout.setter + def scan_timeout(self, value: float) -> None: + self._scan_timeout = self._validate_timeout(value, "scan_timeout") + + def reset(self) -> None: + """Start a new interval at kickoff or after receiving a sample.""" + self._last_data_time = time.monotonic() + self._data_timeout_logged = False + + def check(self, device_name: str, fetch_diagnostics: Callable[[], dict]) -> TimeoutError | None: + """Log a warning or return the exception the flyer should set on its status.""" + if self._last_data_time is None: + return + elapsed = time.monotonic() - self._last_data_time + if elapsed >= self.scan_timeout: + return TimeoutError( + f"No RT scan points received by {device_name} for {elapsed:.1f} s " + f"(scan_timeout={self.scan_timeout:g} s)." + ) + if elapsed < self.data_timeout or self._data_timeout_logged: + return + # Latch before fetching diagnostics so a failed diagnostic cannot cause log spam. + self._data_timeout_logged = True + try: + diagnostics = fetch_diagnostics() + except Exception as exc: + diagnostics = {"error": f"{type(exc).__name__}: {exc}"} + logger.warning( + f"No RT scan data received by {device_name} for {elapsed:.1f} s " + f"(data_timeout={self.data_timeout:g} s). Diagnostics: {diagnostics}" + ) + + class RtSignalBase(SocketSignal): def __init__(self, signal_name, **kwargs): self.signal_name = signal_name diff --git a/csaxs_bec/scans/flomni_fermat_scan.py b/csaxs_bec/scans/flomni_fermat_scan.py index 8aa1ec64..311321b6 100644 --- a/csaxs_bec/scans/flomni_fermat_scan.py +++ b/csaxs_bec/scans/flomni_fermat_scan.py @@ -232,6 +232,7 @@ class FlomniFermatScan(ScanBase): status = self.actions.complete(device=self.dev.rt_positions, wait=False) while not status.done: self.at_each_point() + status.wait() @scan_hook def at_each_point(self): diff --git a/csaxs_bec/scans/lamni_fermat_scan.py b/csaxs_bec/scans/lamni_fermat_scan.py index f152ea82..10935e65 100644 --- a/csaxs_bec/scans/lamni_fermat_scan.py +++ b/csaxs_bec/scans/lamni_fermat_scan.py @@ -226,6 +226,7 @@ class LamniFermatScan(ScanBase): while not status.done: self.at_each_point() time.sleep(1) + status.wait() @scan_hook def at_each_point(self): diff --git a/csaxs_bec/scans/omny_fermat_scan.py b/csaxs_bec/scans/omny_fermat_scan.py index 6c336f8b..e6481dba 100644 --- a/csaxs_bec/scans/omny_fermat_scan.py +++ b/csaxs_bec/scans/omny_fermat_scan.py @@ -207,6 +207,7 @@ class OmnyFermatScan(ScanBase): while not status.done: self.at_each_point() time.sleep(1) + status.wait() @scan_hook def at_each_point(self): diff --git a/tests/tests_devices/test_rt_data_monitor.py b/tests/tests_devices/test_rt_data_monitor.py new file mode 100644 index 00000000..6071f872 --- /dev/null +++ b/tests/tests_devices/test_rt_data_monitor.py @@ -0,0 +1,214 @@ +"""No-data warnings must survive diagnostics failures and rearm when samples resume.""" + +import importlib +import threading +from collections import deque +from types import SimpleNamespace +from unittest import mock + +import pytest + +from csaxs_bec.devices.omny.rt import rt_ophyd + + +@pytest.fixture(params=[("flomni", "RtFlomni"), ("lamni", "RtLamni"), ("omny", "RtOMNY")]) +def flyer(request, monkeypatch): + suffix, class_prefix = request.param + module = importlib.import_module(f"csaxs_bec.devices.omny.rt.rt_{suffix}_ophyd") + controller = mock.Mock() + controller._lock = threading.RLock() + controller._min_scan_buffer_reached = True + controller.command_history = deque(["[PUT]: sd", "[GET]: scan started"]) + controller.get_scan_status.return_value = (2, 10, 6) + controller.feedback_is_running.return_value = True + controller.socket_put_and_receive.side_effect = lambda command: ",".join( + [command[1:]] + ["0"] * 24 + ) + monkeypatch.setattr(module, f"{class_prefix}Controller", lambda **kwargs: controller) + device = getattr(module, f"{class_prefix}Flyer")(name="rt_positions", data_timeout=3) + monkeypatch.setattr(device.data, "set", mock.Mock()) + monkeypatch.setattr(device.progress, "put", mock.Mock()) + yield device + device.destroy() + + +@pytest.fixture +def clock(monkeypatch): + clock = SimpleNamespace(now=100) + monkeypatch.setattr(rt_ophyd, "time", SimpleNamespace(monotonic=lambda: clock.now)) + return clock + + +def run_readout(flyer, clock, states, *, finish=False, failure=False): + """Poll scripted (elapsed seconds, mode, position) without sleeping or hardware.""" + origin = clock.now + states = iter(states) + + def next_status(): + try: + elapsed, mode, position = next(states) + except StopIteration: + flyer.shutdown_event.set() + return (2, 10, 0) + clock.now = origin + elapsed + return mode, 10, position + + flyer.controller.get_scan_status.side_effect = next_status + flyer.shutdown_event = mock.Mock(wraps=threading.Event()) + flyer.shutdown_event.wait.side_effect = lambda timeout: flyer.shutdown_event.is_set() + status = flyer.kickoff() + assert status.done and status.success + status = flyer.complete() + flyer.readout_thread.join(timeout=2) + assert not flyer.readout_thread.is_alive() + if failure: + with pytest.raises(TimeoutError, match="scan_timeout"): + status.wait(timeout=1) + assert status.done and not status.success + else: + status.wait(timeout=1) + assert status.success + if finish: + assert not flyer.shutdown_event.is_set() + return status + + +@pytest.mark.parametrize("mode", [2, 3]) +def test_warns_once_at_configured_timeout_without_samples(flyer, clock, mode): + with ( + mock.patch.object( + flyer, "fetch_diagnostics", return_value={"feedback_running": False} + ) as fetch, + mock.patch.object(rt_ophyd.logger, "warning") as warning, + ): + run_readout(flyer, clock, [(0, mode, 0), (2.999, mode, 1)]) + warning.assert_not_called() + fetch.assert_not_called() + + run_readout(flyer, clock, [(0, mode, 0), (3, mode, 1), (4, mode, 2), (20, mode, 2)]) + warning.assert_called_once() + fetch.assert_called_once_with() + assert "data_timeout=3 s" in warning.call_args.args[0] + assert "'feedback_running': False" in warning.call_args.args[0] + flyer.data.set.assert_not_called() + + +def test_samples_rearm_warning_and_new_scan_resets_timeout(flyer, clock): + with ( + mock.patch.object(flyer, "fetch_diagnostics", return_value={}) as fetch, + mock.patch.object(rt_ophyd.logger, "warning") as warning, + ): + run_readout( + flyer, clock, [(0, 2, 0), (3, 2, 0), (4, 2, 6), (6.999, 2, 6), (7, 2, 6), (20, 2, 6)] + ) + assert flyer.data.set.call_count == 5 + assert warning.call_count == fetch.call_count == 2 + warning.reset_mock() + run_readout(flyer, clock, [(0, 2, 0), (2.999, 2, 0)]) + warning.assert_not_called() + run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0)]) + warning.assert_called_once() + + +def test_available_samples_are_read_before_timeout_check(flyer, clock): + with mock.patch.object(rt_ophyd.logger, "warning") as warning: + run_readout(flyer, clock, [(0, 2, 0), (4, 2, 6), (6.999, 2, 6)]) + warning.assert_not_called() + assert flyer.data.set.call_count == 5 + + +def test_finished_scan_drains_samples_without_stall_warning(flyer, clock): + with mock.patch.object(rt_ophyd.logger, "warning") as warning: + run_readout(flyer, clock, [(0, 2, 0), (70, 0, 3)], finish=True) + warning.assert_not_called() + assert flyer.data.set.call_count == 10 + + +def test_diagnostics_include_status_feedback_and_history(flyer): + assert flyer.fetch_diagnostics() == { + "scan_status": {"mode": 2, "planned_positions": 10, "current_position": 6}, + "feedback_running": True, + "command_history": ["[PUT]: sd", "[GET]: scan started"], + } + + +def test_diagnostics_preserve_other_fields_when_a_query_fails(flyer): + flyer.controller.get_scan_status.side_effect = TimeoutError("status unavailable") + diagnostics = flyer.fetch_diagnostics() + assert diagnostics["scan_status_error"] == "TimeoutError: status unavailable" + assert diagnostics["feedback_running"] is True + assert diagnostics["command_history"] + flyer.controller.feedback_is_running.side_effect = RuntimeError("feedback unavailable") + assert flyer.fetch_diagnostics()["feedback_error"] == "RuntimeError: feedback unavailable" + + +def test_failed_diagnostics_do_not_abort_or_repeat_warning(flyer, clock): + with ( + mock.patch.object( + flyer, "fetch_diagnostics", side_effect=RuntimeError("diagnostic failure") + ), + mock.patch.object(rt_ophyd.logger, "warning") as warning, + ): + run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0), (20, 2, 0)]) + warning.assert_called_once() + assert "RuntimeError: diagnostic failure" in warning.call_args.args[0] + + +def test_stopped_device_does_not_warn(flyer, clock): + flyer._data_monitor.reset() + clock.now += 10 + flyer.shutdown_event.set() + with mock.patch.object(flyer, "fetch_diagnostics") as fetch: + status = mock.Mock() + flyer.read_positions_from_sampler(status) + status.set_finished.assert_called_once_with() + fetch.assert_not_called() + + +@pytest.mark.parametrize("mode", [2, 3]) +def test_scan_timeout_fails_status_without_stopping_axes(flyer, clock, mode): + flyer.scan_timeout = 8 + with ( + mock.patch.object(flyer, "fetch_diagnostics", return_value={}), + mock.patch.object(rt_ophyd.logger, "warning") as warning, + ): + status = run_readout( + flyer, clock, [(0, mode, 0), (3, mode, 0), (7.999, mode, 0), (8, mode, 0)], failure=True + ) + warning.assert_called_once() + assert "scan_timeout=8 s" in str(status.exception()) + assert flyer.shutdown_event.is_set() + assert flyer.scan_done_event.is_set() + flyer.controller.stop_all_axes.assert_not_called() + + +def test_received_points_reset_scan_timeout(flyer, clock): + flyer.scan_timeout = 8 + with mock.patch.object(flyer, "fetch_diagnostics", return_value={}): + status = run_readout( + flyer, clock, [(0, 2, 0), (7, 2, 6), (14.999, 2, 6), (15, 2, 6)], failure=True + ) + assert "for 8.0 s" in str(status.exception()) + assert flyer.data.set.call_count == 5 + + +def test_default_scan_timeout_is_sixty_seconds(flyer, clock): + assert flyer.scan_timeout == 60 + with mock.patch.object(flyer, "fetch_diagnostics", return_value={}): + status = run_readout(flyer, clock, [(0, 2, 0), (59.999, 2, 0), (60, 2, 0)], failure=True) + assert "scan_timeout=60 s" in str(status.exception()) + + +def test_diagnostic_failure_does_not_prevent_scan_timeout(flyer, clock): + flyer.scan_timeout = 8 + with mock.patch.object( + flyer, "fetch_diagnostics", side_effect=RuntimeError("diagnostic failure") + ): + run_readout(flyer, clock, [(0, 2, 0), (3, 2, 0), (8, 2, 0)], failure=True) + + +@pytest.mark.parametrize("name", ["data_timeout", "scan_timeout"]) +@pytest.mark.parametrize("timeout", [0, -1, float("nan"), float("inf")]) +def test_invalid_timeouts_are_rejected(name, timeout): + with pytest.raises(ValueError, match="finite, positive"): + rt_ophyd.RtDataMonitor(**{name: timeout}) diff --git a/tests/tests_scans/test_rt_scan_timeout.py b/tests/tests_scans/test_rt_scan_timeout.py new file mode 100644 index 00000000..d3051753 --- /dev/null +++ b/tests/tests_scans/test_rt_scan_timeout.py @@ -0,0 +1,44 @@ +"""RT completion errors must propagate out of the scan's polling loop.""" + +from types import SimpleNamespace +from unittest import mock + +import pytest +from bec_lib import messages +from bec_server.scan_server.errors import DeviceInstructionError +from bec_server.scan_server.scan_stubs import ScanStubStatus + +from csaxs_bec.scans.flomni_fermat_scan import FlomniFermatScan +from csaxs_bec.scans.lamni_fermat_scan import LamniFermatScan +from csaxs_bec.scans.omny_fermat_scan import OmnyFermatScan + + +@pytest.mark.parametrize("scan_cls", [FlomniFermatScan, LamniFermatScan, OmnyFermatScan]) +@pytest.mark.parametrize("failed", [False, True]) +def test_rt_scan_checks_completed_readout_result(scan_cls, failed): + status = ScanStubStatus(instruction_handler=mock.Mock()) + if failed: + status.set_failed( + messages.ErrorInfo( + error_message="No RT scan points received within scan_timeout", + compact_error_message="RT scan timed out", + exception_type="TimeoutError", + device="rt_positions", + ) + ) + else: + status.set_done() + scan = SimpleNamespace( + actions=mock.Mock(), + dev=SimpleNamespace(rt_positions="rt_positions"), + at_each_point=mock.Mock(), + _scan_modifier=None, + _scan_modifier_hooks={}, + ) + scan.actions.complete.return_value = status + if failed: + with pytest.raises(DeviceInstructionError, match="RT scan timed out"): + scan_cls.scan_core(scan) + else: + scan_cls.scan_core(scan) + scan.actions.complete.assert_called_once_with(device="rt_positions", wait=False) -- 2.54.0