add preliminary PBMotionAnalyzer.py

This commit is contained in:
2016-12-22 10:13:32 +01:00
parent b98456b6ad
commit 6e31c7e705
6 changed files with 235 additions and 181 deletions

232
python/PBMotionAnalyzer.py Normal file
View File

@@ -0,0 +1,232 @@
import os,sys
import wx
import wx.py
import numpy as np
#from hdfTree import *
#from hdfGrid import *
#from hdfAttrib import *
#from hdfImage import *
import utilities as ut
class Path():
@staticmethod
def GetImage():
path=__file__
try:symPath=os.readlink(path) #follow symbolic link
except (AttributeError,OSError) as e:pass
else:
path=symPath
path=os.path.abspath(path)
path=os.path.dirname(path)
return os.path.join(path,'images')
class AboutFrame(wx.Frame):
def __init__(self,parent):
wx.Frame.__init__(self,parent,-1,'About MotionAnalyzer',size=(300,330))
imgDir=Path.GetImage()
icon = wx.Icon(os.path.join(imgDir,'PBMA.ico'), wx.BITMAP_TYPE_ICO)
self.SetIcon(icon)
self.Centre()
panel=wx.Panel(self,-1)
#import pkg_resources
#v=pkg_resources.get_distribution("h5pyViewer")
v='my version info'
s='Version:'+str(v)+'\n(c) www.psi.ch\n Author: Thierry Zamofing\n thierry.zamofing@psi.ch'
st0=wx.StaticText(panel,-1,s,(30,10))
bmp = wx.StaticBitmap(panel,-1,wx.Bitmap(os.path.join(imgDir,'PBMA.png'), wx.BITMAP_TYPE_ANY ), (30,st0.Position[1]+st0.Size[1]+10))
for k,v in os.environ.iteritems():
print k,'=',v
class MAMainFrame(wx.Frame):
def OpenFile(self,fn_npz):
try:
self.fh=fh=np.load(fn_npz)
except IOError as e:
sys.stderr.write('Unable to open File: '+fn_npz+'\n')
else:
pass
s='content of numpy file: '+fn_npz+'\n'
for k,v in fh.iteritems():
s+=' '+k+': '+str(v.dtype)+' '+str(v.shape)+'\n'
self.wxTxt.SetLabel(s)
#self.wxTree.ShowHirarchy(self.fid)
def CloseFile(self):
#http://docs.wxwidgets.org/2.8/wx_windowdeletionoverview.html#windowdeletionoverview
#print 'CloseFile'
try:
self.fh.close()
del self.fh
except AttributeError as e:
pass
def __init__(self, parent, title):
wx.Frame.__init__(self, parent, title=title, size=wx.Size(650, 350))
imgDir=Path.GetImage()
icon = wx.Icon(os.path.join(imgDir,'PBMA.ico'), wx.BITMAP_TYPE_ICO)
self.SetIcon(icon)
#wxSplt = wx.SplitterWindow(self, -1)
#wxTree = HdfTreeCtrl(wxSplt, 1, wx.DefaultPosition, (-1,-1), wx.TR_HAS_BUTTONS)
#wxTree.Bind(wx.EVT_TREE_SEL_CHANGED, self.OnSelChanged, id=1)
#wxTree.Bind(wx.EVT_TREE_ITEM_MENU, self.OnMenu, id=1)
#wx.EVT_TREE_ITEM_MENU(id, func)
#wxTxt = wx.StaticText(wxSplt, -1, '',(10,10) )#, style=wx.ALIGN_CENTRE)
self.wxTxt = wx.StaticText(self, wx.ID_ANY, "MyLabel", wx.DefaultPosition, wx.DefaultSize, 0 )
#wxSplt.SplitVertically(wxTree, wxTxt)
#wxSplt.SetMinimumPaneSize(320)
#wxLstCtrl=HdfAttrListCtrl(wxSplt)
#wxSplt.SplitVertically(wxTree, wxLstCtrl)
self.BuildMenu()
self.Centre()
#self.wxTree=wxTree
#self.display=wxTxt
def __del__(self):
self.CloseFile()
def OnOpen(self, event):
dlg = wx.FileDialog(self, "Choose a file", os.getcwd(), '','numpy files (*.npz;*.npy)|*.npz;*.npy|all (*.*)|*.*', wx.OPEN|wx.FD_CHANGE_DIR)
if dlg.ShowModal() == wx.ID_OK:
path = dlg.GetPath()
#mypath = os.path.basename(path)
#self.SetStatusText("You selected: %s" % mypath)
self.CloseFile()
self.OpenFile(path)
#print 'OnOpen',path
dlg.Destroy()
def OnCloseWindow(self, event):
#print 'OnCloseWindow'
self.Destroy()
def OnAbout(self,event):
frame=AboutFrame(self)
frame.Show()
def BuildMenu(self):
#http://wiki.wxpython.org/AnotherTutorial#wx.MenuBar
mnBar = wx.MenuBar()
#-------- File Menu --------
mn = wx.Menu()
mnItem=mn.Append(wx.ID_OPEN, '&Open', 'Open a new document');self.Bind(wx.EVT_MENU, self.OnOpen, mnItem)
#mnSub = wx.Menu()
#mnItem=mnSub.Append(wx.ID_ANY, 'SubMenuEntry', 'My SubMenuEntry')
#mn.AppendMenu(wx.ID_ANY, 'SubMenu', mnSub)
mn.AppendSeparator()
mnItem=mn.Append(wx.ID_EXIT, '&Quit', 'Quit the Application');self.Bind(wx.EVT_MENU, self.OnCloseWindow, mnItem)
mnBar.Append(mn, '&File')
self.Bind(wx.EVT_CLOSE, self.OnCloseWindow)
#-------- Edit Menu --------
mn = wx.Menu()
mnItem = mn.Append(wx.ID_ANY, 'Show &XY-Path', 'Show XY-path motion path');self.Bind(wx.EVT_MENU, self.OnShowXYPath, mnItem)
mnItem = mn.Append(wx.ID_ANY, 'Show &Error', 'Show error of motion path');self.Bind(wx.EVT_MENU, self.OnShowError, mnItem)
mnItem = mn.Append(wx.ID_ANY, 'Show &Velocity', 'Show velocity of motion path');self.Bind(wx.EVT_MENU, self.OnShowVelocity, mnItem)
mnItem = mn.Append(wx.ID_ANY, '&Python Shell', 'Opens an interactive python shell"');self.Bind(wx.EVT_MENU, self.OnShell, mnItem)
mnBar.Append(mn, '&Window')
#-------- Help Menu --------
mn = wx.Menu()
#mnItem=mn.Append(wx.ID_HELP,'Help','Application Help')
mnItem=mn.Append(wx.ID_ABOUT,'About','Application About');self.Bind(wx.EVT_MENU, self.OnAbout, mnItem)
mnBar.Append(mn, '&Help')
#mn.AppendSeparator()
#mnItem = wx.MenuItem(mn, 105, '&Quit\tCtrl+Q', 'Quit the Application')
#mnItem.SetBitmap(wx.Image('stock_exit-16.png', wx.BITMAP_TYPE_PNG).ConvertToBitmap())
#mn.AppendItem(mnItem)
self.SetMenuBar(mnBar)
self.CreateStatusBar()
def OnShowXYPath(self, event):
pass
def OnShowError(self, event):
pass
def OnShowVelocity(self, event):
pass
def OnShell(self, event):
frame = wx.Frame(self, -1, "wxPyShell",size=wx.Size(800, 500))
imgDir=Path.GetImage()
icon = wx.Icon(os.path.join(imgDir,'PBMA.ico'), wx.BITMAP_TYPE_ICO)
frame.SetIcon(icon)
frame.Centre()
fh=app.GetTopWindow().fh
wnd=app.GetTopWindow()
loc={'app' :app,
'fh' :fh,
'pts' :fh['pts'],
'rec' :fh['rec']
}
introText='''Shell to the HDF5 objects
app: application object
fh: numpy file
pts: desired motion points
rec: recorded data
#Examples:
pts
rec
#using user defined modules
#import userSample as us;reload(us);us.test1(hid)
'''
shell=wx.py.shell.Shell(frame, introText=introText,locals=loc)
frame.Show(True)
#if loc is None, all variables are visible. the context is global
#shell.push('wnd=app.GetTopWindow()')
#for cmd in [
# 'wnd=app.GetTopWindow();wxTree=wnd.wxTree',
# 'wxNode=wnd.wxTree.GetSelection()',
# 'print wnd.fid',
# 'lbl=wxTree.GetItemText(wxNode)',
# 'hid=wxTree.GetPyData(wxNode)']:
# shell.run(cmd, prompt=False)
if __name__ == '__main__':
def GetArgs():
import sys,argparse #since python 2.7
exampleCmd='/tmp/shapepath.npz'
parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
description=__doc__,
epilog='Example:\n'+os.path.basename(sys.argv[0])+' '+exampleCmd+'\n ')
parser.add_argument('npzFile', nargs='?', help='the npz file with motion path data to show',default='/tmp/shapepath.npz')
args = parser.parse_args()
return args
class MyApp(wx.App):
def OnInit(self):
args=GetArgs()
frame = MAMainFrame(None, 'PBMotionAnalyzer')
if args.npzFile:
frame.OpenFile(args.npzFile)
frame.Show(True)
self.SetTopWindow(frame)
return True
#------------------ Main Code ----------------------------------
#redirect stdout/stderr:
#http://www.blog.pythonlibrary.org/2009/01/01/wxpython-redirecting-stdout-stderr/
#https://groups.google.com/forum/#!topic/wxpython-users/S9uSKIYdYoo
#https://17677433047266577941.googlegroups.com/attach/e4d343dc6a751906/REDIRECT.PY?part=2&view=1&vt=ANaJVrFeyCjCMydKnkyfFbYJM7ip07mE-ozUIBxJ5A1QuK1GhycJYJsPTxpAaNk5L2LpXvGhzRPInxDt8_WUcUyK2Ois28Dq8LNebfYoWG9Yxr-tujf5Jk4
#http://www.wxpython.org/docs/api/wx.PyOnDemandOutputWindow-class.html
rd=not sys.stdout.isatty()#have a redirect window, if there is no console
#rd=True #force to open a redirect window
rd=False #avoid a redirect window
app = MyApp(redirect=rd)
app.MainLoop()

