#!/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()