John Henry BealeandClaude Opus 5.5
e05ebeb76b
Detector, collimator, post-tube in/out, move home/yag, the backlight button and the sample exchange/alignment/data collection transitions blocked the GUI thread for up to ~20s per move. - MoveWorker (same thread as the collection worker, so sequences and collections can not overlap) runs (label, callable) steps with a progress dialog; all controls are locked meanwhile (no nudges, zoom stays usable) - move functions raise MoveError instead of showing dialogs; the GUI shows the message when the sequence ends - abort stops all motors (motor record STOP, backlight) and ends the sequence; position checks (assert_tweaker_positions abort=) stop waiting - safety: backlight moves wait and check the position before the next step (post-tube in used to start while the backlight was still moving out) - tandem nudges stay non blocking in the GUI thread; collimator still moves X then Y - sample alignment: backlight in now also checks the detector is clear of the post-tube path (as the backlight button does) Tested in simulation and at the beamline. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
import logging
|
|
_log=logging.getLogger(__name__)
|
|
|
|
import time
|
|
from epics.ca import pend_event
|
|
|
|
class PositionsNotReached(Exception):
|
|
pass
|
|
|
|
|
|
def assert_tweaker_positions(targets, timeout=60.0, abort=None):
|
|
"""
|
|
wait for each of the given motors to have specified positions within tolerance
|
|
:param targets: [
|
|
(motor, target_position, tolerance),
|
|
...
|
|
]
|
|
:param abort: optional threading.Event: stop waiting as soon as it is set
|
|
:return: None
|
|
:raises: PositionsNotReached
|
|
"""
|
|
num_motors=len(targets)
|
|
|
|
timeisup=timeout+time.time()
|
|
|
|
while time.time()<timeisup:
|
|
if abort is not None and abort.is_set():
|
|
raise PositionsNotReached('aborted')
|
|
count=0
|
|
summary=[]
|
|
for i, m in enumerate(targets):
|
|
mot_tw, target, tolerance=m
|
|
label=mot_tw._label
|
|
notSim=not type(mot_tw).__name__.startswith('Sim')
|
|
|
|
if notSim: pend_event()
|
|
cur=mot_tw.get_rbv()
|
|
done=mot_tw.is_done()
|
|
s=f"check {label} {cur:.5g} == {target:.5g} [done={done}]"
|
|
_log.debug(s)
|
|
summary.append(s)
|
|
if done and tolerance>=abs(cur-target):
|
|
count+=1
|
|
if notSim: pend_event(0.1)
|
|
if count==num_motors:
|
|
break
|
|
if notSim: pend_event(0.1)
|
|
|
|
if count!=num_motors:
|
|
raise PositionsNotReached("failed to reach target positions: {}".format("#".join(summary)))
|
|
|
|
|
|
def assert_motor_positions(targets, timeout=60.0):
|
|
"""
|
|
wait for each of the given motors to have specified positions within tolerance
|
|
:param targets: [
|
|
(motor, target_position, tolerance),
|
|
...
|
|
]
|
|
:return: None
|
|
:raises: PositionsNotReached
|
|
"""
|
|
num_motors=len(targets)
|
|
|
|
timeisup=timeout+time.time()
|
|
|
|
while time.time()<timeisup:
|
|
count=0
|
|
summary=[]
|
|
for i, m in enumerate(targets):
|
|
motor, target, tolerance=m
|
|
name=motor._prefix
|
|
pend_event()
|
|
cur=motor.get_rbv(readback=True)
|
|
done=motor.done_moving
|
|
_log.debug("check {}[done={}]: {} == {}".format(name, done, cur, target))
|
|
summary.append("{}[done={}]: {} == {}".format(name, done, cur, target))
|
|
if done and tolerance>=abs(cur-target):
|
|
count+=1
|
|
pend_event(0.1)
|
|
|
|
if count==num_motors:
|
|
break
|
|
pend_event()
|
|
|
|
if count!=num_motors:
|
|
raise PositionsNotReached("failed to reach target positions: {}".format("#".join(summary)))
|