bec/csaxs_bec
0
0
Fork 1
Code Issues 13 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

fix: Cleanup of jungfrau_joch_client

Browse Source
This commit is contained in:
appel_c
2026-01-29 10:21:36 +01:00
parent 665eba9168
commit 30c49aef25
1 changed files with 48 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py
View File

@@ -9,6 +9,7 @@ import traceback
from typing import TYPE_CHECKING
import requests
import yaml
from bec_lib.logger import bec_logger
from jfjoch_client.api.default_api import DefaultApi
from jfjoch_client.api_client import ApiClient
@@ -19,7 +20,7 @@ from jfjoch_client.models.detector_settings import DetectorSettings
logger = bec_logger.logger
if TYPE_CHECKING:
if TYPE_CHECKING: # pragma: no cover
from ophyd import Device
@@ -29,9 +30,6 @@ class JungfrauJochClientError(Exception):
"""Base class for exceptions in this module."""
"Inactive", "Idle", "Busy", "Measuring", "Pedestal", "Error"
class DetectorState(str, enum.Enum):
"""
Enum states of the BrokerStatus state. The pydantic model validates in runtime,
@@ -48,7 +46,7 @@ class DetectorState(str, enum.Enum):
class JungfrauJochClient:
"""
Jungfrau Joch API client wrapper. It provides a few thin wrappers around the API client,
Jungfrau Joch API client wrapper. It provides a thin wrapper methods around the API client,
that allow to connect, initialise, wait for state changes, set settings, start and stop acquisitions.
Args:
@@ -71,6 +69,7 @@ class JungfrauJochClient:
response = self.api.status_get()
return BrokerStatus(**response.to_dict())
# pylint: disable=missing-function-docstring
@property
def initialised(self) -> bool:
return self._initialised
@@ -79,39 +78,46 @@ class JungfrauJochClient:
def initialised(self, value: bool) -> None:
self._initialised = value
# pylint: disable=missing-function-docstring
@property
def detector_state(self) -> DetectorState:
"""Detector state of JungfrauJoch."""
return DetectorState(self.jjf_state.state)
def connect_and_initialise(self, timeout: int = 10, **kwargs) -> None:
def connect_and_initialise(self, timeout: int = 10) -> None:
"""
Connect and initialise the JungfrauJoch detector. The detector must be in
IDLE state to become initialised. This is a blocking call.
IDLE state to become initialised. This is a blocking call, the timeout parameter
will be passed to the HTTP requests timeout method of the wait_for_idle method.
Args:
timeout (int): Timeout in seconds for the initialisation and waiting for IDLE state.
"""
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, request_timeout=timeout)
self.initialised = True
def set_detector_settings(self, settings: dict | DetectorSettings, timeout: int = 10) -> None:
"""Set the detector settings. JungfrauJoch must be in IDLE, Error or Inactive state.
Note, the full settings have to be provided, otherwise the settings will be overwritten with default values.
"""
Set the detector settings. The state of JungfrauJoch must be in IDLE, Error or Inactive state.
Note, a full set of setttings has to be provided, otherwise the settings will be overwritten with default values.
Args:
settings (dict): dictionary of settings
timeout (int): Timeout in seconds for the HTTP request to set the settings.
"""
state = self.detector_state
if state not in [DetectorState.IDLE, DetectorState.ERROR, DetectorState.INACTIVE]:
logger.info(
f"JungfrauJoch backend fo device {self._parent_name} is not in IDLE state, waiting 1s before retrying..."
)
time.sleep(1) # Give the detector 1s to become IDLE, retry
state = self.detector_state
if state not in [DetectorState.IDLE, DetectorState.ERROR, DetectorState.INACTIVE]:
raise JungfrauJochClientError(
f"Error in {self._parent_name}. Detector must be in IDLE, ERROR or INACTIVE state to set settings. Current state: {state}"
f"Error on {self._parent_name}. Detector must be in IDLE, ERROR or INACTIVE state to set settings. Current state: {state}"
)
if isinstance(settings, dict):
@@ -119,31 +125,31 @@ class JungfrauJochClient:
try:
self.api.config_detector_put(detector_settings=settings, _request_timeout=timeout)
except requests.exceptions.Timeout:
raise TimeoutError(f"Timeout while setting detector settings for {self._parent_name}")
raise TimeoutError(
f"Timeout on device {self._parent_name} while setting detector settings {yaml.dump(settings, indent=4)}."
)
except Exception:
content = traceback.format_exc()
logger.error(
f"Error while setting detector settings for {self._parent_name}: {content}"
f"Error on device {self._parent_name} while setting detector settings {yaml.dump(settings, indent=4)}. Error traceback: {content}"
)
raise JungfrauJochClientError(
f"Error while setting detector settings for parent device {self._parent_name}."
f"Error on device {self._parent_name} while setting detector settings {yaml.dump(settings, indent=4)}. Full traceback: {content}."
)
def start(self, settings: dict | DatasetSettings, request_timeout: float = 10) -> None:
"""Start the mesaurement. DatasetSettings must be provided, and JungfrauJoch must be in IDLE state.
The method call is blocking and JungfrauJoch will be ready to measure after the call resolves.
"""
Start the acquisition with the provided dataset settings. The detector must be in IDLE state.
Settings must always provide a full set of parameters, missing parameters will be set to default values.
Args:
settings (dict): dictionary of settings
Please check the DataSettings class for the available settings. Minimum required settings are
beam_x_pxl, beam_y_pxl, detector_distance_mm, incident_energy_keV.
settings (dict | DatasetSettings): Dataset settings to start the acquisition with.
request_timeout (float): Timeout in seconds for the HTTP request to start the acquisition.
"""
state = self.detector_state
if state != DetectorState.IDLE:
raise JungfrauJochClientError(
f"Error in {self._parent_name}. Detector must be in IDLE state to set settings. Current state: {state}"
f"Error on device {self._parent_name}. Detector must be in IDLE state to start acquisition. Current state: {state}"
)
if isinstance(settings, dict):
@@ -155,18 +161,18 @@ class JungfrauJochClient:
except requests.exceptions.Timeout:
content = traceback.format_exc()
logger.error(
f"TimeoutError in JungfrauJochClient for parent device {self._parent_name} during 'start' call: {content}"
f"Timeout error after {request_timeout} seconds on device {self._parent_name} during 'start' call with dataset settings: {yaml.dump(settings, indent=4)}. Traceback: {content}"
)
raise TimeoutError(
f"TimeoutError in JungfrauJochClient for parent device {self._parent_name} for 'start' call"
f"Timeout error after {request_timeout} seconds on device {self._parent_name} during 'start' call with dataset settings: {yaml.dump(settings, indent=4)}."
)
except Exception:
content = traceback.format_exc()
logger.error(
f"Error in JungfrauJochClient for parent device {self._parent_name} during 'start' call: {content}"
f"Error on device {self._parent_name} during 'start' post with dataset settings: {yaml.dump(settings, indent=4)}. Traceback: {content}"
)
raise JungfrauJochClientError(
f"Error in JungfrauJochClient for parent device {self._parent_name} during 'start' post."
f"Error on device {self._parent_name} during 'start' post with dataset settings: {yaml.dump(settings, indent=4)}. Full traceback: {content}."
)
def stop(self, request_timeout: float = 0.5) -> None:
@@ -178,49 +184,49 @@ class JungfrauJochClient:
except requests.exceptions.Timeout:
content = traceback.format_exc()
logger.error(
f"Timeout in JungFrauJochClient for device {self._parent_name} during stop: {content}"
f"Timeout error after {request_timeout} seconds on device {self._parent_name} during stop: {content}"
)
except Exception:
content = traceback.format_exc()
logger.error(
f"Error in JungFrauJochClient for device {self._parent_name} during stop: {content}"
)
logger.error(f"Error on device {self._parent_name} during stop: {content}")
thread = threading.Thread(
target=_stop_call, daemon=True, args=(self,), name="stop_jungfraujoch_thread"
)
thread.start()
def wait_for_idle(
self, timeout: int = 10, request_timeout: float | None = None, raise_on_timeout: bool = True
) -> bool:
"""Wait for JungfrauJoch to be in Idle state. Blocking call with timeout.
def wait_for_idle(self, timeout: int = 10, raise_on_timeout: bool = True) -> bool:
"""
Method to wait until the detector is in IDLE state. This is a blocking call with a
timeout that can be specified. The additional parameter raise_on_timeout can be used to
raise an exception on timeout instead of returning boolean True/False.
Args:
timeout (int): timeout in seconds
raise_on_timeout (bool): If True, raises an exception on timeout. Default is True.
Returns:
bool: True if the detector is in IDLE state, False if timeout occurred
"""
if request_timeout is None:
request_timeout = timeout
try:
self.api.wait_till_done_post(timeout=timeout, _request_timeout=request_timeout)
self.api.wait_till_done_post(timeout=timeout, _request_timeout=timeout)
except requests.exceptions.Timeout:
content = traceback.format_exc()
logger.debug(
f"HTTP request timeout in wait_for_idle for {self._parent_name}: {content}"
logger.info(
f"Timeout after {timeout} seconds on device {self._parent_name} in wait_for_idle: {content}"
)
if raise_on_timeout:
raise TimeoutError(
f"HTTP request timeout in wait_for_idle for {self._parent_name}."
f"Timeout after {timeout} seconds on device {self._parent_name} in wait_for_idle."
)
return False
except Exception as exc:
content = traceback.format_exc()
logger.debug(f"Waiting for device {self._parent_name} to become IDLE: {content}")
logger.info(
f"Error on device {self._parent_name} in wait_for_idle. Full traceback: {content}"
)
if raise_on_timeout:
raise JungfrauJochClientError(
f"Error in wait_for_idle for {self._parent_name}: {content}"
f"Error on device {self._parent_name} in wait_for_idle: {content}"
) from exc
return False
return True