diff --git a/csaxs_bec/devices/utils/__init__.py b/csaxs_bec/devices/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/csaxs_bec/devices/utils/utils.py b/csaxs_bec/devices/utils/utils.py new file mode 100644 index 0000000..1475961 --- /dev/null +++ b/csaxs_bec/devices/utils/utils.py @@ -0,0 +1,26 @@ +"""Utility functions for the devices.""" + +from copy import deepcopy + +import numpy as np +from bec_lib.devicemanager import ScanInfo +from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo +from pydantic import ValidationError + + +def fetch_scan_info(scan_info: ScanInfo) -> ScanServerScanInfo: + """Fetch the scan parameters from the scan_info object and return them as a ScanServerScanInfo object.""" + info = scan_info.msg.info + if isinstance(info["positions"], list): + info["positions"] = np.array(info["positions"]) + try: + msg = ScanServerScanInfo.model_validate(info) + except ValidationError: # This means we have an old scan_info object. + info = deepcopy(info) + # We need to convert a few parameters manually. + info["scan_type"] = ( + "hardware_triggered" if info["scan_type"] == "fly" else "software_triggered" + ) + msg = ScanServerScanInfo.model_validate(info) + + return msg