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

Added app support for adaptive compression

Browse Source
This commit is contained in:
Sven Reiche
2026-02-07 10:54:18 +01:00
parent 5a9a849b0d
commit a621c7f54b
4 changed files with 271 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/__init__.py
View File

@@ -1,5 +1,6 @@
from .adaptiveorbit import AdaptiveOrbit
from .adaptiveorbit_new import AdaptiveOrbitNew
from .adaptivecompression import AdaptiveCompression
#from .adaptiveorbit_new import AdaptiveOrbitNew
from .spectralanalysis import SpectralAnalysis
from .hero import LaserPower,EnergyModulation
from .dispersiontools import Dispersion

134
app/adaptivecompression.py Normal file
View File

@@ -0,0 +1,134 @@
import time
import numpy as np
from bstrd import BSCache
from bstrd.bscache import is_available
from epics import PV
class AdaptiveCompression:
"""
Wrapper class to bundle all daq/io needed for adaptive orbit feedback.
"""
def __init__(self):
self.BS=None
self.PVKick=[]
self.PVFB = []
self.BSChannels=[]
self.BS=None
self.PVKick=None
self.PVFB=None
def getChannels(self,beamline,bunch):
if beamline == 'Aramis':
ch0 = ['SARFE10-PBIG050-EVR0:CALCI']
pvFB = ['SFB_COMPRESSION_BC1_AR:SP1','SFB_COMPRESSION_BC2_AR:SP1']
elif beamline == 'Athos':
ch0 = ['SATFE10-PEPG046-EVR0:CALCI']
pvFB = ['SFB_COMPRESSION_BC1_AT:SP1','SFB_COMPRESSION_BC2_AT:SP1']
else:
return False
self.pvPhase = []
if bunch == 1:
ch1 = ['SINBC02-DBCM410:LM-AG-CH2-B1', 'S10MA01-DCDR080:LM-AG-CH1-B1']
pvPhase = ['SINDLH:SET-BEAM-PHASE-OP','S10:SET-BEAM-PHASE-OP','SFB_COMPRESSION_BC1_AR:ONOFF1','SFB_COMPRESSION_BC2_AR:ONOFF1']
elif bunch == 2:
ch1 = ['SINBC02-DBCM410:LM-AG-CH2-B2','S10MA01-DCDR080:LM-AG-CH1-B2']
pvPhase = ['SINDLH:SET-BEAM-PHASE-OP','S10:SET-BEAM-PHASE-OP','SFB_COMPRESSION_BC1_AT:ONOFF1','SFB_COMPRESSION_BC2_AT:ONOFF1']
else:
return False
self.monchannel=ch1
print("Initializing BSStream for beamline %s and bunch %d" % (beamline,bunch))
self.BS = self.initBSStream(ch0+ch1)
if self.BS is None:
print('Failed to establish BS-stream')
return False
print("Initializing EPICS Channels")
self.PVFB=self.initPV(pvFB)
self.PVPhase=self.initPV(pvPhase)
if self.PVFB is None or self.PVPhase is None:
print('Failed to establish PV-channels')
return False
return True
def initBSStream(self,channels):
ok = True
for chn in channels:
ok &= is_available(chn)
if ok:
bs = BSCache(1000,receive_timeout=1000) # 1000 second time out, capazity for 1000 second.
bs.get_vars(channels)
else:
for chn in channels:
if not is_available(chn):
print('ERROR: BS-Channel %s not found in BS-Dispatcher' % ch)
bs = None
self.BSChannels.clear()
self.BSChannels=channels
return bs
def initPV(self,chx):
pvs = []
for x in chx:
pvs.append(PV(x))
con = [pv.wait_for_connection(timeout=0.2) for pv in pvs]
ok = True
for i, val in enumerate(con):
ok &=val
if val is False:
name = pv[i].pvname
print("ERROR: PV-Channel %s is not available" % name)
if ok:
return pvs
return None
def flush(self):
self.BS.flush()
def terminate(self):
print('Stopping BSStream Thread...')
self.BS.stop()
self.BS.pt.running.clear() # for some reason I have to
def read(self):
if self.BS is None:
return None
data=self.BS.__next__()
return np.array([data['pid']]+[data[cnl] for cnl in self.BSChannels])
def readPV(self):
if self.PVFB is None:
return []
return [pv.value for pv in self.PVFB]
def setPV(self,fbval):
if self.PVFB is None:
return
for i in range(len(fbval)):
self.PVFB[i].value = fbval[i]
def getPVNames(self):
if self.PVFB is None:
return []
return [pv.pvname for pv in self.PVFB]
def getDetectorName(self):
return self.BSChannels[0]
def getPhase(self):
if self.PVPhase is None:
return None
return [pv.value for pv in self.PVPhase]
def setPhase(self,vals):
if self.PVPhase is None:
return
for i,val in enumerate(vals):
self.PVPhase[i].value = val

