pulseParser

Description: CoMPASS binary data parser Author: Ming Fang Date: 2022-08-22 19:00:26 LastEditors: Ming Fang LastEditTime: 2022-08-23 17:55:59

  1'''
  2Description: CoMPASS binary data parser
  3Author: Ming Fang
  4Date: 2022-08-22 19:00:26
  5LastEditors: Ming Fang
  6LastEditTime: 2022-08-23 17:55:59
  7'''
  8from pathlib import Path
  9import numpy as np
 10
 11
 12class WaveBinFile:
 13    """Paser for CoMPASS binary file.
 14    """
 15    def __init__(self, p, version=2) -> None:
 16        """Initialize with input file path and version number.
 17
 18        Args:
 19            p (str | Path): Path to the binary data file saved by CoMPASS.
 20            version (int): Version of the CoMPASS software. Must be 1 or 2. Default is 2.
 21
 22        Raises:
 23            ValueError: If input file is not accessible, or the version is incorrect.
 24        """
 25        self._filePath = Path(p)  # file path
 26        if not self._filePath.is_file():
 27            raise ValueError(f"{str(p)} is not accessible.")
 28        if version != 1 and version != 2:
 29            raise ValueError(f"Version number must be either 1 or 2, but got {version}.")
 30        self._headersize = 24  # header size in bytes
 31        if version == 2:
 32            self._headersize += 1
 33        self._version = version  # CoMPASS version number
 34        self._numberOfSamples = 0  # number of samples per pulse
 35        self._boardNumber = -1  # board number
 36        self._channelNumber = -1  # channel number
 37        self._getCommonHeader()
 38        self._fileObject = open(self._filePath, 'rb')  # data file object
 39        if version == 2:
 40            # skip first 2 bytes
 41            self._fileObject.read(2)
 42        self._pulseSize = self._headersize + 2 * self._numberOfSamples  # pulse size in bytes
 43        self._numberOfPulses = int(self._filePath.stat().st_size / self._pulseSize)  # total number of pulses in the file
 44        if version == 1:
 45            self._pulseType = np.dtype([('Board', np.int16),
 46                                       ('Channel', np.int16),
 47                                       ('Time Stamp', np.int64),
 48                                       ('Energy', np.int16),
 49                                       ('Energy Short', np.int16),
 50                                       ('Flags', np.int32),
 51                                       ('Number of Samples', np.int32),
 52                                       ('Samples', np.uint16, (self._numberOfSamples, ))])  # custom data type for one pulse
 53        else:
 54            self._pulseType = np.dtype([('Board', np.int16),
 55                                       ('Channel', np.int16),
 56                                       ('Time Stamp', np.int64),
 57                                       ('Energy', np.int16),
 58                                       ('Energy Short', np.int16),
 59                                       ('Flags', np.int32),
 60                                       ('Waveform Code', np.int8),
 61                                       ('Number of Samples', np.int32),
 62                                       ('Samples', np.uint16, (self._numberOfSamples, ))])
 63            # print(self.pulseType['Samples'])
 64        self._numberOfPulseUnread = self._numberOfPulses  # number of pulses that have not been read
 65
 66    def skipNextPulse(self):
 67        """Skip the next pulse.
 68        """
 69        if self._numberOfPulseUnread > 0:
 70            self._fileObject.seek(self._pulseSize, 1)
 71            self._numberOfPulseUnread -= 1
 72
 73    def skipNextNPulses(self, num: int):
 74        """Skip the next N pulses.
 75
 76        Args:
 77            num (int): number of pulses to skip.
 78        """
 79        if self._numberOfPulseUnread > num:
 80            self._fileObject.seek(self._pulseSize*num, 1)
 81            self._numberOfPulseUnread -= num
 82        else:
 83            self._fileObject.seek(0, 2)
 84            self._numberOfPulseUnread = 0
 85
 86    def readNextPulse(self):
 87        """Read the next pulse.
 88
 89        Returns:
 90            self.pulseType: Custom data type for pulse object.
 91            
 92            None: If all pulses have been read
 93            
 94            The data members of self.PulseType can be accessed with the following string keys:
 95            
 96                "Board": np.int16, board number
 97                "Channel": np.int16, channel number
 98                "Time Stamp": np.int64, unit: ps
 99                "Energy": np.int16
100                "Energy short": np.int16
101                "Flags": np.int32
102                "Waveform Code": np.int8, version 2 only
103                "Number of Samples" np.int32
104                "Samples": numpy array of np.uint16
105        """
106        if self._numberOfPulseUnread > 0:
107            self._numberOfPulseUnread -= 1
108            buffer = self._fileObject.read(self._pulseSize)
109            return (np.frombuffer(buffer, dtype=self._pulseType))[0]
110        else:
111            return None
112
113    def readNextNPulses(self, num: int):
114        """Read the next N pulses.
115        Note:
116            Be carefule about the memory when reading a large file. 
117            If the user read the whole file and there is not enough memory, program will crash.
118            If that's the case, read the pulses in smaller chunks.
119
120        Args:
121            num (int): number of pulses to read.
122
123        Returns:
124            np.ndarray: Array of pulses. Number of pulses is at most `num`.
125                        If reached EOF, less than `num` pulses will be returned.
126            
127            None: If all pulses have been read
128        """
129        if self._numberOfPulseUnread <= 0:
130            return None
131        elif self._numberOfPulseUnread > num:
132            self._numberOfPulseUnread -= num
133        else:
134            num = self._numberOfPulseUnread
135            self._numberOfPulseUnread = 0
136        buffer = self._fileObject.read(self._pulseSize*num)
137        return np.frombuffer(buffer, dtype=self._pulseType)
138        
139    def rewind(self):
140        """Go to the begnning of the file.
141        """
142        self._fileObject.seek(0)
143        self._numberOfPulseUnread = self._numberOfPulses
144
145    @property
146    def versionNumber(self):
147        """The CoMPASS version number.
148
149        Returns:
150            int
151        """
152        return self._version
153    
154    @property
155    def numberOfSamplesPerPulse(self):
156        """The number of samples per pulse.
157
158        Returns:
159            int
160        """
161        return self._numberOfSamples
162
163    @property
164    def boardNumber(self):
165        """Board number.
166
167        Returns:
168            int
169        """
170        return self._boardNumber
171
172    @property
173    def channelNumber(self):
174        """Channel number.
175
176        Returns:
177            int
178        """
179        return self._channelNumber
180
181    @property
182    def totalNumberOfPulses(self):
183        """The total number of pulses recorded on file.
184
185        Returns:
186            int
187        """
188        return self._numberOfPulses
189
190    @property
191    def numberOfPulsesUnread(self):
192        """The number of pulses that haven't been read.
193
194        Returns:
195            int
196        """
197        return self._numberOfPulseUnread
198    
199    
200    def _getCommonHeader(self):
201        """Read common headers of all pulses.
202        """
203        with open(self._filePath, 'rb') as f:
204            if self._version == 2:
205                f.read(2)
206            # read first header
207            self._boardNumber = int.from_bytes(f.read(2), 'little')
208            self._channelNumber = int.from_bytes(f.read(2), 'little')
209            if self._version == 1:
210                f.read(16)
211            else:
212                f.read(17)
213            self._numberOfSamples = int.from_bytes(f.read(4), 'little')
214
215    def __del__(self):
216        self._fileObject.close()
class WaveBinFile:
 13class WaveBinFile:
 14    """Paser for CoMPASS binary file.
 15    """
 16    def __init__(self, p, version=2) -> None:
 17        """Initialize with input file path and version number.
 18
 19        Args:
 20            p (str | Path): Path to the binary data file saved by CoMPASS.
 21            version (int): Version of the CoMPASS software. Must be 1 or 2. Default is 2.
 22
 23        Raises:
 24            ValueError: If input file is not accessible, or the version is incorrect.
 25        """
 26        self._filePath = Path(p)  # file path
 27        if not self._filePath.is_file():
 28            raise ValueError(f"{str(p)} is not accessible.")
 29        if version != 1 and version != 2:
 30            raise ValueError(f"Version number must be either 1 or 2, but got {version}.")
 31        self._headersize = 24  # header size in bytes
 32        if version == 2:
 33            self._headersize += 1
 34        self._version = version  # CoMPASS version number
 35        self._numberOfSamples = 0  # number of samples per pulse
 36        self._boardNumber = -1  # board number
 37        self._channelNumber = -1  # channel number
 38        self._getCommonHeader()
 39        self._fileObject = open(self._filePath, 'rb')  # data file object
 40        if version == 2:
 41            # skip first 2 bytes
 42            self._fileObject.read(2)
 43        self._pulseSize = self._headersize + 2 * self._numberOfSamples  # pulse size in bytes
 44        self._numberOfPulses = int(self._filePath.stat().st_size / self._pulseSize)  # total number of pulses in the file
 45        if version == 1:
 46            self._pulseType = np.dtype([('Board', np.int16),
 47                                       ('Channel', np.int16),
 48                                       ('Time Stamp', np.int64),
 49                                       ('Energy', np.int16),
 50                                       ('Energy Short', np.int16),
 51                                       ('Flags', np.int32),
 52                                       ('Number of Samples', np.int32),
 53                                       ('Samples', np.uint16, (self._numberOfSamples, ))])  # custom data type for one pulse
 54        else:
 55            self._pulseType = np.dtype([('Board', np.int16),
 56                                       ('Channel', np.int16),
 57                                       ('Time Stamp', np.int64),
 58                                       ('Energy', np.int16),
 59                                       ('Energy Short', np.int16),
 60                                       ('Flags', np.int32),
 61                                       ('Waveform Code', np.int8),
 62                                       ('Number of Samples', np.int32),
 63                                       ('Samples', np.uint16, (self._numberOfSamples, ))])
 64            # print(self.pulseType['Samples'])
 65        self._numberOfPulseUnread = self._numberOfPulses  # number of pulses that have not been read
 66
 67    def skipNextPulse(self):
 68        """Skip the next pulse.
 69        """
 70        if self._numberOfPulseUnread > 0:
 71            self._fileObject.seek(self._pulseSize, 1)
 72            self._numberOfPulseUnread -= 1
 73
 74    def skipNextNPulses(self, num: int):
 75        """Skip the next N pulses.
 76
 77        Args:
 78            num (int): number of pulses to skip.
 79        """
 80        if self._numberOfPulseUnread > num:
 81            self._fileObject.seek(self._pulseSize*num, 1)
 82            self._numberOfPulseUnread -= num
 83        else:
 84            self._fileObject.seek(0, 2)
 85            self._numberOfPulseUnread = 0
 86
 87    def readNextPulse(self):
 88        """Read the next pulse.
 89
 90        Returns:
 91            self.pulseType: Custom data type for pulse object.
 92            
 93            None: If all pulses have been read
 94            
 95            The data members of self.PulseType can be accessed with the following string keys:
 96            
 97                "Board": np.int16, board number
 98                "Channel": np.int16, channel number
 99                "Time Stamp": np.int64, unit: ps
100                "Energy": np.int16
101                "Energy short": np.int16
102                "Flags": np.int32
103                "Waveform Code": np.int8, version 2 only
104                "Number of Samples" np.int32
105                "Samples": numpy array of np.uint16
106        """
107        if self._numberOfPulseUnread > 0:
108            self._numberOfPulseUnread -= 1
109            buffer = self._fileObject.read(self._pulseSize)
110            return (np.frombuffer(buffer, dtype=self._pulseType))[0]
111        else:
112            return None
113
114    def readNextNPulses(self, num: int):
115        """Read the next N pulses.
116        Note:
117            Be carefule about the memory when reading a large file. 
118            If the user read the whole file and there is not enough memory, program will crash.
119            If that's the case, read the pulses in smaller chunks.
120
121        Args:
122            num (int): number of pulses to read.
123
124        Returns:
125            np.ndarray: Array of pulses. Number of pulses is at most `num`.
126                        If reached EOF, less than `num` pulses will be returned.
127            
128            None: If all pulses have been read
129        """
130        if self._numberOfPulseUnread <= 0:
131            return None
132        elif self._numberOfPulseUnread > num:
133            self._numberOfPulseUnread -= num
134        else:
135            num = self._numberOfPulseUnread
136            self._numberOfPulseUnread = 0
137        buffer = self._fileObject.read(self._pulseSize*num)
138        return np.frombuffer(buffer, dtype=self._pulseType)
139        
140    def rewind(self):
141        """Go to the begnning of the file.
142        """
143        self._fileObject.seek(0)
144        self._numberOfPulseUnread = self._numberOfPulses
145
146    @property
147    def versionNumber(self):
148        """The CoMPASS version number.
149
150        Returns:
151            int
152        """
153        return self._version
154    
155    @property
156    def numberOfSamplesPerPulse(self):
157        """The number of samples per pulse.
158
159        Returns:
160            int
161        """
162        return self._numberOfSamples
163
164    @property
165    def boardNumber(self):
166        """Board number.
167
168        Returns:
169            int
170        """
171        return self._boardNumber
172
173    @property
174    def channelNumber(self):
175        """Channel number.
176
177        Returns:
178            int
179        """
180        return self._channelNumber
181
182    @property
183    def totalNumberOfPulses(self):
184        """The total number of pulses recorded on file.
185
186        Returns:
187            int
188        """
189        return self._numberOfPulses
190
191    @property
192    def numberOfPulsesUnread(self):
193        """The number of pulses that haven't been read.
194
195        Returns:
196            int
197        """
198        return self._numberOfPulseUnread
199    
200    
201    def _getCommonHeader(self):
202        """Read common headers of all pulses.
203        """
204        with open(self._filePath, 'rb') as f:
205            if self._version == 2:
206                f.read(2)
207            # read first header
208            self._boardNumber = int.from_bytes(f.read(2), 'little')
209            self._channelNumber = int.from_bytes(f.read(2), 'little')
210            if self._version == 1:
211                f.read(16)
212            else:
213                f.read(17)
214            self._numberOfSamples = int.from_bytes(f.read(4), 'little')
215
216    def __del__(self):
217        self._fileObject.close()

