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

refactor: client refactoring, adding tests for jfj_client models

Browse Source
This commit is contained in:
appel_c
2025-02-26 11:22:14 +01:00
parent 4907c5db27
commit ccf8bb8474
2 changed files with 122 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/devices/jungfraujoch/jungfrau_joch_client.py
+68 -60
View File
@@ -1,8 +1,16 @@
""" Module with client interface for the Jungfrau Joch detector API"""
import enum
import math
import jfjoch_client
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.models.broker_status import BrokerStatus
from jfjoch_client.models.dataset_settings import DatasetSettings
from jfjoch_client.models.detector_settings import DetectorSettings
from ophyd import Device
logger = bec_logger.logger
@@ -12,9 +20,7 @@ class JungfrauJochClientError(Exception):
class DetectorState(enum.StrEnum):
"""Detector states for Jungfrau Joch detector
['Inactive', 'Idle', 'Busy', 'Measuring', 'Pedestal', 'Error']
"""
"""Possible Detector states for Jungfrau Joch detector"""
INACTIVE = "Inactive"
IDLE = "Idle"
@@ -25,10 +31,10 @@ class DetectorState(enum.StrEnum):
class ResponseWaitDone(enum.IntEnum):
"""Response state for Jungfrau Joch detector wait till done"""
"""Response state for Jungfrau Joch detector wait till done call"""
DETECTOR_IDLE = 200
TIMEOUT_PARAM_OUT_OF_RANGE = 400
TIMEOUT_PARAM_OUT_OF_BOUNDS = 400
JUNGFRAU_ERROR = 500
DETECTOR_INACTIVE = 502
TIMEOUT_REACHED = 504
@@ -37,11 +43,20 @@ class ResponseWaitDone(enum.IntEnum):
class JungfrauJochClient:
"""Thin wrapper around the Jungfrau Joch API client"""
def __init__(self, host: str = "http://sls-jfjoch-001:8080") -> None:
def __init__(
self, host: str = "http://sls-jfjoch-001:8080", parent: Device | None = None
) -> None:
self._initialised = False
configuration = jfjoch_client.Configuration(host=host)
api_client = jfjoch_client.ApiClient(configuration)
self.api = jfjoch_client.DefaultApi(api_client)
configuration = Configuration(host=host)
api_client = ApiClient(configuration)
self.api = DefaultApi(api_client)
self._parent_name = parent.name if parent else self.__class__.__name__
@property
def jjf_state(self) -> BrokerStatus:
"""Get the status of JungfrauJoch"""
response = self.api.status_get()
return BrokerStatus(**response.to_dict())
@property
def initialised(self) -> bool:
@@ -53,19 +68,19 @@ class JungfrauJochClient:
"""Set the connected status"""
self._initialised = value
def get_jungfrau_joch_status(self) -> DetectorState:
def get_detector_state(self) -> DetectorState:
"""Get the status of JungfrauJoch"""
return self.api.status_get().state
return DetectorState(self.jjf_state.state)
def connect_and_initialise(self, timeout: int = 5) -> None:
"""Check if JungfrauJoch is connected and ready to receive commands"""
status = self.api.status_get().state
status = self.get_detector_state()
if status != DetectorState.IDLE:
self.api.initialize_post()
self.wait_till_done(timeout)
self.wait_till_done(timeout) # Blocking call
self.initialised = True
def set_detector_settings(self, settings: dict | jfjoch_client.DatasetSettings) -> None:
def set_detector_settings(self, settings: dict | DetectorSettings) -> 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.
@@ -75,79 +90,72 @@ class JungfrauJochClient:
state = self.api.status_get().state
if state not in [DetectorState.IDLE, DetectorState.ERROR, DetectorState.INACTIVE]:
raise JungfrauJochClientError(
f"Detector must be in IDLE, ERROR or INACTIVE state to set settings. Current state: {state}"
f"Error in {self._parent_name}. Detector must be in IDLE, ERROR or INACTIVE state to set settings. Current state: {state}"
)
if isinstance(settings, dict):
settings = jfjoch_client.DatasetSettings(**settings)
self.api.config_detector_put(settings)
settings = DetectorSettings(**settings)
self.api.config_detector_put(detector_settings=settings)
def set_mesaurement_settings(self, settings: dict | jfjoch_client.DatasetSettings) -> None:
"""Set the measurement settings. JungfrauJoch must be in IDLE state.
def start_mesaurement(self, settings: dict | DatasetSettings) -> 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.
Please check the DataSettings class for the available settings.
The minimum required settings are:
beam_x_pxl: StrictFloat | StrictInt,
beam_y_pxl: StrictFloat | StrictInt,
detector_distance_mm: float | int,
incident_energy_keV: float | int,
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.
"""
state = self.api.status_get().state
state = self.get_detector_state()
if state != DetectorState.IDLE:
raise JungfrauJochClientError(
f"Detector must be in IDLE state to set settings. Current state: {state}"
f"Error in {self._parent_name}. Detector must be in IDLE state to set settings. Current state: {state}"
)
if isinstance(settings, dict):
settings = jfjoch_client.DatasetSettings(**settings)
settings = DatasetSettings(**settings)
try:
res = self.api.start_post_with_http_info(dataset_settings=settings)
if res.status_code != 200:
logger.error(
f"Error while setting measurement settings {settings}, response: {res}"
)
raise JungfrauJochClientError(
f"Error while setting measurement settings {settings}, response: {res}"
)
response = f"Error in {self._parent_name}, while setting measurement settings {settings}, response: {res}"
raise JungfrauJochClientError(response)
except JungfrauJochClientError as e:
logger.error(e)
raise e
except Exception as e:
logger.error(
f"Error while setting measurement settings {settings}. Exception raised {e}"
)
raise JungfrauJochClientError(
f"Error while setting measurement settings {settings}. Exception raised {e}"
) from e
response = f"Error in {self._parent_name}, while setting measurement settings {settings}, exception: {e}"
logger.error(response)
raise JungfrauJochClientError(response) from e
def wait_till_done(self, timeout: int = 5) -> None:
"""Wait until JungfrauJoch is done.
"""Wait for JungfrauJoch to be in Idle state. Blocking call with timeout.
Args:
timeout (int): timeout in seconds
"""
success = False
try:
response = self.api.wait_till_done_post_with_http_info(math.ceil(timeout / 2))
if response.status_code != ResponseWaitDone.DETECTOR_IDLE:
logger.info(
f"Waitin for JungfrauJoch to be done, status: {ResponseWaitDone(response.status_code)}; response msg {response}"
)
response = self.api.wait_till_done_post_with_http_info(math.floor(timeout / 2))
if response.status_code == ResponseWaitDone.DETECTOR_IDLE:
success = True
return
logger.info(
f"Waiting for device {self._parent_name}, jungfrau joch to become IDLE, "
f"status: {ResponseWaitDone(response.status_code)}; response msg {response}"
)
response = self.api.wait_till_done_post_with_http_info(math.floor(timeout / 2))
if response.status_code == ResponseWaitDone.DETECTOR_IDLE:
return
except Exception as e:
logger.error(f"Error while waiting for JungfrauJoch to initialise: {e}")
logger.error(
f"Error in device {self._parent_name} while waiting for JungfrauJoch to initialise: {e}"
)
raise JungfrauJochClientError(
f"Error while waiting for JungfrauJoch to initialise: {e}"
f"Error in device {self._parent_name} while waiting for JungfrauJoch to initialise. Exception: {e}"
) from e
else:
if success is False:
logger.error(
f"Failed to initialise JungfrauJoch with status: {response.status_code}; response msg {response}"
)
raise JungfrauJochClientError(
f"Failed to initialise JungfrauJoch with status: {response.status_code}; response msg {response}"
)
# If the response is IDLE, this part is never reached. We will raise a TimeoutError.
msg = (
f"TimeoutError in device {self._parent_name}, failed to initialise JungfrauJoch with status:"
f"{response.status_code}; response msg {response}"
)
logger.error(msg)
raise TimeoutError(msg)
tests/tests_devices/test_jungfrau_joch.py
+54
View File
@@ -0,0 +1,54 @@
import pytest
from jfjoch_client.api_response import ApiResponse
from jfjoch_client.models.broker_status import BrokerStatus
from jfjoch_client.models.dataset_settings import DatasetSettings
from jfjoch_client.models.detector_settings import DetectorSettings
from csaxs_bec.devices.jungfraujoch.jungfrau_joch_client import DetectorState, ResponseWaitDone
def test_jungfrau_joch_client_models_broker_status():
"""Test BrokerStatus model from JJF client"""
# Test broker status model
broker_status = BrokerStatus
assert "state" in broker_status.model_fields
assert "progress" in broker_status.model_fields
# Test that all DetectorStates are valid BrokerStatus states. This will not raise if
for state in DetectorState:
broker_status = BrokerStatus(state=state.value)
# Test an invalid state
with pytest.raises(ValueError):
broker_status = BrokerStatus(state="wrong")
def test_jungfrau_joch_client_models_dataset_settings():
"""Test DatasetSettings model from JJF client"""
# Test detector state model
settings = {
"beam_x_pxl": 0,
"beam_y_pxl": 0,
"detector_distance_mm": 100,
"incident_energy_keV": 10.00,
}
# Try creating DatasetSettings object with minimal required settigns
dataset_settings = DatasetSettings(**settings)
# Test that image_time_ns and ntrigger are still available
settings["image_time_us"] = 1000
settings["ntrigger"] = 100
dataset_settings = DatasetSettings(**settings)
def test_jungfrau_joch_client_models_api_response():
"""Test APIResponse model from JJF client.
We can only check that all http status code responses are valid.
"""
# Check if all ResponseWaitDone http status codes are valid for the APIResponse model
for state in ResponseWaitDone:
response = ApiResponse(status_code=state.value, data="", headers=None, raw_data=b"")
def test_jungfrau_joch_client_models_detector_settigns():
"""Test DetectorSettings model from JJF client"""
# Must be initialized with frame_time_us
settings = {"frame_time_us": 450}
DetectorSettings(**settings) # type:ignore