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

Merge branch 'feature/acquire_gas_detector_with_spectra' into 'master'

Browse Source 
new spectralanalysis with gas detector readings

See merge request slic/sfbd!2
This commit is contained in:
reiche
2024-06-27 09:05:14 +02:00
parent 6fa15dd9d6 2906c9cc44
commit a07f1bcf42
1 changed files with 27 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
app/spectralanalysis.py
View File

@ -1,23 +1,28 @@
import time
import numpy as np
from bstrd import BSCache
from epics import PV
class SpectralAnalysis:
"""
Wrapper class to bundle all daq/io needed for adaptive orbit feedback.
"""
def __init__(self):
athos_uncalibrated = 'SATFE10-PEPG046-EVR0:CALCI'
athos_calibrated = 'SATFE10-PEPG046:FCUP-INTENSITY-CAL'
aramis_uncalibrated = 'SARFE10-PBIG050-EVR0:CALCI'
aramis_calibrated = 'SARFE10-PBPG050:HAMP-INTENSITY-CAL'
channel_dict = {
'PSSS': ['SARFE10-PSSS059:SPECTRUM_Y', aramis_uncalibrated, aramis_calibrated],
'PMOS Maloja': ['SATOP21-PMOS127-2D:SPECTRUM_Y', athos_uncalibrated, athos_calibrated],
'PMOS Furka': ['SATOP31-PMOS132-2D:SPECTRUM_Y', athos_uncalibrated, athos_calibrated],
'PSSS LB': ['SARFE10-PSSS059-LB:SPECTRUM_Y', aramis_uncalibrated, aramis_calibrated],
}
names = ['PSSS', 'PMOS Maloja', 'PMOS Furka', 'PSSS LB']
self.bs = BSCache(100000,10000) # 100 second timeout, size for 100 second data taken
class SpectralAnalysis:
def __init__(self):
self.bs = BSCache(100000,10000) # 100 second timeout, size for 100 second data taken
self.bs.stop()
self.channel = ''
self.channels = ['SARFE10-PSSS059:SPECTRUM_Y',
'SATOP21-PMOS127-2D:SPECTRUM_Y',
'SATOP31-PMOS132-2D:SPECTRUM_Y']
self.isConnected = False
self.channel = None
self.channels = [channel_dict[x] for x in names]
def connect_name(self, name):
index = names.index(name)
self.connect(index)
def connect(self,ich):
if ich < 0 or ich >= len(self.channels):
@ -25,25 +30,24 @@ class SpectralAnalysis:
self.channel = self.channels[ich]
print('Connecting to BS-Channel:',self.channel)
self.bs.channels.clear()
self.bs.get_var(self.channel) # this starts the stream into the cache
self.pv = PV(self.channel.replace('_Y','_X'))
self.bs.get_vars(self.channel) # this starts the stream into the cache
self.pv = PV(self.channel[0].replace('_Y','_X'))
def terminate(self):
print('Stopping BSStream Thread...')
print('Stopping BSStream Thread...')
self.bs.stop()
self.bs.pt.running.clear() # for some reason I have to
self.pv.disconnect()
def flush(self):
self.bs.flush()
def read(self):
data=self.bs.__next__()
return data['pid'],data[self.channel]
return next(self.bs)
def readPV(self):
def read_spectrum_axis(self):
return self.pv.value
def getSpectrometerName(self):
return self.channel