Paser for CoMPASS binary file.

WaveBinFile(p, version=2)
16    def __init__(self, p, version=2) -> None:
17        """Initialize with input file path and version number.
18
19        Args:
20            p (str | Path): Path to the binary data file saved by CoMPASS.
21            version (int): Version of the CoMPASS software. Must be 1 or 2. Default is 2.
22
23        Raises:
24            ValueError: If input file is not accessible, or the version is incorrect.
25        """
26        self._filePath = Path(p)  # file path
27        if not self._filePath.is_file():
28            raise ValueError(f"{str(p)} is not accessible.")
29        if version != 1 and version != 2:
30            raise ValueError(f"Version number must be either 1 or 2, but got {version}.")
31        self._headersize = 24  # header size in bytes
32        if version == 2:
33            self._headersize += 1
34        self._version = version  # CoMPASS version number
35        self._numberOfSamples = 0  # number of samples per pulse
36        self._boardNumber = -1  # board number
37        self._channelNumber = -1  # channel number
38        self._getCommonHeader()
39        self._fileObject = open(self._filePath, 'rb')  # data file object
40        if version == 2:
41            # skip first 2 bytes
42            self._fileObject.read(2)
43        self._pulseSize = self._headersize + 2 * self._numberOfSamples  # pulse size in bytes
44        self._numberOfPulses = int(self._filePath.stat().st_size / self._pulseSize)  # total number of pulses in the file
45        if version == 1:
46            self._pulseType = np.dtype([('Board', np.int16),
47                                       ('Channel', np.int16),
48                                       ('Time Stamp', np.int64),
49                                       ('Energy', np.int16),
50                                       ('Energy Short', np.int16),
51                                       ('Flags', np.int32),
52                                       ('Number of Samples', np.int32),
53                                       ('Samples', np.uint16, (self._numberOfSamples, ))])  # custom data type for one pulse
54        else:
55            self._pulseType = np.dtype([('Board', np.int16),
56                                       ('Channel', np.int16),
57                                       ('Time Stamp', np.int64),
58                                       ('Energy', np.int16),
59                                       ('Energy Short', np.int16),
60                                       ('Flags', np.int32),
61                                       ('Waveform Code', np.int8),
62                                       ('Number of Samples', np.int32),
63                                       ('Samples', np.uint16, (self._numberOfSamples, ))])
64            # print(self.pulseType['Samples'])
65        self._numberOfPulseUnread = self._numberOfPulses  # number of pulses that have not been read

