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

Make sure that parameters are set before validating

Browse Source
This commit is contained in:
gac-x01da
2025-06-25 15:55:54 +02:00
parent ed6d64c7f9
commit e6586ceab2
1 changed files with 85 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
debye_bec/devices/mo1_bragg/mo1_bragg.py
View File

@@ -123,10 +123,11 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object.
"""
# self._check_scan_msg(ScanControlLoadMessage.PENDING)
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING)
self.cancel_on_stop(status)
status.wait(timeout=self.timeout_for_pvwait)
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_info.msg.scan_name
self._update_scan_parameter()
@@ -204,11 +205,15 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
)
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
self.scan_control.scan_load.put(1)
# Wait for params to be checked from controller
status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.SUCCESS)
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
@@ -230,6 +235,10 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
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)
@@ -251,7 +260,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
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()
scan_duration = self.scan_parameter.scan_duration
# TODO implement better logic for infinite scans, at least bring it up with Debye
start_func = (
self.scan_control.scan_start_infinite.put
@@ -291,25 +300,20 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
high (float): High energy/angle value of the scan
scan_time (float): Time for a half oscillation
"""
self.scan_settings.s_scan_energy_lo.put(low)
self.scan_settings.s_scan_energy_hi.put(high)
self.scan_settings.s_scan_scantime.put(scan_time)
def wait_for_signal(self, signal: Cpt, value: Any, timeout: float | None = None) -> None:
"""Wait for a signal to reach a certain value."""
if timeout is None:
timeout = self.timeout_for_pvwait
start_time = time.time()
while time.time() - start_time < timeout:
if signal.get() == value:
return None
if self.stopped is True: # Should this check be optional or configurable?!
raise DeviceStopError(f"Device {self.name} was stopped while waiting for signal")
time.sleep(0.1)
# If we end up here, the status did not resolve
raise TimeoutError(
f"Device {self.name} run into timeout after {timeout}s for signal {signal.name} with value {signal.get()}, expected {value}"
)
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(
@@ -326,15 +330,19 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
"""
self.calculator.calc_reset.put(0)
self.calculator.calc_reset.put(1)
self.wait_for_signal(self.calculator.calc_done, 0)
status = CompareStatus(self.calculator.calc_done, 0)
self.cancel_on_stop(status)
status.wait(self.timeout_for_pvwait)
if mode == "AngleToEnergy":
self.calculator.calc_angle.put(inp)
elif mode == "EnergyToAngle":
self.calculator.calc_energy.put(inp)
self.wait_for_signal(self.calculator.calc_done, 1)
time.sleep(0.25) # Needed due to update frequency of softIOC
status = CompareStatus(self.calculator.calc_done, 1)
self.cancel_on_stop(status)
status.wait(self.timeout_for_pvwait)
time.sleep(0.25) #TODO needed still? Needed due to update frequency of softIOC
if mode == "AngleToEnergy":
return self.calculator.calc_energy.get()
elif mode == "EnergyToAngle":
@@ -365,9 +373,19 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
scan_time=scan_time,
)
self.scan_settings.a_scan_pos.set(pos)
self.scan_settings.a_scan_vel.set(vel)
self.scan_settings.a_scan_time.set(dt)
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,
@@ -390,12 +408,30 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
cycle_low (int): Cycle for low energy/angle
cycle_high (int): Cycle for high energy/angle
"""
self.scan_settings.trig_ena_hi_enum.put(int(enable_high))
self.scan_settings.trig_ena_lo_enum.put(int(enable_low))
self.scan_settings.trig_time_hi.put(exp_time_high)
self.scan_settings.trig_time_lo.put(exp_time_low)
self.scan_settings.trig_every_n_hi.put(cycle_high)
self.scan_settings.trig_every_n_lo.put(cycle_low)
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(exp_time_high))
self.cancel_on_stop(status_list[-1])
status_list.append(self.scan_settings.trig_time_lo.set(exp_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])
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.
@@ -405,8 +441,18 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
scan_duration (float): Duration of the scan
"""
val = ScanControlMode(mode).value
self.scan_control.scan_mode_enum.put(val)
self.scan_control.scan_duration.put(scan_duration)
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)
def _update_scan_parameter(self):
"""Get the scan_info parameters for the scan."""
@@ -416,39 +462,3 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner):
for key, value in self.scan_info.msg.request_inputs["kwargs"].items():
if hasattr(self.scan_parameter, key):
setattr(self.scan_parameter, key, value)
def _check_scan_msg(self, target_state: ScanControlLoadMessage) -> None:
"""Check if the scan message is gettting available
Args:
target_state (ScanControlLoadMessage): Target state to check for
Raises:
TimeoutError: If the scan message is not available after the timeout
"""
try:
self.wait_for_signal(self.scan_control.scan_msg, target_state, timeout=1)
except TimeoutError as exc:
logger.warning(
f"Resetting scan validation in stage for state: {ScanControlLoadMessage(self.scan_control.scan_msg.get())}, "
f"retry .get() on scan_control: {ScanControlLoadMessage(self.scan_control.scan_msg.get())} and sleeping 1s"
)
current_scan_msg = self.scan_control.scan_msg.get()
def callback(*, old_value, value, **kwargs):
if old_value == current_scan_msg and value == target_state:
return True
return False
status = SubscriptionStatus(self.scan_control.scan_msg, callback=callback)
self.scan_control.scan_val_reset.put(1)
status.wait(timeout=self.timeout_for_pvwait)
# try:
# self.wait_for_signal(self.scan_control.scan_msg, target_state, timeout=4)
# except TimeoutError as exc:
# raise TimeoutError(
# f"Timeout after {self.timeout_for_pvwait} while waiting for scan status,"
# f" current state: {ScanControlScanStatus(self.scan_control.scan_msg.get())}"
# ) from exc