5
app/adaptiveorbit.py
View File

@@ -18,6 +18,7 @@ class AdaptiveOrbit:
self.bsAR = self.initBSStream([self.ARch0]+self.ARchx+self.ARchy)
self.pvAR = self.initPV(self.ARchx)
self.kickerAR = self.initPV(['SARMA02-MCRX050:I-SET','SARMA02-MCRY050:I-SET','SARUN02-MCRX080:I-SET','SARUN02-MCRY080:I-SET','SFB_ORBIT_SAR:ONOFF1'])
# Athos Channels
self.ATch0 = 'SATFE10-PEPG046-EVR0:CALCI'
@@ -28,7 +29,7 @@ class AdaptiveOrbit:
if bpm == 5 or bpm ==14:
idx='410'
self.ATchx.append('SATUN%2.2d-DBPM%s:X1' % (bpm,idx))
self.ATchy.append('SATUN%2.2d-DBPM%s:Y1' % (bpm,idx))
self.ATchy.append('SATUN%2.2d-DBPM%s:Y1' % (bpm,idx))
self.bsAT = self.initBSStream([self.ATch0]+self.ATchx+self.ATchy)
self.pvAT = self.initPV(self.ATchx)
self.kickerAT = self.initPV(['SATMA01-MCRX610:I-SET','SATMA01-MCRY610:I-SET','SATUN05-MCRX420:I-SET','SATUN05-MCRY420:I-SET','SFB_ORBIT_SAT:ONOFF1'])
@@ -82,7 +83,9 @@ class AdaptiveOrbit:
def initBSStream(self,channels):
print("Initializing BSstream")
bs = BSCache(1000,receive_timeout=1000) # 1000 second time out, capazity for 1000 second.
print("Getting Channels")
bs.get_vars(channels)
print("Done")
return bs
def intitPV_New(self,chx):

131
app/adaptiveorbit_temp.py Normal file
View File

