#!/usr/bin/env python3
'''
.. module:: HMUX
:synopsis: Read X-ray event records from a XIFU-FDM format structure HDF5 file.
.. moduleauthor:: Cor de Vries <c.p.de.vries@sron.nl>
'''
from tesfdmtools.utils.widgets.filechooser import getfile
from tesfdmtools.utils.cu import cu
import h5py
import sys
import numpy
# def cu(var):
# '''
# convert string to ASCII string when type is bytes (needed for python3)
# '''
# if type(var) is bytes:
# return var.decode('UTF-8')
# return var
[docs]class channelfreq:
'''
Holds the trigger record objects for a given channel frequency.
Used within class **muxchannel**.
Args:
* `chanfreq` = muxchannel object in HDF5 file.
* `ifreq` = frequency number of object.
* `nevents` = number of records in channel.
'''
def __init__(self,chanfreq,ifreq,nevents):
self.chanfreq=chanfreq
''' triggers object in HDF5 file '''
self.attrs=chanfreq.attrs
''' triggers object attributes '''
self.nrec=nevents
''' maximum possible record number '''
self.irec=0
self.ifreq=ifreq
# print("channelfreq for freq: ",ifreq)
try:
ri=next(iter(self.chanfreq)) # first available record item in frequency
except StopIteration:
# print("Stopiteration")
self.nrec=0
self.rec0=numpy.zeros(1)
return
rr=self.chanfreq[ri]
self.rec0=numpy.zeros_like(rr) # define dummy record for non-existent records
self.attrs=rr.attrs # set frequency attributes equal to first record attrs.
def __iter__(self):
self.irec=0
return self
def __getitem__(self,index):
'''
Indexing function
Returns:
* Data record for given trigger index. Holds **attrs** property.
'''
if index >= self.nrec:
raise IndexError("Record does not exist: %d " % index)
recname="%9.9d" % index
try:
return self.chanfreq[recname]
except:
return self.rec0
def __next__(self):
'''
Iterator function
Returns:
* Data record for next trigger in iterator sequence. Holds **attrs** property.
'''
if self.irec >= self.nrec:
raise StopIteration
recname="%9.9d" % self.irec
self.irec=self.irec+1
try:
return self.chanfreq[recname]
except:
return self.rec0
[docs] def overview(self):
'''
Prints attributes overview for this object.
'''
if self.nrec == 0:
return
print(" Frequency: %d" % self.ifreq)
for key in self.attrs:
try:
aa=cu(self.chanfreq.attrs[key])
except:
aa='--'
print(" %15.15s: %s" % (key,aa))
[docs]class muxchannel:
'''
Muxchannel object. Holds the trigger records for the designated channel.
Used within class **HMUX**.
Args:
* `muxchannel` = muxchannel object of HDF5 file
* `configuration` = configuration table object in the muxchannel section
* `ichan` = channel number
'''
def __init__(self,muxchannel,configuration,ichan):
self.attrs=muxchannel.attrs
''' channel attributes '''
self.configuration=configuration
self.freq_conf=None
''' frequency configuration table '''
if list(self.configuration.keys()).count("freq_conf") == 1:
self.freq_conf=self.configuration["freq_conf"][...]
self.freq_conf_keys=list(self.freq_conf.dtype.fields.keys())
self.ichannel=ichan
self.freq={}
''' frequency objects '''
self.sample=None
self.samplename=None
self.nevents=self.attrs['Nevents']
self.freqs=[]
''' list of available frequency identifiers '''
for freq in muxchannel:
if freq.find('freq_') >= 0:
try:
ifreq=int(freq.split('_')[1])
except:
sys.exit('Error in freq name: %s, channel %d' % (freq,ichan))
self.freq[ifreq]=channelfreq(muxchannel[freq],ifreq,self.nevents)
if self.freq[ifreq].nrec != 0:
self.freqs.append(ifreq)
if freq.find('A') >= 0:
self.samplename=freq
self.sample=channelfreq(muxchannel[freq],0,self.nevents)
[docs] def overview(self):
'''
Prints overview of channel attributes.
'''
print(" CHANNEL: %d" % self.ichannel)
print(" number of records: %d" % self.nevents)
for key in self.attrs:
print(" %15.15s: %s" % (key,cu(self.attrs[key])))
for ff in self.freq.keys():
self.freq[ff].overview()
[docs]class HMUX:
'''
Open HDF5 file and make the XIFU-defined structure available for reading of X-ray event records.
Kwargs:
* `path` = filepath where to find the file. Default: /stage/xmmdat10/Athena/raw_data
* `filename` = HDF5 filename to open. If None filename will be asked for.
* `driver` = I/O driver to use for HDF5 (see H5py documentation)
properties::
attrs = file attributes
mux.attrs = mux section attributes
channels = list of actual channel id's
channel = the indivudual channel objects
each channel has the following properties:
attrs = channel attributes
configuration = configuration section of this channel
freq_conf = frequency configuration data
freq_conf_keys = list of columns in frequency configuration data
sample = iterator class for sample numbers
samplename = ID of the sampledata if present (ADC1/DAC1/DAC2)
freqs = iterator of frequency numbers
freq = iterator class of actual frequencies; returns event records
each frequency has the following properties:
attrs = frequency section attributes
nevents = number of event records
each event record has the 'attrs' property
example of use::
hfile=HMUX()
for channel in hfile.channels: # go through all channels of file
for freq in hfile.channel[channel].freqs: # go through all frequencies in channel
for event in hfile.channel[channel].freq[freq]: # go through all event records in frequency
pixel=event.attrs["pixel_index"]
event_time=event.attrs["event_time"]
Idata=event[:,0]
Qdata=event[:,1]
...process event ...
or access a record individually::
hfile=HMUX()
Idata=hfile.channel[0].freq[8][100][:,0] # access Idata of record 100 of channel-0, freq-8
'''
def __init__(self,path=None,filename=None,driver=None):
if path is None:
# filepath='/home/jorisvr/safari/hdf5_test/'
filepath='/stage/ltsboldat5'
else:
filepath=path
if filename is None:
self.hdffile=getfile(pattern="*.h5",path=filepath)
if self.hdffile is None:
sys.exit('No hdf5 file selected')
else:
self.hdffile=filename
''' filename '''
self.hd5=h5py.File(self.hdffile,'r',driver=driver) # open file
self.attrs=self.hd5.attrs
''' file attributes '''
if list(self.hd5.keys()).count("mux") < 1:
sys.exit("File %s does not contain mux output data" % self.hdffile )
self.mux=self.hd5["mux"]
if list(self.hd5.keys()).count("configuration") == 1:
self.configuration=self.hd5["configuration"]
else:
self.configuration=None
self.channel={}
''' individual channel objects in the dataset '''
self.channels=[]
''' list of channel id's in dataset '''
for chan in self.mux:
if chan.find('channel_') >= 0:
try:
ichan=int(chan.split('_')[1])
except:
sys.exit('Error in channel name: %s' % chan)
config=None
if list(self.configuration.keys()).count(chan) == 1:
config=self.configuration[chan]
self.channel[ichan]=muxchannel(self.mux[chan],config,ichan)
self.channels.append(ichan)
def close(self):
self.hd5.close()
[docs] def overview(self):
'''
Prints overview of file attributes all muxchannels with their attributes.
In addition, print overviews of all channels.
'''
print("FILE: %s" % self.hdffile)
for att in self.hd5.attrs:
print("%15.15s: %s" % (att,cu(self.hd5.attrs[att])))
print(" MUX:")
for att in self.mux.attrs:
print(" %15.15s: %s" % (att,cu(self.mux.attrs[att])))
for ch in self.channel.keys():
self.channel[ch].overview()
# ==============================================================================
if __name__ == "__main__":
'''
Demonstrator program for the HMUX class.
'''
import matplotlib.pyplot as plt
# demonstrator program
hdata=HMUX() # open HDF5 file
hdata.overview() # print info of file
if hdata.channel[0].sample is not None: # see if there are sample data ADC1/DAC1/DAC2
print("sample data: %s" % hdata.channel[0].samplename)
i=0
f,ax=plt.subplots(1)
ax.set_xlabel('Sample #')
ax.set_ylabel('Output')
ax.set_title(hdata.channel[0].samplename)
for record in hdata.channel[0].sample: # go through all sample records
for att in record.attrs:
print(" %15.15s: %s" % (att,record.attrs[att])) # print record attributes
print(record[:,0][0:10]) # print sample record
ax.plot(numpy.arange(record[:,0].size),record[:,0])
i=i+1
if i > 10:
break
plt.show()
plt.close('all')
sys.exit() # if sample data there are not further X-ray data->exit
print("Channel numbers: ",hdata.channels) # show some X-ray data
f,ax=plt.subplots(2,sharex=True)
ax[0].set_xlabel('Sample #')
ax[0].set_ylabel('I-signal')
ax[1].set_xlabel('Sample #')
ax[1].set_ylabel('Q-signal')
ax[0].set_title('X-ray events')
i=0
ichan=hdata.channels[0]
ifreq=hdata.channel[ichan].freqs[0]
for record in hdata.channel[ichan].freq[ifreq]: # go through all records of 1st channel,1st freq
print(record)
if cu(record.attrs['data_type']) == 'IQ': # check if data type is 'IQ'
idata=record[:,0]
qdata=record[:,1]
print("I-data[0-10]: ",idata[0:10]) # print some I data
ax[0].plot(numpy.arange(idata.size),idata)
ax[1].plot(numpy.arange(qdata.size),qdata)
for att in record.attrs:
print(" %15.15s: %s" % (att,cu(record.attrs[att]))) # print record attributes
i=i+1
if i > 10:
break
plt.show()
plt.close('all')
print("freq_conf configuration: ")
for ckey in hdata.channel[ichan].freq_conf_keys: # go through all columns of freq-conf data of 1st channel
print(" ",ckey,":",hdata.channel[ichan].freq_conf[ckey])
for att in hdata.channel[0].configuration['freq_conf'].attrs:
print(" %15.15s: %s" % (att,cu(hdata.channel[ichan].configuration['freq_conf'].attrs[att])))
print("====================================")
index=0
print("record: ",index)
rrr=hdata.channel[ichan].freq[ifreq][index]
print(rrr)
for att in rrr.attrs:
print(" %15.15s: %s" % (att,cu(rrr.attrs[att]))) # print record attributes
hdata.close()