Public Member Functions |
Public Attributes |
Static Public Attributes |
Protected Member Functions |
Protected Attributes |
List of all members
trgtools.TCReader.TCReader Class Reference
Inheritance diagram for trgtools.TCReader.TCReader:
Collaboration diagram for trgtools.TCReader.TCReader:
Public Member Functions | |
| None | __init__ (self, str filename, int verbosity=0, bool batch_mode=False) |
| NDArray[tc_dt] | __getitem__ (self, int|str key) |
| None | __setitem__ (self, int|str key, NDArray[tc_dt] value) |
| int | __len__ (self) |
| NDArray | read_fragment (self, str fragment_path) |
| None | clear_data (self) |
 Public Member Functions inherited from trgtools.HDF5Reader.HDF5Reader | |
| list[str] | get_fragment_paths (self) |
| None | set_fragment_paths (self, list[str] fragment_paths) |
| None | read_all_fragments (self) |
| None | reset_fragment_paths (self) |
Public Attributes | |
| tc_data = np.array([], dtype=self.tc_dt) | |
| list | ta_data = [] |
| Public Attributes inherited from trgtools.HDF5Reader.HDF5Reader | |
| run_id = self._h5_file.get_int_attribute('run_number') | |
| file_index = self._h5_file.get_int_attribute('file_index') | |
Static Public Attributes | |
| tc_dt | |
| ta_dt | |
Protected Member Functions | |
| None | _filter_fragment_paths (self) |
| Protected Member Functions inherited from trgtools.HDF5Reader.HDF5Reader | |
Protected Attributes | |
| _fragment_paths = [] | |
| Protected Attributes inherited from trgtools.HDF5Reader.HDF5Reader | |
| _h5_file = HDF5RawDataFile(filename) | |
| _fragment_paths = self._h5_file.get_all_fragment_dataset_paths() | |
| _verbosity = verbosity | |
| int | _num_empty = 0 |
| _batch_mode = batch_mode | |
Additional Inherited Members | |
| Static Protected Attributes inherited from trgtools.HDF5Reader.HDF5Reader | |
| str | _FAIL_TEXT_COLOR = '\033[91m' |
| str | _WARNING_TEXT_COLOR = '\033[93m' |
| str | _BOLD_TEXT = '\033[1m' |
| str | _END_TEXT_COLOR = '\033[0m' |
Detailed Description
Class that reads a given HDF5 data file and can process the TC fragments within. Loading fragments appends to :self.tc_data: and :self.ta_data:. NumPy dtypes of :self.tc_data: and :self.ta_data: are available as :TCReader.tc_dt: and :TCReader.ta_dt:. TC reading can print information that is relevant about the loading process by specifying the verbose level. 0 for errors only. 1 for warnings. 2 for all information.
Definition at line 13 of file TCReader.py.
Constructor & Destructor Documentation
◆ __init__()
| None trgtools.TCReader.TCReader.__init__ | ( | self, | |
| str | filename, | ||
| int | verbosity = 0, | ||
| bool | batch_mode = False ) |
Loads a given HDF5 file.
Parameters:
filename (str): HDF5 file to open.
verbosity (int): Verbose level. 0: Only errors. 1: Warnings. 2: All.
Returns nothing.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 55 of file TCReader.py.
55 def __init__(self, filename: str, verbosity: int = 0, batch_mode: bool = False) -> None:
56 """
57 Loads a given HDF5 file.
58
59 Parameters:
60 filename (str): HDF5 file to open.
61 verbosity (int): Verbose level. 0: Only errors. 1: Warnings. 2: All.
62
63 Returns nothing.
64 """
65 super().__init__(filename, verbosity, batch_mode)
66 self.tc_data = np.array([], dtype=self.tc_dt) # Will concatenate new TCs
67 self.ta_data = [] # ta_data[i] will be a np.ndarray of TAs from the i-th TC
68 return None
69
Member Function Documentation
◆ __getitem__()
| NDArray[tc_dt] trgtools.TCReader.TCReader.__getitem__ | ( | self, | |
| int | str | key ) |
Definition at line 70 of file TCReader.py.
70 def __getitem__(self, key: int | str) -> NDArray[tc_dt]:
71 return self.tc_data[key]
72
◆ __len__()
| int trgtools.TCReader.TCReader.__len__ | ( | self | ) |
Definition at line 77 of file TCReader.py.
77 def __len__(self) -> int:
78 return len(self.tc_data)
79
◆ __setitem__()
| None trgtools.TCReader.TCReader.__setitem__ | ( | self, | |
| int | str | key, | ||
| NDArray[tc_dt] | value ) |
Definition at line 73 of file TCReader.py.
73 def __setitem__(self, key: int | str, value: NDArray[tc_dt]) -> None:
74 self.tc_data[key] = value
75 return
76
◆ _filter_fragment_paths()
|
protected |
Filter the fragment paths for TCs.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 80 of file TCReader.py.
80 def _filter_fragment_paths(self) -> None:
81 """ Filter the fragment paths for TCs. """
82 fragment_paths = []
83
84 # TC fragment paths contain their name in the path.
85 for path in self._fragment_paths:
86 if "Trigger_Candidate" in path:
87 fragment_paths.append(path)
88
89 self._fragment_paths = fragment_paths
90 return None
91
◆ clear_data()
| None trgtools.TCReader.TCReader.clear_data | ( | self | ) |
Clear the contents of the member data.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 169 of file TCReader.py.
169 def clear_data(self) -> None:
170 self.tc_data = np.array([], dtype=self.tc_dt)
171 self.ta_data = []
◆ read_fragment()
| NDArray trgtools.TCReader.TCReader.read_fragment | ( | self, | |
| str | fragment_path ) |
Read from the given data fragment path. Returns a np.ndarray of the first TC that was read and appends all TCs in the fragment to :self.tc_data:.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 92 of file TCReader.py.
92 def read_fragment(self, fragment_path: str) -> NDArray:
93 """
94 Read from the given data fragment path.
95
96 Returns a np.ndarray of the first TC that was read and appends all TCs in the fragment to :self.tc_data:.
97 """
98 if self._verbosity >= 2:
99 print("="*60)
100 print(f"INFO: Reading from the path\n{fragment_path}")
101
102 fragment = self._h5_file.get_frag(fragment_path)
103 fragment_data_size = fragment.get_data_size()
104
105 if fragment_data_size == 0: # Empty fragment
106 self._num_empty += 1
107 if self._verbosity >= 1:
108 print(
109 self._FAIL_TEXT_COLOR
110 + self._BOLD_TEXT
111 + "WARNING: Empty fragment."
112 + self._END_TEXT_COLOR
113 )
114 print("="*60)
115 return np.array([], dtype=self.tc_dt)
116
117 tc_idx = 0 # Debugging output.
118 byte_idx = 0 # Variable TC sizing, must do a while loop.
119 while byte_idx < fragment_data_size:
120 if self._verbosity >= 2:
121 print(f"INFO: Fragment Index: {tc_idx}.")
122 tc_idx += 1
123 print(f"INFO: Byte Index / Frag Size: {byte_idx} / {fragment_data_size}")
124
125 # Process TC data
126 tc_datum = trgdataformats.TriggerCandidate(fragment.get_data(byte_idx))
127 np_tc_datum = np.array([(
128 tc_datum.data.algorithm,
129 tc_datum.data.detid,
130 tc_datum.n_inputs(),
131 tc_datum.data.time_candidate,
132 tc_datum.data.time_end,
133 tc_datum.data.time_start,
134 tc_datum.data.type,
135 tc_datum.data.version)],
136 dtype=self.tc_dt)
137
138 self.tc_data = np.hstack((self.tc_data, np_tc_datum))
139
140 byte_idx += tc_datum.sizeof()
141 if self._verbosity >= 2:
142 print(f"INFO: Upcoming byte index: {byte_idx}.")
143
144 # Process TA data
145 np_ta_data = np.zeros(np_tc_datum['num_tas'], dtype=self.ta_dt)
146 for ta_idx, ta in enumerate(tc_datum):
147 np_ta_data[ta_idx] = np.array([(
148 ta.adc_integral,
149 ta.adc_peak,
150 ta.algorithm,
151 ta.channel_end,
152 ta.channel_peak,
153 ta.channel_start,
154 np.uint16(ta.detid),
155 ta.time_activity,
156 ta.time_end,
157 ta.time_peak,
158 ta.time_start,
159 ta.type,
160 ta.version)],
161 dtype=self.ta_dt)
162 self.ta_data.append(np_ta_data) # Jagged array
163
164 if self._verbosity >= 2:
165 print("INFO: Finished reading.")
166 print("="*60)
167 return np_tc_datum
168
Member Data Documentation
◆ _fragment_paths
|
protected |
Definition at line 85 of file TCReader.py.
◆ ta_data
| list trgtools.TCReader.TCReader.ta_data = [] |
Definition at line 67 of file TCReader.py.
◆ ta_dt
|
static |
Initial value:
= np.dtype([
('adc_integral', np.uint64),
('adc_peak', np.uint64),
('algorithm', trgdataformats.TriggerActivityData.Algorithm),
('channel_end', np.int32),
('channel_peak', np.int32),
('channel_start', np.int32),
('detid', np.uint16),
('time_activity', np.uint64),
('time_end', np.uint64),
('time_peak', np.uint64),
('time_start', np.uint64),
('type', trgdataformats.TriggerActivityData.Type),
('version', np.uint16)
])
Definition at line 39 of file TCReader.py.
◆ tc_data
| trgtools.TCReader.TCReader.tc_data = np.array([], dtype=self.tc_dt) |
Definition at line 66 of file TCReader.py.
◆ tc_dt
|
static |
Initial value:
= np.dtype([
('algorithm', trgdataformats.TriggerCandidateData.Algorithm),
('detid', np.uint16),
('num_tas', np.uint64), # Greedy
('time_candidate', np.uint64),
('time_end', np.uint64),
('time_start', np.uint64),
('type', trgdataformats.TriggerCandidateData.Type),
('version', np.uint16),
])
Definition at line 27 of file TCReader.py.
The documentation for this class was generated from the following file:
- /github/workspace/dunedaq/sourcecode/trgtools/python/trgtools/TCReader.py
Generated on Sat Jun 28 2025 for DUNE-DAQ by
1.12.0