Initialize with input file path and version number.

Args
  • p (str | Path): Path to the binary data file saved by CoMPASS.
  • version (int): Version of the CoMPASS software. Must be 1 or 2. Default is 2.
Raises
  • ValueError: If input file is not accessible, or the version is incorrect.
def skipNextPulse(self):
67    def skipNextPulse(self):
68        """Skip the next pulse.
69        """
70        if self._numberOfPulseUnread > 0:
71            self._fileObject.seek(self._pulseSize, 1)
72            self._numberOfPulseUnread -= 1

Skip the next pulse.

def skipNextNPulses(self, num: int):
74    def skipNextNPulses(self, num: int):
75        """Skip the next N pulses.
76
77        Args:
78            num (int): number of pulses to skip.
79        """
80        if self._numberOfPulseUnread > num:
81            self._fileObject.seek(self._pulseSize*num, 1)
82            self._numberOfPulseUnread -= num
83        else:
84            self._fileObject.seek(0, 2)
85            self._numberOfPulseUnread = 0

Skip the next N pulses.

Args
  • num (int): number of pulses to skip.
def readNextPulse(self):
 87    def readNextPulse(self):
 88        """Read the next pulse.
 89
 90        Returns:
 91            self.pulseType: Custom data type for pulse object.
 92            
 93            None: If all pulses have been read
 94            
 95            The data members of self.PulseType can be accessed with the following string keys:
 96            
 97                "Board": np.int16, board number
 98                "Channel": np.int16, channel number
 99                "Time Stamp": np.int64, unit: ps
100                "Energy": np.int16
101                "Energy short": np.int16
102                "Flags": np.int32
103                "Waveform Code": np.int8, version 2 only
104                "Number of Samples" np.int32
105                "Samples": numpy array of np.uint16
106        """
107        if self._numberOfPulseUnread > 0:
108            self._numberOfPulseUnread -= 1
109            buffer = self._fileObject.read(self._pulseSize)
110            return (np.frombuffer(buffer, dtype=self._pulseType))[0]
111        else:
112            return None

