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

reworked function to get rid of (potentionally infinite) loop.

Browse Source
This commit is contained in:
x01da
2025-10-08 15:44:15 +02:00
committed by appel_c
parent 1c0c9ad53e
commit e9e7d84e60
1 changed files with 37 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
debye_bec/devices/pilatus/pilatus.py
View File

@@ -371,63 +371,46 @@ class Pilatus(PSIDeviceBase, ADBase):
def _calculate_trigger(self, scan_msg: ScanStatusMessage) -> Tuple[float, float]:
self._update_scan_parameter()
total_osc = 0
calc_duration = 0
total_trig_lo = 0
total_trig_hi = 0
calc_duration = 0
n_trig_lo = 1
n_trig_hi = 1
init_lo = 1
init_hi = 1
lo_done = 0
hi_done = 0
if not self.scan_parameter.break_enable_low:
lo_done = 1
if not self.scan_parameter.break_enable_high:
hi_done = 1
start_time = time.time()
while True:
# TODO, we should not use infinite loops, for now let's add the escape Timeout of 20s, but should eventually be reviewed.
if time.time() - start_time > 20:
raise RuntimeError(
f"Calculating the number of triggers for scan {scan_msg.scan_name} took more than 20 seconds, aborting."
)
# Switching high/low is intended as angle is inverse to energy and settings in BEC are always in energy
loc_break_enable_low = self.scan_parameter.break_enable_high
loc_break_time_low = self.scan_parameter.break_time_high
loc_cycle_low = self.scan_parameter.cycle_high
loc_break_enable_high = self.scan_parameter.break_enable_low
loc_break_time_high = self.scan_parameter.break_time_low
loc_cycle_high = self.scan_parameter.cycle_low
if not loc_break_enable_low:
loc_break_time_low = 0
loc_cycle_low = 1
if not loc_break_enable_high:
loc_break_time_high = 0
loc_cycle_high = 1
total_osc = self.scan_parameter.scan_duration / (
self.scan_parameter.scan_time +
loc_break_time_low / (2 * loc_cycle_low) +
loc_break_time_high / (2 * loc_cycle_high)
)
total_osc = np.ceil(total_osc)
total_osc = total_osc + total_osc % 2 # round up to the next even number
if loc_break_enable_low:
total_trig_lo = np.floor(total_osc / (2 * loc_cycle_low))
if loc_break_enable_high:
total_trig_hi = np.floor(total_osc / (2 * loc_cycle_high))
calc_duration = total_osc * self.scan_parameter.scan_time + total_trig_lo * loc_break_time_low + total_trig_hi * loc_break_time_high
if calc_duration < self.scan_parameter.scan_duration:
# Due to inaccuracy in formula, this can happen, we then need to manually add two oscillations and recalculate the triggers
total_osc = total_osc + 2
calc_duration = calc_duration + 2 * self.scan_parameter.scan_time
if self.scan_parameter.break_enable_low and n_trig_lo >= self.scan_parameter.cycle_low:
n_trig_lo = 1
calc_duration = calc_duration + self.scan_parameter.break_time_low
if init_lo:
lo_done = 1
init_lo = 0
else:
n_trig_lo += 1
if (
self.scan_parameter.break_enable_high
and n_trig_hi >= self.scan_parameter.cycle_high
):
n_trig_hi = 1
calc_duration = calc_duration + self.scan_parameter.break_time_high
if init_hi:
hi_done = 1
init_hi = 0
else:
n_trig_hi += 1
if lo_done and hi_done:
n = np.floor(self.scan_parameter.scan_duration / calc_duration)
total_osc = total_osc * n
if self.scan_parameter.break_enable_low:
total_trig_lo = n + 1
if self.scan_parameter.break_enable_high:
total_trig_hi = n + 1
calc_duration = calc_duration * n
lo_done = 0
hi_done = 0
if calc_duration >= self.scan_parameter.scan_duration:
break
if loc_break_enable_low:
total_trig_lo = np.floor(total_osc / (2 * loc_cycle_low))
if loc_break_enable_high:
total_trig_hi = np.floor(total_osc / (2 * loc_cycle_high))
calc_duration = total_osc * self.scan_parameter.scan_time + total_trig_lo * loc_break_time_low + total_trig_hi * loc_break_time_high
return total_trig_lo, total_trig_hi