slic/sfbd
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial debug of slic scan for the dispersion tool

Browse Source
This commit is contained in:
reiche
2023-07-25 09:20:43 +02:00
parent 10c7b37b99
commit 25e717e132
3 changed files with 106 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
app/dispersiontools.py
View File

@ -10,6 +10,8 @@ from slic.core.acquisition import BSAcquisition, PVAcquisition
from slic.core.scanner import Scanner
from sfbd.ext import CounterAdjustable
from sfbd.ext import BSCAcquisition
from sfbd.interface import SlicScan
from bstrd import BSCache
def getAux(pvs=None):
if not pvs:
@ -19,7 +21,6 @@ def getAux(pvs=None):
for i,pv in enumerate(pvs):
if val[i]: # filter out None values
ret[pv]=float(val[i])
epics.ca.clear_cache()
return ret
@ -30,21 +31,32 @@ def getBSChannels(regexp):
if prog.match(bs['name']):
res.append(bs['name'])
return res
def getRFCalibrationChannels(sensors,energy):
aux=[]
for sen in sensors:
if 'PHASE-VS' in sen:
aux.append(sen.replace('PHASE-VS','GET-VSUM-PHASE-OFFSET').replace('RLLE-DSP','RSYS'))
aux.append(sen.replace('PHASE-VS','GET-VSUM-AMPLT-SCALE').replace('RLLE-DSP','RSYS'))
aux.append(sen.replace('PHASE-VS','SM-GET').replace('RLLE-DSP','RMSM'))
aux.append(energy)
return aux
class Dispersion:
def __init__(self, branch = 'Aramis'):
self.scanname = 'Dispersion'
self.branch = 'None'
self.branch = None
dirpath= datetime.datetime.now().strftime('measurements/%Y/%m/%d/slic_sfbd')
self.basedir = '/sf/data/%s/%s' % (dirpath,self.scanname)
self.pgroup='%s/%s' % (dirpath,self.scanname)
self.scandir='/sf/data/'+self.pgroup
self.setBranch(branch)
self.sc = None
self.setBranch('Athos Dump') # enfore Athos dump settings for now
self.Nsteps = 2
self.Nsamples = 1
def setBranch(self,branch = 'Aramis'):
@ -56,29 +68,61 @@ class Dispersion:
elif branch == 'Athos':
self.setupAthos()
else:
self.scanner = None
self.branch = None
def setup(self,scl = 1, Nsteps=5, Nsamples=10):
# the setup is done from the main thread. Therefore I have to define the BSC here
self.N = Nsteps
self.Ns= Nsamples
self.scale = scl
# define stream
self.bsc = BSCache(100000,10000) # 100 second timeout, size for 100 second data taken
self.bsc.get_vars(self.sensor) # this starts the stream into the cache
def scan(self):
# core routine to do all action. Will be a thread of the slic scanner wrapper
if not self.branch:
return
# path is define in the various set-up procedures
self.scanner = Scanner(data_base_dir=self.basedir,scan_info_dir=self.basedir,
make_scan_sub_dir=True,
default_acquisitions=self.acq)
# define acquisition channels wrapper for BSQueue
pgroup = '%s-%s' % (self.pgroup,self.branch)
acq = [BSCAcquisition(".",pgroup, default_channels=[self.bsc])]
# define the scanner
self.scanner = Scanner(data_base_dir=self.basedir, # root directory of data location e.g. /sf/data/2023/02/02/Dispersion
scan_info_dir=self.basedir, # write also scan info there
make_scan_sub_dir=True, # put each scan in its own directory
default_acquisitions=acq) # adjustable to use
# define adjustable
self.adj = PVAdjustable(self.adjSV,pvname_readback = self.adjRB, accuracy = 0.1,ID = self.adjSV, name = self.name)
val = self.adj.get_current_value(readback=False)
# dval = self.amp*scl
# self.values=np.linspace(val-dval,val+dval,num=self.N)
dval = 0
values = np.linspace(val-dval,val+dval, num = self.N)
# create scanner backend
self.sc=self.scanner.ascan_list(self.adj, values, # list of adjustables and their values
filename=self.branch, # relative directory for data
start_immediately = False, # wait for execution to performe pre-action items
n_pulses=self.Ns, # steps
return_to_initial_values=True)# return to initial values
# self.preaction() ######
self.sc.run() # run scan
# self.auxdata = getAux(self.aux)
# self.postaction() #######
self.bsc.stop() # stop the queue
def getRFCalibrationChannels(self,sensor2,energy):
aux=[]
for sen in sensor2:
if 'PHASE-VS' in sen:
aux.append(sen.replace('PHASE-VS','GET-VSUM-PHASE-OFFSET').replace('RLLE-DSP','RSYS'))
aux.append(sen.replace('PHASE-VS','GET-VSUM-AMPLT-SCALE').replace('RLLE-DSP','RSYS'))
aux.append(sen.replace('PHASE-VS','SM-GET').replace('RLLE-DSP','RMSM'))
aux.append(energy)
return aux
def setupAthosDump(self):
# pathes and name tags
self.branch='Athos_Dump'
pgroup = '%s-%s' % (self.pgroup,self.branch)
self.path = '/sf/data/'+pgroup
self.name ='SATCB01-Linac'
# pre-scan item - needs an adjustment of the dipole current of about 2.3e-3 - to be implemented later.
self.pre = {}
self.pre['SFB_BEAM_DUMP_AT:ONOFF1']={'Val':0,'InitVal':0}
@ -87,18 +131,22 @@ class Dispersion:
self.pre['SFB_ORBIT_SAT_%2.2d:ONOFF1' % i ]={'Val':0,'InitVal':0}
for pv in self.pre.keys():
self.pre[pv]['adj']=PVAdjustable(pv)
# adjustable
self.adjSV = 'SATCB01-RSYS:SET-BEAM-PHASE'
self.adjRB = 'SATCB01-RSYS:GET-BEAM-PHASE'
self.adj = PVAdjustable(self.adjSV,pvname_readback = self.adjRB, accuracy = 0.1,ID = self.adjSV, name = "SATCB01-Linac")
self.amp = 30 # the amplitude of the scan, which can be scaled
# acquisition
sensor1 = getBSChannels('SATBD02-DBPM.*:Y2$')
sensor2 = getBSChannels('SATCB.*-RLLE-DSP:.*-VS$')
self.sensor = sensor1+sensor2
self.acq = [BSAcquisition(".",pgroup, default_channels=self.sensor)]
# auxiliar data to be read one
self.aux = self.getRFCalibrationChannels(sensor2,'SATCL01-MBND100:ENERGY-OP')
self.aux = getRFCalibrationChannels(sensor2,'SATCL01-MBND100:ENERGY-OP')
def setupAramis(self):
self.branch='Aramis'
@ -157,13 +205,6 @@ class Dispersion:
self.aux = self.getRFCalibrationChannels(sensor2,'S10BC02-MBND100:ENERGY-OP')
def setup(self,scl = 1, Nsteps=5, Nsamples=5):
val = self.adj.get_current_value(readback=False)
# dval = self.amp*scl
dval = 0 ######## edit this
self.N = Nsteps
self.Ns= Nsamples
self.values=np.linspace(val-dval,val+dval,num=self.N)
def preaction(self):
@ -175,39 +216,44 @@ class Dispersion:
for key in self.pre.keys():
self.pre[key]['adj'].set_target_value(self.pre[key]['InitVal'])
def scan(self):
self.sc=self.scanner.ascan_list(self.adj,self.values,
filename=self.branch,start_immediately = False,
n_pulses=self.Ns,return_to_initial_values=True)
# self.preaction() ######
self.sc.run()
# self.auxdata = getAux(self.aux)
# self.postaction() #######
def stop(self):
if self.sc is None:
return
self.sc.stop()
def running(self):
if self.sc is None:
return False
return self.sc.running
def status(self):
if self.sc is None:
return 0,0
si = self.sc.scan_info.to_dict()
steps = 0
if 'scan_values' in si:
steps=len(si['scan_values'])
steps=len(si['scan_values'])
return steps,self.N
def info(self):
if self.sc is None:
return None
return self.sc.scan_info.to_dict()
#--------------------
# for debugging
if __name__ == '__main__':
daq = Dispersion()
daq.setup(1,10,100)
# threaded execution
scanner=SlicScan()
scanner.start(daq,True)

