Line data Source code
1 : #include "trgtools/TAEmulationUnit.hpp"
2 : #include "trgtools/TCEmulationUnit.hpp"
3 :
4 : #include "CLI/App.hpp"
5 : #include "CLI/Config.hpp"
6 : #include "CLI/Formatter.hpp"
7 :
8 : #include <filesystem>
9 : #include <fmt/chrono.h>
10 : #include <fmt/core.h>
11 : #include <fmt/format.h>
12 :
13 : #include "detchannelmaps/TPCChannelMap.hpp"
14 : #include "hdf5libs/HDF5RawDataFile.hpp"
15 : #include "trgdataformats/TriggerPrimitive.hpp"
16 : #include "triggeralgs/TriggerActivityFactory.hpp"
17 : #include "triggeralgs/TriggerCandidateFactory.hpp"
18 : #include "triggeralgs/TriggerObjectOverlay.hpp"
19 :
20 : using namespace dunedaq;
21 :
22 : class TimeSliceProcessor
23 : {
24 : private:
25 : /* data */
26 : std::unique_ptr<hdf5libs::HDF5RawDataFile> m_input_file;
27 : std::unique_ptr<hdf5libs::HDF5RawDataFile> m_output_file;
28 :
29 : void open_files(std::string input_path, std::string output_path);
30 : void close_files();
31 :
32 : void process(daqdataformats::TimeSlice& tls);
33 :
34 : // Can modify?
35 : std::function<void(daqdataformats::TimeSlice&)> m_processor;
36 :
37 : public:
38 : TimeSliceProcessor(std::string input_path, std::string output_path);
39 : ~TimeSliceProcessor();
40 :
41 : void set_processor(std::function<void(daqdataformats::TimeSlice&)> processor);
42 : void loop(uint64_t num_records = 0, uint64_t offset = 0, bool quiet = false);
43 : };
44 :
45 : //-----------------------------------------------------------------------------
46 0 : TimeSliceProcessor::TimeSliceProcessor(std::string input_path, std::string output_path)
47 : {
48 0 : this->open_files(input_path, output_path);
49 0 : }
50 :
51 : //-----------------------------------------------------------------------------
52 0 : TimeSliceProcessor::~TimeSliceProcessor()
53 : {
54 0 : this->close_files();
55 0 : }
56 :
57 : //-----------------------------------------------------------------------------
58 : void
59 0 : TimeSliceProcessor::open_files(std::string input_path, std::string output_path)
60 : {
61 : // Open input file
62 0 : m_input_file = std::make_unique<hdf5libs::HDF5RawDataFile>(input_path);
63 :
64 0 : if (!m_input_file->is_timeslice_type()) {
65 0 : fmt::print("ERROR: input file '{}' not of type 'TimeSlice'\n", input_path);
66 0 : throw std::runtime_error(fmt::format("ERROR: input file '{}' not of type 'TimeSlice'", input_path));
67 : }
68 :
69 0 : auto run_number = m_input_file->get_attribute<daqdataformats::run_number_t>("run_number");
70 0 : auto file_index = m_input_file->get_attribute<size_t>("file_index");
71 0 : auto application_name = m_input_file->get_attribute<std::string>("application_name");
72 :
73 0 : fmt::print("Run Number: {}\nFile Index: {}\nApp name: '{}'\n", run_number, file_index, application_name);
74 :
75 0 : if (!output_path.empty()) {
76 : // Open output file
77 0 : m_output_file = std::make_unique<hdf5libs::HDF5RawDataFile>(
78 : output_path,
79 0 : m_input_file->get_attribute<daqdataformats::run_number_t>("run_number"),
80 0 : m_input_file->get_attribute<size_t>("file_index"),
81 0 : m_input_file->get_attribute<std::string>("application_name"),
82 0 : m_input_file->get_file_layout().get_file_layout_params(),
83 0 : m_input_file->get_srcid_geoid_map());
84 : }
85 0 : }
86 :
87 : //-----------------------------------------------------------------------------
88 : void
89 0 : TimeSliceProcessor::close_files()
90 : {
91 : // Do something?
92 0 : }
93 :
94 : //-----------------------------------------------------------------------------
95 : void
96 0 : TimeSliceProcessor::set_processor(std::function<void(daqdataformats::TimeSlice&)> processor)
97 : {
98 0 : m_processor = processor;
99 0 : }
100 :
101 : //-----------------------------------------------------------------------------
102 : void
103 0 : TimeSliceProcessor::process(daqdataformats::TimeSlice& tls)
104 : {
105 0 : if (m_processor)
106 0 : m_processor(tls);
107 0 : }
108 :
109 : //-----------------------------------------------------------------------------
110 : void
111 0 : TimeSliceProcessor::loop(uint64_t num_records, uint64_t offset, bool quiet)
112 : {
113 :
114 : // Replace with a record selection?
115 0 : auto records = m_input_file->get_all_record_ids();
116 :
117 0 : if (!num_records) {
118 0 : num_records = (records.size() - offset);
119 : }
120 :
121 0 : uint64_t first_rec = offset, last_rec = offset + num_records;
122 :
123 0 : uint64_t i_rec(0);
124 0 : for (const auto& rid : records) {
125 :
126 0 : if (i_rec < first_rec || i_rec >= last_rec) {
127 0 : ++i_rec;
128 0 : continue;
129 : }
130 :
131 0 : if (!quiet)
132 0 : fmt::print("\n-- Processing TSL {}:{}\n\n", rid.first, rid.second);
133 0 : auto tsl = m_input_file->get_timeslice(rid);
134 : // Or filter on a selection here using a lambda?
135 :
136 : // if (!quiet)
137 : // fmt::print("TSL number {}\n", tsl.get_header().timeslice_number);
138 :
139 : // Add a process method
140 0 : this->process(tsl);
141 :
142 0 : if (m_output_file)
143 0 : m_output_file->write(tsl);
144 :
145 0 : ++i_rec;
146 0 : if (!quiet)
147 0 : fmt::print("\n-- Finished TSL {}:{}\n\n", rid.first, rid.second);
148 0 : }
149 0 : }
150 :
151 : //-----------------------------------------------------------------------------
152 : int
153 0 : main(int argc, char const* argv[])
154 : {
155 :
156 0 : CLI::App app{ "tapipe" };
157 : // argv = app.ensure_utf8(argv);
158 :
159 0 : std::string input_file_path;
160 0 : app.add_option("-i", input_file_path, "Input TPStream file path")->required();
161 0 : std::string output_file_path;
162 0 : app.add_option("-o", output_file_path, "Output TPStream file path");
163 0 : std::string channel_map_name = "VDColdboxTPCChannelMap";
164 0 : app.add_option("-m", channel_map_name, "Detector Channel Map");
165 0 : std::string config_name;
166 0 : app.add_option("-j", config_name, "Trigger Activity and Candidate config JSON to use.")->required();
167 0 : uint64_t skip_rec(0);
168 0 : app.add_option("-s", skip_rec, "Skip records");
169 0 : uint64_t num_rec(0);
170 0 : app.add_option("-n", num_rec, "Process records");
171 :
172 0 : bool quiet = false;
173 0 : app.add_flag("--quiet", quiet, "Quiet outputs.");
174 :
175 0 : bool latencies = false;
176 0 : app.add_flag("--latencies", latencies, "Saves latencies per TP into csv");
177 0 : CLI11_PARSE(app, argc, argv);
178 :
179 0 : if (!quiet)
180 0 : fmt::print("TPStream file: {}\n", input_file_path);
181 :
182 0 : TimeSliceProcessor rp(input_file_path, output_file_path);
183 :
184 : // TP source id (subsystem)
185 0 : auto tp_subsystem_requirement = daqdataformats::SourceID::Subsystem::kTrigger;
186 :
187 0 : auto channel_map = dunedaq::detchannelmaps::make_tpc_map(channel_map_name);
188 :
189 : // Read configuration
190 0 : std::ifstream config_stream(config_name);
191 0 : nlohmann::json config = nlohmann::json::parse(config_stream);
192 :
193 : // Only use the first plugin for now.
194 0 : nlohmann::json ta_algo = config["trigger_activity_plugin"][0];
195 0 : nlohmann::json ta_config = config["trigger_activity_config"][0];
196 :
197 0 : nlohmann::json tc_algo = config["trigger_candidate_plugin"][0];
198 0 : nlohmann::json tc_config = config["trigger_candidate_config"][0];
199 :
200 : // Finally create a TA maker
201 0 : std::unique_ptr<triggeralgs::TriggerActivityMaker> ta_maker =
202 0 : triggeralgs::TriggerActivityFactory::get_instance()->build_maker(ta_algo);
203 0 : ta_maker->configure(ta_config);
204 0 : std::unique_ptr<trgtools::TAEmulationUnit> ta_emulator = std::make_unique<trgtools::TAEmulationUnit>();
205 0 : ta_emulator->set_maker(ta_maker);
206 : // TODO: Use a better file naming scheme for CSV.
207 0 : if (latencies) {
208 0 : std::filesystem::path output_path(output_file_path);
209 0 : ta_emulator->set_timing_file(
210 0 : (output_path.parent_path() / ("ta_timings_" + output_path.stem().string() + ".csv")).string());
211 0 : ta_emulator->write_csv_header("TP Time Start,TP ADC Integral,Time Diffs,Is Last TP In TA");
212 0 : }
213 :
214 : // Finally create a TA maker
215 0 : std::unique_ptr<triggeralgs::TriggerCandidateMaker> tc_maker =
216 0 : triggeralgs::TriggerCandidateFactory::get_instance()->build_maker(tc_algo);
217 0 : tc_maker->configure(tc_config);
218 0 : std::unique_ptr<trgtools::TCEmulationUnit> tc_emulator = std::make_unique<trgtools::TCEmulationUnit>();
219 0 : tc_emulator->set_maker(tc_maker);
220 : // TODO: Use a better file naming scheme for CSV.
221 0 : if (latencies) {
222 0 : std::filesystem::path output_path(output_file_path);
223 0 : tc_emulator->set_timing_file(
224 0 : (output_path.parent_path() / ("tc_timings_" + output_path.stem().string() + ".csv")).string());
225 0 : tc_emulator->write_csv_header("Time Diffs");
226 0 : }
227 :
228 : // Generic filter hook
229 0 : std::function<bool(const trgdataformats::TriggerPrimitive&)> tp_filter;
230 :
231 0 : auto z_plane_filter = [&](const trgdataformats::TriggerPrimitive& tp) -> bool {
232 0 : return (channel_map->get_plane_from_offline_channel(tp.channel) != 2);
233 0 : };
234 :
235 0 : tp_filter = z_plane_filter;
236 :
237 0 : rp.set_processor([&](daqdataformats::TimeSlice& tsl) -> void {
238 0 : const std::vector<std::unique_ptr<daqdataformats::Fragment>>& frags = tsl.get_fragments_ref();
239 0 : const size_t num_frags = frags.size();
240 0 : if (!quiet)
241 0 : fmt::print("The number of fragments: {}\n", num_frags);
242 :
243 0 : uint64_t average_ta_time = 0;
244 0 : uint64_t average_tc_time = 0;
245 :
246 0 : size_t num_tas = 0;
247 0 : size_t num_tcs = 0;
248 :
249 : // Need a static for-loop: adding fragments to tsl will mutate frags even though it's const.
250 0 : for (size_t i = 0; i < num_frags; i++) {
251 0 : const auto& frag = frags[i];
252 :
253 : // The fragment has to be for the trigger (not e.g. for retreival from readout)
254 0 : if (frag->get_element_id().subsystem != tp_subsystem_requirement) {
255 0 : if (!quiet)
256 0 : fmt::print(" Warning, got non kTrigger SourceID {}\n", frag->get_element_id().to_string());
257 0 : continue;
258 : }
259 :
260 : // The fragment has to be TriggerPrimitive
261 0 : if (frag->get_fragment_type() != daqdataformats::FragmentType::kTriggerPrimitive) {
262 0 : if (!quiet)
263 0 : fmt::print(" Error: FragmentType is: {}!\n", fragment_type_to_string(frag->get_fragment_type()));
264 0 : continue;
265 : }
266 :
267 : // This bit should be outside the loop
268 0 : if (!quiet)
269 0 : fmt::print(" Fragment id: {} [{}]\n",
270 0 : frag->get_element_id().to_string(),
271 0 : daqdataformats::fragment_type_to_string(frag->get_fragment_type()));
272 :
273 : // Pull tps out
274 0 : size_t n_tps = frag->get_data_size() / sizeof(trgdataformats::TriggerPrimitive);
275 0 : if (!quiet) {
276 0 : fmt::print(" TP fragment size: {}\n", frag->get_data_size());
277 0 : fmt::print(" Num TPs: {}\n", n_tps);
278 : }
279 :
280 : // Create a TP buffer
281 0 : std::vector<trgdataformats::TriggerPrimitive> tp_buffer;
282 : // Prepare the TP buffer, checking for time ordering
283 0 : tp_buffer.reserve(tp_buffer.size() + n_tps);
284 :
285 : // Populate the TP buffer
286 0 : trgdataformats::TriggerPrimitive* tp_array = static_cast<trgdataformats::TriggerPrimitive*>(frag->get_data());
287 0 : uint64_t last_ts = 0;
288 0 : for (size_t i(0); i < n_tps; ++i) {
289 0 : auto& tp = tp_array[i];
290 0 : if (tp.time_start <= last_ts && !quiet) {
291 0 : fmt::print(" ERROR: {} {} ", +tp.time_start, last_ts);
292 : }
293 0 : tp_buffer.push_back(tp);
294 : }
295 :
296 : // Print some useful info
297 0 : uint64_t d_ts = tp_array[n_tps - 1].time_start - tp_array[0].time_start;
298 0 : if (!quiet)
299 0 : fmt::print(" TS gap: {} {} ms\n", d_ts, d_ts * 16.0 / 1'000'000);
300 :
301 : //
302 : // TA Processing
303 : //
304 :
305 0 : const auto ta_start = std::chrono::steady_clock::now();
306 0 : std::unique_ptr<daqdataformats::Fragment> ta_frag = ta_emulator->emulate_vector(tp_buffer);
307 0 : const auto ta_end = std::chrono::steady_clock::now();
308 :
309 0 : if (ta_frag == nullptr) // Buffer was empty.
310 0 : continue;
311 0 : num_tas += ta_emulator->get_last_output_buffer().size();
312 :
313 : // TA time calculation.
314 0 : const uint64_t ta_diff = std::chrono::nanoseconds(ta_end - ta_start).count();
315 0 : average_ta_time += ta_diff;
316 0 : if (!quiet) {
317 0 : fmt::print("\tTA Time Process: {} ns.\n", ta_diff);
318 : }
319 :
320 0 : daqdataformats::FragmentHeader frag_hdr = frag->get_header();
321 :
322 : // Customise the source id (add 1000 to id)
323 0 : frag_hdr.element_id =
324 0 : daqdataformats::SourceID{ daqdataformats::SourceID::Subsystem::kTrigger, frag->get_element_id().id + 1000 };
325 :
326 0 : ta_frag->set_header_fields(frag_hdr);
327 0 : ta_frag->set_type(daqdataformats::FragmentType::kTriggerActivity);
328 :
329 0 : tsl.add_fragment(std::move(ta_frag));
330 : //
331 : // TA Processing Ends
332 : //
333 :
334 : //
335 : // TC Processing
336 : //
337 :
338 0 : std::vector<triggeralgs::TriggerActivity> ta_buffer = ta_emulator->get_last_output_buffer();
339 0 : const auto tc_start = std::chrono::steady_clock::now();
340 0 : std::unique_ptr<daqdataformats::Fragment> tc_frag = tc_emulator->emulate_vector(ta_buffer);
341 0 : const auto tc_end = std::chrono::steady_clock::now();
342 :
343 0 : if (tc_frag == nullptr) // Buffer was empty.
344 0 : continue;
345 0 : num_tcs += tc_emulator->get_last_output_buffer().size();
346 :
347 : // TC time calculation.
348 0 : const uint64_t tc_diff = std::chrono::nanoseconds(tc_end - tc_start).count();
349 0 : average_tc_time += tc_diff;
350 0 : if (!quiet) {
351 0 : fmt::print("\tTC Time Process: {} ns.\n", tc_diff);
352 : }
353 :
354 : // Shares the same frag_hdr.
355 0 : tc_frag->set_header_fields(frag_hdr);
356 0 : tc_frag->set_type(daqdataformats::FragmentType::kTriggerCandidate);
357 :
358 0 : tsl.add_fragment(std::move(tc_frag));
359 :
360 0 : } // Fragment for loop
361 :
362 0 : if (num_tas == 0)
363 0 : average_ta_time = 0;
364 : else
365 0 : average_ta_time /= num_tas;
366 0 : if (num_tcs == 0)
367 0 : average_tc_time = 0;
368 : else
369 0 : average_tc_time /= num_tcs;
370 0 : if (!quiet) {
371 0 : fmt::print("\t\tAverage TA Time Process ({} TAs): {} ns.\n", num_tas, average_ta_time);
372 0 : fmt::print("\t\tAverage TC Time Process ({} TCs): {} ns.\n", num_tcs, average_tc_time);
373 : }
374 0 : });
375 :
376 0 : rp.loop(num_rec, skip_rec);
377 :
378 : /* code */
379 0 : return 0;
380 0 : }
|
