Merge branch 'main' into fix/tensor-row-device-lookup

csaxs_bec/devices/allied_vision_cameras/allied_vision_aravis_camera.py

@@ -247,6 +247,8 @@ class AlliedVisionAravisCamera(PSIDeviceBase):
 
     def on_destroy(self):
         """Clean up resources when the device is destroyed."""
+        # Stop acquisition before disconnecting so the live thread cannot keep polling.
+        self.stop_live_mode()
         self.cam.on_disconnect()
         super().on_destroy()
 

csaxs_bec/devices/epics/delay_generator_csaxs/ddg_2.py

@@ -22,7 +22,10 @@ Burst mode is enabled:
 - Burst period is set to the exposure time.
 """
 
+from __future__ import annotations
+
 import time
+from typing import TYPE_CHECKING
 
 from bec_lib.logger import bec_logger
 from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
@@ -42,6 +45,9 @@ from csaxs_bec.devices.epics.delay_generator_csaxs.delay_generator_csaxs import
 )
 from csaxs_bec.devices.utils.utils import fetch_scan_info
 
+if TYPE_CHECKING:  # pragma: no cover
+    from bec_lib.devicemanager import DeviceManagerBase, ScanInfo
+
 logger = bec_logger.logger
 
 ########################
@@ -113,6 +119,96 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS):
     distributes the trigger to the detectors. The DDG2 is triggered by the DDG1 through the EXT/EN channel.
     """
 
+    # Methods exposed to the BEC client, i.e. callable as dev.ddg2.<name>()
+    USER_ACCESS = ["get_readout_times", "set_readout_times", "effective_readout_times"]
+
+    def __init__(
+        self,
+        name: str,
+        prefix: str = "",
+        scan_info: ScanInfo | None = None,
+        device_manager: DeviceManagerBase | None = None,
+        readout_times: dict[str, float] | None = None,
+        **kwargs,
+    ):
+        """
+        Args:
+            readout_times: per-channel-pair readout times in seconds, merged over
+                DEFAULT_READOUT_TIMES. Only the pairs given are overridden, e.g.
+                ``{"ab": 3.0e-3}``. See set_readout_times() for what this controls.
+        """
+        super().__init__(
+            name=name, prefix=prefix, scan_info=scan_info, device_manager=device_manager, **kwargs
+        )
+        self._readout_times: dict[str, float] = dict(DEFAULT_READOUT_TIMES)
+        if readout_times:
+            self.set_readout_times(**readout_times)
+
+    def effective_readout_times(self) -> dict[str, float]:
+        """Return the readout times actually in force for the upcoming scan.
+
+        The scan may request a readout time of its own; it wins when it is larger than
+        this device's configured value. Scans default readout_time to 0, so the
+        configured value is the floor rather than a fallback.
+
+        The result defines the exposure window -- period minus gap -- that every
+        detector in the scan is expected to integrate over, whether it reads the gate
+        (Falcon) or runs its own timer from the rising edge (Eiger).
+
+        Returns:
+            dict[str, float]: effective readout time per channel pair, in seconds.
+        """
+        scan_readout = 0.0
+        if self.scan_parameters is not None:
+            scan_readout = float(getattr(self.scan_parameters, "readout_time", 0.0) or 0.0)
+        return {ch: max(scan_readout, rt) for ch, rt in self._readout_times.items()}
+
+    def get_readout_times(self) -> dict[str, float]:
+        """Return the readout times per channel pair, in seconds.
+
+        Returns:
+            dict[str, float]: a copy of the effective readout times.
+        """
+        return dict(self._readout_times)
+
+    def set_readout_times(self, **channels: float) -> dict[str, float]:
+        """Set the readout time of one or more channel pairs, in seconds.
+
+        The readout time is subtracted from the exposure time to obtain the pulse
+        width, while the burst period stays at the exposure time. Raising it therefore
+        widens the gap between consecutive detector triggers and shortens the exposure
+        by the same amount; the frame rate is unaffected.
+
+        This matters for detectors that need a minimum dead time between frames. The
+        default of 0.2 ms is shorter than some detectors can handle -- FalconcSAXS, for
+        instance, declares MIN_READOUT = 3 ms -- and a value that is too short shows up
+        as silently dropped frames rather than as an error.
+
+        Channel pair 'ab' is the one multiplexed to the detectors, so it is normally the
+        only one worth changing. The new value applies from the next on_stage, i.e. the
+        next scan; it does not alter a scan already staged.
+
+        Args:
+            **channels: channel pair to readout time in seconds, e.g. ``ab=3.0e-3``.
+
+        Returns:
+            dict[str, float]: the effective readout times after the update.
+
+        Raises:
+            ValueError: for an unknown channel pair or a non-positive value.
+        """
+        for channel, value in channels.items():
+            if channel not in self._readout_times:
+                raise ValueError(
+                    f"Unknown channel pair {channel!r}; expected one of "
+                    f"{sorted(self._readout_times)}"
+                )
+            if value <= 0:
+                raise ValueError(f"readout time for {channel!r} must be > 0, got {value}")
+            self._readout_times[channel] = float(value)
+        logger.info(f"DDG {self.name} readout times set to {self._readout_times}")
+        return self.get_readout_times()
+
     def on_init(self) -> None:
         """Initialize the device"""
         self.scan_parameters: ScanServerScanInfo | None = None
@@ -193,9 +289,14 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS):
 
         # NOTE Check if the exposure time is longer than all readout times.
         # Raise a ValueError if requested exposure time is too short.
