Furka/EsfRixsApps
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
EsfRixsApps/ARESvis/ARESvis.py
Thierry Zamofing 80976e5b63 SFELPHOTON-1527: Refactor: Improve ARESvis.py for robustness and thread-safety 
Addressed several issues in ARESvis.py:
- Implemented thread-safe GUI updates using pyqtSignals for EPICS callbacks.
- Enhanced  function with null checks and value clipping.
- Improved image loading in  with file existence checks and logging.
- Refactored  to use widget dictionaries, avoiding .
- Cleaned up  and  methods, removing dead code and using widget dictionaries.
- Updated Python version check to exit on failure.
2025-11-18 08:55:37 +01:00

790 lines
24 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python
# *-----------------------------------------------------------------------*
# | |
# | Copyright (c) 2024 by Paul Scherrer Institute (http://www.psi.ch) |
# | |
# | Author Thierry Zamofing (thierry.zamofing@psi.ch) |
# *-----------------------------------------------------------------------*
"""
Furka ARES chamber visualization
For simulated motor IOC:
/home/zamofing_t/Documents/prj/SwissFEL/test_ioc/MotorSim/iocBoot/ARESvis/ARESvis.cmd
For motor ui:
caQtDM ~/Documents/prj/SwissFEL/test_ioc/MotorSim/iocBoot/ARESvis/ARESvis.ui&
self.pv_angles = [epics.PV("SATES30-ARES:MOT_SRY.RBV"), epics.PV("SATES30-ARES:MOT_DRY.RBV"), epics.PV("SATES30-ARES:MOT_2TRY.RBV")]
SATES30-RIXS:MOT_RY.RBV # sliding seal
SATES30-ARES:MOT_JFRY.RBV # jungfrau detector angle:
SATES30-ARES:MOT_2TRY.RBV # 2thetha angle: foc.mirror Diode2 Diode3
SATES30-ARES:MOT_DRY.RBV # detector angle: Diode1 Mirror
SATES30-ARES:MOT_STX.RBV # sample TX
SATES30-ARES:MOT_STZ.RBV # sample TZ
SATES30-ARES:MOT_SRY.RBV # sample rotation
SATES30-MCS001:MOT_6 # Parabola TX
SATES30-ACSFM:MOT_TZ # Foc. Mirror TZ
bitmask for simulation:
0x01: EPICS motors
0x02:
0x04:
0x08:
0x10:
0x20:
0x40:
0x80:
"""
import os.path
from PyQt5.QtWidgets import QApplication, QWidget, QLabel, QPushButton, QSlider, QLineEdit,\
QCheckBox, QHBoxLayout, QVBoxLayout, QGroupBox, QGridLayout, QComboBox
from PyQt5.QtGui import QPainter, QColor, QPen, QBrush, QPolygon, QPolygonF, QTransform, QPainterPath, \
QPixmap, QMouseEvent, QImage
from PyQt5.QtCore import QPoint, QPointF, Qt,pyqtSignal,QObject
#import PyQt5.QtGui as QtGui
#import PyQt5.QtCore as QtCore
import numpy as np
import sys, logging, copy
import epics
if sys.version_info < (3, 6):
sys.exit(f"Must be using Python 3.6 or newer. Try e.g. /opt/gfa/python-3.8/latest/bin/python {sys.argv[0]}")
_log=logging.getLogger(__name__)
import logging
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
R = '\033[1;31m' #bold, red
G = '\033[1;32m' #bold, green
Y = '\033[1;33m' #bold, yellow
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)
def adjust_transparency(pixmap, transparency_factor):
"Adjust the transparency of a QPixmap."
image = pixmap.toImage().convertToFormat(QImage.Format_ARGB32)
if image.isNull():
return pixmap
# Convert QImage to a NumPy array for efficient manipulation
width = image.width()
height = image.height()
ptr = image.bits()
ptr.setsize(image.byteCount()) # Make the sip.voidptr object aware of its size
arr = np.frombuffer(ptr, dtype=np.uint8).reshape((height, width, 4))
# Modify the alpha channel, ensuring values are within the valid 0-255 range
arr[:, :, 3] = (arr[:, :, 3] * transparency_factor).clip(0, 255).astype(np.uint8)
# The numpy array modifies the QImage's buffer in place. Return a new QPixmap from the modified image.
return QPixmap.fromImage(image)
class ARESdevice():
def __init__(self,name,**kwargs):
self._name=name
self._paint=p={
'ofs':(820, 350), # location of device
'rArm':int(540), # 540mm inner radius of chamber (RIXS arm)
'r2Th':int(375), # 375mm radius 2Theta platfform
'rJFr':int(270), # 270mm radius of Jungfrau
'rDet':int(207.5), # 207.5mm radius of detector table
'rTrg':int(168.5), # 168.5mm radius of target
'szArm':(20, 50),
'aArm':100, # angle RIXS arm
'aJFr':140, # angle detector
'a2Th':116, # angle 2thetha
'aDet':100, # angle detector
'txTrg':0, # tx target
'tzTrg':0, # tz target
'aTrg':10, # angle target
'txPar':0, # parabola translation x
'tyPar':50.8,# parabola translation y
'tzFM':10, # focussing mirror translation z
'sclTrf':(2**(6/2), 2**(-2/2)), # scaling transfformation [angle, distance]
}
p.update(kwargs)
self._geo=g={
}
self.setGeometry(g)
self._pic=pic=dict()
base=os.path.join(os.path.dirname(os.path.realpath(__file__)),'pic')
_log.debug(f"Loading pictures from: {base}")
pic_files = {
'ir':'ARES_2Theta_InnerRing.png',
'or':'ARES_2Theta_OuterRing.png',
'bl':'ARES_Below.png',
'df':'ARES_Diffractometer.png',
'di':'ARES_Diode.png',
'jf':'ARES_Jungfrau.png',
'lm':'ARES_LaserMirror.png',
'ma':'ARES_Master.png',
'mi':'ARES_Mirrors.png',
'pb':'ARES_Parabola.png',
}
for k, v in pic_files.items():
path = os.path.join(base, v)
if os.path.exists(path):
pic[k] = adjust_transparency(QPixmap(path), .7)
else:
_log.warning(f"Pixmap file not found: {path}")
def setGeometry(self,geo):
self._geo=geo
p=self._paint
def geometry2motor(self):
# returns raw motor positions
# offset detector plane to deflected beam: 34deg
geo=self._geo
def containsPoint(self,point):
try:
pg=self._polygon
except AttributeError:
return False
return pg.containsPoint(point,Qt.OddEvenFill)
@staticmethod
def plotOrig(qp):
penR=QPen(Qt.red, 2, Qt.SolidLine)
penG=QPen(Qt.green, 2, Qt.SolidLine)
pOrig=qp.pen()
qp.setPen(penR)
qp.drawLine(-20, 0, 20, 0)
qp.drawLine( 20, 0, 16, 2)
qp.setPen(penG)
qp.drawLine(0, -20, 0, 20)
qp.drawLine(0, 20, 2, 16)
qp.setPen(pOrig)
def paint(self,qp):
# qp QPainter to paint on
# ofs caanter to draw
# scl scaling for x and y translation of coordinate systems
# paintMode: mode how to paint the diffraction beam
p=self._paint
pic=self._pic
ofs=p['ofs']
rArm=p['rArm']
r2Th=p['r2Th']
rJFr=p['rJFr']
rDet=p['rDet']
rTrg=p['rTrg']
aArm=p['aArm']
a2Th=180-p['a2Th'] #opposite direction zero at bottom
aJFr=p['aJFr']
aDet=p['aDet']
txTrg=p['txTrg']
tzTrg=p['tzTrg']
aTrg=p['aTrg']
txPar=-p['txPar'] #opposite direction
tyPar=p['tyPar']
tzFM =p['tzFM']
sclTrf=p['sclTrf']
#x=p.get('x',0)
#y=p.get('y',0)
#print(x,y)
tickW,tickL=3,20 # tick px-width, tick len
diW,diH=40,20 # diodes
miW,miH=50,10 # mirrors
tgW,tgH=80,30 # target
# --- prepare transformations ---
# tf0: target not rotated
# tfArm: target center rotated angle aaArm
# tf2Th: target center rotated angle tf2Th
# tfDet: target center rotated angle aaDet
# tfTrg: target center rotated angle aaTrg
tf0=QTransform()
tf0.translate(ofs[0], ofs[1])
tf0.scale(sclTrf[1],sclTrf[1])
tfArm=copy.copy(tf0) # center
tfArm.rotate(-aArm)
tfJFr=copy.copy(tf0)
tfJFr.rotate(-aJFr)
tf2Th=copy.copy(tf0)
tf2Th.rotate(-a2Th)
tfDet=copy.copy(tf0)
tfDet.rotate(-aDet)
tfTrg=copy.copy(tf0)
tfTrg.rotate(-aTrg)
#tfd.translate(r2,0).rotate(-cc)
penBk=QPen(Qt.black, 0, Qt.SolidLine)
penWt=QPen(Qt.white, 1, Qt.SolidLine)
penYl=QPen(Qt.yellow, 1, Qt.SolidLine)
penBl=QPen(Qt.blue, 1, Qt.SolidLine)
penRd=QPen(Qt.red, 1, Qt.SolidLine)
# --- visualize ---
#qp.setRenderHints(QPainter.HighQualityAntialiasing)
# setup and plot dragable region
self._polygon=QPolygon([
QPoint(*tf0.map(-rArm,-rArm )),
QPoint(*tf0.map(-rArm,+rArm+100)),
QPoint(*tf0.map(+rArm,+rArm+100)),
QPoint(*tf0.map(+rArm,-rArm )),
])
qp.setBrush(QColor(0, 0, 0,64))
qp.drawPolygon(self._polygon)
qp.setTransform(tf0)
qp.setPen(penBk)
qp.setBrush(QColor(128, 128, 128, 128)) #r,g,b,a
#circles of rotation
#qp.drawEllipse(-rArm, -rArm, 2*rArm, 2*rArm) # ARES chamber
#qp.drawEllipse(-r2Th, -r2Th, 2*r2Th, 2*r2Th) # 2theta
#qp.drawEllipse(-rJFr, -rJFr, 2*rJFr, 2*rJFr) # jungfrau
#qp.drawEllipse(-rDet, -rDet, 2*rDet, 2*rDet) # detector
#qp.drawEllipse(-rTrg, -rTrg, 2*rTrg, 2*rTrg) # target
for r1,r2,col in (
(rArm,r2Th,QColor(255, 0, 0, 32)), #2theta
(r2Th,rDet,QColor( 0,255, 0, 32)), #Jungfrau
#(r2Th,rJFr,QColor( 0,255, 0, 32)), #Jungfrau
#(rJFr,rDet,QColor(255, 0,255, 32)), #detector
(rDet,rTrg,QColor( 0, 0,255, 32)), #target
):
path=QPainterPath()
qp.setBrush(col) #r,g,b,a
path.addEllipse(-r1, -r1, 2*r1, 2*r1)
path.addEllipse(-r2, -r2, 2*r2, 2*r2) # Jungfrau
qp.drawPath(path)
qp.setBrush(Qt.NoBrush)
qp.drawEllipse(-rJFr, -rJFr, 2*rJFr, 2*rJFr) # jungfrau
qp.setBrush(QColor(0,255,255,32)) # r,g,b,a
qp.drawEllipse(-rTrg, -rTrg, 2*rTrg, 2*rTrg) # target
#beam arrow
qp.setPen(QPen(Qt.black, 3, Qt.SolidLine))
qp.drawLine(0,+rArm+100,0,rArm)
qp.drawPolygon(QPolygon([QPoint(0,rArm),QPoint(-5,rArm+20),QPoint(+5,rArm+20),]))
#crosshair
qp.setPen(penBk)
qp.drawLine(0,-rArm,0,rArm)
qp.drawLine(-rArm,0,rArm,0)
#self.plotOrig(qp)
#qp.setPen(penRd)
#qp.drawRect(-10, -10+rArm, 20, 20)
#--- pixmaps ---
pm=pic['bl'] # bellow
w,h=pm.width(),pm.height()
qp.setTransform(tfArm);qp.translate(0, -rArm);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h),pm)
pm=pic['mi'] # foccussing mirror
w,h=pm.width(),pm.height()
qp.setTransform(tf2Th);qp.translate(-39.6,-273.6);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['di'] #diode 2
w,h=pm.width(),pm.height()
qp.setTransform(tf2Th);qp.rotate(54);qp.translate(0,-240);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['df'] # diffrantometer
w,h=pm.width(),pm.height()
qp.setTransform(tfTrg);qp.translate(22, 17);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['jf'] # jungfrau detector
w,h=pm.width(),pm.height()
qp.setTransform(tfJFr);qp.translate(-25, -225);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['lm'] # laser mirror
w,h=pm.width(),pm.height()
qp.setTransform(tfDet);qp.rotate(181-7);qp.translate(13, -193);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['di'] # diode det1
w,h=pm.width(),pm.height()
qp.setTransform(tfDet);qp.rotate(165.4);qp.translate(-0, -205);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
pm=pic['di'] # diode det2
w,h=pm.width(),pm.height()
qp.setTransform(tfDet);qp.rotate(-96);qp.translate(-0, -205);qp.scale(.64,.64)
qp.drawPixmap(QPointF(-w/2,-h/2),pm)
#--- RIXS-arm devices ---
qp.setTransform(tfArm)
qp.setPen(QPen(Qt.red, tickW, Qt.SolidLine))
qp.drawLine(0,-rArm,0,-rArm+tickL) #tick
qp.setPen(penBk)
qp.setBrush(QColor(255,0,0,128))
qp.drawRect(-60, -150-rArm, 120, 150) # tube
qp.drawRect(-300, -600-rArm, 600, 450) # grating chamber
#--- 2-theta devices ---
qp.setTransform(tf2Th)
qp.setPen(QPen(Qt.green, tickW, Qt.SolidLine))
qp.drawLine(0,-r2Th,0,-r2Th+tickL) #tick
qp.setPen(penBk)
qp.setBrush(QColor(0,255,0,192))
#mirror:310x30mm, 10-40mm dist, 40mm outside 2th
qp.translate(20,-r2Th-40);qp.rotate(2);qp.translate(0,tzFM)
qp.drawRect(0, 0, 30, 310) # foc. mirror 1
qp.setTransform(tf2Th);qp.translate(-20,-r2Th-40);qp.rotate(-2);qp.translate(0,tzFM)
qp.drawRect(-30, 0, 30, 310) # foc. mirror 2
qp.setTransform(tf2Th);qp.rotate(54);qp.translate(-diW/2,-rDet-diH) #-r2Th-50
qp.drawRect(0, 0, diW, diH) # diode2
#qp.setTransform(tf2Th);qp.rotate(80);qp.translate(-diW/2,-r2Th);
#qp.drawRect(0, 0, diW, diH) # diode3
#--- Jungfrau devices ---
qp.setTransform(tfJFr)
qp.setPen(QPen(Qt.magenta, tickW, Qt.SolidLine))
qp.drawLine(0,-rJFr,0,-rJFr+tickL) #tick
qp.setPen(penBk)
qp.setBrush(QColor(255,0,255,192))
jfW,jfH=80,20 # jungfrau detector
qp.setTransform(tfJFr);qp.translate(-jfW/2,-rJFr);
qp.drawRect(0, -jfH, jfW, jfH) # detector mount sample
#--- detector devices ---
qp.setTransform(tfDet)
qp.setPen(QPen(Qt.blue, tickW, Qt.SolidLine))
qp.drawLine(0,-rDet,0,-rDet+tickL) #tick
qp.setPen(penBk)
qp.setBrush(QColor(0,0,255,192))
qp.setTransform(tfDet);qp.rotate(165.4);qp.translate(-diW/2,-rDet)
qp.drawRect(0, 0, diW, diH) # diode det1
qp.setTransform(tfDet);qp.rotate(-96);qp.translate(-diW/2,-rDet)
qp.drawRect(0, 0, diW, diH) # diode det2
qp.setTransform(tfDet);qp.rotate(181);qp.translate(0,-rDet+miW/2);qp.rotate(-45)
qp.drawRect(-int(miW/2), 0, miW, miH) # mirror1
#--- target devices ---
qp.setTransform(tfTrg)
qp.setPen(QPen(Qt.cyan, tickW, Qt.SolidLine))
qp.drawLine(0,-rTrg,0,-rTrg+tickL) #tick
qp.setPen(penBk)
qp.setBrush(QColor(0,255,255,192))
qp.drawRect(int(-(tgW/2)-txTrg), int(-tgH+tzTrg), tgW, tgH) # target mount sample
# --- parabola mirror ---
qp.setTransform(tf0)
qp.setPen(penBk)
qp.setBrush(QColor(224,192,0,224))
#x,x0,y,y0=3*50.8,3*12,3*50.8,3*12
x,x0,y,y0=50.8,12,50.8,4
qp.translate(txPar-x/2+61, tyPar)
path=QPainterPath()
path.moveTo(0,0)
path.cubicTo(0,y/2,x/2,y,x,y)
path.lineTo(x,y+y0)
path.lineTo(-x0,y+y0)
path.lineTo(-x0,0)
path.lineTo(0,0)
qp.drawPath(path)
qp.drawLine(int(x/2),int(y+y0)+20,int(x/2),0)
# --- parabola beam path ---
if txPar>-65 and txPar<-55: #show beam path only within +-5mm
qp.setPen(penBl)
qp.drawLine(0,0,int(x/2),int(-tyPar))
qp.drawLine(0,0,int(rArm+x/2-txPar),0)
qp.drawLine(int(x),int(y),int(x/2),int(-tyPar))
qp.drawLine(int(x),int(y),int(rArm+x/2-txPar),int(y))
# --- print angles ---
#rArm= r2Th= rJFr= rDet= rTrg= aArm= a2Th=180-p['a2Th'] # opposite direction zero at bottom aJFr= aDet= aTrg=
#penBl=QPen(Qt.blue, 3, Qt.SolidLine)
bg_col=QColor(255,255,255, 128) # background color
fg_col=QColor( 0, 0, 0,255) # foreground color
qp.setPen(penBl)
qp.setTransform(QTransform())
for r,a,tf in ((rArm,aArm,tfArm),(r2Th,a2Th,tf2Th),(rDet,aDet,tfDet),(rJFr,aJFr,tfJFr),(rTrg/2,aTrg,tfTrg),):
p=QPoint(*tf.map(0, int(-r+40)))
txt=f'{a:.5g}°'
txr=qp.boundingRect(p.x(), p.y(), 0, 0, Qt.AlignLeft, txt)
txr.moveTop(int(txr.y()-txr.height()/2))
# Draw opaque background
qp.setBrush(bg_col)
qp.setPen(Qt.NoPen)
qp.drawRect(txr)
# Draw text
qp.setPen(fg_col)
qp.drawText(txr, Qt.AlignLeft, txt)
class WndVisualize(QWidget):
updateSignal = pyqtSignal(str, float, str)
connectionSignal = pyqtSignal(str, bool)
_pv2key={
'SATES30-RIXS:MOT_RY.RBV' :'aArm',
'SATES30-ARES:MOT_2TRY.RBV':'a2Th',
'SATES30-ARES:MOT_JFRY.RBV':'aJFr',
'SATES30-ARES:MOT_DRY.RBV' :'aDet',
'SATES30-ARES:MOT_STZ.RBV' :'tzTrg',
'SATES30-ARES:MOT_STX.RBV' :'txTrg',
'SATES30-ARES:MOT_SRY.RBV' :'aTrg',
'SATES30-MCS001:MOT_6.RBV' :'txPar',
'SATES30-ACSFM:MOT_TZ.RBV' :'tzFM',
}
#updSignal=pyqtSignal(str, float, str)
def __init__(self):
super().__init__()
self.initUI()
self.updateSignal.connect(self.vis_update)
self.connectionSignal.connect(self.update_connection_status)
self.connectEPICS()
# self.update() must be called in main thread. Else gui may block after some time
# PV monitoring is done in a separate thread. Therefore functions as:
# OnConnectionChange, OnValueChange MUST NOT call update, but use this Signal helper class.
def initUI(self):
self.setGeometry(560, 100, 1300, 800)
self.setWindowTitle('Visualize')
app=QApplication.instance()
dev=app._dev
self._wdGrpDraw=w=QGroupBox(dev._name,self)
w.move(10,10)
row=0
lg=QGridLayout(w)
pDev=dev._paint
self.labels = {}
self.sliders = {}
for key, rng, tk in (
('aArm' ,( 0,360,), 30), # angle ARES sliding seal
('a2Th' ,( 0,360,), 30), # angle 2thetha
('aJFr' ,( 0,360,), 30), # angle Jungfrau
('aDet' ,( 0,360,), 30), # angle detector
('aTrg' ,( 0,360,), 30), # angle target
('txTrg',( -5, 5,), 30), # tx target
('tzTrg',(-10, 1,), 30), # tz target
('txPar',(-40, 80,), 10), # parabola translation x
('tzFM' ,(-50, 50,), 10), # focussing mirror translation z
('sclA', (-8, 8), 1),
('sclD', (-8, 8), 1),
#('x', (-100, 100), 10),
#('y', (-100, 100), 10),
):
wLb=QLabel(key, objectName=key)
self.labels[key] = wLb
wSl=QSlider(Qt.Horizontal,objectName=key)
wSl.setFixedWidth(200);wSl.setMinimum(rng[0]);wSl.setMaximum(rng[1])
if key.startswith('scl'):
if key[-1]=='A':
v=pDev['sclTrf'][0]
else:
v=pDev['sclTrf'][1]
v=int(round(np.log2(v)*2))
#elif key in ('x','y'):
# v=0
else:
v=pDev[key]
wSl.setValue(v)
wSl.setTickPosition(QSlider.TicksBelow);wSl.setTickInterval(tk)
wSl.valueChanged.connect(lambda val,key=key: self.sldChanged(key,val))
self.sliders[key] = wSl
lg.addWidget(wLb, row, 0)
lg.addWidget(wSl, row, 1);row+=1
w=QPushButton('sync motors')
w.clicked.connect(self.btnSyncMotors)
lg.addWidget(w, row, 1)
#self.event_update.connect(self.cb_update)
self.show()
def btnSyncMotors(self):
# reads the epics motor and updates the vizualization
_log.info('')
self.liveView()
def cb_update(self,*args,**kwargs):
_log.debug(f'{args} {kwargs}')
def connectEPICS(self):
_log.info('connect PVs')
self._pvDict=pvd=dict()
self._pvConnected=0
for pvn in self._pv2key.keys():
pv=epics.get_pv(pvn,connection_callback=self.OnConnectionChange,callback=self.OnValueChange)
pvd[pvn]=pv
_log.info(f'{pv}')
#epics.Motor checks the record type and will fail if the record is not online
#therefore use epics.Device
#epics.Device will force to connect the PV in Device.add_callback
#therefore use epics.PV.add_callback to acc callback
#as soon as the devices are online, they are connected
#but with epics.Device creating PV is not fully flexible. epics.get_pv proviles connection and value change callbacks that is way more flexible.
#therefore the lowest level of the library (only pvs is the best suited
#m=epics.Motor(rec_name)
#m=epics.Device(rec_name, delim='.',with_poll=False,attrs=('VAL', 'RBV', 'DESC', 'RVAL','LVIO', 'HLS', 'LLS'))
#m.add_callback('RBV', self.OnChangedRBV)
#pv=m.PV('RBV',connect=False)
#pv.add_callback(self.OnChangedRBV)
#pv.connection_callbacks
#if not pv.connected:
# disconnected.add(rec_name)
#print(pv.connected)
#devs.add(m)
def OnConnectionChange(self, pvname=None, conn=None, **kws):
pvc=self._pvConnected
if conn:
pvc+=1
else:
if pvc>0: pvc-=1
_log.info(f'PV connection {pvc}/{len(self._pvDict)}: {pvname} {conn}')
self._pvConnected=pvc
self.connectionSignal.emit(pvname, conn)
def update_connection_status(self, pvname, conn):
key = self._pv2key.get(pvname)
if not key:
return
wLb = self.labels.get(key)
if not wLb:
return
if not conn:
v=f"<font color='#a00000'>{key}</font>"
wLb.setText(v)
else:
v=f"<font color='#00a000'>{key}</font>"
wLb.setText(v)
def OnValueChange(self, pvname, value, **kw):
_log.info(f"PV val:{pvname}:{value}")
try:
key=self._pv2key[pvname]
except KeyError as e:
_log.warning(f"can't handle PV: {pvname}:{value}")
return
self.updateSignal.emit(key,value,'008000')
def vis_update(self,key,value,col='000000'):
wSl = self.sliders.get(key)
if wSl:
wSl.blockSignals(True)
wSl.setValue(int(value)) # move the slider without emiting a signal
wSl.blockSignals(False)
self.sldChanged(key,value) # emit the signal
wLb = self.labels.get(key)
if wLb:
v=f"<font color='#{col}'>{key}</font>"
wLb.setText(v)
def liveView(self):
# try to live update all PVs
_log.info('')
p2k=self._pv2key
for pv in self._pvDict.values():
pvn=pv.pvname
key=p2k[pvn]
if pv.connected:
value=pv.get()
self.vis_update(key,value,'008000')
print (pvn,key,value)
else:
print (pvn,key)
pass
def destroy(self, destroyWindow, destroySubWindows): #overloaded function
_log.info('destroy')
def closeEvent(self, event): #overloaded function
_log.info('closeEvent')
def sldChanged(self,key,val,*args,**kwargs):
app=QApplication.instance()
dev=app._dev
p=dev._paint
if key.startswith('scl'):
if key[-1]=='A':
p['sclTrf']=(2**(val/2),p['sclTrf'][1])
else:
p['sclTrf']=(p['sclTrf'][0],2**(val/2))
print(p['sclTrf'])
dev.setGeometry(dev._geo)
else:
p[key]=val
g=dev._geo
g[key]=val
wLb = self.labels.get(key)
if wLb:
v=f"<font color='#0000ff'>{key}</font>"
wLb.setText(v)
self.update()
def mouseReleaseEvent(self, a0):
try:
del self._mouseDrag
except AttributeError:
pass
def mousePressEvent(self, a0):
app=QApplication.instance()
mousePos=a0.pos()
#print(a0.type)
if a0.type()!=QMouseEvent.MouseButtonPress:
return
wGrp=self._wdGrpDraw
#if wGrp.underMouse(): #draging sliders?
if wGrp.geometry().contains(mousePos):
self._mouseDrag={'obj':wGrp, 'start':mousePos}
else:
dev=app._dev
if dev.containsPoint(mousePos):
self._devSel=dev
self._mouseDrag={'obj':dev,'start':(mousePos,dev._paint['ofs'])}
try:
_log.info(f'{self._mouseDrag}')
except AttributeError:
_log.info(f'no object to drag')
def mouseMoveEvent(self, a0):
try:
md=self._mouseDrag
except AttributeError:
return
obj=md['obj']
s=md['start']
if obj==self._wdGrpDraw:
p=a0.pos()
md['start']=p
p=obj.geometry().topLeft()+p-s
obj.move(p)
return
p=a0.pos()
ofs=QPoint(*s[1])+p-s[0]
_log.info(f'{p} {ofs}')
obj._paint['ofs']=(ofs.x(),ofs.y())
self.update()
def paintEvent(self, e):
qp = QPainter()
qp.begin(self)
qp.setRenderHints(QPainter.HighQualityAntialiasing)
app=QApplication.instance()
dev=app._dev
app._dev.paint(qp)
qp.end()
def updateDevice(self, dev):
self._devSel=dev
devP=dev._paint
wGrp=self._wdGrpDraw
wGrp.setTitle(dev._name)
for key, wSl in self.sliders.items():
if key.startswith('scl'):
if key[-1]=='A':
v=devP['sclTrf'][0]
else:
v=devP['sclTrf'][1]
v=int(round(np.log2(v)*2))
else:
v=devP[key]
wSl.blockSignals(True)
wSl.setValue(v)
wSl.blockSignals(False)
self.update()
def OnEvent(self,*args,**kwargs):
#test event
print(f'OnEvent: {args} ,{kwargs}')
if __name__ == '__main__':
import argparse
logging.basicConfig(level=logging.DEBUG, handlers=[logHandler()], format='%(levelname)s:%(module)s:%(lineno)d:%(funcName)s:%(message)s ')
def main():
epilog=__doc__ # +'\nExamples:'+''.join(map(lambda s:cmd+s, exampleCmd))+'\n'
parser=argparse.ArgumentParser(epilog=epilog, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument('--mode', '-m', type=lambda x:int(x, 0), help='mode (see bitmasks) default=0x%(default)x', default=1)
parser.add_argument("--sim", "-s", type=lambda x: int(x,0), help="simulate devices (see bitmasks) default=0x%(default)x", default=0x01)
args=parser.parse_args()
_log.info('Arguments:{}'.format(args.__dict__))
app=QApplication(sys.argv)
app._args=args
app._dev=dev=ARESdevice('Furka-ARES')
if args.mode&0x01:
app._wndVisualize=wnd=WndVisualize()
wnd.show()
sys.exit(app.exec_())
main()
Reference in New Issue View Git Blame Copy Permalink