BIN
python/images/PBMA.ico Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.2 KiB

BIN
python/images/PBMA.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.4 KiB

272
python/shapepath.py Executable file
View File

@@ -0,0 +1,272 @@
#!/usr/bin/env python
# *-----------------------------------------------------------------------*
# | |
# | Copyright (c) 2016 by Paul Scherrer Institute (http://www.psi.ch) |
# | |
# | Author Thierry Zamofing (thierry.zamofing@psi.ch) |
# *-----------------------------------------------------------------------*
'''
shape an optimal path with given points
verbose bits:
1 basic info
2 plot sorting steps
4 list program
4 upload progress
8 plot gather path
#config file example:
{
"points": [
[100,523],[635,632],[756,213],
"sequencer":[
'gen_grid_points(w=10,h=10,pitch=100,rnd=.2)',
'sort_points()',
'gen_prog(file="/tmp/shapepath.prg")',
'plot_gather()']
}
Sequencer functions are:
- generate points (if not in the 'points' configuration)
gen_rand_points(self,n=107,scale=1000)
gen_grid_points(w=10,h=10,pitch=100,rnd=.2)
- sorting points:
sort_points(self)
- generate/download/execute motion progran, upload trace of motors (gather data)
gen_prog(self,prgId=2,file=None,host=None)
if host=None nothing will be downloaded/executed and trace of motors will not be uploaded
if file=None the program will not be saved and nothing will be executed
- plot gathered data
plot_gather()
this makes only sence, if motion has been executed and data can be gathered from the powerbrick
Acquired time is:MaxSamples*Period*.2
'''
import os, sys, json
import numpy as np
import matplotlib as mpl
import matplotlib.pyplot as plt
import subprocess as sprc
import telnetlib
from utilities import *
class ShapePath:
def __init__(self,args):
if args.cfg:
fh=open(args.cfg,'r')
s=fh.read()
cfg=json.loads(s, object_hook=ConvUtf8)
s=json.dumps(cfg, indent=2, separators=(',', ': '));print s
else:
#cfg={"points": [[100,523],[635,632],[756,213]],"sequencer":['sort_points()','gen_prog(file="/tmp/shapepath.prg",host="SAROP11-CPPM-MOT6871")','plot_gather()']}
#cfg={"sequencer":['gen_rand_points(n=107, scale=1000)','sort_points()','gen_prog(file="/tmp/shapepath.prg",host="SAROP11-CPPM-MOT6871")','plot_gather()']}
#cfg={"sequencer":['gen_grid_points(w=10,h=10,pitch=100,rnd=.2)','sort_points()','gen_prog(file="/tmp/shapepath.prg",host="SAROP11-CPPM-MOT6871")','plot_gather()']}
#cfg={"sequencer":['gen_grid_points(w=10,h=10,pitch=100,rnd=0.2)','sort_points()','gen_prog(file="/tmp/shapepath.prg")','plot_gather()']}
#cfg = {"sequencer": ['gen_rand_points(n=107, scale=1000)', 'sort_points()','plot_gather()']}
cfg={"sequencer":['gen_grid_points(w=20,h=20,pitch=50,rnd=.2)','sort_points()','gen_prog(file="/tmp/shapepath.prg",host="SAROP11-CPPM-MOT6871")','plot_gather()']}
cfg={"sequencer":['gen_grid_points(w=20,h=20,pitch=50,rnd=.2)','sort_points()','gen_prog(file="/tmp/shapepath.prg")','plot_gather()']}
#cfg={"sequencer":['gen_rand_points(n=400, scale=1000)','sort_points()','gen_prog(file="/tmp/shapepath.prg",host="SAROP11-CPPM-MOT6871")','plot_gather()']}
self.cfg=dotdict(cfg)
self.args=args
def run(self):
print('args='+str(self.args))
print('cfg='+str(self.cfg))
try:
self.points=np.array(self.cfg.points)
except AttributeError:
pass
try:
sequencer= self.cfg.pop('sequencer')
except KeyError:
print('no command sequence to execute')
else:
dryrun=self.args.dryrun
for cmd in sequencer:
print '>'*5+' '+cmd+' '+'<'*5
if not dryrun:
eval('self.' + cmd)
def gen_rand_points(self,n=107,scale=1000):
np.random.seed(0)
#data=np.random.randint(0,1000,(30,2))
pts=np.random.rand(n,2)*scale
self.points=pts
def gen_grid_points(self,w=10,h=10,pitch=100,rnd=.2):
np.random.seed(0)
xx,yy=np.meshgrid(range(w), range(h))
pts=np.array([xx.reshape(-1),yy.reshape(-1)],dtype=np.float).transpose()*pitch
if rnd != 0:
pts+=(np.random.rand(pts.shape[0],2)*(rnd*pitch))
self.points=pts
def gen_prog(self,prgId=2,file=None,host=None,mode=0):
prg=[]
gather={"MaxSamples":100000, "Period":10 }
#channels=["Motor[1].ActPos","Motor[2].ActPos","Motor[3].ActPos"]
channels=["Motor[1].ActPos","Motor[2].ActPos","Motor[3].ActPos","Motor[1].DesPos","Motor[2].DesPos","Motor[3].DesPos"]
prg.append('Gather.Enable=0')
prg.append('Gather.Items=%d'%len(channels))
for k,v in gather.iteritems():
prg.append('Gather.%s=%d'%(k,v))
for i,c in enumerate(channels):
prg.append('Gather.Addr[%d]=%s.a'%(i,c))
prg.append('open prog %d'%(prgId))
# this uses Coord[1].Tm and limits with MaxSpeed
prg.append('Gather.Enable=2')
if mode==0:
prg.append(' linear abs')
data=self.points
for idx in range(data.shape[0]):
prg.append('X(%f) Y(%f)'%tuple(data[idx,:]))
prg.append('dwell 100')
if mode==1:
pass
prg.append('Gather.Enable=0')
prg.append('close')
prg.append('&1\nb%dr\n'%prgId)
if self.args.verbose & 4:
for ln in prg:
print(ln)
if file is not None:
fh=open(file,'w')
fh.write('\n'.join(prg))
fh.close()
if host is not None:
cmd ='gpasciiCommander --host '+host+' '+ file
print cmd
p = sprc.Popen(cmd, shell=True)#, stdout=sprc.PIPE, stderr=sprc.STDOUT)
#res=p.stdout.readlines(); print res
retval = p.wait()
#gather -u /var/ftp/gather/out.txt
cmd ='PBGatherPlot -m24 -v7 --host '+host
print cmd
p = sprc.Popen(cmd, shell=True)#, stdout=sprc.PIPE, stderr=sprc.STDOUT)
retval = p.wait()
self.prg=prg
def sort_points(self):
pts=self.points
verb=self.args.verbose
#if verb&2:
# self.plot_points(pts)
#sort points along y
pts=pts[pts[:, 1].argsort()]
#if verb&2:
# self.plot_points(pts)
#group sorting
cnt=pts.shape[0]
idx=np.ndarray(cnt,dtype=np.int32)
grp_cnt=int(np.sqrt(cnt))
grp_sz=int(np.ceil(float(cnt)/grp_cnt))
for i in range(grp_cnt):
a=i*grp_sz
#print a,a+grp_sz
if i%2:
idx[a:a+grp_sz]=a+pts[a:a+grp_sz,0].argsort()[::-1]
else:
idx[a:a+grp_sz]=a+pts[a:a+grp_sz,0].argsort()
#print(idx)
pts=pts[idx]
if verb&2:
self.plot_points(pts)
plt.show()
self.points=pts
@staticmethod
def onclick(event):
print 'button=%s, x=%d, y=%d, xdata=%f, ydata=%f'%(
event.button, event.x, event.y, event.xdata, event.ydata)
obj=event.canvas.figure.obj
def plot_points(self,pts):
fig=plt.figure()
ax = fig.add_subplot(1,1,1)
#hl=ax[0].plot(x, y, color=col)
hl=ax.plot(pts[:,0],pts[:,1],'r.')
hl=ax.plot(pts[:,0],pts[:,1],'y--')
cid = fig.canvas.mpl_connect('button_press_event', self.onclick)
fig.obj=self
self.ax=ax
self.hl=hl
def plot_gather(self,fnLoc='/tmp/gather.txt',fnOut='/tmp/shapepath.npz'):
pts=self.points
rec = np.genfromtxt(fnLoc, delimiter=' ')
if fnOut:
np.savez_compressed(fnOut, rec=rec, pts=pts)
fig=plt.figure()
ax = fig.add_subplot(1,1,1)
#hl=ax[0].plot(x, y, color=col)
hl=ax.plot(pts[:,0],pts[:,1],'r.')
hl=ax.plot(pts[:,0],pts[:,1],'y--')
hl = ax.plot(rec[:, 4], rec[:, 5], 'b-')
hl=ax.plot(rec[:,1],rec[:,2],'g-')
ax.xaxis.set_label_text('x-pos um')
ax.yaxis.set_label_text('y-pos um')
cid = fig.canvas.mpl_connect('button_press_event', self.onclick)
fig.obj=self
self.ax=ax
self.hl=hl
fig = plt.figure()
ax = fig.add_subplot(1, 1, 1)
err=np.sqrt((rec[:,1]-rec[:,4])**2+(rec[:,1]-rec[:,4])**2)
hl = ax.plot(err, 'r-')
ax.xaxis.set_label_text('time (10x servo cycle)')
ax.yaxis.set_label_text('pos-error um')
plt.show()
if __name__=='__main__':
from optparse import OptionParser, IndentedHelpFormatter
class MyFormatter(IndentedHelpFormatter):
'helper class for formating the OptionParser'
def __init__(self):
IndentedHelpFormatter.__init__(self)
def format_epilog(self, epilog):
if epilog:
return epilog
else:
return ""
def parse_args():
'main command line interpreter function'
#usage: gpasciiCommunicator.py --host=PPMACZT84 myPowerBRICK.cfg
(h, t)=os.path.split(sys.argv[0]);cmd='\n '+(t if len(h)>3 else sys.argv[0])+' '
exampleCmd=('--host=PPMAC1391 -m 63 --cfg gather.cfg',
'samplePowerBrick.cfg',
'-n stackCheck1.cfg',
'--host=PPMACZT84 stackCheck1.cfg',
'--host=PPMACZT84 stackCheck1.cfg -v15',
)
epilog=__doc__+'''
Examples:'''+''.join(map(lambda s:cmd+s, exampleCmd))+'\n '
fmt=MyFormatter()
parser=OptionParser(epilog=epilog, formatter=fmt)
parser.add_option('-v', '--verbose', type="int", dest='verbose', help='verbosity bits (see below)', default=0)
parser.add_option('-n', '--dryrun', action='store_true', help='dryrun to stdout')
parser.add_option('--cfg', help='config file containing json configuration structure')
(args, other)=parser.parse_args()
args.other=other
sp=ShapePath(args)
sp.run()
#------------------ Main Code ----------------------------------
#ssh_test()
ret=parse_args()
exit(ret)

