diff --git a/csaxs_bec/devices/jungfraujoch/eiger.py b/csaxs_bec/devices/jungfraujoch/eiger.py index bc915928..789168af 100644 --- a/csaxs_bec/devices/jungfraujoch/eiger.py +++ b/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) diff --git a/csaxs_bec/devices/jungfraujoch/eiger_1_5m.py b/csaxs_bec/devices/jungfraujoch/eiger_1_5m.py index 92608502..f5bbcbf5 100644 --- a/csaxs_bec/devices/jungfraujoch/eiger_1_5m.py +++ b/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, diff --git a/csaxs_bec/devices/jungfraujoch/eiger_9m.py b/csaxs_bec/devices/jungfraujoch/eiger_9m.py index 13f1a4c6..4419c61f 100644 --- a/csaxs_bec/devices/jungfraujoch/eiger_9m.py +++ b/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, diff --git a/tests/tests_devices/test_eiger.py b/tests/tests_devices/test_eiger.py index 20dcf483..285dff5b 100644 --- a/tests/tests_devices/test_eiger.py +++ b/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