bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity
Files
637abe50c075191b8d15ca2ba4bce8cc2c2bf64b
debye_bec/debye_bec/devices/mo1_bragg/mo1_bragg.py
T

577 lines
24 KiB
Python
Raw Blame History

"""Module for the Mo1 Bragg positioner of the Debye beamline.
The softIOC is reachable via the EPICS prefix X01DA-OP-MO1:BRAGG: and connected
to a motor controller via web sockets. The Mo1 Bragg positioner is not only a
positioner, but also a scan controller to setup XAS and XRD scans. A few scan modes
are programmed in the controller, e.g. simple and advanced XAS scans + XRD triggering mode.
Note: For some of the Epics PVs, in particular action buttons, the put_complete=True is
used to ensure that the action is executed completely. This is believed
to allow for a more stable execution of the action."""
import time
from typing import Literal
from bec_lib.devicemanager import ScanInfo
from bec_lib.logger import bec_logger
from bec_server.device_server.devices.devicemanager import DeviceManagerDS
from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo
from ophyd import Component as Cpt
from ophyd import DeviceStatus, StatusBase
from ophyd.status import WaitTimeoutError
from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
from pydantic import BaseModel, Field
from typeguard import typechecked
from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggPositioner
# pylint: disable=unused-import
from debye_bec.devices.mo1_bragg.mo1_bragg_enums import (
MoveType,
ScanControlLoadMessage,
ScanControlMode,
ScanControlScanStatus,
TriggerControlMode,
TriggerControlSource,
)
from debye_bec.devices.mo1_bragg.mo1_bragg_utils import compute_spline
from debye_bec.devices.utils.utils import fetch_scan_info
# Initialise logger
logger = bec_logger.logger
########### Exceptions ###########
class Mo1BraggError(Exception):
"""Exception for the Mo1 Bragg positioner"""
########### Mo1 Bragg Motor Class ###########
class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
"""Mo1 Bragg motor for the Debye beamline.
The prefix to connect to the soft IOC is X01DA-OP-MO1:BRAGG:
"""
progress_signal = Cpt(ProgressSignal, name="progress_signal")
USER_ACCESS = ["set_advanced_xas_settings", "set_xtal", "convert_angle_energy"]
def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, device_manager: DeviceManagerDS | None = None, **kwargs): # type: ignore
"""
Initialize the PSI Device Base class.
Args:
name (str) : Name of the device
scan_info (ScanInfo): The scan info to use.
"""
super().__init__(
name=name, scan_info=scan_info, prefix=prefix, device_manager=device_manager, **kwargs
)
self.scan_parameters: ScanServerScanInfo = None
self.timeout_for_pvwait = 7.5
self.valid_scan_names = [
"xas_simple_scan",
"xas_simple_scan_with_xrd",
"xas_advanced_scan",
"xas_advanced_scan_with_xrd",
"nidaq_continuous_scan",
]
########################################
# Beamline Specific Implementations #
########################################
def on_init(self) -> None:
"""
Called when the device is initialized.
No signals are connected at this point. If you like to
set default values on signals, please use on_connected instead.
"""
def on_connected(self) -> None:
"""
Called after the device is connected and its signals are connected.
Default values for signals should be set here.
"""
self.scan_control.scan_progress.subscribe(self._progress_update, run=False)
def on_stage(self) -> DeviceStatus | StatusBase | None:
"""
Called while staging the device.
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
"""
self.scan_parameters = fetch_scan_info(self.scan_info)
if self.scan_control.scan_msg.get() != ScanControlLoadMessage.PENDING:
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
self.cancel_on_stop(status)
self.scan_control.scan_val_reset.put(1)
status.wait(timeout=self.timeout_for_pvwait)
scan_name = self.scan_parameters.scan_name
if self._check_if_scan_name_is_valid(self.scan_parameters):
if self.scan_parameters.positions is not None:
start, stop = (
self.scan_parameters.positions
if len(self.scan_parameters.positions) == 2
else (None, None)
)
else:
start, stop = (None, None)
scan_time = self.scan_parameters.additional_scan_parameters.get("scan_time", None)
scan_duration = self.scan_parameters.additional_scan_parameters.get(
"scan_duration", None
)
if scan_name == "xas_simple_scan":
if any(param is None for param in [start, stop, scan_time, scan_duration]):
raise Mo1BraggError(
f"Missing scan parameters for xas_simple_scan. Required parameters: start, stop, scan_time, scan_duration in additional_scan_parameters dict {self.scan_parameters.additional_scan_parameters}"
)
self.set_xas_settings(low=start, high=stop, scan_time=scan_time)
self.set_trig_settings(
enable_low=False,
enable_high=False,
break_time_low=0,
break_time_high=0,
cycle_low=0,
cycle_high=0,
exp_time=0,
n_of_trigger=0,
)
self.set_scan_control_settings(
mode=ScanControlMode.SIMPLE, scan_duration=scan_duration
)
elif scan_name == "xas_simple_scan_with_xrd":
break_enable_low = self.scan_parameters.additional_scan_parameters.get(
"break_enable_low", None
)
break_enable_high = self.scan_parameters.additional_scan_parameters.get(
"break_enable_high", None
)
break_time_low = self.scan_parameters.additional_scan_parameters.get(
"break_time_low", None
)
break_time_high = self.scan_parameters.additional_scan_parameters.get(
"break_time_high", None
)
cycle_low = self.scan_parameters.additional_scan_parameters.get("cycle_low", None)
cycle_high = self.scan_parameters.additional_scan_parameters.get("cycle_high", None)
exp_time = self.scan_parameters.exp_time
n_of_trigger = self.scan_parameters.additional_scan_parameters.get(
"n_of_trigger", None
)
if any(
param is None
for param in [
start,
stop,
scan_time,
scan_duration,
break_enable_low,
break_enable_high,
break_time_low,
break_time_high,
cycle_low,
cycle_high,
exp_time,
n_of_trigger,
]
):
raise Mo1BraggError(
f"Missing scan parameters for xas_simple_scan_with_xrd. Required parameters: start, stop, scan_time, scan_duration, break_enable_low, break_enable_high, break_time_low, break_time_high, cycle_low, cycle_high, exp_time, n_of_trigger in additional_scan_parameters dict {self.scan_parameters.additional_scan_parameters}"
)
self.set_xas_settings(low=start, high=stop, scan_time=scan_time)
self.set_trig_settings(
enable_low=break_enable_low,
enable_high=break_enable_high,
break_time_low=break_time_low,
break_time_high=break_time_high,
cycle_low=cycle_low,
cycle_high=cycle_high,
exp_time=exp_time,
n_of_trigger=n_of_trigger,
)
self.set_scan_control_settings(
mode=ScanControlMode.SIMPLE, scan_duration=scan_duration
)
elif scan_name == "xas_advanced_scan":
p_kink = self.scan_parameters.additional_scan_parameters.get("p_kink", None)
e_kink = self.scan_parameters.additional_scan_parameters.get("e_kink", None)
if any(
param is None
for param in [start, stop, scan_time, scan_duration, p_kink, e_kink]
):
raise Mo1BraggError(
f"Missing scan parameters for xas_advanced_scan. Required parameters: start, stop, scan_time, scan_duration, p_kink, e_kink in additional_scan_parameters dict {self.scan_parameters.additional_scan_parameters}"
)
self.set_advanced_xas_settings(
low=start, high=stop, scan_time=scan_time, p_kink=p_kink, e_kink=e_kink
)
self.set_trig_settings(
enable_low=False,
enable_high=False,
break_time_low=0,
break_time_high=0,
cycle_low=0,
cycle_high=0,
exp_time=0,
n_of_trigger=0,
)
self.set_scan_control_settings(
mode=ScanControlMode.ADVANCED, scan_duration=scan_duration
)
elif scan_name == "xas_advanced_scan_with_xrd":
p_kink = self.scan_parameters.additional_scan_parameters.get("p_kink", None)
e_kink = self.scan_parameters.additional_scan_parameters.get("e_kink", None)
break_enable_low = self.scan_parameters.additional_scan_parameters.get(
"break_enable_low", None
)
break_enable_high = self.scan_parameters.additional_scan_parameters.get(
"break_enable_high", None
)
break_time_low = self.scan_parameters.additional_scan_parameters.get(
"break_time_low", None
)
break_time_high = self.scan_parameters.additional_scan_parameters.get(
"break_time_high", None
)
cycle_low = self.scan_parameters.additional_scan_parameters.get("cycle_low", None)
cycle_high = self.scan_parameters.additional_scan_parameters.get("cycle_high", None)
exp_time = self.scan_parameters.exp_time
n_of_trigger = self.scan_parameters.additional_scan_parameters.get(
"n_of_trigger", None
)
if any(
param is None
for param in [
start,
stop,
scan_time,
scan_duration,
p_kink,
e_kink,
break_enable_low,
break_enable_high,
break_time_low,
break_time_high,
cycle_low,
cycle_high,
exp_time,
n_of_trigger,
]
):
raise Mo1BraggError(
f"Missing scan parameters for xas_advanced_scan_with_xrd. Required parameters: start, stop, scan_time, scan_duration, p_kink, e_kink, break_enable_low, break_enable_high, break_time_low, break_time_high, cycle_low, cycle_high, exp_time, n_of_trigger in additional_scan_parameters dict {self.scan_parameters.additional_scan_parameters}"
)
self.set_advanced_xas_settings(
low=start, high=stop, scan_time=scan_time, p_kink=p_kink, e_kink=e_kink
)
self.set_trig_settings(
enable_low=break_enable_low,
enable_high=break_enable_high,
break_time_low=break_time_low,
break_time_high=break_time_high,
cycle_low=cycle_low,
cycle_high=cycle_high,
exp_time=exp_time,
n_of_trigger=n_of_trigger,
)
self.set_scan_control_settings(
mode=ScanControlMode.ADVANCED, scan_duration=scan_duration
)
else:
return # Should never happen.
else:
return
# Setting scan duration seems to lag behind slightly in the backend, include small sleep
logger.info(f"Sleeping for one second")
time.sleep(1)
logger.info(f"Device {self.name}, done sleeping")
# Load the scan parameters to the controller
status = CompareStatus(
self.scan_control.scan_msg,
ScanControlLoadMessage.SUCCESS,
failure_value=[m for m in ScanControlLoadMessage if m.name.startswith("ERR_")],
)
self.cancel_on_stop(status)
self.scan_control.scan_load.put(1)
# Wait for params to be checked from controller
status.wait(self.timeout_for_pvwait)
return None
def on_unstage(self) -> DeviceStatus | StatusBase | None:
"""Called while unstaging the device."""
if self.stopped is True:
logger.warning(f"Resetting stopped in unstage for device {self.name}.")
self._stopped = False
if self.scan_control.scan_msg.get() in [
ScanControlLoadMessage.STARTED,
ScanControlLoadMessage.SUCCESS,
]:
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
self.cancel_on_stop(status)
try:
status.wait(2)
return None
except WaitTimeoutError:
logger.warning(
f"Timeout in on_unstage of {self.name} after {self.timeout_for_pvwait}s, current scan_control_message : {self.scan_control.scan_msg.get()}"
)
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
self.cancel_on_stop(status)
self.scan_control.scan_val_reset.put(1)
status.wait(timeout=self.timeout_for_pvwait)
else:
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
self.cancel_on_stop(status)
self.scan_control.scan_val_reset.put(1)
status.wait(timeout=self.timeout_for_pvwait)
return None
def on_pre_scan(self) -> DeviceStatus | StatusBase | None:
"""Called right before the scan starts on all devices automatically."""
def on_trigger(self) -> DeviceStatus | StatusBase | None:
"""Called when the device is triggered."""
def on_complete(self) -> DeviceStatus | StatusBase | None:
"""Called to inquire if a device has completed a scans."""
time_started = time.time()
logger.info(f"Calling complete on {self.name}.")
status = CompareStatus(self.scan_control.scan_done, 1)
status.add_callback(self._status_callback)
self.cancel_on_stop(status)
logger.info(f"Finished calling complete on {self.name} within {time.time()-time_started}s.")
return status
def _status_callback(self, status, **kwargs):
logger.info(f"Complete finished on mo1bragg with {status.done} and {status.success}")
def on_kickoff(self) -> DeviceStatus | StatusBase | None:
"""Called to kickoff a device for a fly scan. Has to be called explicitly."""
scan_duration = self.scan_control.scan_duration.get()
# TODO implement better logic for infinite scans, at least bring it up with Debye
start_func = (
self.scan_control.scan_start_infinite.put
if scan_duration < 0.1
else self.scan_control.scan_start_timer.put
)
status = TransitionStatus(
self.scan_control.scan_status,
transitions=[ScanControlScanStatus.READY, ScanControlScanStatus.RUNNING],
strict=True,
failure_states=[ScanControlScanStatus.PARAMETER_WRONG],
)
self.cancel_on_stop(status)
start_func(1)
return status
def on_stop(self) -> None:
"""Called when the device is stopped."""
self.stopped = True # Needs to be set to stop motion
######### Utility Methods #########
def _check_if_scan_name_is_valid(self, scan_parameters: ScanServerScanInfo) -> bool:
"""Check if the scan is within the list of scans for which the backend is working"""
if scan_parameters.scan_name in self.valid_scan_names:
return True
return False
def _progress_update(self, value, old_value, **kwargs) -> None:
"""Callback method to update the scan progress, runs a callback
to SUB_PROGRESS subscribers, i.e. BEC.
Args:
value (int) : current progress value
"""
if self.scan_info.msg is None:
return
if self.scan_info.msg.status != "open":
if old_value == value:
return
# TODO check if logic is true
self.progress_signal.put(value=100, max_value=100, done=True)
max_value = 100
self.progress_signal.put(value=value, max_value=max_value, done=bool(max_value == value))
def set_xas_settings(self, low: float, high: float, scan_time: float) -> None:
"""Set XAS parameters for upcoming scan.
Args:
low (float): Low energy/angle value of the scan
high (float): High energy/angle value of the scan
scan_time (float): Time for a half oscillation
"""
status_list = []
status_list.append(self.scan_settings.s_scan_energy_lo.set(low))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.s_scan_energy_hi.set(high))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.s_scan_scantime.set(scan_time))
self.cancel_on_stop(status_list[-1])
for s in status_list:
s.wait(timeout=self.timeout_for_pvwait)
@typechecked
def convert_angle_energy(
self, mode: Literal["AngleToEnergy", "EnergyToAngle"], inp: float
) -> float:
"""Calculate energy to angle or vice versa
Args:
mode (Literal["AngleToEnergy", "EnergyToAngle"]): Mode of calculation
input (float): Either angle or energy
Returns:
output (float): Converted angle or energy
"""
self.calculator.calc_reset.put(0)
self.calculator.calc_reset.put(1)
status = CompareStatus(self.calculator.calc_done, 0)
self.cancel_on_stop(status)
status.wait(self.timeout_for_pvwait)
self.calculator.calc_reset.put(0)
if mode == "AngleToEnergy":
in_signal = self.calculator.calc_angle
out_signal = self.calculator.calc_energy
elif mode == "EnergyToAngle":
in_signal = self.calculator.calc_energy
out_signal = self.calculator.calc_angle
else:
raise Mo1BraggError(f"Unknown mode {mode}")
in_signal.put(inp)
status = CompareStatus(self.calculator.calc_done, 1)
self.cancel_on_stop(status)
status.wait(self.timeout_for_pvwait)
status = CompareStatus(out_signal, 0, operation_success=">")
self.cancel_on_stop(status)
status.wait(self.timeout_for_pvwait)
return out_signal.get()
def set_advanced_xas_settings(
self, low: float, high: float, scan_time: float, p_kink: float, e_kink: float
) -> None:
"""Set Advanced XAS parameters for upcoming scan.
Args:
low (float): Low angle value of the scan in eV
high (float): High angle value of the scan in eV
scan_time (float): Time for a half oscillation in s
p_kink (float): Position of kink in %
e_kink (float): Energy of kink in eV
"""
e_kink_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=e_kink)
# Angle and Energy are inverse proportional!
high_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=low)
low_deg = self.convert_angle_energy(mode="EnergyToAngle", inp=high)
pos, vel, dt = compute_spline(
low_deg=low_deg,
high_deg=high_deg,
p_kink=p_kink,
e_kink_deg=e_kink_deg,
scan_time=scan_time,
)
status_list = []
status_list.append(self.scan_settings.a_scan_pos.set(pos))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.a_scan_vel.set(vel))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.a_scan_time.set(dt))
self.cancel_on_stop(status_list[-1])
for s in status_list:
s.wait(timeout=self.timeout_for_pvwait)
def set_trig_settings(
self,
enable_low: bool,
enable_high: bool,
break_time_low: float,
break_time_high: float,
cycle_low: int,
cycle_high: int,
exp_time: float,
n_of_trigger: int,
) -> None:
"""Set TRIG settings for the upcoming scan.
Args:
enable_low (bool): Enable TRIG for low energy/angle
enable_high (bool): Enable TRIG for high energy/angle
break_time_low (float): Exposure time for low energy/angle
break_time_high (float): Exposure time for high energy/angle
cycle_low (int): Cycle for low energy/angle
cycle_high (int): Cycle for high energy/angle
exp_time (float): Length of 1 trigger period in seconds
n_of_trigger (int): Amount of triggers to be fired during brake
"""
status_list = []
status_list.append(self.scan_settings.trig_ena_hi_enum.set(int(enable_high)))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_ena_lo_enum.set(int(enable_low)))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_time_hi.set(break_time_high))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_time_lo.set(break_time_low))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_every_n_hi.set(cycle_high))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_every_n_lo.set(cycle_low))
self.cancel_on_stop(status_list[-1])
status_list.append(self.trigger_settings.xrd_trig_period.set(exp_time))
self.cancel_on_stop(status_list[-1])
status_list.append(self.trigger_settings.xrd_n_of_trig.set(n_of_trigger))
self.cancel_on_stop(status_list[-1])
for s in status_list:
s.wait(timeout=self.timeout_for_pvwait)
def set_scan_control_settings(self, mode: ScanControlMode, scan_duration: float) -> None:
"""Set the scan control settings for the upcoming scan.
Args:
mode (ScanControlMode): Mode for the scan, either simple or advanced
scan_duration (float): Duration of the scan
"""
val = ScanControlMode(mode).value
status_list = []
status_list.append(self.scan_control.scan_mode_enum.set(val))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_control.scan_duration.set(scan_duration))
self.cancel_on_stop(status_list[-1])
for s in status_list:
s.wait(timeout=self.timeout_for_pvwait)
Reference in New Issue View Git Blame Copy Permalink