bec/debye_bec
0
0
Fork 0
Code Issues 9 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor: refactored motor and scan at Debye

Browse Source
This commit is contained in:
gac-x01da (Resp. Clark Adam Hugh)
2024-07-18 16:44:43 +02:00
committed by appel_c
parent b441f5db65
commit 8eab8e0c6a
2 changed files with 65 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
debye_bec/devices/mo1_bragg.py
View File

@@ -26,14 +26,21 @@ from ophyd_devices.utils import bec_scaninfo_mixin, bec_utils
logger = bec_logger.logger
# TODO check this with the PVs
class ScanControlScanStatus(int, enum.Enum):
"""Enum class for the scan status of the Bragg positioner"""
WAITING_FOR_PARAMETER = 0
PARAMETER_VALIDATED = 1
RUNNING = 2
FINISHED = 3
PARAMETER_WRONG = 0
VALIDATION_PENDING = 1
READY = 2
RUNNING = 3
class ScanControlLoadMessage(int, enum.Enum):
""" Validation of loading the scan parameters """
PENDING = 0
STARTED = 1
SUCCESS = 2
#TODO add here specific errors above 3-14
INVALID_SCAN_MODE = 15
class Mo1BraggError(Exception):
@@ -173,7 +180,7 @@ class Mo1BraggScanControl(Device):
scan_duration = Cpt(
EpicsSignalWithRBV, suffix="scan_duration", kind="normal", auto_monitor=True
)
scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="normal")
scan_load = Cpt(EpicsSignal, suffix="scan_load.PROC", kind="normal", put_complete=True)
scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="normal", auto_monitor=True)
scan_start_infinite = Cpt(EpicsSignal, suffix="scan_start_infinite", kind="normal")
scan_start_timer = Cpt(EpicsSignal, suffix="scan_start_timer", kind="normal")
@@ -185,9 +192,12 @@ class Mo1BraggScanControl(Device):
EpicsSignalRO, suffix="scan_time_left_RBV", kind="normal", auto_monitor=True
)
scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="normal", auto_monitor=True)
scan_val_reset = Cpt(EpicsSignal, suffix="scan_val_reset", kind="config")
scan_val_reset = Cpt(EpicsSignal, suffix="scan_val_reset.PROC", kind="config", put_complete=True)
scan_progress = Cpt(EpicsSignalRO, suffix="scan_progress_RBV", kind="normal", auto_monitor=True)
scan_spectra_done = Cpt(EpicsSignalRO, suffix="scan_n_osc_RBV", kind="normal", auto_monitor=True)
scan_spectra_left = Cpt(EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="normal", auto_monitor=True)
#TODO Review kind for all PVs
class Mo1Bragg(Device, PositionerBase):
"""Class for the Mo1 Bragg positioner of the Debye beamline"""
@@ -229,7 +239,7 @@ class Mo1Bragg(Device, PositionerBase):
)
# Execute motion
move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config")
move_abs = Cpt(EpicsSignal, suffix="move_abs.PROC", kind="config", put_complete=True)
move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config")
motor_is_moving = Cpt(
EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True
@@ -237,6 +247,7 @@ class Mo1Bragg(Device, PositionerBase):
SUB_READBACK = "readback"
_default_sub = SUB_READBACK
SUB_PROGRESS = 'progress'
def __init__(
self, prefix="", *, name: str, kind: Kind = None, device_manager=None, parent=None, **kwargs
@@ -254,6 +265,7 @@ class Mo1Bragg(Device, PositionerBase):
self.scan_duration = None
self.start = None
self.end = None
self.timeout_for_pvwait = 2.5
if device_manager:
self.device_manager = device_manager
@@ -262,6 +274,15 @@ class Mo1Bragg(Device, PositionerBase):
self.connector = self.device_manager.connector
self._update_scaninfo()
self._init()
def _init(self):
self.scan_control.scan_progress.subscribe(self._progress_update, run=False)
def _progress_update(self, value, **kwargs):
max_value=100
logger.info(f"Progress at {value}")
self._run_subs(sub_type=self.SUB_PROGRESS, value=value, max_value = max_value, done=bool(max_value == value))
def _update_scaninfo(self) -> None:
"""Update scaninfo from BecScaninfoMixing
@@ -316,6 +337,8 @@ class Mo1Bragg(Device, PositionerBase):
def stop(self, *, success=False) -> None:
"""Stop any motion on the positioner"""
# Stop the motion
#TODO decide appropriate stop procedure
self.stop_scan()
self.move_stop.put(1)
# Move thread is stopped too
self._stopped = True
@@ -338,6 +361,7 @@ class Mo1Bragg(Device, PositionerBase):
target_pos (float) : target position for the motion
status (DeviceStatus) : status object to set the status of the motion
"""
#TODO if called in sequence, the move resolves directly.
success = True
exception = None
try:
@@ -346,7 +370,7 @@ class Mo1Bragg(Device, PositionerBase):
# Start motion
self.move_abs.put(1)
# Sleep needed due to updated frequency to soft IOC
time.sleep(0.3)
time.sleep(0.5)
while self.motor_is_moving.get() == 0:
val = read_cpt.get()
self._run_subs(sub_type=self.SUB_READBACK, value=val)
@@ -355,12 +379,14 @@ class Mo1Bragg(Device, PositionerBase):
success = False
break
time.sleep(update_frequency)
# pylint: disable=broad-except
except Exception as exc:
content = traceback.format_exc()
logger.error(f"Error in move thread of device {self.name}: {content}")
# status._finished(success=success)
exception = exc
status.set_exception(exc=exception)
#status.set_exception(exc=exc)
finally:
if exception:
status.set_exception(exc=exception)
@@ -478,24 +504,12 @@ class Mo1Bragg(Device, PositionerBase):
mode (Literal["simple", "advanced"]): Pick a mode for the oscillation motion, either simple or advanced (needs to be configured)
scan_duration (float): Duration of the scan
"""
timeout = kwargs.get("timeout", 1.5)
# TODO check if maybe we want an extra argument for infinite or finite motion
self.set_xas_settings(low=low, high=high, scan_time=scan_time)
self.set_xrd_settings(False)
self.set_scan_control_settings(mode=mode, scan_duration=scan_duration)
self.scan_control.scan_load.put(1)
# TODO The wait/sleep here could be move to pre_scan actions to avoid potentially slow signals
time.sleep(0.2)
if not self.wait_for_signals(
signal_conditions=[
(self.scan_control.scan_status.get, ScanControlScanStatus.PARAMETER_VALIDATED)
],
timeout=timeout,
check_stopped=True,
):
raise TimeoutError(
f"Scan parameter validation run into timeout after {timeout} with {self.scan_control.scan_status.get()}"
)
def kickoff(self):
"""Kickoff the device, called from BEC."""
@@ -504,7 +518,6 @@ class Mo1Bragg(Device, PositionerBase):
# If the scan_duration is less than 0.1s, the scan will be started infinite
# This logic should move in future to the controller, but for now it is implemented here
# It is necessary since the current NIDAQ implementation relies on it
timeout = 3
scan_duration = self.scan_control.scan_duration.get()
start_func = (
self.scan_control.scan_start_infinite.put
@@ -512,11 +525,9 @@ class Mo1Bragg(Device, PositionerBase):
else self.scan_control.scan_start_timer.put
)
start_func(1)
# TODO check if sleep is needed
time.sleep(0.2)
status = self.wait_with_status(
signal_conditions=[(self.scan_control.scan_done.get, ScanControlScanStatus.RUNNING)],
timeout=timeout,
signal_conditions=[(self.scan_control.scan_status.get, ScanControlScanStatus.RUNNING)],
timeout=self.timeout_for_pvwait,
check_stopped=True,
)
return status
@@ -551,16 +562,17 @@ class Mo1Bragg(Device, PositionerBase):
"""Actions to be executed when the device is staged."""
if not self.scaninfo.scan_type == "xas":
return
# Could become redudant if we use this in unstage
self.scan_val_reset.put(1)
# Assume that scan_val_reset was usee in unstage, if required, do it again but then we have to introduce a sleep ~1
if self.scan_control.scan_msg.get() != ScanControlLoadMessage.PENDING:
self.scan_control.scan_val_reset.put(1)#.set()
time.sleep(1)
self._get_scaninfo_parameters()
# Brief sleep to allow for PV to update (FYI, reliable set makes this redundant)
time.sleep(0.2)
# Wait for check to go to Pending
if not self.wait_for_signals(
signal_conditions=[
(self.scan_control.scan_status.get, ScanControlScanStatus.WAITING_FOR_PARAMETER)
(self.scan_control.scan_msg.get, ScanControlLoadMessage.PENDING)
],
timeout=1.5,
timeout=self.timeout_for_pvwait,
check_stopped=True,
):
raise TimeoutError(
@@ -574,6 +586,17 @@ class Mo1Bragg(Device, PositionerBase):
mode=ScanControlMode.SIMPLE,
scan_duration=self.scan_duration,
)
#Wait for params to be checked from controller
if not self.wait_for_signals(
signal_conditions=[
(self.scan_control.scan_msg.get, ScanControlLoadMessage.SUCCESS)
],
timeout=self.timeout_for_pvwait,
check_stopped=True,
):
raise TimeoutError(
f"Scan parameter validation run into timeout after {self.timeout_for_pvwait} with {self.scan_control.scan_status.get()}"
)
def complete(self) -> None:
"""Complete the acquisition, called from BEC.
@@ -597,7 +620,7 @@ class Mo1Bragg(Device, PositionerBase):
"""
# Create DeviceStatus object with hook to finishing the scan.
status = self.wait_with_status(
signal_conditions=[(self.scan_control.scan_done.get, ScanControlScanStatus.FINISHED)],
signal_conditions=[(self.scan_control.scan_done.get, 1)],
timeout=None,
check_stopped=True,
)
@@ -605,8 +628,9 @@ class Mo1Bragg(Device, PositionerBase):
def stop_scan(self) -> None:
"""Stop the scan manually."""
# TODO Check this signal
self.scan_control.scan_stop.put(1)
#TODO Check if a single variable for stop scan and stop motion is sufficient.
self._stopped = True
def unstage(self) -> list[object]:
"""
@@ -625,7 +649,7 @@ class Mo1Bragg(Device, PositionerBase):
def on_unstage(self) -> None:
"""Actions to be executed when the device is unstaged."""
# Reset setting of paramters after scan to be ready and clean for the next scan
self.scan_val_reset.put(1)
self.scan_control.scan_val_reset.put(1)
def check_scan_id(self) -> None:
"""Checks if scan_id has changed and set stopped flagged to True if it has."""

4
debye_bec/scans/mono_bragg_scans.py
View File

@@ -18,6 +18,7 @@ class MonoBraggOscillationScan(AsyncFlyScanBase):
scan_name = "energy_oscillation_scan"
scan_type = "xas"
scan_report_hint = "device_progress"
required_kwargs = []
use_scan_progress_report = False
pre_move = False
@@ -66,8 +67,9 @@ class MonoBraggOscillationScan(AsyncFlyScanBase):
"""
Return the instructions for the scan report.
"""
yield from self.stubs.scan_report_instruction({"device_progress" : [self.motor]})
# TODO implement which instructions the scan should listen to for the table printout if wanted.
yield None
# yield None
def scan_core(self):
"""Run the scan core.