@@ -0,0 +1,131 @@
import time
import numpy as np
from bstrd import BSCache
from bstrd.bscache import make_channel_config, is_available
from epics import PV
class AdaptiveOrbit:
"""
Wrapper class to bundle all daq/io needed for adaptive orbit feedback.
"""
def __init__(self):
# Aramis Channels
self.ARch0 = 'SARFE10-PBIG050-EVR0:CALCI'
self.ARchx = ['SARUN%2.2d-DBPM070:X1' % id for id in range(1,17)]
self.ARchy = ['SARUN%2.2d-DBPM070:Y1' % id for id in range(1,17)]
self.bsAR = self.initBSStream([self.ARch0]+self.ARchx+self.ARchy)
self.pvAR = self.initPV(self.ARchx)
self.kickerAR = self.initPV(['SARMA02-MCRX050:I-SET','SARMA02-MCRY050:I-SET','SARUN02-MCRX080:I-SET','SARUN02-MCRY080:I-SET','SFB_ORBIT_SAR:ONOFF1'])
# Athos Channels
self.ATch0 = 'SATFE10-PEPG046-EVR0:CALCI'
self.ATchx=['SATUN04-DBPM010:X1','SATUN04-DBPM040:X1']
self.ATchy=['SATUN04-DBPM010:Y1','SATUN04-DBPM040:Y1']
for bpm in range(5,23):
idx = '070'
if bpm == 5 or bpm ==14:
idx='410'
self.ATchx.append('SATUN%2.2d-DBPM%s:X1' % (bpm,idx))
self.ATchy.append('SATUN%2.2d-DBPM%s:Y1' % (bpm,idx))
self.bsAT = self.initBSStream([self.ATch0]+self.ATchx+self.ATchy)
self.pvAT = self.initPV(self.ATchx)
self.kickerAT = self.initPV(['SATMA01-MCRX610:I-SET','SATMA01-MCRY610:I-SET','SATUN05-MCRX420:I-SET','SATUN05-MCRY420:I-SET','SFB_ORBIT_SAT:ONOFF1'])
# select first beamline
self.isAramis = True
def initBSStream(self,channels):
print("Initializing BSstream")
bs = BSCache(100000,receive_timeout=100000) # 1000 second time out, capazity for 1000 second.
bs.get_vars(channels)
return bs
def initPV(self,chx):
print("Initializing EPICS Channels")
pvs = []
for x in chx:
if ':X1' in x:
pvs.append(PV(x.replace(':X1',':X-REF-FB')))
pvs.append(PV(x.replace(':X1',':Y-REF-FB')))
elif ':X2' in x:
pvs.append(PV(x.replace(':X2',':X-REF-FB')))
pvs.append(PV(x.replace(':X2',':Y-REF-FB')))
else:
pvs.append(PV(x))
con = [pv.wait_for_connection(timeout=0.2) for pv in pvs]
for i, val in enumerate(con):
if val is False:
name = pv[i].pvname
raise ValueError(f"PV-Channel {name} is not available")
return pvs
def flush(self):
self.bsAR.flush()
self.bsAT.flush()
def terminate(self):
print('Stopping BSStream Thread...')
self.bsAR.stop()
self.bsAR.pt.running.clear() # for some reason I have to
self.bsAT.stop()
self.bsAT.pt.running.clear() # for some reason I have to
def getBeamline(self,beamline):
if beamline == 'Aramis':
self.isAramis = True
else:
self.isAramis = False
# all routine for accessing the machine (read/write)
def read(self):
if self.isAramis:
data=self.bsAR.__next__()
return data['pid'],data[self.ARch0],np.array([data[cnl] for cnl in self.ARchx]),np.array([data[cnl] for cnl in self.ARchy])
data=self.bsAT.__next__()
return data['pid'],data[self.ATch0],np.array([data[cnl] for cnl in self.ATchx]),np.array([data[cnl] for cnl in self.ATchy])
def readPV(self):
if self.isAramis:
return [pv.value for pv in self.pvAR]
return [pv.value for pv in self.pvAT]
def setPV(self,fbval):
if self.isAramis:
for i in range(len(fbval)):
self.pvAR[i].value = fbval[i]
else:
for i in range(len(fbval)):
self.pvAT[i].value = fbval[i]
def getPVNames(self):
if self.isAramis:
return [pv.pvname for pv in self.pvAR]
return [pv.pvname for pv in self.pvAT]
def getDetectorName(self):
if self.isAramis:
return self.ARch0
return self.ATch0
def getKicker(self):
if self.isAramis:
return [pv.value for pv in self.kickerAR]
return [pv.value for pv in self.kickerAT]
def setKicker(self,vals):
if self.isAramis:
for i,val in enumerate(vals):
self.kickerAR[i].value = val
return
for i,val in enumerate(vals):
self.kickerAT[i].value = val