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()
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.
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.
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.
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.
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
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 thannum
pulses will be returned.None: If all pulses have been read