-        if any(exp_time <= rt for rt in DEFAULT_READOUT_TIMES.values()):
+        # NOTE The gap is the scan's readout_time when the scan asks for one, otherwise this
+        # device's configured value. Scans default readout_time to 0, hence the max().
+        # Every detector must derive its exposure window from the same number -- see
+        # effective_readout_times().
+        readouts = self.effective_readout_times()
+        if any(exp_time <= rt for rt in readouts.values()):
             raise ValueError(
-                f"Exposure time {exp_time} is too short for the readout times {DEFAULT_READOUT_TIMES}"
+                f"Exposure time {exp_time} is too short for the readout times {readouts}"
             )
 
         #########################################
@@ -208,8 +309,13 @@ class DDG2(PSIDeviceBase, DelayGeneratorCSAXS):
         if self.burst_period.get() != exp_time:
             self.burst_period.put(exp_time)
 
-        # Calculate the pulse width for the channel pair 'ab'
-        burst_pulse_width = exp_time - DEFAULT_READOUT_TIMES["ab"]
+        # Calculate the pulse width for the channel pair 'ab'. The readout time is the
+        # gap left between consecutive triggers; see set_readout_times().
+        burst_pulse_width = exp_time - readouts["ab"]
+        logger.info(
+            f"DDG {self.name} exposure window: period {exp_time}s, gap {readouts['ab']}s, "
+            f"pulse width {burst_pulse_width}s"
+        )
 
         # Trigger detectors with delay 0, and pulse width = exp_time - readout_time
         self.set_delay_pairs(channel="ab", delay=0, width=burst_pulse_width)

csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py

@@ -70,11 +70,14 @@ def suppress_mca_callbacks(mcs_card: MCSCard, restore_after_timeout: None | floa
     """
     with mcs_card._rlock:
         mcs_card._omit_mca_callbacks.set()  # pylint: disable=protected-access
-        try:
-            yield
-        finally:
-            if restore_after_timeout is not None:
-                time.sleep(restore_after_timeout)
+    try:
+        # MCA callbacks must be able to acquire the data lock while suppression
+        # is active so they are discarded instead of queued until afterward.
+        yield
+    finally:
+        if restore_after_timeout is not None:
+            time.sleep(restore_after_timeout)
+            with mcs_card._rlock:
                 mcs_card._omit_mca_callbacks.clear()  # pylint: disable=protected-access
 
 
@@ -467,7 +470,16 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
                         logger.info(
                             f"Software triggered scan: {self._current_data_index}/{self.scan_parameters.num_points} points received."
                         )
-                        if self._current_data_index == self.scan_parameters.num_points:
+                        if self._current_data_index > self.scan_parameters.num_points:
+                            exception = RuntimeError(
+                                f"MCS card {self.name} emitted {self._current_data_index} "
+                                f"data points, but the scan requested "
+                                f"{self.scan_parameters.num_points}."
+                            )
+                            logger.error(str(exception))
+                            for callback in self._scan_done_callbacks:
+                                callback(exception=exception)
+                        elif self._current_data_index == self.scan_parameters.num_points:
                             for callback in self._scan_done_callbacks:
                                 callback(exception=None)
                     else:

csaxs_bec/devices/ids_cameras/ids_camera.py

@@ -289,6 +289,8 @@ class IDSCamera(PSIDeviceBase):
 
     def on_destroy(self):
         """Clean up resources when the device is destroyed."""
+        # Stop acquisition before disconnecting so the live thread cannot keep polling.
+        self.stop_live_mode()
         self.cam.on_disconnect()
         super().on_destroy()
 

csaxs_bec/devices/jungfraujoch/eiger.py

@@ -8,10 +8,21 @@ A few notes on setup and operation of the Eiger detectors through the JungfrauJo
 - Changes in energy may take time,  good to implement logic that only resets energy if needed.
 - For the Eiger, the frame_time_us in DetectorSettings is ignored, only the frame_time_us in
   the DatasetSettings is relevant
-- The bit_depth will be adjusted automatically based on the exp_time. Here, we need to ensure
-  that subsequent triggers properly consider the readout_time of the boards. For the Eiger detectors
-  at cSAXS, a readout time of 20us is configured through the JungfrauJoch deployment config. This
-  setting is sufficiently large for the detectors if they run in parallel mode.
+- The bit_depth will be adjusted automatically based on the exp_time (32 bit above 2622 us),
+  and in 32 bit the detector accumulates ~2.3 ms subframes. JungfrauJoch's own deployment config
+  budgets 20 us of board readout between frames, which is right for 16 bit parallel mode but far
+  smaller than the subframe granularity -- so a frame can outlast the trigger period without
+  anything reporting an error.
+- The Eiger is TRIGGERED rather than gated (DetectorTiming.TRIGGER), for stability: gating
+  it proved unreliable. It therefore ignores the gate width and runs its own timer from the
+  rising edge. The pulse train is deliberately shaped so that this timer coincides with the
+  gate -- the Eiger is meant to behave *as if* it were gated, so that every detector in the
+  scan integrates the same window. Sending it anything other than the gated window breaks
+  that, silently, and is only visible on a scope.
+- scan_info's exp_time is the trigger PERIOD, not the exposure. The delay generator gates for
+  exp_time - readout_time, and on_stage sends that same window as image_time_us so the detector
+  is not still acquiring when the next trigger arrives. The readout time must agree with DDG2's
+  DEFAULT_READOUT_TIMES["ab"]; a scan-level readout_time overrides both when larger.
 - beam_center and detector settings are required input arguments, thus, they may be set to wrong
   values for acquisitions to start. Please keep this in mind.
 
@@ -65,7 +76,32 @@ if TYPE_CHECKING:  # pragma no cover
 
 logger = bec_logger.logger
 
-EIGER_READOUT_TIME_US = 500e-6  # 500 microseconds in s
+# NOTE Two distinct numbers that used to be one. MIN_EXP_TIME is a validation floor
+# (its value is inherited from the former EIGER_READOUT_TIME_US and has no documented
+# provenance). EIGER_READOUT_TIME is the gap left between the end of one acquisition and
+# the next trigger; it must match the gap the delay generator leaves in the pulse train,
+# DDG2's DEFAULT_READOUT_TIMES["ab"], or the detector is told to acquire for longer than
+# the trigger period allows.
+MIN_EXP_TIME = 500e-6  # s
+EIGER_READOUT_TIME = 2e-4  # s -- matches DDG2 DEFAULT_READOUT_TIMES["ab"]
+
+# NOTE The JungfrauJoch broker reports packet loss as a plain error with no error code,
+# so the only way to recognise it is the wording of its status message. If JFJoch ever
+# rephrases this, the match stops working and the error raises again -- which is the safe
+# direction to fail in.
+MISSING_PACKETS_MESSAGE = "Missing packets in data collection"
+
+# NOTE Default for raise_on_missing_packets, i.e. whether a missing-packet error from the
+# broker fails the scan. False while the 9M is losing packets during the 2026-09 beamtime
+# and scans must not die on it. Set it back to True once that is resolved, and put
+# "raise_on_missing_packets: false" in the deviceConfig of the detector that still needs
+# it -- per detector, rather than for every Eiger at once.
+RAISE_ON_MISSING_PACKETS = False
+# NOTE Each model declares its own value and passes it here; that duplication is
+# deliberate, since readout is a property of the detector (a 9M has more modules than a
+# 1.5M, and the Falcon needs 3 ms). They all happen to be 2e-4 today only because no
+# measured per-model number exists yet. This constant is the fallback for a model that
+# does not state one; deviceConfig's readout_time overrides it per deployment.
 
 
 class EigerError(Exception):
@@ -83,6 +119,8 @@ class Eiger(PSIDeviceBase):
         port (int): Port of the Jungfrau Joch server.
         scan_info (ScanInfo): The scan info to use.
         device_manager (DeviceManagerDS): The device manager to use.
+        raise_on_missing_packets (bool): Whether a 'missing packets' error from the
+            JungfrauJoch broker fails the scan. See RAISE_ON_MISSING_PACKETS.
         **kwargs: Additional keyword arguments.
     """
 
@@ -92,6 +130,9 @@ class Eiger(PSIDeviceBase):
         "print_detector_distance_and_center",
         "set_azi_settings",
         "get_azi_settings",
+        "get_raise_on_missing_packets",
+        "set_raise_on_missing_packets",
+        "get_missing_packet_events",
     ]
 
     file_event = Cpt(FileEventSignal, name="file_event")
