Source code for csaxs_bec.devices.utils.utils

"""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"])
    info["num_monitored_readouts"] = scan_info.msg.num_monitored_readouts
    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