sf-mx/SwissMX
0
0 Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dev
SwissMX/swissmx.py
T

3111 lines
116 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python
# *-----------------------------------------------------------------------*
# | |
# | Copyright (c) 2022 by Paul Scherrer Institute (http://www.psi.ch) |
# | Based on Zac great first implementation |
# | Author Thierry Zamofing (thierry.zamofing@psi.ch) |
# *-----------------------------------------------------------------------*
"""
SwissMx experiment application.
Based on Zac great first implementation
bitmask for simulation:
0x001: backlight
0x002: illumination
0x004: zoom
0x008: camera
0x010: EPICS motors Deltatau
0x020: EPICS motors SmarAct
0x040: shutter
0x080: Jungfrau
0x100: Deltatau motion code
"""
import logging,time,os,sys,socket
class col:
d='\033[0m' # default
r='\033[31m' # red
g='\033[32m' # green
y='\033[33m' # yellow
rr='\033[91m' # red(bright)
gg='\033[92m' # green(bright)
yy='\033[93m' # yellow(bright)
b='\033[1m' # bold
u='\033[4m' # underline
def listLoggers():
lines=[]
rootlogger=logging.getLogger()
lines.append(f'{rootlogger}')
for h in rootlogger.handlers:
lines.append(' %s'%h)
for nm, lgr in logging.Logger.manager.loggerDict.items():
lines.append('+ [%-20s] %s '%(nm, lgr))
if not isinstance(lgr, logging.PlaceHolder):
for h in lgr.handlers:
lines.append(' %s'%h)
logging.getLogger('swissmx').debug('loggers:\n'+'\n'.join(lines))
class logHandler(logging.StreamHandler):
def __init__(self):
logging.StreamHandler.__init__(self)
def emit(self, record):
'''override function of base class'''
try:
msg=self.format(record)
# print(record.__dict__)
if record.levelno<=10:
c=col.g
elif record.levelno<=20:
c=col.y
elif record.levelno<=30:
c=col.yy
elif record.levelno<=40:
c=col.r
else:
c=col.rr+col.b
msg=c+msg+col.d
stream=self.stream
stream.write(msg+self.terminator)
self.flush()
except RecursionError:
raise
except Exception:
self.handleError(record)
logging.basicConfig(level=logging.INFO, format='%(levelname)s:%(module)s:%(lineno)d:%(funcName)s:%(message)s',
handlers=[logHandler()])
_log = logging.getLogger("swissmx")
if __name__=="__main__":
if socket.gethostname()=='ganymede':
base=os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, os.path.abspath(os.path.join(base,'../PBSwissMX/python')))
sys.path.insert(0, os.path.abspath(os.path.join(base,'../../PBTools')))
else:
base=os.path.abspath(os.path.dirname(__file__))
sys.path.insert(0, os.path.join(base,'PBTools'))
sys.path.insert(0, os.path.join(base,'PBSwissMX/python'))
class timestamp():
def __init__(self):
self.t=time.time()
def log(self,txt):
t=time.time()
_log.info(txt+f'{t-self.t:6.3g}')
self.t=t
ts=timestamp()
ts.log('Import part 1/8:')
import sys, os
import yaml
import signal, subprocess
import matplotlib as mpl
import matplotlib.pyplot as plt
mpl.use('Qt5Agg') # needed to avoid blocking of ui !
TASK_JUNGFRAU_SETTINGS = "jungfrau_settings"
TASK_SETUP_GEOMETRY_CALIB = "geometry_calib"
TASK_SETUP_PPM_CALIBRATION_TBOX = "ppm_calibration_tbox"
TASK_SETUP_BEAM_CENTER = "beamcenter_calibration"
TASK_SETUP_CAMERA = "setup_camera"
TASK_SETUP_ROI = "setup_rois"
TASK_SCREENING = "screening"
TASK_FIX_TARGET = "fix_trg"
# during a data collection only these motors can be nudged, in total by at most NUDGE_MAX_STEP (mm)
# from their position at the start of the collection
NUDGE_TWEAKERS = ("colli_x", "colli_y", "tube_usx", "tube_usy", "tube_dsx", "tube_dsy", "tube_z")
NUDGE_MAX_STEP = 0.5
ts.log('Import part 2/8:')
import ModuleFixTarget
ts.log('Import part 3/8:')
import qtawesome
import qutilities
from PyQt5 import QtCore, QtGui
from PyQt5.QtCore import Qt, pyqtSlot, QSize, pyqtSignal, QObject, QRectF,QT_VERSION_STR, QRegularExpression, QTimer
from PyQt5.QtGui import QKeySequence, QPixmap, QRegularExpressionValidator
from PyQt5.QtWidgets import (
QAction, QApplication, QCheckBox, QDoubleSpinBox, QFileDialog, QGridLayout, QHBoxLayout, QLabel, QLineEdit,
QComboBox, QMessageBox, QProgressBar, QPushButton, QShortcut, QSizePolicy, QSpinBox,
QSplashScreen, QToolBox, QVBoxLayout, QWidget,)
from PyQt5.uic import loadUiType
ts.log('Import part 4/8:')
import pyqtUsrObj as UsrGO
from epics_widgets.MotorTweak import MotorTweak
from epics_widgets.SimMotorTweak import SimMotorTweak
ts.log('Import part 5/8:')
import numpy as np
np.set_printoptions(suppress=True,linewidth=196)
import pyqtgraph as pg
import pyqtgraph.exporters
# use antialias for draw lines, interpret image data as row-major instead of col-major
pg.setConfigOptions(antialias=True,imageAxisOrder='row-major')
ts.log('Import part 6/8:')
import app_utils
from app_config import AppCfg,WndParameter #settings, option, toggle_option
import epics
import camera,backlight,zoom,illumination,geometry
ts.log('Import part 7/8:')
import psi_device
import daq_worker
ts.log('Import part 8/8:')
def tdstamp():
return time.strftime("%Y%m%dH%H%M%S")
def datestamp():
return time.strftime("%Y%m%d")
def get_version(path='.'):
#sys.stdout.write('getVersion() -> using git command -> ')
p = subprocess.Popen(f'git -C {path} describe --match ''*.*.*'' --long --tags --dirty', shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
retval = p.wait()
res=p.stdout.readline()
p.stdout.close()
#res=res.decode()[1:-1].split('-',2)
res=res.decode()[:-1].split('-',2)
ver='.'.join(res[:2])
gitcmt=res[2][1:]
return (ver,gitcmt)
def sigint_handler(*args):
"""Handler for the SIGINT signal."""
app=QApplication.instance()
app.quit()
class AcquisitionAbortedException(Exception):
pass
class Sequencer(QObject):
finished = pyqtSignal()
timeoutExpired = pyqtSignal()
errorHappened = pyqtSignal(str)
def __init__(self, steps):
super().__init__()
self.steps = steps
def run_sequence(self):
num_steps = len(self.steps)
for n, step in enumerate(self.steps):
_log.info("runing step {}/{}".format(n, num_steps))
step()
_log.info("emiting finished")
self.finished.emit()
class StartupSplash:
def __init__(self):
splash_pix = QPixmap("artwork/logo/256x256.png")
self._wnd=splash = QSplashScreen(splash_pix, Qt.WindowStaysOnTopHint)
splash.setWindowFlags(Qt.WindowStaysOnTopHint | Qt.FramelessWindowHint)
splash.setEnabled(False)
self._prgsBar=prgs=QProgressBar(splash)
prgs.setMaximum(100)
prgs.setGeometry(0, splash_pix.height() - 50, splash_pix.width(), 20)
splash.show()
def set(self,i,msg):
self._prgsBar.setValue(i)
self._wnd.showMessage(f"<font color='red'>{msg}</font></h1>", int(Qt.AlignBottom|Qt.AlignCenter), Qt.black)
app=QApplication.instance()
app.processEvents()
time.sleep(.1)
Ui_MainWindow, QMainWindow = loadUiType(os.path.join(os.path.dirname(__file__),"swissmx.ui"))
class WndSwissMx(QMainWindow, Ui_MainWindow):
sigNewCamImg = pyqtSignal() # index in self._grids
#pixelsPerMillimeter = pyqtSignal(float)
#beamCameraCoordinatesChanged = pyqtSignal(float, float)
addGridRequest = pyqtSignal(float, float)
#zoomChanged = pyqtSignal(float)
#folderChanged = pyqtSignal(str)
prefixChanged = pyqtSignal(str)
projectChanged = pyqtSignal(str)
gridUpdated = pyqtSignal(int) # index in self._grids
gonioMoveRequest = pyqtSignal(float, float, float, float, float)
#fast_x_position = pyqtSignal(float)
#fast_y_position = pyqtSignal(float)
#fast_dx_position = pyqtSignal(float)
#fast_dy_position = pyqtSignal(float)
fiducialPositionSelected = pyqtSignal(float, float, float, float) # camx, camy, gonx, gony
appendPrelocatedPosition = pyqtSignal(float, float, float, float)
appendPrelocatedFiducial = pyqtSignal(bool, float, float, float, float)
increaseRunNumberRequest = pyqtSignal()
daqAborted = pyqtSignal()
def __init__(self,):
super(WndSwissMx, self).__init__()
self.setupUi(self)
app=QApplication.instance()
if os.getenv("DEVELOPMENT_VERSION"):
title = "[DEVELOPMENT] SwissMX - Solid support scanning for FEL sources"
else:
title = "SwissMX - Solid support scanning for FEL sources"
self.setWindowTitle(title)
self._do_quit = False
self._collecting = False
path=os.path.join(os.path.dirname(__file__),"fonts")
qtawesome.load_font("material", "MaterialIcons-Regular.ttf", "MaterialIcons-Regular.json", path)
QtGui.QFontDatabase.addApplicationFont("fonts/Inconsolata-Bold.ttf")
QtGui.QFontDatabase.addApplicationFont("fonts/Baloo-Regular.ttf")
self.init_settings()
# self.create_escape_toolbar()
self.tweakers = {}
self.setup_sliders()
self.init_graphics()
self._esc_state ="Maintenance"
self._pin_mounting_offset = 0.0
self._mouse_tracking = False
self._rois = []
# add a generic ellipse
# pen = QtGui.QPen(QColor(Qt.yellow))
# pen.setWidth(5)
# ellipse1 = QtWidgets.QGraphicsEllipseItem(650, 700, 500, 300)
# ellipse1.setFlag(QGraphicsItem.ItemIsSelectable)
# ellipse1.setFlag(QGraphicsItem.ItemIsMovable)
# ellipse1.setPen(pen)
# self.vb.addItem(ellipse1)
# self.vb.addItem(Grid(pos=(650, 700), size=(100, 200), cell_size=(3, 4), parent=self))
# self.vb.addItem(Retangulo(pos=(750, 700), size=(100, 200), pen=pg.fn.mkPen(color=[230, 150, 200], width=4.7)))
# self.vb.addItem(pg.PolyLineROI([[80, 60], [90, 30], [60, 40]], pen=(6,9), closed=True))
# test
# x2 = np.linspace(-100, 100, 1000)
# data2 = np.sin(x2) / x2
# # p8 = self.glw.addPlot(row=1, col=0, title="Region Selection")
# plitem = pg.PlotItem()
# plitem.plot(data2, pen=(255, 255, 255, 200))
# # lr = pg.LinearRegionItem([400, 700])
# # lr.setZValue(-10)
# self.vb.addItem(plitem)
self._message_critical_fault = QLabel(None)
self._message_critical_fault.setAccessibleName("device_fault")
self.statusbar.insertWidget(0, self._message_critical_fault)
self._lb_coords=lbl=QLabel(None)
#f=QFont('monospace', 10)
#lbl.setFont(f)
self.statusbar.addPermanentWidget(lbl)
self._fel_status = QLabel(None)
self._fel_status.setAccessibleName("fel_status_statusbar")
self.statusbar.addPermanentWidget(self._fel_status)
#swissfel.statusStringUpdated.connect(self._fel_status.setText) #ZAC: orig. code
self._status_task = QLabel(None)
self._status_task.setAccessibleName("status_task_label")
self.statusbar.addPermanentWidget(self._status_task)
# internal state for camera live update handling
self._camera_monitor_running = False
self.sigNewCamImg.connect(self.cb_update_img, Qt.QueuedConnection)
# data collection runs in its own thread, results come back with queued signals
self._daq_thread=thrd=QtCore.QThread(self)
thrd.setObjectName('daq')
self._daq_worker=w=daq_worker.CollectionWorker()
w.moveToThread(thrd)
thrd.started.connect(w.init_thread)
w.sigProgress.connect(self.cb_daq_progress, Qt.QueuedConnection)
w.sigRunInfo.connect(self.cb_daq_run_info, Qt.QueuedConnection)
w.sigPlotGather.connect(self.cb_daq_plot_gather, Qt.QueuedConnection)
w.sigFinished.connect(self.cb_daq_finished, Qt.QueuedConnection)
self._daq_dlg=None
# blocking motor sequences run in the same thread: they can not overlap with a collection
self._move_worker=mw=daq_worker.MoveWorker(self.stop_all_motors)
mw.moveToThread(thrd)
mw.sigProgress.connect(self.cb_daq_progress, Qt.QueuedConnection)
mw.sigFinished.connect(self.cb_moves_finished, Qt.QueuedConnection)
thrd.start()
#self.add_beam_marker()
#self._beam_markers = {}
# ppm calibration
#self._zoom_to_ppm = {}
#self._ppm_click = None
self.load_stylesheet()
#self.folderChanged.connect(lambda t: print("folder now: {}".format(t))) # FIXME
self.init_actions()
#self.prepare_microscope_page()
self.prepare_wd_tabs_left()
#self.update_beam_marker(qoptic_zoom.get_sp()) #ZAC: orig. code
self._wd_stack.setCurrentIndex(0)
self._wd_stack.currentChanged.connect(self.cb_update_center_widget)
#self._OLD_init_validators()
#self.init_settings_tracker() ? not needed, was for TELL ?
#self._OLD_wire_storage()
self.cb_update_center_widget(0) # start camera updater
curzoom = app._zoom.get_val()
_log.debug(f"starting app with zoom at {curzoom}")
self.cb_zoom_changed(curzoom)
self._tabs_daq_methods.currentChanged.connect(self.cb_switch_task)
self.cb_switch_task()
def init_settings(self):
app = QApplication.instance()
cfg = app._cfg
keys = cfg.allKeys()
#if EXPERIMENT_PGROUP not in keys:
# self.update_user_and_storage()
if cfg.WINDOW_GEOMETRY in keys:
self.restoreGeometry(cfg.value(cfg.WINDOW_GEOMETRY, ""))
self.restoreState(cfg.value(cfg.WINDOW_STATE, ""))
if cfg.WINDOW_SPLITTER in keys:
sz = [int(s) for s in cfg.value(cfg.WINDOW_SPLITTER)]
else:
sz=(500, 1200)
self._wd_splitter.setSizes(sz)
# Data Storage / Experimental Parameters: editable DAQ fields (also in the preferences)
self.build_daq_fields()
self._le_runnumber.setReadOnly(True)
self.cb_update_run_number()
self._le_datapath.setReadOnly(True)
self.cb_update_data_path()
#if AppCfg.BEAM_MARKER_POSITIONS in keys:
# self._beam_markers = s.value(AppCfg.BEAM_MARKER_POSITIONS)
# self.update_beam_marker_fitters()
# _log.info("read beam markers {}".format(self._beam_markers))
#if AppCfg.CAMERA_TRANSFORMATIONS in keys:
# app._camera.set_transformations(s.value(AppCfg.CAMERA_TRANSFORMATIONS))
#if AppCfg.CAMERA_ZOOM_TO_PPM in keys:
# self._zoom_to_ppm = s.value(AppCfg.CAMERA_ZOOM_TO_PPM)
# _log.info(f"{AppCfg.CAMERA_ZOOM_TO_PPM} updated: {self._zoom_to_ppm}")
# self.update_ppm_fitters()
def init_graphics(self):
app = QApplication.instance()
cfg = app._cfg
geo = app._geometry
self.glw = pg.GraphicsLayoutWidget()
self.microscope_page.setLayout(QVBoxLayout())
self.microscope_page.layout().addWidget(self.glw)
self.glw.show()
self.glw.scene().sigMouseMoved.connect(self.cb_mouse_move)
self.glw.scene().sigMouseClicked.connect(self.cb_mouse_click)
#--- viewbox ---
self.vb=vb=self.glw.addViewBox(invertY=False,border='r',enableMenu=True)
#TODO: vb.enableAutoRange(enable=True), vb.autoRange() does not work for ItemGroups
#therefore set the vieweRange manually
pad=10
vb.setRange(QRectF(-1200-pad,-1000-pad,1200+2*pad,1000+2*pad))
vb.setAspectLocked(True)
vb.setBackgroundColor((120, 90, 90))
tr=QtGui.QTransform() # prepare ImageItem transformation:
opt_ctr=app._geometry._opt_ctr
#--- image group ---
# uses image transformation
# contains image and opticalcenter
self._goImgGrp=grp=pg.ItemGroup()
self.vb.addItem(grp)
trf=cfg.value(AppCfg.GEO_CAM_TRF)
# be aware: setTransform is transposed!
# Qt uses: p'=(p.T*A.T).T , but I am used: p'=A*p, with p=[x,y,1].T
tr.setMatrix(trf[0,0],trf[1,0],trf[2,0], #(-1, 0, 0,
trf[0,1],trf[1,1],trf[2,1], # 0,-1, 0,
trf[0,2],trf[1,2],trf[2,2]) # 0, 0, 1)
grp.setTransform(tr) # assign transform
#--- image ---
self._goImg=img=pg.ImageItem() #border=pg.mkPen('r',width=2))
grp.addItem(img)
#--- opctical center ----
oc_sz=np.array((50,50))
#self._goOptCtr=obj=UsrGO.Marker(-opt_ctr+oc_sz/2, oc_sz,mode=1)
self._goOptCtr=obj=UsrGO.Marker(opt_ctr-oc_sz/2, oc_sz,mode=1, movable=False)
obj.sigRegionChangeFinished.connect(self.cb_marker_moved)
grp.addItem(obj)
#--- grid ---
try:
self._goGrid=grid=pg.GridItem(pen=(0,255,0),textPen=(0,255,0)) #green grid and labels
except NameError:
_log.debug('workaround for typo in pyqtgraph:0.11.0')
from PyQt5.QtGui import QPen,QColor
self._goGrid=grid=pg.GridItem() # green grid and labels
grid.opts['pen']=QPen(QColor(0, 255, 0))
grid.opts['textPen']=QPen(QColor(0, 255, 0))
#tr.reset()
#grid.setTransform(tr) # assign transform
vb.addItem(grid)
#--- fixed group ---
# uses pix2pos transformation with a fixed fx,fy value =(0,0)
# contains beam marker
self._goFixGrp=grp=pg.ItemGroup()
geo.interp_zoom(1)
pix2pos=geo._pix2pos
A=np.asarray(pix2pos.I)
tr=grp.transform()
p1=np.hstack((opt_ctr,1)) #position of optical center on image item
p2=np.matmul(trf, p1) #position of optical center on viewbox
tr.setMatrix(A[0,0], A[0,1], 0,
A[1,0], A[1,1], 0,
p2[0], p2[1], 1) # translate dx,dy
grp.setTransform(tr)
self.vb.addItem(grp)
#--- beam marker ---
size_eu=cfg.value(AppCfg.GEO_BEAM_SZ)/1000 # convert from um to mm
pos_eu=cfg.value(AppCfg.GEO_BEAM_POS)/1000 # convert from um to mm
self._goBeamMarker=obj=UsrGO.Marker(pos_eu-size_eu/2,size_eu,mode=0,movable=False)
obj.sigRegionChangeFinished.connect(self.cb_marker_moved)
#bm.setTransform(tr) # assign transform
grp.addItem(obj)
#--- testing scan grid ---
#self.track_objects() # first call is needed to initialize the structure self._goTracked
# #go=UsrGO.Grid((120, -100), (200, 150), (30, 22), 2)
# go=UsrGO.Grid((120, -100), (1000, 500), (10, 5), 2)
# go.setTransform(tr) # assign transform
# vb.addItem(go)
# self._goTracked['objLst'].append(go)
# self.track_objects() #tracking now the objects
# #UsrGO.obj_tree(vb)
def init_actions(self):
app = QApplication.instance()
cfg = app._cfg
self.toolBar.setToolButtonStyle(Qt.ToolButtonTextUnderIcon)
self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.SHIFT + Qt.Key_S), self)
self.shortcut.activated.connect(self.cb_save_cam_image)
self.shortcut = QShortcut(QKeySequence(Qt.Key_F2), self)
self.shortcut.activated.connect(self.cb_save_cam_image)
self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.Key_S), self)
self.shortcut.activated.connect(self.cb_save_cam_image)
for k in range(10):
qkey = k + Qt.Key_0
self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.SHIFT + qkey), self)
self.shortcut.activated.connect(lambda key=k: self.saveBookmark(key))
self.shortcut = QShortcut(QKeySequence(Qt.CTRL + qkey), self)
self.shortcut.activated.connect(lambda key=k: self.gotoBookmark(key))
self._button_collect.clicked.connect(self.cb_execute_collection)
# Toolbar buttons
icon_size = QSize(50, 50)
icon = qtawesome.icon("material.photo_camera")
action = QAction(icon, "Save Original", self)
action.setToolTip("(Ctrl+Shift+S) Take a screenshot of the sample image, without markers. Saves in current folder.")
action.triggered.connect(self.cb_save_cam_image)
self.toolBar.addAction(action)
action = QAction(icon, "Save View", self)
action.setToolTip("(Ctrl+S) Take a screenshot of the currently visible sample image, including markers. Saves in current folder.")
action.triggered.connect(lambda: self.cb_save_cam_image(True))
self.toolBar.addAction(action)
if os.getenv("DEVELOPMENT_VERSION"):
icon = qtawesome.icon("material.style")
action = QAction(icon, "Reload Stylesheet", self)
action.triggered.connect(self.load_stylesheet)
self.toolBar.addAction(action)
# icon = qtawesome.icon("material.shutter_speed")
#action = QAction("Use Pulse Picker", self)
#action.setCheckable(True)
#action.setChecked(cfg.option(AppCfg.ACTIVATE_PULSE_PICKER))
#action.triggered.connect(lambda: cfg.toggle_option(AppCfg.ACTIVATE_PULSE_PICKER))
#self.toolBar.addAction(action)
#icon = qtawesome.icon("material.timeline")
#action = QAction(icon, "Add Line", self)
#action.triggered.connect(self._OLD_roi_add_line)
#self.toolBar.addAction(action)
#icon = qtawesome.icon("material.tab_unselected")
#action = QAction(icon, "Add Rect", self)
#action.triggered.connect(self._OLD_roi_add_rect)
#self.toolBar.addAction(action)
#icon = qtawesome.icon("material.play_for_work")
#action = QAction(icon, "TELL Mount", self)
#action.setToolTip("BookMark(0) => Move stages CZ, CX, FX, FY, and Omega to Tell sample changer mount position")
#action.triggered.connect(self._OLD_escape_goToTellMountPosition)
#self.toolBar.addAction(action)
icon = qtawesome.icon("material.play_for_work")
action = QAction(icon, "Auto\nFocus", self)
action.triggered.connect(self.cb_autofocus)
self.toolBar.addAction(action)
action = QAction(icon, "Find\nFiducial", self)
action.triggered.connect(self.cb_find_fiducial)
self.toolBar.addAction(action)
action = QAction(icon, "Test\nCode", self)
action.triggered.connect(self.cb_testcode)
self.toolBar.addAction(action)
action = QAction(icon, "Move to\nYAG", self)
action.setToolTip("Move goniometer stages to the configured YAG position.")
action.triggered.connect(self.cb_move_yag)
self.toolBar.addAction(action)
action = QAction(icon, "Move to\nHome", self)
action.setToolTip("Move the fast stages (fast X, fast Y) back to 0, 0.")
action.triggered.connect(self.cb_move_home)
self.toolBar.addAction(action)
self.toolBar.addWidget(qutilities.horiz_spacer())
icon = qtawesome.icon("material.sync")
action = QAction(icon, "Sample\nExchange", self)
action.setToolTip("Move devices so a sample can be exchanged.")
action.setObjectName("action_SampleExchange")
action.triggered.connect(self.cb_esc_sample_exchange)
self.toolBar.addAction(action)
self.toolBar.widgetForAction(action).setAccessibleName("action_SampleExchange")
icon = qtawesome.icon("material.my_location")
action = QAction(icon, "Sample\nAlignment", self)
action.setToolTip("Move devices so a sample can be aligned.")
action.triggered.connect(self.cb_esc_sample_alignment)
self.toolBar.addAction(action)
self.toolBar.widgetForAction(action).setAccessibleName("action_SampleAlignment")
icon = qtawesome.icon("material.fingerprint")
action = QAction(icon, "Data\nCollection", self)
action.setToolTip("Move devices so a sample can be collected.")
action.triggered.connect(self.cb_esc_data_collection)
self.toolBar.addAction(action)
self.toolBar.widgetForAction(action).setAccessibleName("action_DataCollection")
self.actionQuit.triggered.connect(self.cb_really_quit)
self.actionPreferences.triggered.connect(self.cb_modify_app_param)
self.actionHome_Fast_Stages.triggered.connect(self.cb_deltatau_home_faststages)
action=QAction("Home Coarse Y", self)
action.triggered.connect(self.cb_deltatau_home_coarse_y)
self.menuAdvanced.addAction(action)
self.actionAbout.triggered.connect(self.cb_about)
#self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.SHIFT + Qt.Key_L), self)
#self.shortcut.activated.connect(self._OLD_roi_add_line)
#self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.SHIFT + Qt.Key_R), self)
#self.shortcut.activated.connect(self._OLD_roi_add_rect)
#self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.SHIFT + Qt.Key_M), self)
#self.shortcut.activated.connect(self._OLD_toggle_mouse_tracking)
#self.shortcut = QShortcut(QKeySequence(Qt.Key_F11), self)
#self.shortcut.activated.connect(self._OLD_toggle_maximized)
#self.shortcut = QShortcut(QKeySequence(Qt.Key_F12), self)
#self.shortcut.activated.connect(self._OLD_show_window_configuration)
#self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.Key_B), self)
#self.shortcut.activated.connect(lambda: self._OLD_beammark.toggle_handle())
#self.shortcut = QShortcut(QKeySequence(Qt.CTRL + Qt.Key_P), self)
#self.shortcut.activated.connect(self._OLD_camera_pause_toggle)
# adding a menu entry to one of the menus
#action = QAction("parameters", self)
#action.setToolTip("modify application parameters")
#action.setStatusTip("modify application parameters")
#action.triggered.connect(self.cb_modify_app_param)
#self.menuAdvanced.addAction(action)
#action = QAction("geometry", self)
#action.setToolTip("Update optical center, beam marker size etc.")
#action.setStatusTip("Update optical center, beam marker size etc.")
#action.triggered.connect(lambda x: cfg.dlg_geometry(self))
#self.menuAdvanced.addAction(action)
#action = QAction("Backlight Reference Positions", self)
#action.setToolTip("Update the reference positions for the backlight")
#action.setStatusTip("Update the reference positions for the backlight")
#action.triggered.connect(cfg.dlg_backlight_positions)
#self.menuAdvanced.addAction(action)
#action = QAction("Collimator Reference Positions", self)
#action.setToolTip("Update the reference positions for the collimator")
#action.setStatusTip("Update the reference positions for the collimator")
#action.triggered.connect(cfg.dlg_collimator_reference_positions)
#self.menuAdvanced.addAction(action)
#action = QAction("Post sample tube Reference Positions", self)
#action.setToolTip("Update the reference positions for the post tube")
#action.setStatusTip("Update the reference positions for the post tube")
#action.triggered.connect(cfg.dlg_posttube_references)
#self.menuAdvanced.addAction(action)
#action = QAction("Delta Tau Parameters", self)
#action.setToolTip("Parameters affecting the Delta Tau")
#action.setStatusTip("Parameters affecting the Delta Tau")
#action.triggered.connect(cfg.dlg_deltatau_parameters)
#self.menuAdvanced.addAction(action)
#action = QAction("Cryojet Reference Positions", self)
#action.setToolTip("Update the reference positions for the cryojet nozzle position")
#action.setStatusTip("Update the reference positions for the cryojet nozzle position")
#action.triggered.connect(self.set_cryojet_positions_dialog)
#self.menuAdvanced.addAction(action)
#action = QAction("Tell Mount Positions", self)
#action.setToolTip("Mount positions for TELL sample changer")
#action.setStatusTip("Parameters affecting the Delta Tau")
#action.triggered.connect(cfg.dlg_tell_mount_positions)
#self.menuAdvanced.addAction(action)
def closeEvent(self, event): #overloaded function
"""this is called when the user clicks the window's cross icon"""
if self._daq_worker.busy or self._move_worker.busy:
QMessageBox.warning(self, "busy", "Abort or wait for the data collection or motor sequence before quitting.")
event.ignore()
return
if (self._do_quit or QMessageBox.question( self, "", "Are you sure you want to quit?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No,) == QMessageBox.Yes):
app=QApplication.instance()
cfg=app._cfg
self._daq_thread.quit()
self._daq_thread.wait(2000)
try:
for pv in app._pv_mon_lst:
try:
pv.clear_auto_monitor() # disconnect PV monitor callback -> program exit faster.
except Exception:
pv.auto_monitor=False
except AttributeError:
_log.warning('disconnect PV callback failed.')
_log.info('disconnect PV callback')
cfg.setValue(cfg.WINDOW_GEOMETRY, self.saveGeometry())
cfg.setValue(cfg.WINDOW_STATE, self.saveState())
cfg.setValue(cfg.WINDOW_SPLITTER, self._wd_splitter.sizes())
cfg.setValue('last_active', time.time())
_log.info('save settings')
#QMainWindow.closeEvent(self, event)
_log.info('closeEvent done')
else:
event.ignore()
def cb_update_center_widget(self, index):
_log.debug('TODO: check to not connect smultiple igNewCamImg')
if index > 0: # not showing camera image
_log.warning("listening to zescape")
pass
#self.timer.stop()
#zescape.start_listening()
#self.timer = QtCore.QTimer(self)
#self.timer.timeout.connect(self.check_zescape)
#self.timer.start(20)
else:
app=QApplication.instance()
try:
self.cb_new_frame_sim()
except AttributeError:
cam=app._camera
cfg=app._cfg
param=cam.get_param()
paramSv=cfg.value(AppCfg.GEO_CAM_PARAM)
if not paramSv:
cfg.setValue(AppCfg.GEO_CAM_PARAM, param)
else:
diff=False
for k,v in param.items():
if v!=paramSv[k]:
if type(v)==tuple:
if v==tuple(paramSv[k]):
continue
_log.debug(f'diff:{k},{v} != {paramSv[k]}')
diff=True
if diff:
cam.set_param(**paramSv)
if 'pic' not in cam._pv:
cam.run(self.cb_new_frame_pv)
app._pv_mon_lst.append(cam._pv['pic'])
self._camera_monitor_running = True
def cb_new_frame_pv(self, **kwargs):
try: #pv-monitor-func
app=QApplication.instance()
cam=app._camera
sz=cam._sz
if kwargs['count']==sz[0]*sz[1]:
pic=kwargs['value'].reshape(sz[::-1])
else:
sz=app._camera.update_size()
pic=kwargs['value'].reshape(sz[::-1])
if pic.dtype==np.int16:
pic.dtype=np.uint16
camera.epics_cam.set_fiducial(pic, 255)
cam._pic=pic
cam._timestamp=kwargs['timestamp']
try:
cam.process()
except AttributeError:
pass
# self._goImg.setImage(cam._pic) caused some deadlocks.
# therefore try to update the image with signals instead
self.sigNewCamImg.emit()
except Exception as e:
_log.critical(f'{e}')
def cb_update_img(self):
#thrd=threading.current_thread()
#_log.debug(f'thread:{thrd.getName()}, {thrd.native_id}')
app=QApplication.instance()
cam=app._camera
pic=cam._pic
if pic.shape!=getattr(self,'_goImgShape',None): # binning or roi changed
self._goImgShape=pic.shape
self.update_img_binning()
self._goImg.setImage(pic)
#vb.setRange(QRectF(-1300,-1100,1400,1200))
#force a segmentation fault
#try:
# dbgSegFault=self._dbgSegFault
#except AttributeError as e:
# dbgSegFault=self._dbgSegFault=0
#_log.warning(f'dbgSegFault:{dbgSegFault}')
#if dbgSegFault>100:
# _log.critical('Xforce a segmentation fault')
# import ctypes
# ctypes.string_at(0)
#self._dbgSegFault+=1
def update_img_binning(self):
# pix2pos, opt_ctr etc. are in pixels at geometry.CALIB_BINNING.
# scale the image so that its pixels match these calibration pixels at any binning
app=QApplication.instance()
cam=app._camera
try:
b=(cam.getPv('BINX').get(),cam.getPv('BINY').get())
s=np.array(b,np.float64)/geometry.CALIB_BINNING
except Exception as e:
_log.warning(f'binning unknown, image not scaled: {e}')
s=(1.,1.)
_log.debug(f'image binning scale: {s}')
self._goImg.setTransform(QtGui.QTransform.fromScale(s[0],s[1]))
def cb_new_frame_sim(self, **kwargs):
app=QApplication.instance()
sim=app._camera._sim
imgSeq=sim['imgSeq']
idx=sim['imgIdx']
sim['imgIdx']=(idx+1)
imgIdx=idx%imgSeq.shape[0]
# _log.info('simulated idx:{}'.format(idx))
pic=imgSeq[imgIdx]
self._goImg.setImage(pic)
delay=500 # ms -> 2fps
QtCore.QTimer.singleShot(delay, self.cb_new_frame_sim)
#force a segmentation fault
#_log.warning(f'imgIdx:{imgIdx},idx:{idx}')
#if idx==20:
# _log.critical('force a segmentation fault')
# import ctypes
# ctypes.string_at(0)
def load_stylesheet(self):
with open(os.path.join(os.path.dirname(__file__),"swissmx.css"), "r") as sheet:
self.setStyleSheet(sheet.read())
def setup_sliders(self):
cont = self._wd_right
self._tweak_container = QWidget()
self._tweak_container.setLayout(QVBoxLayout())
cont.layout().addWidget(self._tweak_container)
layout = self._tweak_container.layout()
layout.setSpacing(0)
layout.setContentsMargins(0, 0, 0, 0)
self.zoombox = zoom.Zoom()
self.zoombox.init_settings()
self.zoombox.zoomChanged.connect(self.cb_zoom_changed)
self.zoombox.moveBacklight.connect(self.cb_move_backlight_safe)
layout.addWidget(self.zoombox)
self._tbox_motors = toolbox = QToolBox()
layout.addWidget(toolbox)
self._wd_nudge_groups = [] # toolbox pages that stay enabled during data collection
self._wd_nudge_macros = [] # buttons in these pages that are disabled during data collection
self.build_group_faststage(toolbox)
# self.build_slits_group(toolbox)
self.build_group_collimator(toolbox)
self.build_group_posttube(toolbox)
#self.build_cryo_group(toolbox)
# monitor all axis for an axis fault
for key, tweaker in self.tweakers.items():
tweaker.event_axis_fault.connect(self.cb_axis_fault)
#self.tweakers["fast_x"].event_readback.connect(lambda alias, value, kw: self.fast_x_position.emit(value))
#self.tweakers["fast_y"].event_readback.connect(lambda alias, value, kw: self.fast_y_position.emit(value))
self.tweakers["fast_x"].event_rbv.connect(lambda rec_name, kw: self.track_objects())
self.tweakers["fast_y"].event_rbv.connect(lambda rec_name, kw: self.track_objects())
# layout.addStretch()
def cb_marker_moved(self,obj,*args,**kwargs):
_log.debug(args)
_log.debug(kwargs)
app=QApplication.instance()
cfg=app._cfg
geo=app._geometry
opt_ctr=geo._opt_ctr
if obj==self._goOptCtr:
oc_pos=obj.pos()
oc_sz=obj.size()
geo._opt_ctr=opt_ctr=np.array(oc_pos+oc_sz/2)
if (QMessageBox.question(self, "save config", f"save optical center position {opt_ctr} permanently?",
QMessageBox.Yes|QMessageBox.No, QMessageBox.No, )==QMessageBox.Yes):
cfg.setValue(AppCfg.GEO_OPT_CTR,opt_ctr)
elif obj==self._goBeamMarker:
bm_pos=obj.pos()
bm_sz=obj.size()
pos=np.array(bm_pos+bm_sz/2)*1000 # convert from mm to um
if (QMessageBox.question(self, "save config", f"save beam marker position {pos} permanently?",
QMessageBox.Yes|QMessageBox.No, QMessageBox.No, )==QMessageBox.Yes):
cfg.setValue(AppCfg.GEO_BEAM_POS,pos)
self.track_objects()
def cb_zoom_changed(self, value):
#self.zoomChanged.emit(value)
app=QApplication.instance()
geo=app._geometry
try:
geo.interp_zoom(value)
except AttributeError as e:
_log.warning(e)
else:
grp=self._goFixGrp
opt_ctr=geo._opt_ctr
A=np.asarray(geo._pix2pos.I)
#trf=cfg.value(AppCfg.GEO_CAM_TRF)
tr=self._goImgGrp.transform()
p=tr.map(opt_ctr[0],opt_ctr[1])
tr.setMatrix(A[0, 0], A[0, 1], 0,
A[1, 0], A[1, 1], 0,
p[0], p[1], 1) # translate dx,dy
grp.setTransform(tr)
#bm=self._goBeamMarker
#bm_sz=np.abs(geo.pos2pix(bm._size_eu))
#bm_pos=-opt_ctr-geo.pos2pix(bm._pos_eu)-bm_sz/2
#bm.blockSignals(True) # avoid to call cb_marker_moved
#bm.setPos(bm_pos,finish=False)
#bm.setSize(bm_sz)
#bm.blockSignals(False)
#_log.debug(f"zoom->{value} beam marker pos:{bm_pos} sz:{bm_sz})")
self.track_objects()
def cb_move_backlight_safe(self, pos):
self.run_moves(f'backlight {pos}', [
('detector out', lambda: self.move_detector("out")),
('post sample tube out', lambda: self.move_post_tube("out")),
(f'backlight {pos}', lambda: self.move_backlight(pos, check_detector=True)),
])
def move_backlight(self, pos, check_detector=False):
'''move the backlight and wait until it arrived (worker thread)'''
app=QApplication.instance()
bl=app._backlight
if check_detector and not self.is_detector_safe():
raise daq_worker.MoveError("Detector is not clear of the post-sample-tube travel path - backlight move aborted.")
bl.move(pos, wait=True)
self._move_worker.check_abort()
if not bl.is_pos(pos):
raise daq_worker.MoveError(f"backlight did not reach the {pos} position")
def get_tweaker(self, rec, mtype=0, **kwargs):
app = QApplication.instance()
sim=app._args.sim
if mtype == 0: # default epics motor record
if sim&0x10:
m_tw=SimMotorTweak()
else:
m_tw=MotorTweak()
elif mtype==1: # smaract motor record
if sim&0x20:
m_tw=SimMotorTweak()
else:
m_tw=MotorTweak()
m_tw.connect_motor(rec, **kwargs)
self.tweakers[kwargs['alias']] = m_tw
return m_tw
def build_post_tube_tandem_tweaker(self):
block = QWidget()
block.setAccessibleName("post tube tamden")
block.setContentsMargins(0, 0, 0, 0)
block.setLayout(QVBoxLayout())
c = QWidget()
glay = QGridLayout()
c.setLayout(glay)
b = QWidget()
blay = QHBoxLayout()
b.setLayout(blay)
blay.addWidget(QLabel("Tweak (mm)"))
self._post_tandem_tweak_val = QLineEdit("1.0")
blay.addWidget(self._post_tandem_tweak_val)
glay.addWidget(b, 0, 0, 1, 2)
icon = qtawesome.icon("material.arrow_back")
but = QPushButton(icon, "")
but.clicked.connect(lambda m, pos="x+": self.move_post_tube(pos))
glay.addWidget(but, 1, 0)
# c.setLayout(glay)
icon = qtawesome.icon("material.arrow_forward")
but = QPushButton(icon, "")
but.clicked.connect(lambda m, pos="x-": self.move_post_tube(pos))
glay.addWidget(but, 1, 1)
# c.setLayout(glay)
icon = qtawesome.icon("material.arrow_downward")
but = QPushButton(icon, "")
but.clicked.connect(lambda m, pos="y-": self.move_post_tube(pos))
glay.addWidget(but, 2, 0)
# c.setLayout(glay)
icon = qtawesome.icon("material.arrow_upward")
but = QPushButton(icon, "")
but.clicked.connect(lambda m, pos="y+": self.move_post_tube(pos))
glay.addWidget(but, 2, 1)
block.layout().addWidget(c)
return block
def cb_axis_fault(self, pvname, kw):
"""
swissmx - {'pvname': 'SAR-EXPMX:MOT_FY.STAT', 'value': 0, 'char_value': 'NO_ALARM', 'status': 0, 'ftype': 17, 'chid': 38980392,
'host': 'SAR-CPPM-EXPMX1.psi.ch:5064', 'count': 1, 'access': 'read-only', 'write_access': False, 'read_access': True,
'severity': 0,
'timestamp': 1537867290.914189, 'precision': None, 'units': None, 'enum_strs': ('NO_ALARM', 'READ', 'WRITE', 'HIHI', 'HIGH', 'LOLO', 'LOW', 'STATE', 'COS', 'COMM', 'TIMEOUT', 'HWLIMIT', 'CALC', 'SCAN', 'LINK', 'SOFT'), 'upper_disp_limit': None, 'lower_disp_limit': None, 'upper_alarm_limit': None, 'lower_alarm_limit': None, 'lower_warning_limit': None, 'upper_warning_limit': None, 'upper_ctrl_limit': None, 'lower_ctrl_limit': None, 'nelm': 1, 'type': 'time_enum', 'typefull': 'time_enum', 'motor_field': 'STAT', 'source_field': 'STAT', 'cb_info': (1, <PV 'SAR-EXPMX:MOT_FY.STAT', count=1, type=time_enum, access=read-only>), 'alias': 'fast Y'}
:param pvname:
:param kw:
:return:
"""
_log.debug(f"AXIS FAULT: {kw}")
if kw["severity"]:
msg = f"axis {pvname} has a fault"
_log.critical(msg)
_log.critical(f"{kw}")
else:
msg = ""
self._message_critical_fault.setText(msg)
@staticmethod
def pv_monitor_add(widgets):
#pv-monitor-func
# this is a function to search for a issue that crashes the application
# during data acquisition
#_log.info('modify monitors')
app=QApplication.instance()
pv_mon_lst=app._pv_mon_lst
for w in widgets[:-1]: #ignore last item
if type(w)==SimMotorTweak: # simulated motor: no PVs to monitor
continue
try:
for pv in w._motor._pvs.values():
pv_mon_lst.append(pv)
except AttributeError as e:
_log.warning(f'{e},{w}')
# for k,pv in w._motor._pvs.items():
# pv.auto_monitor=False
# for k,pv in w._motor._pvs.items():
# pv.auto_monitor=True
def build_group_faststage(self, toolbox):
pfx=QApplication.instance()._cfg.value(AppCfg.GBL_DEV_PREFIX)[0]
c=QWidget()
c.setLayout(QGridLayout())
f=lambda v: lambda x:self.run_moves(f'detector {v}', [(f'detector {v}', lambda: self.move_detector(v))]) #needs nested lambda to work ...
for i,v in enumerate(('in','out')):
but=QPushButton('detector '+v)
but.clicked.connect(f(v))
c.layout().addWidget(but, 0, i)
widgets=[
# self.get_tweaker('f"{pfx}:MOT_BLGT', alias='backlight', label='backlight'),
self.get_tweaker(f"{pfx}:MOT_FY", alias="fast_y", label="fast Y"),
self.get_tweaker(f"{pfx}:MOT_FX", alias="fast_x", label="fast X"),
self.get_tweaker(f"{pfx}:MOT_CY", alias="base_y", label="base Y"),
self.get_tweaker(f"{pfx}:MOT_CX", alias="base_x", label="base X"),
self.get_tweaker(f"{pfx}:MOT_CZ", alias="base_z", label="base Z"),
self.get_tweaker(f"{pfx}:MOT_DET_Z", alias="det_z", label="detector Z"),
c
]
qutilities.add_item_to_toolbox(toolbox,"Fast Stage",widget_list=widgets)
WndSwissMx.pv_monitor_add(widgets)
def build_group_collimator(self, toolbox):
pfx='SARES30-SMX' # SmarAct motors (the 2nd device_prefix in the preferences is outdated)
c=QWidget()
c.setLayout(QGridLayout())
f=lambda v: lambda x:self.run_moves(f'collimator {v}', [(f'collimator {v}', lambda: self.move_collimator(v))]) #needs nested lambda to work ...
for i,v in enumerate(('in','out','ready')):
but=QPushButton(v)
but.clicked.connect(f(v))
c.layout().addWidget(but, 0, i)
self._wd_nudge_macros.append(but)
widgets = [
self.get_tweaker(f"{pfx}:MCS1", alias="colli_x", label="colli X", mtype=1,),
self.get_tweaker(f"{pfx}:MCS2", alias="colli_y", label="colli Y", mtype=1,),
c,
]
idx=qutilities.add_item_to_toolbox(toolbox,"Collimator",widget_list=widgets)
self._wd_nudge_groups.append(toolbox.widget(idx))
WndSwissMx.pv_monitor_add(widgets)
def build_group_posttube(self, toolbox):
pfx='SARES30-SMX' # SmarAct motors (the 2nd device_prefix in the preferences is outdated)
widgets = [
self.get_tweaker(f"{pfx}:MCS4", alias="tube_usx", label="upstream X", mtype=1,),
self.get_tweaker(f"{pfx}:MCS6", alias="tube_usy", label="upstream Y", mtype=1,),
self.get_tweaker(f"{pfx}:MCS5", alias="tube_dsx", label="downstream X", mtype=1,),
self.get_tweaker(f"{pfx}:MCS7", alias="tube_dsy", label="downstream Y", mtype=1,),
self.get_tweaker(f"{pfx}:MCS8", alias="tube_z", label="tube Z", mtype=1),
]
c = QWidget()
c.setLayout(QGridLayout())
but = QPushButton("post tube out")
but.clicked.connect(lambda m: self.run_moves('post sample tube out', [('post sample tube out', lambda: self.move_post_tube("out"))]))
c.layout().addWidget(but, 0, 0)
self._wd_nudge_macros.append(but)
but = QPushButton("post tube in")
but.clicked.connect(lambda m: self.run_moves('post sample tube in', [
('detector out', lambda: self.move_detector("out")),
('backlight out', lambda: self.move_backlight("out")),
('post sample tube in', lambda: self.move_post_tube("in"))]))
c.layout().addWidget(but, 0, 1)
self._wd_nudge_macros.append(but)
widgets.append(c)
label = "Post Sample Tube"
block = QWidget()
block.setAccessibleName(label)
block.setContentsMargins(0, 0, 0, 0)
block.setLayout(QVBoxLayout())
for w in widgets:
block.layout().addWidget(w)
block.layout().addWidget(self.build_post_tube_tandem_tweaker())
block.layout().addStretch()
toolbox.addItem(block, label)
self._wd_nudge_groups.append(block)
def set_active_task(self, task):
_log.info("TASK == {}".format(task))
self._active_task = task
def active_task(self):
return self._active_task
def cb_mouse_move(self, pos):
app = QApplication.instance()
geo = app._geometry
#pos = pixel position on the widget
task = self.active_task()
z = app._zoom.get_val()
bm=self._goBeamMarker
#pos=event.scenePos()
pImg=pg.Point(self._goImg.mapFromScene(pos))
pTrk=pg.Point(self._goTracked.mapFromScene(pos))
pFix=pg.Point(self._goFixGrp.mapFromScene(pos))
pFix-=bm.pos()+bm.size()/2
fx=self.tweakers["fast_x"].get_val()
fy=self.tweakers["fast_y"].get_val()
pRel=pTrk-(fx,fy)
#s=f'pImg{pImg} pTrk{pTrk} bm._pos_eu{bm._pos_eu}'
try:
pln=geo._fitPlane
except AttributeError:
cz=np.nan
else:
cz=pln[0]*pTrk[0]+pln[1]*pTrk[1]+pln[2] # z=ax+by+c
s=\
f'img pix ({pImg[0]:0.1f} {pImg[1]:0.1f})px \u23A2 ' \
f'stage ({pTrk[0]:0.4f} {pTrk[1]:>0.4f} {cz:>0.4f})mm \u23A2 ' \
f'dist to beam ({pFix[0]:>0.4f} {pFix[1]:>0.4f})mm '
#f'dist to beam ({pRel[0]:>0.6g} {pRel[1]:>0.6g}mm) '
#_log.debug(s)
self._lb_coords.setText(s)
#self._lb_coords.setText(
# "\u23FA{:>}:{:>6.0f} {:<.0f}"
# "\u23FA{:>}:{:>6.0f} {:<.0f}"
# "\u23FA{:>}:{:<.0f}"
# "\u23FA{:>}:{:>6.1f} {:<.1f} \u00B5"
# "\u23FA{:>}:{:>7.3f} {:<.3f} mm".format(
# "Beam at", bx, by,
# "Pixel coord ", x, y,
# "PPM", ppm,
# "Distance to beam", 1000 * dx, 1000 * dy,
# "Stage", fx, fy,
# )
#)
def cb_mouse_click_geometry_calib(self, event):
app=QApplication.instance()
zoom = app._zoom.get_val()
fast_x=self.tweakers["fast_x"];fast_y=self.tweakers["fast_y"]
if self._btn_pix2pos.isChecked():
try:
raw=app._raw_pix2pos[zoom]
except KeyError as e:
raw=app._raw_pix2pos[zoom]=list()
imgPos=self._goImgGrp.mapFromScene(event.scenePos()) # calibration pixels (independent of binning)
fx=fast_x.get_rbv();
fy=fast_y.get_rbv()
x=imgPos.x();y=imgPos.y()
_log.debug('append calib: zoom:{} stage_x_y:{}/{} cam_pix_x_y:{}/{}'.format(zoom, fx, fy, x, y))
raw.append((fx, fy, x, y))
self._wd_calib_tree.setData(app._raw_pix2pos)
elif self._btn_opt_ctr.isChecked():
try:
raw=app._raw_opt_ctr[zoom]
except KeyError as e:
raw=app._raw_opt_ctr[zoom]=list()
imgPos=self._goImgGrp.mapFromScene(event.scenePos()) # calibration pixels (independent of binning)
x=imgPos.x();y=imgPos.y()
_log.debug('append calib: zoom:{} cam_pix_x_y:{}/{}'.format(zoom, x, y))
raw.append(( x, y))
self._wd_calib_tree.setData(app._raw_opt_ctr)
else:
QMessageBox.warning(self, "calibration", "no calibration started yet.\nPress 'pix2pos' or 'opt_ctr' button first")
def cb_mouse_click(self, event):
#_log.debug("{}".format(event))
#_log.debug("screen pos {}".format(event.screenPos())) #pixel position on the whole screen
#_log.debug("scene pos {}".format(event.scenePos())) #pixel position on the scene (including black frame)
#_log.debug(" pos {}".format(event.pos())) #pixel position of the ckicked object mapped to its coordinates
#p=event.scenePos()
#_log.debug(f"vb pos {self.vb.mapFromScene(p)}") #pixel position on the scene (including black frame)
#for o in self.vb.childGroup.childItems():
# _log.debug(f"{type(o)} pos {o.mapFromScene(p)}") #pixel position on the scene (including black frame)
#_log.debug(f"currentItem:{event.currentItem}")
if self._collecting:
_log.info('mouse click ignored during data collection')
return
app=QApplication.instance()
mod=event.modifiers()
btn=event.button()
task = self.active_task()
if task==TASK_SETUP_GEOMETRY_CALIB:
if btn==Qt.LeftButton and not mod&Qt.ControlModifier:
self.cb_mouse_click_geometry_calib(event)
return
################################
# Ctrl-left : move to position
# Ctrl-right : object tree
################################
#dblclick = event.double()
if btn==Qt.LeftButton:
if mod&Qt.ShiftModifier:
pass
elif mod&Qt.ControlModifier:
pos=event.scenePos()
_log.debug(f'move to position :scene pos {pos}')
geo=app._geometry
bm=self._goBeamMarker
pImg=pg.Point(self._goImg.mapFromScene(pos))
pTrk=pg.Point(self._goTracked.mapFromScene(pos))
pFix=pg.Point(self._goFixGrp.mapFromScene(pos))
pFix-=bm.pos()+bm.size()/2
#create trace
self.add_camera_trace(pFix)
_log.debug(f'dist to beam ({pFix[0]:>0.6g} {pFix[1]:>0.6g}mm)')
fx_motor=self.tweakers["fast_x"]
fy_motor=self.tweakers["fast_y"]
fx_motor.move_rel(pFix[0])
fy_motor.move_rel(pFix[1])
#fx_motor.move_abs(fx_motor.get_val()+pFix[0])
#fy_motor.move_abs(fy_motor.get_val()+pFix[1])
#if none of the image is visible, recenter it
r1=self.vb.viewRect()
r2=self.vb.itemBoundingRect(self._goImg)
if not r1.intersects(r2):
r1.translate(r2.center()-r1.center())
self.vb.setRange(r1)
try:
pln=geo._fitPlane
except AttributeError: pass
else:
cz_motor=self.tweakers["base_z"]
cz=pln[0]*pTrk[0]+pln[1]*pTrk[1]+pln[2] #z=ax+by+c
if cz:
cz_motor.move_abs(cz)
else:
_log.critical(f'{cz} {pTrk} {pln}')
_log.debug(f'cz={pln[0]:.4g}*{pTrk[0]:.4g}+{pln[1]:.4g}*{pTrk[1]:.4g}+{pln[2]:.4g}={cz}')
pass
elif mod&Qt.AltModifier:
pass
else:
pass
elif btn==Qt.RightButton:
if mod&Qt.ShiftModifier:
pass
elif mod&Qt.ControlModifier:
_log.debug('object tree')
UsrGO.obj_tree(self.vb)
pass
elif mod&Qt.AltModifier:
pass
else:
pass
return
return
def track_objects(self):
#self._goTracked= dictionary with infos about tracked objects
# objLst = list of objects to track
# state = (zoom, fx, fy) (created at first call of self.track_objects() )
app=QApplication.instance()
geo=app._geometry
zoom = app._zoom.get_val()
twk_fx=self.tweakers["fast_x"];twk_fy=self.tweakers["fast_y"]
# TODO: get_position() is delayed as it is the RBV. do same as for the zoom
# TODO: and return the last set_point
fx=twk_fx.get_rbv()
fy=twk_fy.get_rbv()
try:
grp=self._goTracked
grpTrc=self._goTrace
zoom_old,fx_old,fy_old=grp._state
except AttributeError: # at initialization
self._goTracked=grp=pg.ItemGroup()
grp.setZValue(20) #higher than beam marker and opt ctr that have z=10
self.vb.addItem(grp)
grp._state=(zoom,fx,fy)
zoom_old=fx_old=fy_old=None
#--- camera tracing images ---
self._goTrace=grpTrc=pg.ItemGroup() # the transformation is the same as for tracked objects
grpTrc.setZValue(-20) #lower than other objects
self.vb.addItem(grpTrc)
try:
_log.debug(f"zoom:{zoom_old}->{zoom:d} fx:{fx_old:.3f}->{fx:.3f} fy:{fy_old:.3f}->{fy:.3f}")
except TypeError:
_log.debug(f"zoom:{zoom_old}->{zoom} fx:{fx_old}->{fx:.3f} fy:{fy_old}->{fy:.3f}")
if zoom_old!=zoom:
geo.interp_zoom(zoom)
bm=self._goBeamMarker
opt_ctr=geo._opt_ctr
A=np.asarray(geo._pix2pos.I)
# trf=cfg.value(AppCfg.GEO_CAM_TRF)
tr=self._goFixGrp.transform()
p1=bm.pos()+bm.size()/2+(-fx,-fy)
#p=tr.map(opt_ctr[0], opt_ctr[1])
p=tr.map(p1[0], p1[1])
tr.setMatrix(A[0, 0], A[0, 1], 0,
A[1, 0], A[1, 1], 0,
p[0], p[1], 1) # translate dx,dy
grp.setTransform(tr)
grpTrc.setTransform(tr)
def add_camera_trace(self,delta_mm):
# add up to tracelen image when moveing motors to new positions
try:
tracelen=self._goTrace._tracelen
except AttributeError:
app=QApplication.instance()
cfg=app._cfg
tracelen=self._goTrace._tracelen=cfg.value(AppCfg.GBL_MISC)['img_trace_len']
if tracelen==0:
return
trImg=self._goImgGrp.transform()
trTrk=self._goTracked.transform()
tr=self._goImg.transform()*trImg*trTrk.inverted()[0] # TODO: check if oder correct !!!
cm=pg.ColorMap((0,255),((0,0,64),(255,255,128)))#,opacity=.5)
img=pg.ImageItem(self._goImg.image,lut=cm.getLookupTable(0,255))
img.setTransform(tr)
self._goTrace.addItem(img) # automatically added to self.vb
cldLst=self._goTrace.childItems()
if len(cldLst)>tracelen: #check if one image needs to be removed
r1=self.vb.itemBoundingRect(img)
for c in cldLst[:-1]:
#check overlap with newest image, and remove previous if it is more than 30%
r2=self.vb.itemBoundingRect(c)
r3=r1.intersected(r2)
s3=r3.size()
s1=r1.size()
if s3.width()/s1.width()>0.3 and s3.height()/s1.height()>0.3:
self.vb.removeItem(c)
return
self.vb.removeItem(cldLst[0]) #if no one was removes, remove oldest
def cb_modify_app_param(self):
wnd=getattr(self, '_wnd_param', None)
if wnd is None:
self._wnd_param=wnd=WndParameter(self)
else:
wnd.refresh_cam_param() # the camera parameters may have changed since the window was built
wnd.show()
wnd.raise_()
wnd.activateWindow()
# --- editable DAQ fields on the main page -----------------------------------------------------------
# (widget, settings key, field). The values are stored in the same settings as the preferences window.
def build_daq_fields(self):
lay=self._group_data_storage.layout()
alnum=QRegularExpressionValidator(QRegularExpression(r'^[A-Za-z0-9_]*$')) # used in file/directory names
self._daq_fields=fld=[]
def line_edit(key, field):
w=QLineEdit()
w.setValidator(alnum)
w.textEdited.connect(lambda txt, w=w: self.cb_daq_field_changed(w)) # saved at every change (no Enter needed)
fld.append((w, key, field))
return w
# data storage: p-group, protein, prefix (editable), last run, last path (read only)
for w in (self.label_3, self._le_prefix, self.label_4, self._le_runnumber, self.label_dp, self._le_datapath):
lay.removeWidget(w)
self._le_pgroup=line_edit(AppCfg.DAQ_LOC, 'p_group')
self._le_protein=line_edit(AppCfg.DAQ_SAMPLE, 'protein')
self._le_prefix.setValidator(alnum)
self._le_prefix.textEdited.connect(lambda txt: self.cb_daq_field_changed(self._le_prefix))
fld.append((self._le_prefix, AppCfg.DAQ_RUN, 'prefix'))
for lbl,txt in ((self.label_3,'prefix'), (self.label_4,'last run'), (self.label_dp,'last path')):
lbl.setText(txt)
rows=((QLabel('pgroup'), self._le_pgroup), (QLabel('protein'), self._le_protein), (self.label_3, self._le_prefix),
(self.label_4, self._le_runnumber), (self.label_dp, self._le_datapath))
for r,(lbl,w) in enumerate(rows):
lbl.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
lay.addWidget(lbl, r, 0)
lay.addWidget(w, r, 1)
# experimental parameters
lay=self._group_experiment.layout()
def spin(key, field, lo, hi, dec=None):
if dec is None:
w=QSpinBox()
else:
w=QDoubleSpinBox()
w.setDecimals(dec)
w.setRange(lo, hi)
w.valueChanged.connect(lambda v, w=w: self.cb_daq_field_changed(w)) # typing or arrows, no Enter needed
fld.append((w, key, field))
return w
self._sb_energy=spin(AppCfg.DAQ_LOC, 'incident_energy_kev', 0, 100, 3)
self._sb_energy.setSuffix(' keV')
self._sb_evt_trig=spin(AppCfg.DAQ_LOC, 'event_trigger', 1, 256)
self._cb_triggered=cb=QCheckBox()
cb.clicked.connect(lambda: self.cb_daq_field_changed(cb))
fld.append((cb, AppCfg.DAQ_RUN, 'triggered'))
r=0
for txt,w in (('incident energy', self._sb_energy), ('labelling event', self._sb_evt_trig), ('triggered', cb)):
lbl=QLabel(txt)
lbl.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
lay.addWidget(lbl, r, 0)
lay.addWidget(w, r, 1, 1, 3)
r+=1
lbl=QLabel('unit cell')
lbl.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
lay.addWidget(lbl, r, 0)
for i,(txt,lo,hi) in enumerate((('a',0,10000),('b',0,10000),('c',0,10000),('alpha',0,180),('beta',0,180),('gamma',0,180))):
w=spin(AppCfg.DAQ_UC, txt, lo, hi, 3)
w.setPrefix({'alpha':'α ','beta':'β ','gamma':'γ '}.get(txt, f'{txt} '))
lay.addWidget(w, r+i//3, 1+i%3)
r+=2
lbl=QLabel('space group')
lbl.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
lay.addWidget(lbl, r, 0)
lay.addWidget(spin(AppCfg.DAQ_UC, 'space_group_number', 1, 230), r, 1)
self.refresh_daq_fields()
def refresh_daq_fields(self):
'''load the DAQ fields from the settings (e.g. after a change in the preferences window)'''
cfg=QApplication.instance()._cfg
focus=QApplication.focusWidget()
for w, key, field in self._daq_fields:
if w is focus: # do not overwrite a field the operator is editing
continue
v=(cfg.value(key) or {}).get(field)
if v is None:
continue
w.blockSignals(True)
if isinstance(w, QLineEdit):
w.setText(str(v))
elif isinstance(w, QCheckBox):
w.setChecked(bool(v))
elif isinstance(w, QDoubleSpinBox):
w.setValue(float(v))
else:
w.setValue(int(v))
w.blockSignals(False)
def cb_daq_field_changed(self, w):
'''a DAQ field on the main page was edited: save it and update the preferences window'''
app=QApplication.instance()
cfg=app._cfg
for ww, key, field in self._daq_fields:
if ww is w:
break
else:
return
if isinstance(w, QLineEdit):
v=w.text()
elif isinstance(w, QCheckBox):
v=w.isChecked()
else:
v=w.value()
d=cfg.value(key) or {}
if d.get(field)==v:
return
d[field]=v
cfg.setValue(key, d)
cfg.sync()
_log.debug(f'{key}/{field} = {v!r}')
if key==AppCfg.DAQ_LOC:
try:
del app._jungfrau # force a new jungfrau object with the new p-group/energy/trigger
except AttributeError:
pass
wnd=getattr(self, '_wnd_param', None)
if wnd is not None:
wnd.set_daq_value(key, field, v)
# --- chip type / motion on the main page (experimental parameters) -----------------------------
# The Fix Target parameter text box stays the single source: these fields read and rewrite only
# their keys in it (mode, tmove, twait, ssz). tmove is fixed to one FEL pulse (fix_motion_param).
# Text box and chip type are kept in the settings.
MOTION_MODES=((4,'4 continuous'), (5,'5 stop-and-go'), (6,'6 hit-and-return'), (7,'7 stop-and-go hit-and-return'))
MOTION_ALLOWED=tuple(m for m,_ in MOTION_MODES)
def build_motion_fields(self):
app=QApplication.instance()
cfg=app._cfg
mft=self._moduleFixTarget
lay=self._group_experiment.layout()
r=lay.rowCount()
def label(txt):
lbl=QLabel(txt)
lbl.setAlignment(Qt.AlignRight|Qt.AlignVCenter)
return lbl
self._cb_chip=cb=QComboBox()
cb.addItems([mft._cbType.itemText(i) for i in range(mft._cbType.count())])
cb.setToolTip('object type added by "add obj" in the Fix Target tab (same selection)')
cb.currentIndexChanged.connect(mft._cbType.setCurrentIndex)
mft._cbType.currentIndexChanged.connect(cb.setCurrentIndex)
mft._cbType.currentIndexChanged.connect(self.cb_motion_save)
mft._cbType.currentIndexChanged.connect(self.update_motion_delay)
self._cb_mode=cb=QComboBox()
for m,txt in self.MOTION_MODES:
cb.addItem(txt, m)
cb.setToolTip('motion program (mode) of the Fix Target parameters')
cb.currentIndexChanged.connect(self.cb_motion_field_changed)
fel_per=int(cfg.value(AppCfg.DT_MISC)['fel_per'])
self._sb_twait=w=QSpinBox() # ms, snapped to a multiple of fel_per (motion_values)
w.setRange(fel_per, 100000)
w.setSingleStep(fel_per)
w.setValue(fel_per)
w.setSuffix(' ms')
w.setKeyboardTracking(False) # valueChanged on enter/focus out, not at each typed digit
w.setToolTip('twait: wait at each aperture in ms, a multiple of the FEL period\n'
'tmove is fixed to one FEL period')
w.valueChanged.connect(self.cb_motion_field_changed)
def spin(pfx):
w=QSpinBox()
w.setRange(1, 1000)
w.setPrefix(pfx)
w.setValue(2)
w.valueChanged.connect(self.cb_motion_field_changed)
return w
self._sb_ssz=(spin('x '), spin('y '))
self._le_laser_delay=QLineEdit()
self._le_laser_delay.setReadOnly(True)
self._le_laser_delay.setToolTip(f'{psi_device.LASER_DELAY_PV} (ns, negative: laser before the X-rays)')
self._le_delay=QLineEdit()
self._le_delay.setReadOnly(True)
self._le_runtime=QLineEdit()
self._le_runtime.setReadOnly(True)
self._motion_rows={}
for key,txt,ws in (('chip', 'chip type', (self._cb_chip,)), ('mode', 'motion', (self._cb_mode,)),
('t', 'wait', (self._sb_twait,)), ('ssz', 'section size', self._sb_ssz),
('laser', 'laser delay', (self._le_laser_delay,)), ('delay', 'delay', (self._le_delay,)),
('runtime', 'run time', (self._le_runtime,))):
lbl=label(txt)
lay.addWidget(lbl, r, 0)
if len(ws)==1:
lay.addWidget(ws[0], r, 1, 1, 3)
else:
for i,w in enumerate(ws):
lay.addWidget(w, r, 1+i)
self._motion_rows[key]=(lbl,)+tuple(ws)
r+=1
try:
self._pv_laser_delay=epics.PV(psi_device.LASER_DELAY_PV, auto_monitor=True)
except Exception as e:
_log.warning(f'laser delay PV {psi_device.LASER_DELAY_PV}: {e}')
self._pv_laser_delay=None
self._motion_param=None # last valid parse of the text box
mft._txtParam.textChanged.connect(self.cb_motion_text_changed)
d=cfg.value(AppCfg.DAQ_MOTION) or {}
i=mft._cbType.findText(d.get('chip_type', ''))
if i>=0:
mft._cbType.setCurrentIndex(i)
mft._txtParam.setText(d.get('param', ''))
self.cb_motion_text_changed()
p=self._motion_param
if p is not None and ('mode' not in p or self.fix_motion_param(dict(p))!=p): # explicit mode, fixed tmove
self.cb_motion_field_changed()
self._tmr_delay=t=QTimer(self) # polls the monitored PV value in the GUI thread (fel_per may change too)
t.timeout.connect(self.update_motion_delay)
t.start(1000)
def fix_motion_param(self, param):
'tmove fixed to one FEL pulse for mode 5/7: at add obj, update param and the collection'
tmove=psi_device.motion_tmove(param.get('mode', 1), QApplication.instance()._cfg.value(AppCfg.DT_MISC)['fel_per'])
if tmove is not None:
if param.get('tmove', tmove)!=tmove:
_log.warning(f"tmove {param['tmove']} replaced by the fixed {tmove} ms")
param['tmove']=tmove
return param
def motion_values(self):
'mode and its motion parameters as shown in the fields (with the fixed tmove)'
mode=self._cb_mode.currentData()
p={'mode':mode}
if mode in (5, 7):
p['tmove']=psi_device.motion_tmove(mode, QApplication.instance()._cfg.value(AppCfg.DT_MISC)['fel_per'])
p['twait']=self._sb_twait.value() # ms
if mode in (6, 7):
p['ssz']=[w.value() for w in self._sb_ssz]
return p
def cb_motion_text_changed(self, *args):
'the parameter text box changed: update the motion fields (left unchanged while the text is not valid)'
try:
p=ModuleFixTarget.parse_param(self._moduleFixTarget._txtParam.text())
except Exception:
return
self._motion_param=p
ws=(self._cb_mode, self._sb_twait)+self._sb_ssz
for w in ws:
w.blockSignals(True)
self._cb_mode.setCurrentIndex(self._cb_mode.findData(p.get('mode', 1))) # 1: default of get_scan_param, not allowed -> blank
twait=p.get('twait')
if isinstance(twait, (int, float)) and p.get('mode') in (5, 7):
self._sb_twait.setValue(int(twait))
ssz=p.get('ssz')
if isinstance(ssz, list) and len(ssz)==2 and all(isinstance(v, int) for v in ssz):
for w,v in zip(self._sb_ssz, ssz):
w.setValue(v)
for w in ws:
w.blockSignals(False)
self.show_motion_fields()
self.cb_motion_save()
self.update_motion_delay()
def cb_motion_field_changed(self, *args):
'a motion field changed: rewrite its keys in the parameter text box (the other keys are kept)'
txt=self._moduleFixTarget._txtParam
try:
p=ModuleFixTarget.parse_param(txt.text())
except Exception as e:
QMessageBox.warning(self, 'motion parameters', f'Fix the parameter text first:\n{txt.text()}\n{e}')
self.cb_motion_text_changed() # back to the values of the text
return
for k in ('tmove', 'twait', 'ssz'):
p.pop(k, None)
if self._cb_mode.currentIndex()<0:
self._cb_mode.setCurrentIndex(0)
w=self._sb_twait # only multiples of fel_per
fel_per=QApplication.instance()._cfg.value(AppCfg.DT_MISC)['fel_per']
w.blockSignals(True)
w.setValue(int(max(1, int(w.value()/fel_per+.5))*fel_per))
w.blockSignals(False)
p.update(self.motion_values())
txt.setText(ModuleFixTarget.format_param(p)) # -> cb_motion_text_changed
def show_motion_fields(self):
mode=self._cb_mode.currentData()
for key,vis in (('t', mode in (5, 7)), ('ssz', mode in (6, 7))):
for w in self._motion_rows[key]:
w.setVisible(vis)
def cb_motion_save(self, *args):
mft=self._moduleFixTarget
QApplication.instance()._cfg.setValue(AppCfg.DAQ_MOTION, {'chip_type':mft._cbType.currentText(), 'param':mft._txtParam.text()})
def chip_count(self):
'aperture grid count (nx, ny) of the selected chip type, None if it has no grid'
txt=self._cb_chip.currentText()
if txt.startswith('FixTarget('):
tpl=UsrGO.FixTargetFrame.tpl.get(txt[len('FixTarget('):-1])
return tpl['grid']['count'] if tpl else None
if txt=='Grid()':
cnt=(self._motion_param or {}).get('cnt', (30, 22)) # default of module_fix_target_add_obj
return cnt if isinstance(cnt, (list, tuple)) and len(cnt)==2 else None
return None
def update_motion_delay(self, *args):
'''laser delay readback, calculated laser to X-ray delay (same calculation as the JFJ header)
and estimated run time of the selected chip'''
fel_per=QApplication.instance()._cfg.value(AppCfg.DT_MISC)['fel_per']
pv=self._pv_laser_delay
laser=pv.value if pv is not None and pv.connected else None
try:
laser=float(laser)
except (TypeError, ValueError):
laser=None
self._le_laser_delay.setText('n/a (not connected)' if laser is None else f'{laser:.3f} ns')
def show(le, txt, warn, tip):
le.setText(txt)
le.setStyleSheet('background-color: #ffd27f' if warn else '')
le.setToolTip('\n'.join(tip+warn))
mode=self._cb_mode.currentData()
if mode is None: # mode of the text box not allowed
m=(self._motion_param or {}).get('mode', 1)
warn=[f'motion mode {m} is not allowed: select one of {", ".join(map(str, self.MOTION_ALLOWED))}']
show(self._le_delay, f'mode {m} not allowed', warn, [])
show(self._le_runtime, 'n/a', warn, [])
return
p=self.motion_values()
warn=[]
if mode in (5, 7): # as typed in the text box
tmove=p['tmove']
txt=self._motion_param or {}
if txt.get('tmove')!=tmove:
warn.append(f'tmove {txt.get("tmove")} in the text: it is fixed to {tmove:g} ms (set when the object is added)')
twait=txt.get('twait')
if isinstance(twait, (int, float)) and twait%tmove:
warn.append(f'twait {twait:g} is not a multiple of {tmove:g}: the CTA uses twait//tmove = {int(twait//tmove)} pulses')
n=psi_device.motion_delay_pulses(mode, p)
coarse=f'{n} x {fel_per:g} ms'
if laser is None:
txt=f'{-n*fel_per:g} ms + laser ({coarse})'
else:
txt=f'{psi_device.total_delay_ns(n, fel_per, laser)/1e6:.6f} ms ({coarse}, laser {laser:g} ns)'
show(self._le_delay, txt, warn, ['laser to X-ray delay, negative: laser before the X-rays',
'trigger to X-ray pulses x fel_per + laser delay (recorded in the JFJ header)'])
cnt=self.chip_count()
tip=['estimated beam time of one chip: FEL pulses of the CTA sequence x fel_per',
'(without detector/motion setup and the start of the collection)']
if cnt is None:
show(self._le_runtime, 'n/a (no aperture grid)', [], tip)
return
pulses, warn=psi_device.motion_run_pulses(mode, p, cnt)
if pulses is None:
show(self._le_runtime, 'collection would abort', warn, tip)
return
sec=pulses*fel_per/1000
h, rem=divmod(int(round(sec)), 3600)
hms=(f'{h} h ' if h else '')+f'{rem//60} min {rem%60:02d} s'
show(self._le_runtime, f'~ {hms} ({cnt[0]*cnt[1]} apertures, {pulses} pulses)', [], tip)
@pyqtSlot()
def cb_save_cam_image(self,overlays=False):
app=QApplication.instance()
cam=app._camera
filename, _ = QFileDialog.getSaveFileName(self,"Save data file",'', 'PNG files (*.png);;all files (*)')
if not filename:
return
_log.info(f"saving view screenshot: {filename}")
#try:
if not overlays:
import PIL.Image
#img=PIL.Image.fromarray(cam.pic.astype(np.uint8))
try:
pic=cam._pic
mx=pic.max()
if pic.max()>255:
scl=2**int(round(np.log2(mx)-8))
pic=np.array(pic/scl,dtype=np.uint8)
elif pic.dtype!=np.uint8:
pic=np.array(pic, dtype=np.uint8)
except AttributeError:
sim=app._camera._sim
pic=cam._sim['imgSeq'][sim['imgIdx']]
img=PIL.Image.fromarray(pic)
img.save(filename)
else:
exporter = pg.exporters.ImageExporter(self.vb)
# set export parameters if needed
#exporter.parameters()["width"] = 2000 # (note this also affects height parameter)
# save to file
exporter.export(filename)
#except Exception as e:
# _log.warning(e)
# QMessageBox.warning(self, "Screenshot: failed to save image", "Failed to save screenshot!")
def cb_execute_collection(self):
app=QApplication.instance()
geo=app._geometry
#zoom=app._zoom.get_val()
task = self.active_task()
#self._is_aborted = False
#method = self._tabs_daq_methods.currentWidget().accessibleName()
if task != TASK_FIX_TARGET: # the other tasks are not implemented (yet)
QMessageBox.warning(self, "data collection", f"Data collection is only available in the fixed target task (current task: {task}).")
return
if task == TASK_FIX_TARGET:
#self.re_connect_collect_button(callback=self.collect_abort_grid, accessibleName="grid_abort", label="Abort Grid Scan",)
#self._inspect = self._grid_inspect_area
#self._inspect.setPlainText("")
fast_x=self.tweakers["fast_x"]
fast_y=self.tweakers["fast_y"]
fx=fast_x.get_val()
fy=fast_y.get_val()
opt_ctr=geo._opt_ctr
grp=self._goTracked
cldLst=grp.childItems()
jobs=[]
for go in cldLst:
if type(go)==UsrGO.Fiducial:
continue
t=type(go)
if t not in(UsrGO.FixTargetFrame,UsrGO.Path,UsrGO.Grid):
_log.warning(f'{t} not supported for FixTargetFrame ->skipped:{go}')
continue
try:
param=go.get_scan_param() #points in coordinate system of ROI
except AttributeError as e:
_log.warning(f'no scan parameters for object->skipped:{go}')
continue
param['chip_type']=getattr(go, '_chip_type', t.__name__) # recorded in the JFJ header
mode=param.get('mode', 1)
if mode not in self.MOTION_ALLOWED:
QMessageBox.warning(self, "data collection", f"Motion mode {mode} of {param['chip_type']} is not allowed "
f"(only {', '.join(map(str, self.MOTION_ALLOWED))}).\nUpdate the object parameters (mode:...).")
return
self.fix_motion_param(param)
vb=self.vb
grp=self._goTracked
mft=self._moduleFixTarget
try:
tmpGoLst=self._goTmp
except AttributeError:
tmpGoLst=self._goTmp=[]
else:
for go in tmpGoLst:
vb.removeItem(go)
tmpGoLst.clear()
# adding debug fiducials
#n=int(p.shape[0]/100)+1
#for i in range(0,p.shape[0],n):
# fx,fy=p[i, :]/1000
# fx=-fx #X axis has inverted sign !
# l=.06
# go=UsrGO.Fiducial((fx-l/2, fy-l/2), (l, l), i)
# grp.addItem(go)
# tmpGoLst.append(go)
#mft._tree.setData(grp.childItems())
#_log.debug(f'first point:{p[0,:]}')
#_log.debug(f'step to 2nd point:{p[1,:]-p[0,:]}')
#print('DEBUG: difference from point to point:')
#print(np.diff(p, axis=0))
jobs.append(param)
if jobs:
if not self.confirm_collection_positions():
return
self.daq_collect(jobs)
else:
QMessageBox.warning(self, "data collection", "Nothing to collect: add a fixed target, grid or path first.")
def cb_esc_sample_exchange(self):
app=QApplication.instance()
steps = []
#if option(CRYOJET_MOTION_ENABLED):
# steps.append(lambda: self.move_cryojet_nozzle("out"))
steps=[
('detector out', lambda: self.move_detector("out")),
('post sample tube out', lambda: self.move_post_tube("out")),
('backlight out', lambda: self.move_backlight("out")),
('collimator out', lambda: self.move_collimator("out")),
('goniometer mount', lambda: self.move_gonio("mount")),
]
self.esc_run_steps(steps, "Transitioning to Sample Exchange","ManualSampleExchange")
def cb_esc_sample_alignment(self):
app=QApplication.instance()
steps = [
# lambda: sample_selection.tell.set_current(30.0),
('detector out', lambda: self.move_detector("out")),
('collimator out', lambda: self.move_collimator("out")),
('post sample tube out', lambda: self.move_post_tube("out")),
('backlight in', lambda: self.move_backlight("in", check_detector=True)),
('goniometer align', lambda: self.move_gonio("align")),
]
#if option(CRYOJET_MOTION_ENABLED):
# steps.extend([lambda: self.move_cryojet_nozzle("in")])
#steps.extend([lambda: self.move_post_tube("out"), lambda: app._backlight.move("in")])
self.esc_run_steps(steps, "Transitioning to Sample Alignment","SampleAlignment")
def cb_esc_data_collection(self):
app=QApplication.instance()
steps = [
# lambda: sample_selection.tell.set_current(30.0),
('detector out', lambda: self.move_detector("out")),
('backlight out', lambda: self.move_backlight("out")),
('post sample tube in', lambda: self.move_post_tube("in")),
('collimator in', lambda: self.move_collimator("in")),
('detector in', lambda: self.move_detector("in")),
]
#if option(CRYOJET_MOTION_ENABLED):
# steps.extend([lambda: self.move_cryojet_nozzle("in")])
self.esc_run_steps(steps, "Transitioning to Data Collection","DataCollection")
def cb_really_quit(self):
"""called when user Ctrl-Q the app"""
if QMessageBox.question(self, "", "Are you sure you want to quit?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No,) == QMessageBox.Yes:
self._do_quit = True
self.close()
def cb_deltatau_home_faststages(self):
self.deltatau_home_plc(1, 'fast stages (fast X and Y)')
def cb_deltatau_home_coarse_y(self):
self.deltatau_home_plc(2, 'coarse Y (base Y)')
def deltatau_home_plc(self, plc, what):
'''start a homing PLC on the PowerBrick (MX1_home.cfg: PLC 1 = fast X/Y, PLC 2 = coarse Y)'''
app=QApplication.instance()
if QMessageBox.question(self, "homing", f"Home the {what}?\nThe stage moves to its end stop and back to the index.",
QMessageBox.Yes|QMessageBox.No, QMessageBox.No)!=QMessageBox.Yes:
return
if app._args.sim&0x10:
_log.info(f"homing {what}: simulated motors -> nothing done")
return
pfx=app._cfg.value(AppCfg.GBL_DEV_PREFIX)[0]
pvname=f"{pfx}:asyn1.AOUT"
_log.info(f"homing {what}: {pvname} <- enable plc {plc}")
try:
ret=epics.PV(pvname).put(f"enable plc {plc}", wait=True, timeout=2.0)
except Exception as e:
_log.error(f"homing {what}: {e}")
ret=None
if ret is None or ret<0: # pyepics: negative on timeout, None if not connected
_log.error(f"homing {what}: failed to write {pvname}")
QMessageBox.warning(self, "homing", f"Failed to start homing of the {what} ({pvname}).")
def cb_about(self):
try:
ver,gitcmt=get_version()
v_SMX=f'{ver} git:{gitcmt}'
except:
v_SMX='git version failed'
try:
ver,gitcmt=get_version('..')
v_SMXT=f'{ver} git:{gitcmt}'
except:
v_SMXT='git version failed'
txt=f'''About Swissmx:
SwissMX: {v_SMX}
SwissMXTools: {v_SMXT}
qt:{QT_VERSION_STR}
pyqtgraph:{pg.__version__}
numpy:{np.__version__}
matplotlib:{mpl.__version__}
epics:{epics.__version__}
Copyright (c) 2022 by Paul Scherrer Institute
(http://www.psi.ch)
Based on Zac great first implementation
Author Thierry Zamofing (thierry.zamofing@psi.ch)
'''
QMessageBox.about(self, "SwissMX", txt)
pass
def cb_autofocus(self):
app=QApplication.instance()
cfg=app._cfg.value(AppCfg.GEO_AUTOFOC)
geo=app._geometry
#geo.autofocus(app._camera, self.tweakers['base_z'],rng=(-1, 1), n=30,saveImg=True)
af_range=cfg['range']
n=cfg['steps']
vel=cfg['velocity']
rng=(-af_range/2, af_range/2)
#geo.autofocus(app._camera, self.tweakers['base_z'],rng, n,dlg)
af=geometry.autofocus(app._camera, self.tweakers['base_z'])
# af.run(rng, n, dlg)
#with pg.ProgressDialog('Progress', 0, n) as dlg:
af.run_continous(rng=rng, velo=vel)
def cb_find_fiducial(self):
app=QApplication.instance()
geo=app._geometry
cfg=app._cfg.value(AppCfg.GEO_FND_FID)
scl=np.linalg.norm(geo.pos2pix((.001,0))) #pix/um (calibration pixels)
scl/=self._goImg.transform().m11() # -> camera image pixels (binning)
sz=int(cfg['sz']*scl)
brd=int(cfg['brd']*scl)
# sz=( 84, 84); brd=( 7, 7) # zoom 001
# sz=(128, 128); brd=( 9, 9) # zoom 200
# sz=(200, 200); brd=(13, 13) # zoom 400
_log.debug(f'sz:{sz} brd:{brd}')
pos,corr=geo.find_fiducial(app._camera, sz=(sz,sz),brd=(brd,brd))
bz=self.tweakers["base_z"].get_rbv()
pos_scn=self._goImg.mapToScene(pg.Point(pos))
pos_eu=self._goTracked.mapFromScene(pos_scn)
l=.120
#go=UsrGO.Fiducial((fx-l/2, fy-l/2), (l, l), bz)
go=UsrGO.Fiducial((pos_eu.x()-l/2, pos_eu.y()-l/2), (l, l), bz)
go.sigRegionChangeFinished.connect(self.cb_fiducial_update_z)
grp=self._goTracked
grp.addItem(go)
self._moduleFixTarget._tree.setData(grp.childItems())
pass
def cb_move_yag(self):
self.run_moves('goniometer yag', [('goniometer yag', lambda: self.move_gonio("yag"))])
def cb_move_home(self):
self.run_moves('fast stages home', [('fast stages home (0,0)', self.move_fast_home)])
def move_fast_home(self):
_log.info("moving fast stages to home (0,0)")
tw_fx = self.tweakers["fast_x"]
tw_fy = self.tweakers["fast_y"]
tw_fx.move_abs(0.0)
tw_fy.move_abs(0.0)
self.assert_positions([(tw_fx, 0.0, 0.1), (tw_fy, 0.0, 0.1)], 'fast stages home')
def cb_set_daq_prefix(self, prefix):
app=QApplication.instance()
cfg=app._cfg
daq_run=cfg.value(AppCfg.DAQ_RUN)
daq_run['prefix']=prefix
cfg.setValue(AppCfg.DAQ_RUN, daq_run)
self._le_prefix.setText(prefix)
wnd=getattr(self, '_wnd_param', None)
if wnd is not None:
wnd.set_daq_value(AppCfg.DAQ_RUN, 'prefix', prefix)
def cb_update_run_number(self):
app=QApplication.instance()
try:
run_number=app._jungfrau._daq.client.run_number
except AttributeError:
self._le_runnumber.setText('')
return
self._le_runnumber.setText(str(run_number))
def cb_update_data_path(self):
app=QApplication.instance()
try:
path=app._jungfrau.jfj_file_prefix
except AttributeError:
self._le_datapath.setText('')
return
self._le_datapath.setText('/'+path)
def cb_testcode(self):
app=QApplication.instance()
cfg=app._cfg
geo=app._geometry
try:
tc=self._testCode
tc['idx']+=1
except AttributeError:
self._testCode=tc={'idx':0}
step=tc['idx']
def testMatplotlib():
import matplotlib.pyplot as plt
x=np.linspace(0.1, 2*np.pi, 41)
y=np.exp(np.sin(x))
plt.stem(x, y)
plt.show(block=False)
step=6
if step==0:
vb=self.vb
vb.autoRange(items=(self._goImg,))
elif step==1:
testMatplotlib()
elif step==2:
vb=self.vb
grp=pg.ItemGroup()
vb.addItem(grp)
obj=UsrGO.Marker((100, 100), (100, 100), mode=1)
grp.addItem(obj)
obj=UsrGO.Marker((150, 100), (50, 50), mode=1)
grp.addItem(obj)
obj=UsrGO.Marker((200, 100), (100, 100), mode=1)
grp.addItem(obj)
tc['grp']=grp
vb.autoRange(items=(obj,))
elif step==3:
grp=tc['grp']
tr=grp.transform()
# UsrGO.obj_info(tr)
tr.setMatrix(1, .2, 0,
-.2, 1, 0,
0, 0, 1)
grp.setTransform(tr)
elif step==4:
geo.autofocus(app._camera, self.tweakers['base_z'],rng=(-1, 1), n=30,saveImg=True)
elif step==5:
geo.autofocus(app._camera, self.tweakers['base_z'],rng=(-1, 1), n=10)
elif step==6:
geo.find_fiducial(app._camera)
#print(vb.childGroup.childItems())
pass
def prepare_wd_tabs_left(self):
tabs = self._wd_tabs_left
tabs.currentChanged.connect(self.cb_switch_task)
setup_tab = self._tab_setup
exp_tab = self._tab_experiment
# add layouts
#exp_tab.setLayout(QVBoxLayout())
setup_tab.setLayout(QVBoxLayout())
setup_tab.setSizePolicy(QSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding))
self._setup_toolbox = tbox = QToolBox()
tbox.setSizePolicy(QSizePolicy(QSizePolicy.Expanding, QSizePolicy.Ignored))
# tbox.setMinimumWidth(itemWidget.sizeHint().width())
setup_tab.layout().addWidget(tbox)
# -------- DAQ Methods Tabs --------
self.build_tab_module_fix_target()
#self._OLD_build_daq_methods_grid_tab()
#self._OLD_build_daq_methods_prelocated_tab() #ZAC: orig. code
#self._OLD_create_helical_widgets() #ZAC: orig. code
#self._OLD_build_daq_methods_embl_tab()
#self._OLD_build_sample_selection_tab()
# -------- group geometry calibration --------
block = QWidget()
block.setAccessibleName(TASK_SETUP_GEOMETRY_CALIB)
block.setContentsMargins(0, 0, 0, 0)
block.setLayout(QVBoxLayout())
grp = QWidget()
block.layout().addWidget(grp)
block.layout().addStretch()
grp.setLayout(QGridLayout())
tbox.addItem(block, "geometry calibration")
#grp.layout().addWidget(QLabel("feature size (µM)"), 0, 0)
#self._ppm_feature_size_spinbox = QDoubleSpinBox()
#self._ppm_feature_size_spinbox.setRange(5, 10000)
#self._ppm_feature_size_spinbox.setValue(50)
#self._ppm_feature_size_spinbox.setDecimals(0)
#self._ppm_feature_size_spinbox.setSuffix(" µM")
#grp.layout().addWidget(self._ppm_feature_size_spinbox, 0, 1)
self._btn_pix2pos = but = QPushButton("Pixel to position calibration")
but.setCheckable(True)
but.clicked.connect(self.cb_update_pix2pos)
grp.layout().addWidget(but, 0, 0)
self._btn_opt_ctr = but = QPushButton("Optical center calibration")
but.setCheckable(True)
but.clicked.connect(self.cb_update_opt_ctr)
grp.layout().addWidget(but, 1, 0)
self._wd_calib_tree=tree=pg.DataTreeWidget(data=None)
grp.layout().addWidget(tree, 2, 0)
tbox.currentChanged.connect(self.cb_switch_task)
# final stretch
# setup_tab.layout().addStretch()
exp_tab.layout().addStretch()
# Jungfrau parameters
block=QWidget()
block.setAccessibleName(TASK_JUNGFRAU_SETTINGS)
block.setContentsMargins(0, 0, 0, 0)
block.setLayout(QVBoxLayout())
# self.jungfrau = jungfrau_widget.JungfrauWidget() #ZAC: orig. code
# block.layout().addWidget(self.jungfrau) #ZAC: orig. code
block.layout().addStretch()
tbox.addItem(block, "Jungfrau Parameters")
# # -------- group beam box --------
# block = QWidget()
# block.setAccessibleName(TASK_SETUP_BEAM_CENTER)
# block.setContentsMargins(0, 0, 0, 0)
# block.setLayout(QVBoxLayout())
# grp = QWidget()
# block.layout().addWidget(grp)
# block.layout().addStretch()
# grp.setLayout(QGridLayout())
# tbox.addItem(block, "Beam Box")
# # setup_tab.layout().addWidget(grp)
#
# row = grp.layout().rowCount
#
# if self._pv_shutter is not None:
# but = QPushButton("open/close shutter")
# but.setCheckable(True)
# but.setChecked(1 == self._pv_shutter.get())
# but.clicked.connect(self.toggle_shutter)
# self._button_shutter = but
# grp.layout().addWidget(but, 0, 0, 1, 2)
# self._pv_shutter.add_callback(self.update_shutter_label)
# #else: #ZAC: orig. code
# # self._picker = bernina_pulse_picker.PickerWidget()
# # grp.layout().addWidget(self._picker, 0, 0, 1, 2)
# but = QPushButton("clear beam markers")
# but.clicked.connect(self.remove_beam_markers)
# grp.layout().addWidget(but, 1, 1)
# help = QTextBrowser()
# grp.layout().addWidget(help, 2, 0, 5, 2)
# help.setHtml(
# """<body>
# <h1>Updating Beam Mark</h1>
# <ol>
# <li>clear beam markers</li>
# <li>for each zoom level, CTRL-Click (left mouse button) the center of the beam</li>
# </ol>
# </body>
# """
# )
# # -------- group regions --------
# block = QWidget()
# block.setAccessibleName(TASK_SETUP_ROI)
# block.setContentsMargins(0, 0, 0, 0)
# block.setLayout(QVBoxLayout())
# grp = QWidget()
# block.layout().addWidget(grp)
# block.layout().addStretch()
# grp.setLayout(QGridLayout())
# tbox.addItem(block, "Regions of Interest")
# but = QPushButton("add RECT roi")
# but.clicked.connect(self.roi_add_rect)
# grp.layout().addWidget(but, 0, 0)
# but = QPushButton("add LINE roi")
# but.clicked.connect(self.roi_add_line)
# grp.layout().addWidget(but, 0, 1)
#
# # -------- group camera settings --------
# block = QWidget()
# block.setAccessibleName(TASK_SETUP_CAMERA)
# block.setContentsMargins(0, 0, 0, 0)
# block.setLayout(QVBoxLayout())
# grp = QWidget()
# block.layout().addWidget(grp)
# block.layout().addStretch()
# grp.setLayout(QGridLayout())
# tbox.addItem(block, "Camera Settings")
# but = QPushButton("remove transformations")
# but.clicked.connect(lambda bogus: self.modify_camera_transform("remove_all"))
# grp.layout().addWidget(but, 0, 0)
# but = QPushButton("remove last")
# but.clicked.connect(lambda bogus: self.modify_camera_transform("undo_last"))
# grp.layout().addWidget(but, 0, 1)
# but = QPushButton("Turn 90 CCW")
# #but.clicked.connect(
# # lambda bogus, n=camera.Transforms.ROTATE_90: self.modify_camera_transform(n)
# #) #ZAC: orig. code
# grp.layout().addWidget(but, 1, 0)
# but = QPushButton("Turn 90 CW")
# #but.clicked.connect(lambda bogus, n=camera.Transforms.ROTATE_270: self.modify_camera_transform(n)) #ZAC: orig. code
# grp.layout().addWidget(but)
# but = QPushButton("Turn 180 CCW")
# #but.clicked.connect(lambda bogus, n=camera.Transforms.ROTATE_180: self.modify_camera_transform(n)) #ZAC: orig. code
# grp.layout().addWidget(but)
# but = QPushButton("Transpose")
# #but.clicked.connect(lambda bogus, n=camera.Transforms.TRANSPOSE: self.modify_camera_transform(n)) #ZAC: orig. code
# grp.layout().addWidget(but)
# but = QPushButton("Flip L/R")
# #but.clicked.connect(lambda bogus, n=camera.Transforms.FLIP_LR: self.modify_camera_transform(n)) #ZAC: orig. code
# grp.layout().addWidget(but)
# but = QPushButton("Flip U/D")
# #but.clicked.connect(lambda bogus, n=camera.Transforms.FLIP_UD: self.modify_camera_transform(n)) #ZAC: orig. code
# grp.layout().addWidget(but)
# row = grp.layout().rowCount()
# self._label_transforms = QLabel()
# self._label_transforms.setWordWrap(True)
# self.modify_camera_transform(None)
#
# grp.layout().addWidget(self._label_transforms, row, 0, 1, 2)
def build_tab_module_fix_target(self):
#tab = self._tab_daq_method_prelocated
try:
objLst=self._goTracked.childItems()
except AttributeError:
_log.debug('NOT YET INITIALIZED')
objLst=None
self._moduleFixTarget =mft = ModuleFixTarget.WndFixTarget(self,objLst)
tab = self._tabs_daq_methods.insertTab(0,mft,'Fix Target')
mft.setAccessibleName(TASK_FIX_TARGET)
self._tabs_daq_methods.setCurrentWidget(mft) #set this as the active tabs
mft._cbType.addItems(["Fiducial", "FixTarget(12.5x12.5)", "FixTarget(23.0x23.0)", "FixTarget(23.0x23.0_160x156)", "FixTarget(test)", "Grid()", "SwissMX()", "SwissFEL()"])
#print(psi_device.shapepath.ShapePath.setup_motion.__doc__)
mft._txtParam.setToolTip('''\
additional parameter as yaml for object settings and motion generation:
full example for grid() object and motion settings:
size:[3.6,2.664],cnt:[30,22],fiducialSize:0.05,mode:6,scale:.5,ssz:[7,4],smv:[6,4],sdelay:9
further examples:
object settings:
for Grid(): size:[4,3],cnt:[30,22],fiducialSize:0.1
motion param:
for continous: cnt:2,scale:.3,dwell:100
for stop-and-go: tmove:10,twait:20
for hit-and-return: ssz:(4,6)
motion generation:
For details read: psi_device.shapepath.ShapePath.setup_motion.__doc__
mode: select one of following modes: 1,3,4,5,6
mode settings:
all: cnt:1
scale:1.
dwell:10
5: tmove:10
twait:20
6: ssz:(7,4)
smv:(6,4) (delault defined)
sdelay:9 (delault defined)
object settings:
Fiducial:
no param
FixTarget(12.5x12.5), FixTarget(23.0x23.0), FixTarget(23.0x23.0_160x156), FixTarget(test):
no param
Grid():
size:[30,20]
cnt:[30,22]
fiducialSize:0.1
SwissMX():
ofs:[.2,.2]
width:10
fidScl:.02
fiducial:[[.1,.1],[.1,2.7],[10.3,.1],[10.3, 2.7]]
SwissFEL():
ofs:[.2,.2]
width:10
fidScl:.02
fiducial:[[.1,.1],[.1,2.2],[10.3,.1],[10.3,2.2]]
''')
self.build_motion_fields()
mft._btnAdd.clicked.connect(self.module_fix_target_add_obj)
mft._btnDelAll.clicked.connect(self.module_fix_target_del_all_obj)
mft._btnFit.clicked.connect(self.module_fix_target_fit_fiducial)
#tab.layout().addWidget(mft)
mft.prefixSelected.connect(self.cb_set_daq_prefix)
#mft.dataFileLoaded.connect(self._OLD_daq_method_prelocated_update_markers)
#mft.prelocatedDataUpdated.connect(self._OLD_daq_method_prelocated_update_markers)
#mft.markersDeleted.connect(self._OLD_daq_method_prelocated_remove_markers)
#mft.moveFastStageRequest.connect(self._OLD_move_fast_stage)
#self.fiducialPositionSelected.connect(self._OLD_daq_method_prelocated_set_fiducial)
#self.appendPrelocatedPosition.connect(self._OLD_daq_method_prelocated_append_data)
#self._preloc_inspect_area = QPlainTextEdit()
#tab.layout().addWidget(self._preloc_inspect_area)
def cb_switch_task(self, index=0):
stack = self._wd_stack
task = self._wd_tabs_left.currentWidget().accessibleName()
setup_task = self._setup_toolbox.currentWidget().accessibleName()
method = self._tabs_daq_methods.currentWidget().accessibleName()
_log.debug("Task switching: top_tabs={} exp.method={} setup_task={}".format(task, method, setup_task))
if task == "task_experiment":
stack.setCurrentIndex(0)
active_task = method
elif task == "task_setup":
active_task = setup_task
else:
_log.error("BUG: un-handled task")
QMessageBox.warning(self, "BUG unhandled task!", "unhandled task: top_tabs={} method={}".format(task, method),)
self.set_active_task(active_task)
self._status_task.setText(active_task)
def cb_update_pix2pos(self, calib):
app=QApplication.instance()
if calib:
_log.info("received new pix2pos calibration")
QMessageBox.information(self,'Pixel to position calibration','''\
- for each zoom level:
- choose a good fiducial mark
- move the stage x/y at min. 3 different locations
- click (left mouse button) on that fiducial mark
- un-click button "Pixel to position calibration"
''')
app._raw_pix2pos=dict()
else:
s=str(app._raw_pix2pos).replace('),','),\n ').replace('],','],\n')
print(s)
res=QMessageBox.question(self, "calibration", f"use calibration\n{s} ? ")
if res==QMessageBox.Yes:
geo=app._geometry
geo.update_pix2pos(app._raw_pix2pos)
app._cfg.setValue(AppCfg.GEO_PIX2POS, geo._lut_pix2pos)
del app._raw_pix2pos
def cb_update_opt_ctr(self, calib):
app=QApplication.instance()
if calib:
_log.info("received new opt_ctr calibration")
QMessageBox.information(self,'Optical center calibration','''\
- choose the biggest zoom level to start
- choose at least 3 fiducial marks
- for at least two zoom level:
- Click (left mouse button) on the fiducial marks (always in the same order)
- un-click button "Optical center calibration"
''')
app._raw_opt_ctr=dict()
else:
geo=app._geometry
ocOld=geo._opt_ctr
try:
geo.update_optical_center(app._raw_opt_ctr,True)
except BaseException as e:
_log.warning(e)
QMessageBox.warning(self, "calibration", f"rejected\n{e}")
del app._raw_opt_ctr
return
del app._raw_opt_ctr
go=self._goOptCtr
oc_sz=go.size()
go.blockSignals(True) # avoid to call cb_marker_moved
go.setPos(geo._opt_ctr-oc_sz/2)
res=QMessageBox.question(self, "calibration", f"keep new optical center\n{geo._opt_ctr} ? ")
if res==QMessageBox.Yes:
app._cfg.setValue(AppCfg.GEO_OPT_CTR, geo._opt_ctr)
else:
geo._opt_ctr=ocOld
go.setPos(geo._opt_ctr-oc_sz/2)
go.blockSignals(False) # allow to call cb_marker_moved
def cb_fiducial_update_z(self,obj,*vargs):
obj._z=self.tweakers["base_z"].get_rbv()
def module_fix_target_add_obj(self,*args,**kwargs):
mft=self._moduleFixTarget
fx=self.tweakers["fast_x"].get_rbv()
fy=self.tweakers["fast_y"].get_rbv()
bz=self.tweakers["base_z"].get_rbv()
app=QApplication.instance()
geo=app._geometry
oc=geo._opt_ctr
bm_pos=self._goBeamMarker.pos()
bm_sz=self._goBeamMarker.size()
idx=mft._cbType.currentIndex()
param=self.fix_motion_param(mft.get_param())
if idx==0:
#go=UsrGO.Fiducial(bm_pos+bm_sz/2-(20, 20), (40, 40),(fx,fy,bz))
l=.120
go=UsrGO.Fiducial((fx-l/2,fy-l/2), (l, l),bz)
go.sigRegionChangeFinished.connect(self.cb_fiducial_update_z)
elif idx==1:
#v=geo.pos2pix((12.5, 0));l=np.linalg.norm(v);l=12.5
sz=param.pop('size',(12.5, 12.5))
go=UsrGO.FixTargetFrame((fx-sz[0]/2,fy-sz[1]/2), sz, tpl='12.5x12.5',**param)
elif idx==2:
#v=geo.pos2pix((23, 0));l=np.linalg.norm(v)#l=23
sz=param.pop('size',(23, 23))
go=UsrGO.FixTargetFrame((fx-sz[0]/2,fy-sz[1]/2), sz, tpl='23.0x23.0',**param)
elif idx==3:
sz=param.pop('size',(23, 23))
go=UsrGO.FixTargetFrame((fx-sz[0]/2,fy-sz[1]/2), sz, tpl='23.0x23.0_160x156',**param)
elif idx==4:
sz=param.pop('size',(.120*12, .120*8))
go=UsrGO.FixTargetFrame((fx-sz[0]/2,fy-sz[1]/2), sz, tpl='test',**param)
elif idx==5:
w,h=size=param.pop('size',(30, 20))
cnt=param.pop('cnt',(30, 22))
fiducialSize=param.pop('fiducialSize',.1)
go=UsrGO.Grid((fx-w/2,fy-h/2), size, cnt, fiducialSize,**param)
elif idx==6:
ofs=param.pop('ofs',(.2, .2))
width=param.pop('width',10)
fidScl=param.pop('fidScl',.02)
fiducial=param.pop('fiducial',((.1, .1), (.1, 2.7), (10.3, .1), (10.3, 2.7)))
fiducial=np.array(fiducial)
gp=psi_device.shapepath.GenPath(); gp.swissmx(ofs=ofs, width=width, flipy=True)
sz=gp.points.max(0)+np.array(ofs)*2
go=UsrGO.Path((fx-sz[0]/2, fy-sz[1]/2), sz, gp.points, fiducial, fidScl,**param)
elif idx==7:
ofs=param.pop('ofs',(.2, .2))
width=param.pop('width',10)
fidScl=param.pop('fidScl',.02)
fiducial=param.pop('fiducial',((.1, .1), (.1, 2.2), (10.3, .1), (10.3, 2.2)))
fiducial=np.array(fiducial)
gp=psi_device.shapepath.GenPath(); gp.swissfel(ofs=ofs, width=width,flipy=True)
sz=gp.points.max(0)+np.array(ofs)*2
go=UsrGO.Path((fx-sz[0]/2, fy-sz[1]/2), sz, gp.points, fiducial, fidScl,**param)
else:
_log.error('index not handeled')
go._chip_type=mft._cbType.currentText() # recorded in the JFJ header
grp=self._goTracked
grp.addItem(go)
mft._tree.setData(grp.childItems())
def module_fix_target_del_all_obj(self):
app=QApplication.instance()
geo=app._geometry
try: #remove plane fitter if it esists
del geo._fitPlane
except AttributeError:
pass
mft=self._moduleFixTarget
vb=self.vb
grp=self._goTracked
for go in grp.childItems():
vb.removeItem(go)
grp.setFlag(grp.ItemHasNoContents)
mft._tree.setData(None)
def module_fix_target_fit_fiducial(self):
mft=self._moduleFixTarget
vb=self.vb
grp=self._goTracked
cldLst=grp.childItems()
n=0
for go in cldLst:
if type(go)==UsrGO.Fiducial:
n+=1
ptFitTrf=np.ndarray((4,2)) # 4 (x,y) points to fit a transformation of a parallelogram
ptFitPlane=np.ndarray((n,3)) #n (x,y,z) points to fit a plane transformation z=ax+by+c
i=j=0
for go in reversed(cldLst):
if type(go)==UsrGO.FixTargetFrame:
if j==4:
#fid=np.array(((0.1, 0.1), (0.1, 1.1), (1.1, 0.1), (1.1, 1.1)))
fid=np.array(go._dscr['fiducial']['pos'])
sz=np.array(go._dscr['size'])
fid=fid/sz
trf=geometry.geometry.least_square_trf(ptFitTrf,fid)
#_log.debug(f'{trf}')
_log.debug('\n trf='+str(trf).replace('\n','\n '))
tr=go.transform()
tr.setMatrix(1, trf[1,0]/trf[0,0], 0,
trf[0,1]/trf[1,1], 1, 0,
0, 0, 1)
go.setTransform(tr)
pos=np.asarray(trf[:,2]).ravel() # bug fix deprecation of np 1.25
go.setPos(pos[0],pos[1]) # bug fix deprecation of np 1.25
#go.setSize((1,1))
sz=(trf[0,0],trf[1,1])
go.setSize(sz)
j=0
elif type(go)==UsrGO.Path:
if j==4:
fid=go._fiducial
sz=go.szOrig
fid=fid/sz
trf=geometry.geometry.least_square_trf(ptFitTrf, fid)
_log.debug('\n trf='+str(trf).replace('\n','\n '))
tr=go.transform()
tr.setMatrix(1, trf[1, 0]/trf[0, 0], 0,
trf[0, 1]/trf[1, 1], 1, 0,
0, 0, 1)
go.setTransform(tr)
pos=np.asarray(trf[:,2]).ravel() # bug fix deprecation of np 1.25
go.setPos(pos[0],pos[1]) # bug fix deprecation of np 1.25
# go.setSize((1,1))
sz=(trf[0, 0], trf[1, 1])
go.setSize(sz)
j=0
elif type(go)==UsrGO.Fiducial and go.size()[0]==0.12:
if j<ptFitTrf.shape[0]:
ptFitTrf[j]=go.pos()+go.size()/2
j+=1
ptFitPlane[i]=go.ctr()
i+=1
app=QApplication.instance()
geo=app._geometry
if n>=3:
geo.least_square_plane(ptFitPlane)
def daq_collect(self, jobs):
'''
start the data collection in the worker thread (see daq_worker.CollectionWorker._collect)
jobs: list of scan parameter dicts (one per graphical object)
'''
app = QApplication.instance()
cfg = app._cfg
w=self._daq_worker
if w.busy or self._move_worker.busy:
QMessageBox.warning(self, "busy", "A data collection or motor sequence is already running.")
return
#pv-monitor-func: stop monitors
if not cfg.value(AppCfg.GBL_MISC)['live_on_collect']:
for pv in app._pv_mon_lst:
try:
pv.clear_auto_monitor()
except Exception:
pv.auto_monitor=False
self.set_collecting(True)
self._daq_dlg=dlg=daq_worker.CollectionProgressDialog(w, self)
dlg.show()
w.submit(jobs)
def set_collecting(self, on, nudge=True):
'''
grey out everything that must not move during a data collection.
only the collimator and post sample tube motors (limited to NUDGE_MAX_STEP in total) and the zoom stay enabled
'''
self._collecting=on
en=not on
self._wd_tabs_left.setEnabled(en)
self.toolBar.setEnabled(en)
self.menuBar().setEnabled(en)
self.zoombox._top_grid.setEnabled(en) # backlight, lights, exposure. zoom (_button_box) stays enabled
tbox=self._tbox_motors
for i in range(tbox.count()):
w=tbox.widget(i)
if on and nudge and w in self._wd_nudge_groups:
continue # stays usable for nudges during data collection
w.setEnabled(en)
for but in self._wd_nudge_macros:
but.setEnabled(en)
max_step=NUDGE_MAX_STEP if (on and nudge) else None # no cap for motor sequences: they move these motors
for k in NUDGE_TWEAKERS:
self.tweakers[k].set_max_step(max_step)
def cb_daq_progress(self, val, maximum, label):
dlg=self._daq_dlg
if dlg is not None:
dlg.set_progress(val, maximum, label)
def cb_daq_run_info(self):
self.cb_update_run_number()
self.cb_update_data_path()
def cb_daq_plot_gather(self, sp):
dp=psi_device.shapepath.DebugPlot(sp)
dp.plot_gather(mode=11)
plt.show(block=False)
def cb_daq_finished(self, msg):
app = QApplication.instance()
cfg = app._cfg
# pv-monitor-func: start monitors
if not cfg.value(AppCfg.GBL_MISC)['live_on_collect']:
for pv in app._pv_mon_lst:
pv.auto_monitor=True
self.set_collecting(False)
dlg=self._daq_dlg
self._daq_dlg=None
if dlg is not None:
dlg.finish()
if msg and msg!='aborted':
QMessageBox.warning(self, "data collection", msg)
def esc_run_steps(self, steps, title, esc_state):
if self.run_moves(title, steps):
self._esc_state = esc_state
def run_moves(self, title, steps):
'''
run a blocking motor sequence in the worker thread with a progress/abort dialog.
steps: list of (label, callable). All controls are locked meanwhile (no nudges)
'''
if self._daq_worker.busy or self._move_worker.busy:
QMessageBox.warning(self, title, "A data collection or motor sequence is already running.")
return False
self.set_collecting(True, nudge=False)
self._daq_dlg=dlg=daq_worker.CollectionProgressDialog(self._move_worker, self, title=title)
dlg.show()
self._move_worker.submit(steps)
return True
def cb_moves_finished(self, msg):
_log.info(f'motor sequence finished, unlock GUI ({msg!r})')
self.set_collecting(False)
dlg=self._daq_dlg
self._daq_dlg=None
if dlg is not None:
dlg.finish()
if msg and msg!='aborted':
QMessageBox.warning(self, "motor sequence", msg)
elif msg=='aborted':
self.statusbar.showMessage('motor sequence aborted: all motors stopped', 10000)
def stop_all_motors(self):
'''abort of a motor sequence (GUI thread): stop every motor'''
app=QApplication.instance()
for k,tw in self.tweakers.items():
try:
tw.stop()
except Exception as e:
_log.error(f'failed to stop {k}: {e}')
try:
app._backlight.stop()
except Exception as e:
_log.error(f'failed to stop backlight: {e}')
def assert_positions(self, targets, what, timeout=20.0):
'''wait for the motors to reach their targets (worker thread). raises MoveError'''
try:
app_utils.assert_tweaker_positions(targets, timeout=timeout, abort=self._move_worker.abort_event)
except app_utils.PositionsNotReached as e:
self._move_worker.check_abort()
raise daq_worker.MoveError(f"failed to move {what}: {e}") from e
def check_collection_positions(self, tol=0.1):
'''
compare collimator, post sample tube, detector (in) and backlight (out) with their data collection
positions. returns a list of (name, ok, detail)
'''
app=QApplication.instance()
cfg=app._cfg
tw=self.tweakers
res=[]
def check(name, items, cfg_key):
if not cfg.value(cfg_key):
res.append((name, False, 'position not configured'))
return
bad=[]
for alias, target in items:
try:
cur=tw[alias].get_rbv()
except Exception as e:
bad.append(f'{alias}: read failed ({e})')
continue
if target is None:
bad.append(f'{alias}: target not configured')
elif abs(cur-target)>tol:
bad.append(f'{tw[alias]._label} {cur:.3f} (in: {target:.3f})')
res.append((name, not bad, ', '.join(bad)))
col=cfg.value(AppCfg.DFT_POS_COL)
check('collimator', (('colli_x', col.get('x_in')), ('colli_y', col.get('y_in'))), AppCfg.DFT_POS_COL)
pst=cfg.value(AppCfg.DFT_POS_PST)
check('post sample tube', (('tube_usx', pst.get('x_in_us')), ('tube_usy', pst.get('y_in_us')),
('tube_dsx', pst.get('x_in_ds')), ('tube_dsy', pst.get('y_in_ds')),
('tube_z', pst.get('z_in'))), AppCfg.DFT_POS_PST)
det=cfg.value(AppCfg.DFT_POS_DET)
check('detector', (('det_z', det.get('pos_in')),), AppCfg.DFT_POS_DET)
try:
ok=app._backlight.is_pos('out')
res.append(('backlight', ok, '' if ok else 'not out'))
except Exception as e:
res.append(('backlight', False, f'read failed ({e})'))
return res
def confirm_collection_positions(self):
'''ask the operator to confirm if a device is not at its data collection position. returns True to go on'''
res=self.check_collection_positions()
bad=[r for r in res if not r[1]]
if not bad:
return True
lines='\n'.join(f"{'OK' if ok else 'NOT IN POSITION'}: {name}{' - '+detail if detail else ''}" for name, ok, detail in res)
_log.warning('collect: devices not at data collection position: '+'; '.join(f'{n}: {d}' for n, ok, d in bad))
box=QMessageBox(QMessageBox.Warning, "data collection",
"Not everything is in the data collection position:\n\n"+lines+
"\n\nAre you sure you want to start the data collection?",
QMessageBox.Yes|QMessageBox.No, self)
box.setDefaultButton(QMessageBox.No)
if box.exec_()==QMessageBox.Yes:
_log.warning('collect: operator confirmed collection with devices out of position')
return True
return False
def move_gonio(self, pos):
# gonio="{'pos_mount': (0.0,0.0,0.0,0.0,0.0), 'pos_align': (0.0,0.0,0.0,0.0,0.0)}"
# positions are: fast_x,fast_y,base_x,base_y,base_z,(omega=0,det_z, no change)
app=QApplication.instance()
cfg=app._cfg
pos_gonio = cfg.value(AppCfg.DFT_POS_GONIO)
if pos_gonio is None:
raise daq_worker.MoveError("gonio default positions are not configured.")
tw_fx = self.tweakers["fast_x"]
tw_fy = self.tweakers["fast_y"]
tw_cx = self.tweakers["base_x"]
tw_cy = self.tweakers["base_y"]
tw_cz = self.tweakers["base_z"]
try:
p_fx, p_fy, p_cx, p_cy, p_cz,=pos_gonio['pos_'+pos]
except KeyError:
raise ValueError("Goniometer position *{}* is not known!!")
_log.info(f"moving goniometer {pos} to fx:{p_fx:.5g},fy:{p_fy:.5g},cx:{p_cx:.5g},cy:{p_cy:.5g},cz_{p_cz:.5g}") #,ry:{p_ry:.5g}")
tw_fx.move_abs(p_fx)
tw_fy.move_abs(p_fy)
tw_cx.move_abs(p_cx)
tw_cy.move_abs(p_cy)
tw_cz.move_abs(p_cz)
self.assert_positions([
(tw_fx, p_fx, 0.1),
(tw_fy, p_fy, 0.1),
(tw_cx, p_cx, 0.1),
(tw_cy, p_cy, 0.1),
(tw_cz, p_cz, 0.1), ], f'goniometer {pos}')
def move_post_tube(self, pos):
# post_sample_tube: absolute in/out positions x/y_in/out_us/ds, z_in, z_out
# tandem nudges: GUI thread, non blocking (also allowed during data collection)
if pos in ("x+", "x-", "y+", "y-"):
tandem_twv = float(self._post_tandem_tweak_val.text())
d = tandem_twv if pos[1]=='+' else -tandem_twv
tw_us, tw_ds = (self.tweakers["tube_usx"], self.tweakers["tube_dsx"]) if pos[0]=='x' else \
(self.tweakers["tube_usy"], self.tweakers["tube_dsy"])
self.move_together([(tw_us, tw_us.get_rbv()+d), (tw_ds, tw_ds.get_rbv()+d)])
return
# in/out: worker thread (run_moves). detector and backlight safety
if pos == "in":
self.move_detector("out")
self.move_backlight("out") # waits until the backlight is out
if not self.is_detector_safe():
raise daq_worker.MoveError("Detector is not clear of the post-sample-tube travel path - post-tube move aborted.")
app=QApplication.instance()
cfg=app._cfg
pos_pst = cfg.value(AppCfg.DFT_POS_PST)
if pos_pst is None:
raise daq_worker.MoveError("SwissMX *POST-SAMPLE-TUBE* configuration is incomplete!!!")
x_in_us=pos_pst['x_in_us']
y_in_us=pos_pst['y_in_us']
x_in_ds=pos_pst['x_in_ds']
y_in_ds=pos_pst['y_in_ds']
# absolute out positions (fallback for settings that only have the old deltas)
x_out_us=pos_pst.get('x_out_us', x_in_us+pos_pst.get('x_out_delta',0.0))
y_out_us=pos_pst.get('y_out_us', y_in_us+pos_pst.get('y_out_delta',0.0))
x_out_ds=pos_pst.get('x_out_ds', x_in_ds+pos_pst.get('x_out_delta',0.0))
y_out_ds=pos_pst.get('y_out_ds', y_in_ds+pos_pst.get('y_out_delta',0.0))
z_in =pos_pst['z_in']
z_out =pos_pst['z_out']
tw_x_us = self.tweakers["tube_usx"]
tw_y_us = self.tweakers["tube_usy"]
tw_x_ds = self.tweakers["tube_dsx"]
tw_y_ds = self.tweakers["tube_dsy"]
tw_z = self.tweakers["tube_z"]
_log.info(f"move post sample tube {pos}")
if pos == "in":
if not self.move_together([(tw_x_us, x_in_us), (tw_x_ds, x_in_ds), (tw_y_us, y_in_us), (tw_y_ds, y_in_ds)]):
raise daq_worker.MoveError(f"post sample tube {pos}: target refused (soft limits)")
self.assert_positions([
(tw_x_us, x_in_us, 0.1),
(tw_y_us, y_in_us, 0.1),
(tw_x_ds, x_in_ds, 0.1),
(tw_y_ds, y_in_ds, 0.1), ], f'post sample tube {pos}')
tw_z.move_abs(z_in, wait=True)
self.assert_positions([(tw_z, z_in, 0.1)], f'post sample tube {pos}', timeout=60.0)
elif pos == "out":
tw_z.move_abs(z_out, wait=True)
self.assert_positions([(tw_z, z_out, 0.1)], f'post sample tube {pos}', timeout=60.0)
if not self.move_together([(tw_x_us, x_out_us), (tw_x_ds, x_out_ds), (tw_y_us, y_out_us), (tw_y_ds, y_out_ds)]):
raise daq_worker.MoveError(f"post sample tube {pos}: target refused (soft limits)")
self.assert_positions([
(tw_x_us, x_out_us, 0.1),
(tw_y_us, y_out_us, 0.1),
(tw_x_ds, x_out_ds, 0.1),
(tw_y_ds, y_out_ds, 0.1), ], f'post sample tube {pos}')
else:
raise ValueError("post sample tube *{}* is not known!!")
def move_together(self, targets):
'''
start several motors at the same time (absolute targets, e.g. upstream and downstream post tube axes).
all targets are checked first (nudge cap, soft limits), then the moves are sent back to back directly to
the VAL fields, without reads in between. either all motors move or none. returns True if moved
'''
for tw, v in targets:
if not tw._target_allowed(v):
return False
if isinstance(tw, MotorTweak) and not tw._ignore_limits and not tw._motor.within_limits(v):
_log.warning(f'{tw._label}: target {v:.5g} outside the soft limits -> no motor moved')
return False
for tw, v in targets:
if isinstance(tw, MotorTweak):
tw._val=v
tw._motor.put('VAL', v, wait=False)
else:
tw.move_abs(v)
_log.debug('move together: '+', '.join(f'{tw._label}->{v:.5g}' for tw, v in targets))
return True
def move_detector(self, pos):
# detector="{'pos_in': -0.1, 'pos_out': 0.0}"
app=QApplication.instance()
cfg=app._cfg
pos_det = cfg.value(AppCfg.DFT_POS_DET)
if pos_det is None:
raise daq_worker.MoveError("detector default positions are not configured.")
det_z = self.tweakers["det_z"]
try:
p_z=pos_det['pos_'+pos]
except KeyError:
raise ValueError("Collimator position *{}* is not known!!")
cur_z = det_z.get_rbv()
if abs(cur_z-p_z) < 0.1:
_log.info(f"detector already at {pos} position")
return
_log.info(f"moving detector {pos} to {p_z:.5g}")
det_z.move_abs(p_z, wait=True)
self.assert_positions([(det_z, p_z, 0.1)], f'detector {pos}')
def is_detector_safe(self):
# detector must be at/beyond the configured *safe* position to guarantee no post-tube collision
app=QApplication.instance()
cfg=app._cfg
pos_det = cfg.value(AppCfg.DFT_POS_DET)
pos_safe = pos_det.get('pos_safe') if pos_det is not None else None
if pos_safe is None:
raise daq_worker.MoveError("detector *safe* position is not configured.")
det_z = self.tweakers["det_z"]
return det_z.get_rbv() >= pos_safe
def move_collimator(self, pos):
# collimator: absolute positions x_in, y_in, x_out, y_out (ready = x_in, y_out)
app=QApplication.instance()
cfg=app._cfg
cx = self.tweakers["colli_x"]
cy = self.tweakers["colli_y"]
pos_col = cfg.value(AppCfg.DFT_POS_COL)
if pos_col is None:
raise daq_worker.MoveError("COLLIMATOR configuration is incomplete!")
x_in = pos_col['x_in']
y_in = pos_col['y_in']
x_out = pos_col['x_out']
y_out = pos_col['y_out']
if pos == "out":
x_pos=x_out
y_pos=y_out
elif pos == "in":
x_pos=x_in
y_pos=y_in
elif pos == "ready":
x_pos=x_in
y_pos=y_out
else:
raise ValueError("Collimator position *{}* is not known!!")
_log.info(f"moving collimator {pos} to X,Y = {x_pos:.3f}, {y_pos:.3f}")
cx.move_abs(x_pos) # X first, then Y (as before)
self.assert_positions([(cx, x_pos, 0.1)], f'collimator {pos} (X)')
cy.move_abs(y_pos)
self.assert_positions([(cy, y_pos, 0.1)], f'collimator {pos} (Y)')
if __name__=="__main__":
def main():
#_log.info(f'Version: pyqtgraph:{pg.__version__} matplotlib:{mpl.__version__} numpy:{np.__version__} epics:{epics.__version__} qt:{QT_VERSION_STR}' )
_log.info(f'Version: pyqtgraph:{pg.__version__} epics:{epics.__version__} qt:{QT_VERSION_STR}' )
import argparse, socket
hostname=socket.gethostname()
if hostname=='ganymede':
#use EPICS locally
#os.environ['EPICS_CA_ADDR_LIST']='localhost'
#use EPICS if connected to ESC network
os.environ['EPICS_CA_ADDR_LIST'] ='129.129.244.255 sf-saresc-cagw.psi.ch:5062 sf-saresc-cagw.psi.ch:5066'
#(h, t)=os.path.split(sys.argv[0]);cmd='\n '+(t if len(h)>20 else sys.argv[0])+' '
#exampleCmd=('', '-m0xf -v0')
epilog=__doc__ #+'\nExamples:'+''.join(map(lambda s:cmd+s, exampleCmd))+'\n'
parser=argparse.ArgumentParser(epilog=epilog, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument('-l', '--log', type=int, help='debug(1)|info(2)|warn(3)|error(4) default=%(default)d', default=2)
parser.add_argument("--sim", "-s", type=lambda x: int(x,0), help="simulate devices (see bitmasks) default=0x%(default)x", default=0)
args = parser.parse_args()
_log.info('Arguments:{}'.format(args.__dict__))
if args.log!=2:
for l in ('app_config', 'app_utils', 'backlight', 'camera', 'ctadaq', 'daq_worker', 'epics_widgets',
'geometry', 'illumination', 'ModuleFixTarget', 'MXMotion',
'pbtools', 'psi_device', 'pyqtUsrObj', 'shapepath', 'swissmx', 'zoom',):
try:
logging.getLogger(l).setLevel(args.log*10)
except BaseException as e:
_log.error(f'failed set log level ({l}): {e}')
if args.log<2: listLoggers()
from PyQt5.QtWidgets import QApplication
# set app icon
app = QApplication(sys.argv)
app_icon = QtGui.QIcon()
app_icon.addFile("artwork/logo/16x16.png", QtCore.QSize(16, 16))
app_icon.addFile("artwork/logo/24x24.png", QtCore.QSize(24, 24))
app_icon.addFile("artwork/logo/32x32.png", QtCore.QSize(32, 32))
app_icon.addFile("artwork/logo/48x48.png", QtCore.QSize(48, 48))
app_icon.addFile("artwork/logo/256x256.png", QtCore.QSize(256, 256))
app.setWindowIcon(app_icon)
startupWin=StartupSplash()
#for i in range(1, 20):
# startupWin.set(i,f'sample startup text {i}')
# #app.processEvents()
# t = time.time()
# #while time.time() < t + 0.1:
# # app.processEvents()
#app._settings=QSettings("PSI", "SwissMX")
app._args=args
app._pv_mon_lst=list() #pv-monitor-func: list of all pv that are monitored and that should
# be turned on/off during data collection to avoid (hopefully) segmentation fault
startupWin.set(5, f'load settings')
app._cfg=cfg=AppCfg()
app._geometry=geometry.geometry()
app._geometry._lut_pix2pos=cfg.value(cfg.GEO_PIX2POS)
app._geometry._opt_ctr=cfg.value(cfg.GEO_OPT_CTR)
startupWin.set(15, f'connect backlight')
pfx=app._cfg.value(AppCfg.GBL_DEV_PREFIX)[0]
dft_pos_bklgt=app._cfg.value(AppCfg.DFT_POS_BKLGT)
if args.sim&0x01:
app._backlight = backlight.Backlight(None,dft_pos_bklgt)
else:
app._backlight = backlight.Backlight(f'{pfx}:MOT_BLGT',dft_pos_bklgt)
startupWin.set(20, f'connect illumination')
if args.sim&0x02:
app._illumination = illumination.IlluminationControl(None)
else:
app._illumination = illumination.IlluminationControl()
startupWin.set(40, f'connect zoom')
if args.sim&0x04:
app._zoom = zoom.QopticZoom(None)
else:
app._zoom = zoom.QopticZoom()
startupWin.set(60, f'connect camera')
if args.sim&0x08:
app._camera = camera.epics_cam(None)
app._camera.sim_gen(mode=1)
else:
app._camera = camera.epics_cam()
if args.sim&0x80:
app._shutter = psi_device.Shutter(0)
else:
app._shutter = psi_device.Shutter(1)
#app._camera.run() is called in center_piece_update
startupWin.set(60, f'start main window')
app._mainWnd=wnd=WndSwissMx()
wnd.show()
startupWin._wnd.finish(wnd)
# needed so pycharm can restart application
signal.signal(signal.SIGINT, sigint_handler)
#main.update_user_and_storage() #ZAC: orig. code
sys.exit(app.exec_())
main()
Reference in new issue View Git Blame Copy Permalink