InteractiveStreamingExample/jupyter/indraMessage.py

0001 #!/usr/bin/python3.6
0002
0003 # Copyright 2020, Jefferson Science Associates, LLC.
0004 # Subject to the terms in the LICENSE file found in the top-level directory.
0005
0006 import struct
0007 import numpy as np
0008
0009 # define global variables
0010 numChans = 80
0011
0012
0013 class IndraMessage:
0014     """Class to handle decoding of ZMQ INDRA Messages"""
0015
0016     def __init__(self, zmq_msg):
0017         # def __init__(self, zmq_msg, zmq_pub):
0018         # define the data dictionary
0019         self.data_dict = {}
0020         # size up the zmq message
0021         self.msg_size = len(zmq_msg)
0022         # indra message header is a fixed 56 bytes
0023         self.payload_bytes = self.msg_size - 56
0024         # unpack the message according to the indra message protocol
0025         self.msg = struct.unpack('IIIIIIIQqq%ds' % self.payload_bytes, zmq_msg)
0026         # decode the indra message members
0027         self.source_id = self.msg[0]
0028         self.total_bytes = self.msg[1]
0029         self.payload_bytes = self.msg[2]
0030         self.compressed_bytes = self.msg[3]
0031         self.magic = self.msg[4]
0032         self.format_vrsn = self.msg[5]
0033         self.flags = self.msg[6]
0034         self.record_cntr = self.msg[7]
0035         self.time_stamps_sec = self.msg[8]
0036         self.time_stamps_nsec = self.msg[9]
0037         self.payload = self.msg[10]
0038         # print message info
0039         print('INDRA Message received -> event = %d, size = %d bytes' % (self.record_cntr, self.msg_size))
0040         # adc samples are uints of length 4 plus 1 space delimiter, convert to array of ints
0041         self.adc_smpls_str = np.frombuffer(self.payload, dtype='S5')
0042         self.adc_smpls = np.reshape(self.adc_smpls_str.astype(np.int), (1024, 80))
0043         # define data dictionary keys and data types
0044         for chan in range(1, numChans + 1):
0045             self.data_dict.update({'adcSamplesChan_%s' % chan: np.array([])})
0046             self.data_dict.update(
0047                 {'tdcSamplesChan_%s' % chan: np.arange((self.record_cntr - 1) * 1024, self.record_cntr * 1024)})
0048         # enumerate the samples object and populate the data dictionary
0049         for index, sample in np.ndenumerate(self.adc_smpls):
0050             self.data_dict['adcSamplesChan_%s' % str(index[1] + 1)] = \
0051                 np.append(self.data_dict['adcSamplesChan_%s' % str(index[1] + 1)], sample)
0052         # serialize the data dictionary via pickle and publish it
0053         # self.event_data_dict = pickle.dumps(self.data_dict)
0054         # zmq_pub.send_pyobj(self.event_data_dict)
0055         # remove event data after being published
0056         # for chan in range(1, numChans + 1) :
0057         #     self.data_dict.pop('adcSamplesChan_%s' % str(chan), None)
0058         #     self.data_dict.pop('tdcSamplesChan_%s' % str(chan), None)