97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
import numpy as np
|
|
import yaml
|
|
import os
|
|
import datetime
|
|
import epics
|
|
|
|
# things to do:
|
|
# 1. Read a snapshot file (not request file)
|
|
# 2. add parameters and performance channels (e.g. AT photon energies)
|
|
|
|
def parseSnapShotReqYAML(filename):
|
|
# read the snapshot request file
|
|
# returns a list of PV names
|
|
if not filename:
|
|
filename = '/sf/data/applications/snapshot/req/op/SF_settings.yaml'
|
|
pvs = []
|
|
path = os.path.dirname(filename)
|
|
with open(filename) as f:
|
|
try:
|
|
content = yaml.load(f, Loader=yaml.SafeLoader)
|
|
if 'include' in content.keys():
|
|
if len(content['include']) > 0:
|
|
for cont in content['include']:
|
|
retpv = parseSnapShotReqYAML(path+'/'+cont['name'])
|
|
if 'macros' in cont.keys():
|
|
retpv = applyReqMacro(retpv,cont['macros'])
|
|
pvs = pvs + retpv
|
|
if 'pvs' in content.keys():
|
|
if 'list' in content['pvs']:
|
|
for pv in content['pvs']['list']:
|
|
pvs.append(pv['name'])
|
|
return pvs
|
|
return None
|
|
except yaml.YAMLError as e:
|
|
print(e)
|
|
return None
|
|
return None
|
|
|
|
def applyReqMacro(pvs_in,macros):
|
|
pvs = []
|
|
for macro in macros:
|
|
for key in macro:
|
|
tag='$('+key+')'
|
|
for pv in pvs_in:
|
|
if tag in pv:
|
|
pvs.append(pv.replace(tag,macro[key]))
|
|
for pv in pvs_in: # copy the ones without macro
|
|
if not '$(' in pv:
|
|
pvs.append(pv)
|
|
return pvs
|
|
|
|
def getSnap(pvs=None):
|
|
if not isinstance(pvs,list):
|
|
pvs = parseSnapShotReqYAML(pvs)
|
|
if not pvs:
|
|
return
|
|
ret={}
|
|
val = epics.caget_many(pvs)
|
|
for i,pv in enumerate(pvs):
|
|
if not val[i] is None: # filter out None values
|
|
ret[pv]=float(val[i])
|
|
return ret
|
|
|
|
def saveSnap(pvs={},label="", comment = "generated by application",reqfile = "SF_settings.yaml"):
|
|
filename = datetime.datetime.now().strftime('/sf/data/applications/snapshot/SF_settings_%Y%m%d_%H%M%S.snap')
|
|
with open(filename,'w') as fid:
|
|
fid.write('#{"labels":["%s"],"comment":"%s", "machine_parms":{}, "save_time": 0.0, "req_file_name": "%s"}\n' % (label,comment,reqfile))
|
|
for key in pvs.keys():
|
|
if isinstance(pvs[key],int):
|
|
fid.write('%s,{"val": %d}\n' % (key,pvs[key]))
|
|
elif isinstance(pvs[key],float):
|
|
fid.write('%s,{"val": %f}\n' % (key,pvs[key]))
|
|
elif isinstance(pvs[key],str):
|
|
fid.write('%s,{"val": %s}\n' % (key,pvs[key]))
|
|
return filename
|
|
|
|
def loadSnap(filename):
|
|
res={}
|
|
with open(filename,'r') as fid:
|
|
lines=fid.readlines()
|
|
for line in lines[1:]:
|
|
split = line.split(',')
|
|
pv = split[0].strip()
|
|
val = split[1].split('"val":')[1].split('}')[0]
|
|
if '.' in val:
|
|
res[pv]=float(val)
|
|
else:
|
|
res[pv]= val
|
|
return res
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|