@@ -106,7 +147,8 @@ class Eiger(PSIDeviceBase):
         detector_distance: float = 100.0,
         beam_center: tuple[int, int] = (0, 0),
         scan_info: ScanInfo = None,
-        readout_time: float = EIGER_READOUT_TIME_US,
+        readout_time: float = EIGER_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         device_manager=None,
         **kwargs,
     ):
@@ -130,6 +172,47 @@ class Eiger(PSIDeviceBase):
         if self.device_manager is not None:
             self.device_manager: DeviceManagerDS
         self.scan_parameters: ScanServerScanInfo | None = None
+        self._raise_on_missing_packets = bool(raise_on_missing_packets)
+        self._missing_packet_events = 0
+
+    def get_raise_on_missing_packets(self) -> bool:
+        """
+        Whether a 'missing packets' error from the JungfrauJoch broker fails the scan.
+
+        Returns:
+            bool: True if missing packets raise an EigerError, False if they are tolerated.
+        """
+        return self._raise_on_missing_packets
+
+    def set_raise_on_missing_packets(self, value: bool) -> None:
+        """
+        Set whether a 'missing packets' error from the JungfrauJoch broker fails the scan.
+
+        NOTE This is shared by every client of this device server, and it does NOT survive a
+        server restart: '__init__' runs again and restores the value from deviceConfig. To make
+        a choice stick for a beamtime, set 'raise_on_missing_packets' in the deviceConfig of the
+        detector instead.
+
+        Args:
+            value (bool): True to raise on missing packets, False to tolerate them.
+        """
+        self._raise_on_missing_packets = bool(value)
+        logger.warning(
+            f"Device {self.name}: raise_on_missing_packets is now "
+            f"{self._raise_on_missing_packets}."
+        )
+
+    def get_missing_packet_events(self) -> int:
+        """
+        How many missing-packet errors have been tolerated since this device was created.
+
+        Reset by a server restart, like the flag itself. Non-zero means acquisitions completed
+        that the broker had flagged, so the data of those scans may be incomplete.
+
+        Returns:
+            int: Number of tolerated missing-packet errors.
+        """
+        return self._missing_packet_events
 
     def _preview_callback(self, message: dict) -> None:
         """
@@ -291,6 +374,10 @@ class Eiger(PSIDeviceBase):
         # self.jfj_client.connect_and_initialise(timeout=10)
 
         # Setup Detector settings, here we may also set the energy already as this might be time consuming
+        # NOTE frame_time_us is required by the API but ignored for the Eiger ("For EIGER
+        # detector this is default frame time, not used otherwise"); the exposure comes
+        # from image_time_us in DatasetSettings. The 500 is a placeholder -- it is NOT the
+        # readout time, and it is in microseconds, unlike every other time in this module.
         settings = DetectorSettings(frame_time_us=int(500), timing=DetectorTiming.TRIGGER)
         self.jfj_client.set_detector_settings(settings, timeout=5)
 
@@ -344,9 +431,21 @@ class Eiger(PSIDeviceBase):
 
         # Setting up exp_time and num_triggers acquisition parameter
         exp_time = self.scan_parameters.exp_time
-        if exp_time <= self._readout_time:  # Exp_time must be at least the readout time
+        if exp_time <= MIN_EXP_TIME:
             raise ValueError(
-                f"Value error on device {self.name}: Exposure time {exp_time}s is less than readout time {self._readout_time}s."
+                f"Value error on device {self.name}: Exposure time {exp_time}s is below the "
+                f"minimum exposure time {MIN_EXP_TIME}s."
+            )
+        # NOTE exp_time is the trigger PERIOD, not the exposure. The delay generator gates
+        # for exp_time - readout_time, so the detector must acquire for the same window or
+        # it is still busy when the next trigger arrives. Scans default readout_time to 0,
+        # hence the max() against this device's configured value.
+        readout_time = max(self.scan_parameters.readout_time, self._readout_time)
+        acquisition_time = exp_time - readout_time
+        if acquisition_time <= 0:
+            raise ValueError(
+                f"Value error on device {self.name}: readout time {readout_time}s leaves no "
+                f"acquisition time within the trigger period {exp_time}s."
             )
         self._num_triggers = int(
             self.scan_parameters.num_points * self.scan_parameters.frames_per_trigger
@@ -369,7 +468,7 @@ class Eiger(PSIDeviceBase):
 
         # Create dataset settings for API call.
         data_settings = DatasetSettings(
-            image_time_us=int(exp_time * 1e6),
+            image_time_us=int(acquisition_time * 1e6),
             ntrigger=self._num_triggers,
             file_prefix=path,
             beam_x_pxl=float(self._beam_center[0]),
@@ -379,6 +478,10 @@ class Eiger(PSIDeviceBase):
         )
         logger.debug(f"Setting data_settings: {yaml.dump(data_settings.to_dict(), indent=4)}")
         prep_time = time.time()
+        logger.info(
+            f"{self.name} exposure window: period {exp_time}s, readout {readout_time}s, "
+            f"acquisition {acquisition_time}s"
+        )
         self.jfj_client.wait_for_idle(timeout=10)  # Ensure we are in IDLE state
         self.jfj_client.start(settings=data_settings)  # Takes around ~0.6s -> 6s currently.
 
@@ -434,13 +537,28 @@ class Eiger(PSIDeviceBase):
                         logger.info(
                             f"JungfrauJoch broker status: {yaml.dump(broker_status.to_dict(), indent=4)}"
                         )
-                    if broker_status.message_severity == "error":  # Raise on error
-                        raise EigerError(
-                            f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
-                        )
-                        # logger.warning(
-                        #     f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
-                        # )
+                    if broker_status.message_severity == "error":
+                        # NOTE Missing packets can be tolerated per detector, because the frame
+                        # count is still checked below: what is being let through here is "the
+                        # broker flagged packet loss but delivered the expected number of
+                        # images", not a short acquisition.
+                        if (
+                            MISSING_PACKETS_MESSAGE in (broker_status.message or "")
+                            and not self._raise_on_missing_packets
+                        ):
+                            self._missing_packet_events += 1
+                            logger.warning(
+                                f"Device {self.name} reported missing packets and the error is "
+                                f"being tolerated (raise_on_missing_packets is False), so the "
+                                f"data of this scan may be incomplete. That is tolerated event "
+                                f"{self._missing_packet_events} for this device since it was "
+                                f"created. Broker status: "
+                                f"{yaml.dump(broker_status.to_dict(), indent=4)}"
+                            )
+                        else:
+                            raise EigerError(
+                                f"Device {self.name} acquisition completed with error status from JungfrauJoch broker: {yaml.dump(broker_status.to_dict(), indent=4)}"
+                            )
                     # Call API endpoint to get statistics
                     statistics: MeasurementStatistics = (
                         self.jfj_client.api.statistics_data_collection_get(_request_timeout=5)

csaxs_bec/devices/jungfraujoch/eiger_1_5m.py

@@ -9,9 +9,13 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from csaxs_bec.devices.jungfraujoch.eiger import Eiger
+from csaxs_bec.devices.jungfraujoch.eiger import RAISE_ON_MISSING_PACKETS, Eiger
 
-EIGER1_5M_READOUT_TIME_US = 500e-6  # 500 microseconds in s
+# Gap this model needs between the end of an acquisition and the next trigger. Kept
+# per model on purpose -- readout is a detector property, not a beamline constant. The
+# delay generator's gap (DDG2 DEFAULT_READOUT_TIMES["ab"]) must be at least as large as
+# the largest value among the detectors in a scan, or the slowest one overruns.
+EIGER1_5M_READOUT_TIME = 2e-4  # s -- placeholder until a measured value exists
 DETECTOR_NAME = "EIGER 1.5M"
 
 
@@ -40,12 +44,15 @@ class Eiger1_5M(Eiger):
         beam_center: tuple[float, float] = (0.0, 0.0),
         scan_info: ScanInfo = None,
         device_manager: DeviceManagerDS = None,
+        readout_time: float = EIGER1_5M_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         **kwargs,
     ) -> None:
         super().__init__(
             name=name,
             detector_name=DETECTOR_NAME,
-            readout_time=EIGER1_5M_READOUT_TIME_US,
+            readout_time=readout_time,
+            raise_on_missing_packets=raise_on_missing_packets,
             detector_distance=detector_distance,
             beam_center=beam_center,
             scan_info=scan_info,

csaxs_bec/devices/jungfraujoch/eiger_9m.py

@@ -14,13 +14,17 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from csaxs_bec.devices.jungfraujoch.eiger import Eiger
+from csaxs_bec.devices.jungfraujoch.eiger import RAISE_ON_MISSING_PACKETS, Eiger
 
 if TYPE_CHECKING:  # pragma no cover
     from bec_lib.devicemanager import ScanInfo
     from bec_server.device_server.device_server import DeviceManagerDS
 
-EIGER9M_READOUT_TIME_US = 500e-6  # 500 microseconds in s
+# Gap this model needs between the end of an acquisition and the next trigger. Kept
+# per model on purpose -- readout is a detector property, not a beamline constant. The
+# delay generator's gap (DDG2 DEFAULT_READOUT_TIMES["ab"]) must be at least as large as
+# the largest value among the detectors in a scan, or the slowest one overruns.
+EIGER9M_READOUT_TIME = 2e-4  # s -- placeholder until a measured value exists
 DETECTOR_NAME = "EIGER 9M"  # "EIGER 9M""
 
 
@@ -44,12 +48,15 @@ class Eiger9M(Eiger):
         beam_center: tuple[float, float] = (0.0, 0.0),
         scan_info: ScanInfo = None,
         device_manager: DeviceManagerDS = None,
+        readout_time: float = EIGER9M_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         **kwargs,
     ) -> None:
         super().__init__(
             name=name,
             detector_name=DETECTOR_NAME,
-            readout_time=EIGER9M_READOUT_TIME_US,
+            readout_time=readout_time,
+            raise_on_missing_packets=raise_on_missing_packets,
             detector_distance=detector_distance,
             beam_center=beam_center,
             scan_info=scan_info,

tests/tests_devices/test_allied_vision_aravis_camera.py

@@ -23,6 +23,7 @@ def allied_vision_camera():
     camera.cam.width = 2
     camera.cam.height = 2
     yield camera
+    camera.stop_live_mode()
 
 
 def test_mask_setter_getter(allied_vision_camera):

tests/tests_devices/test_delay_generator_csaxs.py

@@ -613,3 +613,95 @@ def test_ddg2_on_stop(mock_ddg2: DDG2):
 #     mock_ddg2.burst_mode.put(1)  # Enable burst mode
 #     mock_ddg2.stop()
 #     assert mock_ddg2.burst_mode.get() == 0  # Burst mode is disabled
+
+
+def test_ddg2_readout_times_default(mock_ddg2: DDG2):
+    """Without an override the module defaults are in force."""
+    assert mock_ddg2.get_readout_times() == DDG2_DEFAULT_READOUT_TIMES
+    # A copy, so a caller cannot mutate the device's state by accident.
+    mock_ddg2.get_readout_times()["ab"] = 1.0
+    assert mock_ddg2.get_readout_times()["ab"] == DDG2_DEFAULT_READOUT_TIMES["ab"]
+
+
+def test_ddg2_set_readout_times_validates(mock_ddg2: DDG2):
+    """Unknown pairs and non-positive values are rejected before they reach a scan."""
+    with pytest.raises(ValueError):
+        mock_ddg2.set_readout_times(xy=1e-3)
+    with pytest.raises(ValueError):
+        mock_ddg2.set_readout_times(ab=0)
+    with pytest.raises(ValueError):
+        mock_ddg2.set_readout_times(ab=-1e-3)
+    assert mock_ddg2.get_readout_times() == DDG2_DEFAULT_READOUT_TIMES
+
+
+def test_ddg2_set_readout_times_widens_the_gap(mock_ddg2: DDG2):
+    """A raised readout time shortens the pulse; the burst period is unchanged."""
+    ddg = mock_ddg2
+    exp_time = 0.1
+    frames_per_trigger = 10
+    ddg.on_connected()
+    ddg.scan_info.msg.info["exp_time"] = exp_time
+    ddg.scan_info.msg.info["frames_per_trigger"] = frames_per_trigger
+
+    ddg.set_readout_times(ab=3e-3)
+    assert ddg.get_readout_times()["ab"] == 3e-3
+    # Other pairs are untouched.
+    assert ddg.get_readout_times()["cd"] == DDG2_DEFAULT_READOUT_TIMES["cd"]
+
+    ddg.stage()
+    assert np.isclose(ddg.ab.width.get(), exp_time - 3e-3)
+    assert ddg.burst_period.get() == exp_time  # frame rate unaffected
+    ddg.unstage()
+
+
+def test_ddg2_readout_times_from_device_config():
+    """deviceConfig can set it per deployment, merged over the defaults."""
+    with patched_device(
+        DDG2, name="ddg2", prefix="test_ddg2:", readout_times={"ab": 3e-3}, _mock_pv_initial_value=0
+    ) as ddg:
+        try:
+            assert ddg.get_readout_times()["ab"] == 3e-3
+            assert ddg.get_readout_times()["gh"] == DDG2_DEFAULT_READOUT_TIMES["gh"]
+        finally:
+            ddg.destroy()
+
+
+def test_ddg2_too_long_readout_time_is_caught_at_stage(mock_ddg2: DDG2):
+    """A readout time exceeding the exposure must fail loudly, not emit a negative pulse."""
+    ddg = mock_ddg2
+    ddg.on_connected()
+    ddg.scan_info.msg.info["exp_time"] = 1e-3
+    ddg.scan_info.msg.info["frames_per_trigger"] = 1
+    ddg.set_readout_times(ab=3e-3)
+    with pytest.raises(ValueError):
+        ddg.stage()
+
+
+def test_ddg2_effective_readout_uses_the_scan_value_when_larger(mock_ddg2: DDG2):
+    """A scan-level readout_time widens the gap; the configured value is the floor."""
+    ddg = mock_ddg2
+    exp_time = 0.1
+    ddg.on_connected()
+    ddg.scan_info.msg.info["exp_time"] = exp_time
+    ddg.scan_info.msg.info["frames_per_trigger"] = 1
+    ddg.scan_info.msg.info["readout_time"] = 3e-3
+
+    ddg.stage()
+    assert ddg.effective_readout_times()["ab"] == 3e-3
+    assert np.isclose(ddg.ab.width.get(), exp_time - 3e-3)
+    ddg.unstage()
+
+
+def test_ddg2_effective_readout_falls_back_to_the_configured_value(mock_ddg2: DDG2):
+    """Scans default readout_time to 0, so the device value must still apply."""
+    ddg = mock_ddg2
+    exp_time = 0.1
+    ddg.on_connected()
+    ddg.scan_info.msg.info["exp_time"] = exp_time
+    ddg.scan_info.msg.info["frames_per_trigger"] = 1
+    ddg.scan_info.msg.info["readout_time"] = 0.0
+
+    ddg.stage()
+    assert ddg.effective_readout_times()["ab"] == DDG2_DEFAULT_READOUT_TIMES["ab"]
+    assert np.isclose(ddg.ab.width.get(), exp_time - DDG2_DEFAULT_READOUT_TIMES["ab"])
+    ddg.unstage()

tests/tests_devices/test_eiger.py

@@ -1,4 +1,5 @@
 # pylint: skip-file
+import inspect
 import os
 import threading
 import time
@@ -22,8 +23,14 @@ from ophyd import Staged
 from ophyd_devices.utils.psi_device_base_utils import DeviceStatus
 
 from csaxs_bec.devices.jungfraujoch.eiger import EigerError
-from csaxs_bec.devices.jungfraujoch.eiger_1_5m import Eiger1_5M
-from csaxs_bec.devices.jungfraujoch.eiger_9m import Eiger9M
+from csaxs_bec.devices.jungfraujoch.eiger import (
+    EIGER_READOUT_TIME,
+    MIN_EXP_TIME,
+    MISSING_PACKETS_MESSAGE,
+    RAISE_ON_MISSING_PACKETS,
+)
+from csaxs_bec.devices.jungfraujoch.eiger_1_5m import EIGER1_5M_READOUT_TIME, Eiger1_5M
+from csaxs_bec.devices.jungfraujoch.eiger_9m import EIGER9M_READOUT_TIME, Eiger9M
 
 if TYPE_CHECKING:  # pragma no cover
     from bec_lib.messages import FileMessage
@@ -327,11 +334,12 @@ def test_eiger_on_complete(eiger_1_5m, raise_timeout):
         assert status.success == False if raise_timeout else True
 
 
-@pytest.mark.skip(
-    reason="This test currently fails because the error should be skipped for now due to HW issues."
-)
 def test_eiger_on_complete_error_message(eiger_1_5m):
-    """Test that on_complete raises if the message severity is error."""
+    """Test that on_complete raises if the message severity is error.
+
+    Unrelated to the missing-packet tolerance: this message is a different error, so it
+    raises whatever raise_on_missing_packets is set to.
+    """
     # Test that on_complete raises if message_severity is error
     eiger = eiger_1_5m
     unblock_wait_for_idle = threading.Event()
@@ -416,7 +424,8 @@ def test_eiger_on_stage(eiger_1_5m):
         assert file_msg.hinted_h5_entries == {"data": "entry/data/data"}
 
         data_settings = DatasetSettings(
-            image_time_us=int(scan_msg.scan_parameters["exp_time"] * 1e6),
+            # exp_time is the trigger period; the detector acquires for the gated window
+            image_time_us=int((scan_msg.scan_parameters["exp_time"] - EIGER_READOUT_TIME) * 1e6),
             ntrigger=int(scan_msg.num_points * scan_msg.scan_parameters["frames_per_trigger"]),
             file_prefix=os.path.relpath(eiger._full_path, start="/sls/x12sa/data").removesuffix(
                 "_master.h5"
@@ -460,3 +469,233 @@ def test_eiger_preview_callback(eiger_1_5m):
                 mock_preview_put.assert_called_once_with(msg["data"]["default"])
             else:
                 mock_preview_put.assert_not_called()
+
+
+def test_eiger_acquisition_matches_the_gated_window(eiger_1_5m):
+    """The detector must acquire for exp_time - readout, not for the whole period.
+
+    exp_time is the trigger period. The delay generator gates for exp_time - readout, so
+    sending the full period leaves the detector still acquiring when the next trigger
+    arrives -- 20 us of margin, which is what JungfrauJoch subtracts internally.
+    """
+    eiger = eiger_1_5m
+    exp_time = eiger.scan_info.msg.scan_parameters["exp_time"]
+
+    with (
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", return_value=True),
+        mock.patch.object(eiger.jfj_client, "start") as mock_start,
+        mock.patch.object(
+            eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle")
+        ),
+    ):
+        eiger.stage()
+
+    sent = mock_start.call_args.kwargs["settings"].image_time_us
+    assert sent == int((exp_time - EIGER_READOUT_TIME) * 1e6)
+    assert sent < int(exp_time * 1e6)
+
+
+def test_eiger_scan_readout_time_wins_when_larger(eiger_1_5m):
+    """A scan asking for a wider gap must widen the acquisition window too."""
+    eiger = eiger_1_5m
+    exp_time = eiger.scan_info.msg.scan_parameters["exp_time"]
+    eiger.scan_info.msg.info["readout_time"] = 3e-3  # > EIGER_READOUT_TIME
+
+    with (
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", return_value=True),
+        mock.patch.object(eiger.jfj_client, "start") as mock_start,
+        mock.patch.object(
+            eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle")
+        ),
+    ):
+        eiger.stage()
+
+    assert mock_start.call_args.kwargs["settings"].image_time_us == int((exp_time - 3e-3) * 1e6)
+
+
+def test_eiger_rejects_a_readout_that_fills_the_period(eiger_1_5m):
+    """A gap wider than the period leaves nothing to acquire; fail rather than send <= 0."""
+    eiger = eiger_1_5m
+    exp_time = eiger.scan_info.msg.scan_parameters["exp_time"]
+    eiger.scan_info.msg.info["readout_time"] = exp_time + 1e-3
+
+    with (
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", return_value=True),
+        mock.patch.object(eiger.jfj_client, "start"),
+        mock.patch.object(
+            eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle")
+        ),
+    ):
+        with pytest.raises(ValueError):
+            eiger.stage()
+
+
+def test_eiger_minimum_exposure_check_is_unchanged(eiger_1_5m):
+    """Splitting readout from the validation floor must not relax the minimum exposure."""
+    eiger = eiger_1_5m
+    eiger.scan_info.msg.info["exp_time"] = MIN_EXP_TIME / 2
+
+    with (
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", return_value=True),
+        mock.patch.object(eiger.jfj_client, "start"),
+        mock.patch.object(
+            eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle")
+        ),
+    ):
+        with pytest.raises(ValueError):
+            eiger.stage()
+
+
+def test_eiger_models_may_declare_their_own_readout_time(eiger_1_5m, eiger_9m):
+    """Readout is a detector property, so each model states its own value.
+
+    They coincide today, but nothing should force them to: a 9M has more modules
+    than a 1.5M, and the Falcon needs 3 ms.
+    """
+    assert eiger_1_5m._readout_time == EIGER1_5M_READOUT_TIME
+    assert eiger_9m._readout_time == EIGER9M_READOUT_TIME
+
+
+def test_eiger_readout_time_is_overridable_per_deployment(mock_scan_info):
+    """deviceConfig's readout_time wins over the model's default."""
+    dev = Eiger1_5M(
+        name="eiger_1_5m", beam_center=(256, 256), detector_distance=100.0, readout_time=1.5e-3
+    )
+    try:
+        assert dev._readout_time == 1.5e-3
+    finally:
+        dev.destroy()
+
+
+def test_eiger_subclasses_name_readout_time_for_the_device_server():
+    """deviceConfig keys reach a device only if the class signature names them.
+
+    bec_server intersects the config keys with the named parameters of the class
+    (device_server/devices/devicemanager.py), so a subclass that only forwards
+    **kwargs silently drops readout_time -- and passing it explicitly used to raise
+    'got multiple values for keyword argument'.
+    """
+    for cls in (Eiger1_5M, Eiger9M):
+        assert "readout_time" in inspect.signature(cls).parameters
+
+
+def _run_on_complete(eiger, broker_status, images_collected=0, timeout=5):
+    """Drive on_complete to a verdict against a given broker status.
+
+    Returns the status once it resolves; raises whatever on_complete raised.
+    """
+    eiger._wait_for_on_complete = 1  # keep the loop short
+    unblock_wait_for_idle = threading.Event()
+
+    def mock_wait_for_idle(timeout: float, raise_on_timeout: bool) -> bool:
+        return unblock_wait_for_idle.wait(timeout)
+
+    with (
+        mock.patch.object(eiger.jfj_client.api, "status_get", return_value=broker_status),
+        mock.patch.object(eiger.jfj_client, "wait_for_idle", side_effect=mock_wait_for_idle),
+        mock.patch.object(
+            eiger.jfj_client.api,
+            "statistics_data_collection_get",
+            return_value=MeasurementStatistics(run_number=1, images_collected=images_collected),
+        ),
+    ):
+        status = eiger.complete()
+        unblock_wait_for_idle.set()
+        status.wait(timeout=timeout)
+    return status
+
+
+def _missing_packets_status():
+    """A broker status carrying the packet-loss error JungfrauJoch actually emits."""
+    return BrokerStatus(
+        state="Idle", message_severity="error", message=f"{MISSING_PACKETS_MESSAGE} (run 42)"
+    )
+
+
+def test_eiger_tolerates_missing_packets_when_configured_to(eiger_1_5m):
+    """With the tolerance on, a missing-packet error completes the scan and is counted."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+
+    status = _run_on_complete(eiger, _missing_packets_status())
+
+    assert status.done is True
+    assert status.success is True
+    assert eiger.get_missing_packet_events() == 1
+
+
+def test_eiger_raises_on_missing_packets_when_asked_to(eiger_1_5m):
+    """With the tolerance off, the same error fails the scan and is not counted."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(True)
+
+    with pytest.raises(EigerError):
+        _run_on_complete(eiger, _missing_packets_status())
+
+    assert eiger.get_missing_packet_events() == 0
+
+
+def test_eiger_tolerance_does_not_extend_to_other_broker_errors(eiger_1_5m):
+    """Tolerating packet loss must not tolerate every error the broker reports."""
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+
+    with pytest.raises(EigerError):
+        _run_on_complete(
+            eiger, BrokerStatus(state="Idle", message_severity="error", message="Detector on fire")
+        )
+
+    assert eiger.get_missing_packet_events() == 0
+
+
+def test_eiger_tolerance_does_not_hide_a_short_acquisition(eiger_1_5m):
+    """The frame-count check still fires: only flagged-but-complete data is let through.
+
+    This is what makes the tolerance defensible -- a scan that actually lost frames still
+    fails, whatever raise_on_missing_packets says.
+    """
+    eiger = eiger_1_5m
+    eiger.set_raise_on_missing_packets(False)
+    eiger._num_triggers = 10
+
+    with pytest.raises(EigerError, match="acquisition incomplete"):
+        _run_on_complete(eiger, _missing_packets_status(), images_collected=9)
+
+
+def test_eiger_missing_packet_tolerance_is_switchable_at_runtime(eiger_1_5m):
+    """The flag is reachable from the client, so a beamtime can change its mind."""
+    eiger = eiger_1_5m
+    assert eiger.get_raise_on_missing_packets() is RAISE_ON_MISSING_PACKETS
+
+    eiger.set_raise_on_missing_packets(True)
+    assert eiger.get_raise_on_missing_packets() is True
+    eiger.set_raise_on_missing_packets(False)
+    assert eiger.get_raise_on_missing_packets() is False
+
+    for name in (
+        "get_raise_on_missing_packets",
+        "set_raise_on_missing_packets",
+        "get_missing_packet_events",
+    ):
+        assert name in type(eiger).USER_ACCESS
+
+
+def test_eiger_missing_packet_tolerance_is_overridable_per_deployment(mock_scan_info):
+    """deviceConfig's raise_on_missing_packets wins over the default, per detector."""
+    dev = Eiger1_5M(
+        name="eiger_1_5m",
+        beam_center=(256, 256),
+        detector_distance=100.0,
+        raise_on_missing_packets=True,
+    )
+    try:
+        assert dev.get_raise_on_missing_packets() is True
+    finally:
+        dev.destroy()
+
+
+def test_eiger_subclasses_name_raise_on_missing_packets_for_the_device_server():
+    """Same rule as readout_time: a deviceConfig key only reaches a device if the
+    class signature names it, so **kwargs alone would silently drop it."""
+    for cls in (Eiger1_5M, Eiger9M):
+        assert "raise_on_missing_packets" in inspect.signature(cls).parameters