23
interface/slic.py
View File

@ -25,22 +25,15 @@ class SlicScan(QObject):
self.data = None
self.act = None
self.snap = None
self.doSnap = False
def start(self,daq,snap=False):
self.clear()
self.doSnap = snap
self.daq=daq
Thread(target=self.Tmonitor).start()
self.startSnap(snap)
def startSnap(self,snap=False):
if snap:
Thread(target=self.Tsnap).start()
def Tsnap(self):
self.snap = getSnap()
self.sigsnap.emit(True)
def Tmonitor(self):
def Tmonitor(self):
mythread = Thread(target=self.Tscanner).start()
time.sleep(1)
ostep = -1
@ -58,10 +51,20 @@ class SlicScan(QObject):
if hasattr(self.daq,'auxdata'):
self.data.update(self.daq.auxdata)
self.sigterm.emit(istep==nstep)
# if requested add snapshot acquisition
if self.doSnap:
self.startSnap()
def Tscanner(self):
self.daq.scan()
def startSnap(self):
Thread(target=self.Tsnap).start()
def Tsnap(self):
self.snap = getSnap()
self.sigsnap.emit(True)
def stop(self):
self.daq.stop()

1
interface/snap.py
View File

@ -60,6 +60,7 @@ def getSnap(pvs=None):
if val[i]: # filter out None values
ret[pv]=float(val[i])
epics.ca.clear_cache()
return ret
def saveSnap(pvs={},label="", comment = "generated by application"):