93
python/utilities.py Executable file
View File

@@ -0,0 +1,93 @@
#!/usr/bin/env python
#*-----------------------------------------------------------------------*
#| |
#| Copyright (c) 2016 by Paul Scherrer Institute (http://www.psi.ch) |
#| |
#| Author Thierry Zamofing (thierry.zamofing@psi.ch) |
#*-----------------------------------------------------------------------*
'''
utilities classes
'''
import logging, h5py, re, zlib, zmq, json
import numpy as np
from libDetXR import *
import time
class dotdict(dict):
"""dot.notation access to dictionary attributes"""
def __init__(self,arg=None,**kwargs):
if arg!=None:
self.__fill__(arg)
self.__fill__(kwargs)
def __fill__(self,kw):
for k,v in kw.iteritems():
if type(v)==dict:
self[k]=dotdict(v)
else:
self[k]=v
if type(v)==list:
for i,w in enumerate(v):
if type(w)==dict:
v[i]=dotdict(w)
pass
def __dir__(self):
l=dir(object)
#l.extend(self.keys())
l.extend(map(str,self.keys()))
return l
def __getattr__(self, attr):
#return self.get(attr)
try:
return self[attr]
except KeyError as e:
raise AttributeError("%r instance has no attribute %r" % (self.__class__, attr))
def __repr__(self):
return '<' + dict.__repr__(self)[1:-1] + '>'
def PrettyPrint(self,indent=0):
for k,v in self.iteritems():
if type(v)==dotdict:
print ' '*indent,str(k)+':'
v.PrettyPrint(indent+2)
else:
print ' '*indent+str(k)+'\t'+str(v)
__setattr__= dict.__setitem__
__delattr__= dict.__delitem__
#__getattr__= dict.__getattr__
def ConvUtf8(s):
'convert unicoded json object to ASCII encoded'
#http://stackoverflow.com/questions/956867/how-to-get-string-objects-instead-of-unicode-ones-from-json-in-python
if isinstance(s, dict):
return {ConvUtf8(key): ConvUtf8(value) for key, value in s.iteritems()}
elif isinstance(s, list):
return [ConvUtf8(element) for element in s]
elif isinstance(s, unicode):
return s.encode('utf-8')
else:
return s
class GpasciiCommunicator():
'''Communicates with the Delta Tau gpascii programm
'''
gpascii_ack="\x06\r\n"
gpascii_inp='Input\r\n'
def connect(self, host, username='root', password='deltatau',prompt='ppmac# '):
p=telnetlib.Telnet(host)
print p.read_until('login: ')
p.write(username+'\n')
print p.read_until('Password: ')
p.write(password+'\n')
print p.read_until(prompt) # command prompt
p.write('gpascii -2\n') # execute gpascii command
print p.read_until(self.gpascii_inp)
return p