feat(eiger): optional auto-reinitialize of JungfrauJoch after detector error

When the detector goes into Error state, JungfrauJoch stays in WrongDAQState until
it is initialised again, so every following scan aborts in on_stage. With the new
auto_reinitialize device option, on_stage re-initialises the detector (like the
WEB UI "Initialize" button) before staging; the scan_repeat retry of the failed
projection then succeeds.

- JungfrauJochClient.recover_from_error(), single attempt, only from Error state
- wait_for_idle reports the broker's HTTP 500 message instead of a pydantic traceback
- on_complete fails immediately if the detector is in Error state
- default off; no re-initialisation during an acquisition

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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

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

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