csaxs_bec/devices/jungfraujoch/eiger.py

@@ -84,6 +84,19 @@ logger = bec_logger.logger
 # 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
@@ -106,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.
     """
 
@@ -115,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")
@@ -130,6 +148,7 @@ class Eiger(PSIDeviceBase):
         beam_center: tuple[int, int] = (0, 0),
         scan_info: ScanInfo = None,
         readout_time: float = EIGER_READOUT_TIME,
+        raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS,
         device_manager=None,
         **kwargs,
     ):
@@ -153,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:
         """
@@ -477,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,7 +9,7 @@ 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
 
 # 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
@@ -45,12 +45,14 @@ class Eiger1_5M(Eiger):
         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=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,7 +14,7 @@ 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
@@ -49,12 +49,14 @@ class Eiger9M(Eiger):
         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=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_eiger.py

@@ -23,7 +23,12 @@ 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 import EIGER_READOUT_TIME, MIN_EXP_TIME
+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
 
@@ -329,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()
@@ -571,3 +577,125 @@ def test_eiger_subclasses_name_readout_time_for_the_device_server():
     """
     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