diff --git a/csaxs_bec/devices/jungfraujoch/eiger.py b/csaxs_bec/devices/jungfraujoch/eiger.py index c597ac10..bd31b2e9 100644 --- a/csaxs_bec/devices/jungfraujoch/eiger.py +++ b/csaxs_bec/devices/jungfraujoch/eiger.py @@ -64,7 +64,10 @@ from ophyd import DeviceStatus from ophyd_devices import FileEventSignal, PreviewSignal from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from csaxs_bec.devices.jungfraujoch.jungfrau_joch_client import JungfrauJochClient +from csaxs_bec.devices.jungfraujoch.jungfrau_joch_client import ( + JungfrauJochClient, + JungfrauJochClientError, +) from csaxs_bec.devices.jungfraujoch.jungfraujoch_preview import JungfrauJochPreview from csaxs_bec.devices.utils.utils import fetch_scan_info @@ -121,6 +124,9 @@ class Eiger(PSIDeviceBase): 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. + auto_reinitialize (bool): Whether the detector is re-initialised on staging if + JungfrauJoch is in ERROR state, instead of requiring a manual 'Initialize' in the + JungfrauJoch WEB UI. Requires the shutter to be closed between scans. **kwargs: Additional keyword arguments. """ @@ -151,6 +157,7 @@ class Eiger(PSIDeviceBase): scan_info: ScanInfo = None, readout_time: float = EIGER_READOUT_TIME, raise_on_missing_packets: bool = RAISE_ON_MISSING_PACKETS, + auto_reinitialize: bool = False, device_manager=None, **kwargs, ): @@ -176,6 +183,7 @@ class Eiger(PSIDeviceBase): self.scan_parameters: ScanServerScanInfo | None = None self._raise_on_missing_packets = bool(raise_on_missing_packets) self._missing_packet_events = 0 + self._auto_reinitialize = bool(auto_reinitialize) def get_raise_on_missing_packets(self) -> bool: """ @@ -415,6 +423,14 @@ class Eiger(PSIDeviceBase): logger.info(f"Device {self.name} on_stage called.") start_time = time.time() self.scan_parameters = fetch_scan_info(self.scan_info) + if self._auto_reinitialize: + try: + self.jfj_client.recover_from_error() + except JungfrauJochClientError as exc: + raise EigerError( + f"Device {self.name} is in ERROR state and could not be re-initialised " + f"automatically, please initialise it in the WEB UI: {self._host}. {exc}" + ) from exc if self.jfj_client.detector_state != DetectorState.IDLE: self.jfj_client.stop(request_timeout=0.5) self.jfj_client.wait_for_idle(timeout=5) @@ -536,6 +552,11 @@ class Eiger(PSIDeviceBase): # NOTE: This adjust the time (s) that should be waited for completion of the scan. timeout = self._wait_for_on_complete while time.time() - start_time < timeout: + if self.jfj_client.detector_state == DetectorState.ERROR: + raise EigerError( + f"Device {self.name} is in ERROR state during acquisition. Broker status: " + f"{yaml.dump(self.jfj_client.jfj_status.to_dict(), indent=4)}" + ) if self.jfj_client.wait_for_idle(timeout=1, raise_on_timeout=False): # Call API endpoints to get broker status broker_status = self.jfj_client.jfj_status diff --git a/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py b/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py index dab8cf6c..2d514dce 100644 --- a/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py +++ b/csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py @@ -3,6 +3,7 @@ from __future__ import annotations import enum +import json import threading import time import traceback @@ -14,6 +15,7 @@ from bec_lib.logger import bec_logger from jfjoch_client.api.default_api import DefaultApi from jfjoch_client.api_client import ApiClient from jfjoch_client.configuration import Configuration +from jfjoch_client.exceptions import ServiceException from jfjoch_client.models.broker_status import BrokerStatus from jfjoch_client.models.dataset_settings import DatasetSettings from jfjoch_client.models.detector_settings import DetectorSettings @@ -60,6 +62,8 @@ class JungfrauJochClient: self, host: str = "http://sls-jfjoch-001:8080", parent: Device | None = None ) -> None: self._initialised = False + self._recover_lock = threading.Lock() + self._recoveries = 0 configuration = Configuration(host=host) api_client = ApiClient(configuration) self.api = DefaultApi(api_client) @@ -97,10 +101,75 @@ class JungfrauJochClient: status = self.detector_state # TODO: #135 Check if the detector has to be in INACTIVE state before initialisation if status != DetectorState.IDLE: - self.api.initialize_post() - self.wait_for_idle(timeout) + self._initialize(timeout) self.initialised = True + def _initialize(self, timeout: float) -> None: + """ + Trigger the initialisation of the detector and wait until it is done. The X-ray shutter + must be closed, and the call can take up to a few minutes depending on the detector. + + Args: + timeout (float): Timeout in seconds to wait for the initialisation to finish. + """ + self.api.initialize_post() + self.wait_for_idle(timeout) + + def recover_from_error(self, timeout: float = 60) -> bool: + """ + Re-initialise the detector if JungfrauJoch is in ERROR state, equivalent to pressing + 'Initialize' in the JungfrauJoch WEB UI. Must only be called between acquisitions, as + the initialisation requires a closed shutter and is not safe during a measurement. + Only a single attempt is made per call. + + Args: + timeout (float): Timeout in seconds to wait for the detector to become IDLE again. + + Returns: + bool: True if the detector was re-initialised, False if it was not in ERROR state. + + Raises: + JungfrauJochClientError: If the detector is not IDLE after the re-initialisation. + """ + with self._recover_lock: + if self.detector_state != DetectorState.ERROR: + return False + self._recoveries += 1 + logger.warning( + f"JungfrauJoch backend for device {self._parent_name} is in ERROR state, " + f"re-initialising the detector (recovery #{self._recoveries} since startup)." + ) + try: + self._initialize(timeout) + except Exception as exc: + raise JungfrauJochClientError( + f"Error on device {self._parent_name}: re-initialisation after detector " + f"error failed: {exc}" + ) from exc + state = self.detector_state + if state != DetectorState.IDLE: + raise JungfrauJochClientError( + f"Error on device {self._parent_name}: detector is in state {state.value} " + "after re-initialisation, expected IDLE." + ) + logger.info(f"Device {self._parent_name} re-initialised and back in IDLE state.") + return True + + @staticmethod + def _api_error_message(exc: Exception) -> str | None: + """ + Extract the message of a JungfrauJoch HTTP 500 error. The response body is a plain + string that jfjoch_client fails to parse into its ErrorMessage model (pydantic + ValidationError), so the message is read from the raw body of the ServiceException. + """ + if not isinstance(exc, ServiceException): + return None + try: + body = json.loads(exc.body) + return f"{body['msg']} ({body['reason']})" + except Exception: + return exc.body or exc.reason + def set_detector_settings(self, settings: dict | DetectorSettings, timeout: int = 10) -> None: """ Set the detector settings. The state of JungfrauJoch must be in IDLE, @@ -237,12 +306,15 @@ class JungfrauJochClient: return False except Exception as exc: content = traceback.format_exc() + message = self._api_error_message(exc) logger.info( - f"Error on device {self._parent_name} in wait_for_idle. Full traceback: {content}" + f"Error on device {self._parent_name} in wait_for_idle. " + f"{message or ''} Full traceback: {content}" ) if raise_on_timeout: raise JungfrauJochClientError( - f"Error on device {self._parent_name} in wait_for_idle: {content}" + f"Error on device {self._parent_name} in wait_for_idle: " + f"{message or content}" ) from exc return False return True diff --git a/tests/tests_devices/test_eiger.py b/tests/tests_devices/test_eiger.py index 979494c2..2b966dc7 100644 --- a/tests/tests_devices/test_eiger.py +++ b/tests/tests_devices/test_eiger.py @@ -10,6 +10,7 @@ import numpy as np import pytest from bec_lib.messages import FileMessage, ScanStatusMessage from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo +from jfjoch_client.exceptions import ServiceException from jfjoch_client.models.broker_status import BrokerStatus from jfjoch_client.models.dataset_settings import DatasetSettings from jfjoch_client.models.detector_list import DetectorList @@ -29,6 +30,11 @@ from csaxs_bec.devices.jungfraujoch.eiger import ( MISSING_PACKETS_MESSAGE, RAISE_ON_MISSING_PACKETS, ) +from csaxs_bec.devices.jungfraujoch.jungfrau_joch_client import ( + DetectorState, + JungfrauJochClient, + JungfrauJochClientError, +) 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 from csaxs_bec.devices.jungfraujoch.eiger2_si_500k import ( @@ -794,3 +800,106 @@ def test_eiger_subclasses_name_raise_on_missing_packets_for_the_device_server(): class signature names it, so **kwargs alone would silently drop it.""" for cls in (Eiger1_5M, Eiger9M, Eiger2Si500K): assert "raise_on_missing_packets" in inspect.signature(cls).parameters + + +def test_eiger_on_stage_auto_reinitialize(eiger_1_5m): + """With auto_reinitialize the detector is recovered on staging, otherwise it is not.""" + eiger = eiger_1_5m + for flag in (True, False): + eiger._auto_reinitialize = flag + 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, "stop"), + mock.patch.object(eiger.jfj_client, "recover_from_error") as mock_recover, + mock.patch.object( + eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle") + ), + ): + eiger.stage() + assert mock_recover.call_count == (1 if flag else 0) + eiger.unstage() + + +def test_eiger_on_stage_auto_reinitialize_failure(eiger_1_5m): + """A failed recovery raises a single EigerError pointing to the WEB UI.""" + eiger = eiger_1_5m + eiger._auto_reinitialize = True + with mock.patch.object( + eiger.jfj_client, "recover_from_error", side_effect=JungfrauJochClientError("still bad") + ): + with pytest.raises(EigerError, match="could not be re-initialised"): + eiger.on_stage() + + +def test_eiger_on_complete_fails_fast_in_error_state(eiger_1_5m): + """on_complete raises right away if the detector is in ERROR state, not after the timeout.""" + eiger = eiger_1_5m + with mock.patch.object( + eiger.jfj_client.api, + "status_get", + return_value=BrokerStatus(state="Error", message_severity="error", message="boom"), + ): + start = time.time() + status = eiger.complete() + with pytest.raises(EigerError): + status.wait(timeout=5) + assert time.time() - start < 5 + + +def _jfj_client() -> JungfrauJochClient: + return JungfrauJochClient(host="http://localhost:1") + + +@pytest.mark.parametrize( + "states, expected, n_init", + [ + (["Idle"], False, 0), + (["Error", "Idle"], True, 1), + ], +) +def test_jfj_client_recover_from_error(states, expected, n_init): + client = _jfj_client() + with ( + mock.patch.object( + JungfrauJochClient, + "detector_state", + new_callable=mock.PropertyMock, + side_effect=[DetectorState(s) for s in states], + ), + mock.patch.object(client.api, "initialize_post") as mock_init, + mock.patch.object(client, "wait_for_idle", return_value=True) as mock_wait, + ): + assert client.recover_from_error() is expected + assert mock_init.call_count == n_init + assert mock_wait.call_count == n_init + + +def test_jfj_client_recover_from_error_still_in_error(): + client = _jfj_client() + with ( + mock.patch.object( + JungfrauJochClient, + "detector_state", + new_callable=mock.PropertyMock, + return_value=DetectorState.ERROR, + ), + mock.patch.object(client.api, "initialize_post"), + mock.patch.object(client, "wait_for_idle", return_value=True), + ): + with pytest.raises(JungfrauJochClientError, match="after re-initialisation"): + client.recover_from_error() + + +def test_jfj_client_wait_for_idle_reports_broker_message(): + """The HTTP 500 body is reported instead of the pydantic validation traceback.""" + client = _jfj_client() + body = ( + '{"msg":"DAQ state error (Detector error (Error in detector operation))",' + '"reason":"WrongDAQState"}' + ) + exc = ServiceException(status=500, reason="Internal Server Error", body=body) + with mock.patch.object(client.api, "wait_till_done_post", side_effect=exc): + with pytest.raises(JungfrauJochClientError, match="WrongDAQState"): + client.wait_for_idle(timeout=1) + assert client.wait_for_idle(timeout=1, raise_on_timeout=False) is False