tests/tests_devices/test_ids_camera.py

@@ -28,6 +28,7 @@ def ids_camera():
     camera.cam.cam.width.value = 2
     camera.cam.cam.height.value = 2
     yield camera
+    camera.stop_live_mode()
 
 
 def test_mask_setter_getter(ids_camera):

tests/tests_devices/test_mcs_card.py

@@ -24,7 +24,7 @@ from csaxs_bec.devices.epics.mcs_card.mcs_card import (
     READMODE,
     MCSCard,
 )
-from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS
+from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS, suppress_mca_callbacks
 from csaxs_bec.devices.utils.utils import fetch_scan_info
 
 
@@ -167,9 +167,10 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS):
     """
     Test complete method of MCSCarcCSAXS.
 
-    Two use cases:
+    Three use cases:
     I. Acquisition is stopped externally
     II. Acquisition completes normally
+    III. Acquisition emits more points than requested
     """
     mcs = mock_mcs_csaxs
     mcs.scan_parameters = fetch_scan_info(mcs.scan_info)
@@ -219,6 +220,38 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS):
     mcs._start_monitor_async_data_emission.wait(2)
     assert not mcs._start_monitor_async_data_emission.is_set()
 
+    ##############################
+    # III. Extra data fails fast #
+    ##############################
+    mcs._current_data_index = 11
+    st = mcs.complete()
+    with pytest.raises(RuntimeError, match="emitted 11 data points, but the scan requested 10"):
+        st.wait(timeout=1)
+
+
+def test_suppress_mca_callbacks_discards_callbacks_without_blocking(mock_mcs_csaxs: MCSCardCSAXS):
+    """A complete MCA callback batch is discarded while suppression is active."""
+    mcs = mock_mcs_csaxs
+    callback_finished = threading.Event()
+
+    def emit_callbacks():
+        for index in range(mcs.NUM_MCA_CHANNELS):
+            counter = getattr(mcs.counters, f"mca{index + 1}")
+            mcs._on_counter_update(index, obj=counter, timestamp=1.0)
+        callback_finished.set()
+
+    worker = threading.Thread(target=emit_callbacks)
+    with mock.patch.object(mcs.mca, "put") as mca_put:
+        with suppress_mca_callbacks(mcs, restore_after_timeout=0):
+            worker.start()
+            assert callback_finished.wait(timeout=1)
+
+        worker.join(timeout=1)
+        assert not worker.is_alive()
+        assert mcs._current_data == {}
+        assert mcs._current_data_index == 0
+        mca_put.assert_not_called()
+
 
 def test_mcs_on_stop(mock_mcs_csaxs: MCSCardCSAXS):
     """Test that on stop sets the omit_mca_callbacks flag. Also test that on stage clears the omit_mca_callbacks flag."""