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 56 of file TCReader.py.
56 def __init__(self, filename: str, verbosity: int = 0, batch_mode: bool = False) -> None:
57 """
58 Loads a given HDF5 file.
59
60 Parameters:
61 filename (str): HDF5 file to open.
62 verbosity (int): Verbose level. 0: Only errors. 1: Warnings. 2: All.
63
64 Returns nothing.
65 """
66 super().__init__(filename, verbosity, batch_mode)
67 self.tc_data = np.array([], dtype=self.tc_dt) # Will concatenate new TCs
68 self.ta_data = [] # ta_data[i] will be a np.ndarray of TAs from the i-th TC
69 return None
70
Member Function Documentation
◆ __getitem__()
| NDArray[tc_dt] trgtools.TCReader.TCReader.__getitem__ | ( | self, | |
| int | str | key ) |
Definition at line 71 of file TCReader.py.
71 def __getitem__(self, key: int | str) -> NDArray[tc_dt]:
72 return self.tc_data[key]
73
◆ __len__()
| int trgtools.TCReader.TCReader.__len__ | ( | self | ) |
Definition at line 78 of file TCReader.py.
78 def __len__(self) -> int:
79 return len(self.tc_data)
80
◆ __setitem__()
| None trgtools.TCReader.TCReader.__setitem__ | ( | self, | |
| int | str | key, | ||
| NDArray[tc_dt] | value ) |
Definition at line 74 of file TCReader.py.
74 def __setitem__(self, key: int | str, value: NDArray[tc_dt]) -> None:
75 self.tc_data[key] = value
76 return
77
◆ _filter_fragment_paths()
|
protected |
Filter the fragment paths for TCs.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 81 of file TCReader.py.
81 def _filter_fragment_paths(self) -> None:
82 """ Filter the fragment paths for TCs. """
83 fragment_paths = []
84
85 # TC fragment paths contain their name in the path.
86 for path in self._fragment_paths:
87 if "Trigger_Candidate" in path:
88 fragment_paths.append(path)
89
90 self._fragment_paths = fragment_paths
91 return None
92
◆ clear_data()
| None trgtools.TCReader.TCReader.clear_data | ( | self | ) |
Clear the contents of the member data.
Reimplemented from trgtools.HDF5Reader.HDF5Reader.
Definition at line 172 of file TCReader.py.
172 def clear_data(self) -> None:
173 self.tc_data = np.array([], dtype=self.tc_dt)
174 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 93 of file TCReader.py.
93 def read_fragment(self, fragment_path: str) -> NDArray:
94 """
95 Read from the given data fragment path.
96
97 Returns a np.ndarray of the first TC that was read and appends all TCs in the fragment to :self.tc_data:.
98 """
99 if self._verbosity >= 2:
100 print("="*60)
101 print(f"INFO: Reading from the path\n{fragment_path}")
102
103 fragment = self._h5_file.get_frag(fragment_path)
104 fragment_data_size = fragment.get_data_size()
105 trigger_number = fragment.get_trigger_number()
106
107 if fragment_data_size == 0: # Empty fragment
108 self._num_empty += 1
109 if self._verbosity >= 1:
110 print(
111 self._FAIL_TEXT_COLOR
112 + self._BOLD_TEXT
113 + "WARNING: Empty fragment."
114 + self._END_TEXT_COLOR
115 )
116 print("="*60)
117 return np.array([], dtype=self.tc_dt)
118
119 tc_idx = 0 # Debugging output.
120 byte_idx = 0 # Variable TC sizing, must do a while loop.
121 while byte_idx < fragment_data_size:
122 if self._verbosity >= 2:
123 print(f"INFO: Fragment Index: {tc_idx}.")
124 tc_idx += 1
125 print(f"INFO: Byte Index / Frag Size: {byte_idx} / {fragment_data_size}")
126
127 # Process TC data
128 tc_datum = trgdataformats.TriggerCandidate(fragment.get_data(byte_idx))
129 np_tc_datum = np.array([(
130 tc_datum.data.algorithm,
131 tc_datum.data.detid,
132 tc_datum.n_inputs(),
133 tc_datum.data.time_candidate,
134 tc_datum.data.time_end,
135 tc_datum.data.time_start,
136 tc_datum.data.type,
137 tc_datum.data.version,
138 trigger_number)],
139 dtype=self.tc_dt)
140
141 self.tc_data = np.hstack((self.tc_data, np_tc_datum))
142
143 byte_idx += tc_datum.sizeof()
144 if self._verbosity >= 2:
145 print(f"INFO: Upcoming byte index: {byte_idx}.")
146
147 # Process TA data
148 np_ta_data = np.zeros(np_tc_datum['num_tas'], dtype=self.ta_dt)
149 for ta_idx, ta in enumerate(tc_datum):
150 np_ta_data[ta_idx] = np.array([(
151 ta.adc_integral,
152 ta.adc_peak,
153 ta.algorithm,
154 ta.channel_end,
155 ta.channel_peak,
156 ta.channel_start,
157 np.uint16(ta.detid),
158 ta.time_activity,
159 ta.time_end,
160 ta.time_peak,
161 ta.time_start,
162 ta.type,
163 ta.version)],
164 dtype=self.ta_dt)
165 self.ta_data.append(np_ta_data) # Jagged array
166
167 if self._verbosity >= 2:
168 print("INFO: Finished reading.")
169 print("="*60)
170 return np_tc_datum
171
Member Data Documentation
◆ _fragment_paths
|
protected |
Definition at line 86 of file TCReader.py.
◆ ta_data
| list trgtools.TCReader.TCReader.ta_data = [] |
Definition at line 68 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 40 of file TCReader.py.
◆ tc_data
| trgtools.TCReader.TCReader.tc_data = np.array([], dtype=self.tc_dt) |
Definition at line 67 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),
('trigger_number', np.uint64)
])
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 Wed Oct 22 2025 for DUNE-DAQ by
1.12.0