John Henry BealeandClaude Opus 5.5
efe18d4204
- move_together(): post-tube in/out and the tandem arrows check all targets first (nudge cap, soft limits), then send the moves back to back to the VAL fields (us X, ds X, us Y, ds Y) without reads in between; either all axes move or none (a refused tandem axis could tilt the tube) - set_collecting(False) re-enables all groups: after a motor sequence the collimator and post-tube groups stayed greyed out - log the end of a motor sequence and the GUI unlock Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
491 lines
16 KiB
Python
491 lines
16 KiB
Python
#!/usr/bin/env python
|
|
# *-----------------------------------------------------------------------*
|
|
# | |
|
|
# | Copyright (c) 2026 by Paul Scherrer Institute (http://www.psi.ch) |
|
|
# | |
|
|
# | Author John Beale (john.beale@psi.ch) |
|
|
# *-----------------------------------------------------------------------*
|
|
"""
|
|
data collection worker running in its own QThread.
|
|
|
|
The worker owns the Deltatau (PPComm is not thread safe) and the Jungfrau/slic objects.
|
|
It never touches widgets: progress, run info, plots and the final result are sent to
|
|
the GUI thread with signals.
|
|
|
|
Abort:
|
|
request_abort() is called directly from the GUI thread. It closes the shutter immediately
|
|
and sets an event that the worker checks between steps and while polling the motion.
|
|
Steps that block inside PBSwissMX (homing, wait_armed, gather upload) can not be interrupted.
|
|
"""
|
|
import logging,threading,time
|
|
import numpy as np
|
|
|
|
from PyQt5.QtCore import Qt, QObject, pyqtSignal, pyqtSlot
|
|
from PyQt5.QtWidgets import QApplication, QDialog, QLabel, QProgressBar, QPushButton, QVBoxLayout
|
|
|
|
import epics
|
|
from app_config import AppCfg
|
|
import psi_device
|
|
|
|
_log=logging.getLogger(__name__)
|
|
|
|
class CollectionAborted(Exception):
|
|
pass
|
|
|
|
class MoveError(Exception):
|
|
'''a motor sequence can not be done (configuration missing, safety condition, position not reached)'''
|
|
pass
|
|
|
|
class MoveAborted(Exception):
|
|
pass
|
|
|
|
class CollectionWorker(QObject):
|
|
sigProgress=pyqtSignal(int, int, str) # value, maximum, label (empty label -> keep current)
|
|
sigRunInfo=pyqtSignal() # run number and data path changed
|
|
sigPlotGather=pyqtSignal(object) # shapepath object to plot in the GUI thread
|
|
sigFinished=pyqtSignal(str) # empty string on success, else error message
|
|
_sigRun=pyqtSignal(object) # internal: list of scan parameter dicts
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self._abort=threading.Event()
|
|
self._busy=False
|
|
self._val=0
|
|
self._max=100
|
|
self._sigRun.connect(self._run, Qt.QueuedConnection)
|
|
|
|
@pyqtSlot()
|
|
def init_thread(self):
|
|
# connected to QThread.started: worker threads must attach to the pyepics CA context
|
|
epics.ca.use_initial_context()
|
|
_log.info('collection worker thread started')
|
|
|
|
@property
|
|
def busy(self):
|
|
return self._busy
|
|
|
|
def submit(self, jobs):
|
|
'''called from the GUI thread: queue a list of scan parameter dicts'''
|
|
if self._busy:
|
|
_log.warning('data collection already running -> request ignored')
|
|
return False
|
|
self._busy=True
|
|
self._abort.clear()
|
|
self._sigRun.emit(jobs)
|
|
return True
|
|
|
|
def request_abort(self):
|
|
'''called from the GUI thread: close the shutter now, the worker stops at the next check'''
|
|
if not self._busy:
|
|
return
|
|
_log.warning('data collection abort requested')
|
|
self._abort.set()
|
|
app=QApplication.instance()
|
|
try:
|
|
app._shutter.close() # shutters are closed by the caput sequence in Shutter.close(), not by the CTA stop
|
|
except Exception as e:
|
|
_log.error(f'failed to close shutter on abort: {e}')
|
|
try:
|
|
jf=app._jungfrau
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
try:
|
|
jf.stop_triggers()
|
|
except Exception as e:
|
|
_log.error(f'failed to stop CTA on abort: {e}')
|
|
|
|
def _check_abort(self):
|
|
if self._abort.is_set():
|
|
raise CollectionAborted()
|
|
|
|
def _progress(self, val=None, label='', inc=0, maximum=None):
|
|
if val is not None:
|
|
self._val=val
|
|
self._val+=inc
|
|
if maximum is not None:
|
|
self._max=maximum
|
|
self.sigProgress.emit(int(self._val), int(self._max), label)
|
|
|
|
@pyqtSlot(object)
|
|
def _run(self, jobs):
|
|
errLst=[]
|
|
try:
|
|
for kwargs in jobs:
|
|
self._check_abort()
|
|
err=self._collect(**kwargs)
|
|
if err:
|
|
errLst.append(err)
|
|
except CollectionAborted:
|
|
_log.warning('data collection aborted by user')
|
|
errLst.append('aborted')
|
|
except Exception as e:
|
|
_log.exception('data collection failed')
|
|
errLst.append(repr(e))
|
|
finally:
|
|
self._busy=False
|
|
self.sigFinished.emit('\n'.join(errLst))
|
|
|
|
def _collect(self, **kwargs):
|
|
'''
|
|
kwargs:
|
|
grid: grid dictionary with orig, pitch, count
|
|
trf: transformation matrix
|
|
points: list of points without transformation
|
|
pts_trf: list of points with transformation
|
|
returns an error message or None
|
|
'''
|
|
app=QApplication.instance()
|
|
cfg=app._cfg
|
|
dt_misc=cfg.value(AppCfg.DT_MISC)
|
|
|
|
geo=app._geometry
|
|
shutter=app._shutter
|
|
fn='/tmp/shapepath'
|
|
try:
|
|
dt=app._deltatau
|
|
except AttributeError:
|
|
app._deltatau=dt=psi_device.Deltatau(app._args.sim&0x100!=0)
|
|
try:
|
|
jf=app._jungfrau
|
|
except AttributeError:
|
|
app._jungfrau=jf=psi_device.Jungfrau(app._args.sim&0x80!=0)
|
|
self._progress(0, 'Configure detector', maximum=100)
|
|
try:
|
|
jf.config(**kwargs)
|
|
except psi_device.DataCollectionAbort as e:
|
|
msg=f"Data collection aborted: {e}"
|
|
_log.warning(msg)
|
|
return msg
|
|
self._check_abort()
|
|
sp=dt._shapepath
|
|
sp.verbose=dt_misc['verbose']
|
|
|
|
sp.meta['fel_per']=dt_misc['fel_per']
|
|
sp.meta['sync_mode']=dt_misc['sync_mode']
|
|
sp.meta['sync_flag']=dt_misc['sync_flag']
|
|
|
|
self._progress(label="Setup Gather/Sync", inc=5)
|
|
mode=kwargs.pop('mode',1)
|
|
sp.setup_sync(verbose=sp.verbose&0x40, timeOfs=dt_misc['time_ofs'], timeCor=dt_misc['time_cor'])
|
|
self._progress(label="Download motion program", inc=5)
|
|
|
|
use_trf=kwargs.get('use_trf', True) # do not use coordinate transformation
|
|
if not use_trf:
|
|
trf=kwargs.pop('trf')
|
|
try:
|
|
sp.setup_motion(fnPrg=fn+'.prg', mode=mode, **kwargs)
|
|
except BaseException as e:
|
|
_log.error(repr(e))
|
|
return repr(e)
|
|
sp.setup_gather()
|
|
if not use_trf:
|
|
kwargs['trf']=trf
|
|
try:
|
|
p=geo._fitPlane
|
|
#TODO: cleanup
|
|
if use_trf:
|
|
trf=kwargs['trf'] # grid-coord -> motor-um
|
|
#(0,0,1)*trf
|
|
# trf*'gridpos in um' -> motor pos in mm
|
|
trf2=np.asmatrix(np.identity(3))
|
|
trf2[:, :2]=trf
|
|
#p1=(0,0,1)*trf2 # =matrix([[-2376.8, 1376.8, 1. ]]) um
|
|
trf3=np.matrix( ((p[0],0,0),(p[1],0,0),(1000*p[2],0,1)) )
|
|
#(0, 0, 1)*trf2*trf3 -> z in um of gridpos(0,0)
|
|
t=(trf2*trf3)[:,0].A.ravel()
|
|
cz=f'{t[0]:+.18g}X{t[1]:+.18g}Y{t[2]:+.18g}'
|
|
else:
|
|
# X has inverted sign !
|
|
# Z is in um -> therefore the offset must be multiplied with 1000 ! Z motor has opposite sign !
|
|
#cz=f'{+p[0]:+.18g}X{-p[1]:+.18g}Y{-p[2]*1000:+.18g}'
|
|
t=p*np.array((1,1,1000))
|
|
cz=f'{t[0]:+.18g}X{t[1]:+.18g}Y{t[2]:+.18g}'
|
|
except AttributeError:
|
|
cz='0'
|
|
_log.warning('no plane fitting done. z does not move')
|
|
|
|
trf=kwargs['trf']
|
|
if use_trf:
|
|
fx=f'{trf[0, 0]:+.18g}X{trf[1, 0]:+.18g}Y{trf[2, 0]:+.18g}'
|
|
fy=f'{trf[0, 1]:+.18g}X{trf[1, 1]:+.18g}Y{trf[2, 1]:+.18g}'
|
|
else:
|
|
fx='X';fy='Y'
|
|
if logging.getLogger('swissmx').level==logging.DEBUG and not ('points' in kwargs and 'pts_trf' in kwargs):
|
|
_log.debug('coordinate log skipped: points/pts_trf not in scan parameters')
|
|
elif logging.getLogger('swissmx').level==logging.DEBUG:
|
|
try: t
|
|
except NameError: t=(np.nan,np.nan,np.nan)
|
|
fnCoord='/tmp/coord.log'
|
|
_log.debug(f'write all coordinates to {fnCoord}')
|
|
if use_trf:
|
|
xy1=np.hstack((kwargs["points"],np.ones((kwargs["points"].shape[0],1))))
|
|
else:
|
|
xy1=np.hstack((kwargs["pts_trf"],np.ones((kwargs["pts_trf"].shape[0],1))))
|
|
pcz=np.matrix(((t[0],t[1],t[2])))*xy1.T
|
|
x_y_fx_fy_cz=np.vstack((kwargs["points"].T, kwargs["pts_trf"].T, pcz)).A
|
|
with open(fnCoord,'w') as fh:
|
|
fh.write(' x y fx fy cz\n')
|
|
for i in range(x_y_fx_fy_cz.shape[1]):
|
|
fh.write('%9.7g %9.7g %9.7g %9.7g %9.7g\n'%tuple(x_y_fx_fy_cz[:,i]))
|
|
|
|
sp.setup_coord_trf(fx, fy, cz) # reset to shape path system
|
|
|
|
comm=sp.comm
|
|
if comm is None:
|
|
_log.info('simulated')
|
|
else:
|
|
comm.gpascii._cb_func=lambda i, sz:self._progress(int(i*90/sz))
|
|
self._check_abort()
|
|
|
|
motion_started=False
|
|
shutter_opened=False
|
|
daq_started=False
|
|
try:
|
|
self._progress(label="Homing and get ready", inc=5)
|
|
sp.homing() # homing if needed
|
|
self._check_abort()
|
|
motion_started=True # set before run(): run() also starts triggerSync
|
|
sp.run() # start motion program
|
|
sp.wait_armed() # wait until motors are at first position
|
|
self._check_abort()
|
|
shutter_opened=True
|
|
shutter.open()
|
|
if self._abort.wait(1.1):
|
|
raise CollectionAborted()
|
|
num_pts=kwargs['num_pts']
|
|
daq_started=True
|
|
jf.acquire(num_pts)
|
|
if self._abort.is_set(): # abort arrived while acquire() was starting the CTA: stop it again
|
|
raise CollectionAborted()
|
|
self.sigRunInfo.emit()
|
|
sp.trigger(1.0) # send a start trigger (if needed) after given time
|
|
_log.info('start trigger sent')
|
|
if dt._comm is None:
|
|
self._progress(0, "run motion/acquisition (simulated)", maximum=num_pts)
|
|
for p in range(0,num_pts,1+int(num_pts/100)):
|
|
self._progress(p)
|
|
if self._abort.wait(.1):
|
|
raise CollectionAborted()
|
|
daq_started=False
|
|
else:
|
|
self._progress(0, "run motion/acquisition", maximum=num_pts)
|
|
while True:
|
|
p=int(sp.progress())
|
|
_log.info(f'progress {p}/{num_pts}')
|
|
if p<0:
|
|
break
|
|
self._progress(min(p,num_pts))
|
|
if self._abort.wait(1.): #wait 1 sec instead of .1...hopefully less segmentation faults
|
|
raise CollectionAborted()
|
|
motion_started=False
|
|
daq_started=False
|
|
jf.gather_upload()
|
|
self._progress(label="upload gather data")
|
|
sp.gather_upload(fnRec=fn+'.npz')
|
|
if dt_misc['show_plots']:
|
|
self.sigPlotGather.emit(sp)
|
|
finally:
|
|
if shutter_opened:
|
|
_log.info('Shutter closed')
|
|
shutter.close()
|
|
if daq_started: # aborted or failed while the CTA/JFJ acquisition was active
|
|
try:
|
|
jf.stop_triggers()
|
|
except Exception as e:
|
|
_log.error(f'failed to stop CTA: {e}')
|
|
if motion_started and dt._comm is not None: # aborted or failed while the motion program was active
|
|
self._recover_motion(sp)
|
|
if daq_started: # cancel JFJ and wait for the retrieval last: this can take a while
|
|
self._progress(label='Aborting: wait for data retrieval and JFJ')
|
|
try:
|
|
jf.cancel()
|
|
except Exception as e:
|
|
_log.error(f'failed to cancel JFJ: {e}')
|
|
if comm is not None:
|
|
try:
|
|
del comm.gpascii._cb_func
|
|
except AttributeError:
|
|
pass
|
|
return None
|
|
|
|
# PowerBrick variables logged before and after the recovery, to diagnose a failed recovery
|
|
PB_STATE_VARS=('Coord[1].ProgRunning', 'Coord[1].DesTimeBase', 'Coord[1].TimeBaseSlew', 'Coord[1].Q[0]',
|
|
'Gather.Enable', 'Sys.ServoPeriod')
|
|
|
|
def _pb_state(self, gpascii):
|
|
state={}
|
|
for v in self.PB_STATE_VARS:
|
|
try:
|
|
state[v]=gpascii.get_variable(v)
|
|
except Exception as e:
|
|
state[v]=f'? ({e})'
|
|
return state
|
|
|
|
def _recover_motion(self, sp):
|
|
'''
|
|
bring the Deltatau back to a usable state after an aborted or failed collection:
|
|
- abort the motion program in coordinate system 1 and stop gathering
|
|
- stop triggerSync (normally closed by gather_upload, which is skipped on abort)
|
|
- restore the time base: the motion program sets Coord[1].DesTimeBase=0 while waiting for the
|
|
start trigger, and TimeBaseSlew=1 (default 1E-4). triggerSync adjusts DesTimeBase while running
|
|
- reset the sync state Coord[1].Q[0]
|
|
'''
|
|
comm=sp.comm
|
|
gpascii=comm.gpascii
|
|
_log.warning(f'recover motion: PowerBrick state before: {self._pb_state(gpascii)}')
|
|
try:
|
|
gpascii.send_block(['&1a', 'Gather.Enable=0'])
|
|
except Exception as e:
|
|
_log.error(f'recover motion: failed to stop motion program: {e}')
|
|
try:
|
|
sync_shell=sp.sync_shell
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
del sp.sync_shell
|
|
try:
|
|
sync_shell._channel.send('\x03') # Ctrl-C to the foreground triggerSync
|
|
time.sleep(.2)
|
|
_log.info(f'recover motion: triggerSync output: {sync_shell.sync()!r}')
|
|
sync_shell._channel.close()
|
|
except Exception as e:
|
|
_log.error(f'recover motion: failed to close triggerSync channel: {e}')
|
|
try:
|
|
comm.shell_command('killall triggerSync')
|
|
except Exception as e:
|
|
_log.error(f'recover motion: killall triggerSync failed: {e}')
|
|
# restore the time base only after triggerSync is stopped, else triggerSync overwrites DesTimeBase
|
|
try:
|
|
gpascii.send_block(['Coord[1].DesTimeBase=Sys.ServoPeriod', 'Coord[1].TimeBaseSlew=1E-4',
|
|
'Coord[1].Q[0]=0']) # Q[0]=-2 (armed) would let the next wait_armed() return early
|
|
except Exception as e:
|
|
_log.error(f'recover motion: failed to restore time base: {e}')
|
|
_log.warning(f'recover motion: PowerBrick state after: {self._pb_state(gpascii)}')
|
|
|
|
|
|
class MoveWorker(QObject):
|
|
'''
|
|
runs blocking motor sequences (post tube, detector, collimator, gonio, ESC transitions) in the worker thread.
|
|
steps: list of (label, callable). The callables run in the worker thread: they must not touch widgets and
|
|
report problems by raising MoveError.
|
|
request_abort() (GUI thread) stops all motors through stop_all and ends the sequence.
|
|
'''
|
|
sigProgress=pyqtSignal(int, int, str) # step, number of steps, label
|
|
sigFinished=pyqtSignal(str) # empty string on success, 'aborted' or error message
|
|
_sigRun=pyqtSignal(object)
|
|
|
|
def __init__(self, stop_all):
|
|
super().__init__()
|
|
self._stop_all=stop_all
|
|
self.abort_event=threading.Event()
|
|
self._busy=False
|
|
self._sigRun.connect(self._run, Qt.QueuedConnection)
|
|
|
|
@property
|
|
def busy(self):
|
|
return self._busy
|
|
|
|
def submit(self, steps):
|
|
if self._busy:
|
|
_log.warning('motor sequence already running -> request ignored')
|
|
return False
|
|
self._busy=True
|
|
self.abort_event.clear()
|
|
self._sigRun.emit(steps)
|
|
return True
|
|
|
|
def request_abort(self):
|
|
if not self._busy:
|
|
return
|
|
_log.warning('motor sequence abort requested: stop all motors')
|
|
self.abort_event.set()
|
|
try:
|
|
self._stop_all()
|
|
except Exception as e:
|
|
_log.error(f'failed to stop motors: {e}')
|
|
|
|
def check_abort(self):
|
|
if self.abort_event.is_set():
|
|
raise MoveAborted()
|
|
|
|
@pyqtSlot(object)
|
|
def _run(self, steps):
|
|
err=''
|
|
try:
|
|
n=len(steps)
|
|
for i,(label,func) in enumerate(steps):
|
|
self.check_abort()
|
|
_log.info(f'motor sequence step {i+1}/{n}: {label}')
|
|
self.sigProgress.emit(i, n, label)
|
|
func()
|
|
self.check_abort()
|
|
except MoveAborted:
|
|
err='aborted'
|
|
except Exception as e:
|
|
if self.abort_event.is_set():
|
|
err='aborted'
|
|
elif isinstance(e, MoveError):
|
|
err=str(e)
|
|
else:
|
|
_log.exception('motor sequence failed')
|
|
err=repr(e)
|
|
finally:
|
|
if err:
|
|
_log.warning(f'motor sequence ended: {err}')
|
|
else:
|
|
_log.info('motor sequence done')
|
|
self._busy=False
|
|
self.sigFinished.emit(err)
|
|
|
|
|
|
class CollectionProgressDialog(QDialog):
|
|
'''
|
|
non modal progress dialog: the main window stays usable for the nudges allowed during collection.
|
|
Abort (button, Esc or window close) calls the worker's request_abort
|
|
'''
|
|
def __init__(self, worker, parent=None, title='Progress'):
|
|
super().__init__(parent)
|
|
self._worker=worker
|
|
self.setWindowTitle(title)
|
|
self.setWindowModality(Qt.NonModal)
|
|
lay=QVBoxLayout(self)
|
|
self._label=lbl=QLabel('Starting...')
|
|
lay.addWidget(lbl)
|
|
self._bar=bar=QProgressBar()
|
|
bar.setRange(0, 100)
|
|
lay.addWidget(bar)
|
|
self._btn=btn=QPushButton('Abort')
|
|
btn.clicked.connect(self.abort)
|
|
lay.addWidget(btn)
|
|
self._done=False
|
|
|
|
def set_progress(self, val, maximum, label):
|
|
if maximum!=self._bar.maximum():
|
|
self._bar.setMaximum(maximum)
|
|
self._bar.setValue(min(val, maximum))
|
|
if label:
|
|
self._label.setText(label)
|
|
|
|
def abort(self):
|
|
self._label.setText('Aborting...')
|
|
self._btn.setEnabled(False)
|
|
self._worker.request_abort()
|
|
|
|
def finish(self):
|
|
self._done=True
|
|
self.accept()
|
|
|
|
def reject(self): # Esc key
|
|
if self._done:
|
|
super().reject()
|
|
else:
|
|
self.abort()
|
|
|
|
def closeEvent(self, event):
|
|
if self._done:
|
|
event.accept()
|
|
else:
|
|
event.ignore()
|
|
self.abort()
|