Read the next pulse.

Returns

self.pulseType: Custom data type for pulse object.

None: If all pulses have been read

The data members of self.PulseType can be accessed with the following string keys:

"Board": np.int16, board number
"Channel": np.int16, channel number
"Time Stamp": np.int64, unit: ps
"Energy": np.int16
"Energy short": np.int16
"Flags": np.int32
"Waveform Code": np.int8, version 2 only
"Number of Samples" np.int32
"Samples": numpy array of np.uint16
def readNextNPulses(self, num: int):
114    def readNextNPulses(self, num: int):
115        """Read the next N pulses.
116        Note:
117            Be carefule about the memory when reading a large file. 
118            If the user read the whole file and there is not enough memory, program will crash.
119            If that's the case, read the pulses in smaller chunks.
120
121        Args:
122            num (int): number of pulses to read.
123
124        Returns:
125            np.ndarray: Array of pulses. Number of pulses is at most `num`.
126                        If reached EOF, less than `num` pulses will be returned.
127            
128            None: If all pulses have been read
129        """
130        if self._numberOfPulseUnread <= 0:
131            return None
132        elif self._numberOfPulseUnread > num:
133            self._numberOfPulseUnread -= num
134        else:
135            num = self._numberOfPulseUnread
136            self._numberOfPulseUnread = 0
137        buffer = self._fileObject.read(self._pulseSize*num)
138        return np.frombuffer(buffer, dtype=self._pulseType)

Read the next N pulses.

Note

Be carefule about the memory when reading a large file. If the user read the whole file and there is not enough memory, program will crash. If that's the case, read the pulses in smaller chunks.

Args
  • num (int): number of pulses to read.
Returns

np.ndarray: Array of pulses. Number of pulses is at most num. If reached EOF, less than num pulses will be returned.

None: If all pulses have been read

def rewind(self):
140    def rewind(self):
141        """Go to the begnning of the file.
142        """
143        self._fileObject.seek(0)
144        self._numberOfPulseUnread = self._numberOfPulses

Go to the begnning of the file.

versionNumber

The CoMPASS version number.

Returns

int

numberOfSamplesPerPulse

The number of samples per pulse.

Returns

int

boardNumber

Board number.

Returns

int

channelNumber

Channel number.

Returns

int

totalNumberOfPulses

The total number of pulses recorded on file.

Returns

int

numberOfPulsesUnread

The number of pulses that haven't been read.

Returns

int