Line data Source code
1 : #include "trgtools/TAEmulationWorker.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 : #include <optional>
13 :
14 : #include "hdf5libs/HDF5RawDataFile.hpp"
15 : #include "hdf5libs/HDF5SourceIDHandler.hpp"
16 :
17 : #include "triggeralgs/TriggerCandidateFactory.hpp"
18 :
19 : //-----------------------------------------------------------------------------
20 :
21 : using namespace dunedaq;
22 : using namespace trgtools;
23 :
24 : /**
25 : * @brief Saves fragments in timeslice format into output HDF5 file
26 : *
27 : * @param _outputfilename Name of the hdf5 file to save the output into
28 : * @param _sourceid_geoid_map sourceid--geoid map required to create a HDF5 file
29 : * @param _frags Map of fragments, with a vector of fragment pointers for each slice id
30 : * @param _quiet Do we want to quiet down the cout?
31 : */
32 : void
33 0 : save_fragments(const std::string& _outputfilename,
34 : const hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t& _sourceid_geoid_map,
35 : std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>> _frags,
36 : bool _quiet)
37 : {
38 0 : std::string output_filename = _outputfilename;
39 0 : if (output_filename.size() < 5 || output_filename.compare(output_filename.size() - 5, 5, ".hdf5") != 0) {
40 0 : output_filename += ".hdf5";
41 : }
42 :
43 : // Create layout parameter object required for HDF5 creation
44 0 : hdf5libs::HDF5FileLayoutParameters layout_params;
45 :
46 : // Create HDF5 parameter path required for the layout
47 0 : hdf5libs::HDF5PathParameters params_trigger;
48 0 : params_trigger.detector_group_type = "Detector_Readout";
49 : /// @todo Maybe in the future we will want to emulate PDS TPs.
50 0 : params_trigger.detector_group_name = "TPC";
51 0 : params_trigger.element_name_prefix = "Link";
52 0 : params_trigger.digits_for_element_number = 5;
53 :
54 : // Fill the HDF5 layout
55 0 : std::vector<hdf5libs::HDF5PathParameters> params;
56 0 : params.push_back(params_trigger);
57 0 : layout_params.record_name_prefix = "TimeSlice";
58 0 : layout_params.digits_for_record_number = 6;
59 0 : layout_params.digits_for_sequence_number = 0;
60 0 : layout_params.record_header_dataset_name = "TimeSliceHeader";
61 0 : layout_params.raw_data_group_name = "RawData";
62 0 : layout_params.view_group_name = "Views";
63 0 : layout_params.path_params_list = { params };
64 :
65 : // Create pointer to a new output HDF5 file
66 0 : std::unique_ptr<hdf5libs::HDF5RawDataFile> output_file =
67 : std::make_unique<hdf5libs::HDF5RawDataFile>(output_filename,
68 0 : _frags.begin()->second[0]->get_run_number(),
69 0 : 0,
70 : "emulate_from_tpstream",
71 : layout_params,
72 0 : _sourceid_geoid_map);
73 :
74 : // Iterate over the time slices & save all the fragments
75 0 : for (auto& [slice_id, vec_frags] : _frags) {
76 : // Create a new timeslice header
77 0 : daqdataformats::TimeSliceHeader tsh;
78 0 : tsh.timeslice_number = slice_id;
79 0 : tsh.run_number = vec_frags[0]->get_run_number();
80 0 : tsh.element_id = dunedaq::daqdataformats::SourceID(dunedaq::daqdataformats::SourceID::Subsystem::kTRBuilder, 0);
81 :
82 : // Create a new timeslice
83 0 : dunedaq::daqdataformats::TimeSlice ts(tsh);
84 0 : if (!_quiet) {
85 0 : std::cout << "Time slice number: " << slice_id << std::endl;
86 : }
87 : // Add the fragments to the timeslices
88 0 : for (std::unique_ptr<daqdataformats::Fragment>& frag_ptr : vec_frags) {
89 0 : if (!_quiet) {
90 0 : std::cout << " Writing elementid: " << frag_ptr->get_element_id()
91 0 : << " trigger number: " << frag_ptr->get_trigger_number()
92 0 : << " trigger_timestamp: " << frag_ptr->get_trigger_timestamp()
93 0 : << " window_begin: " << frag_ptr->get_window_begin()
94 0 : << " sequence_no: " << frag_ptr->get_sequence_number() << std::endl;
95 : }
96 0 : ts.add_fragment(std::move(frag_ptr));
97 : }
98 :
99 : // Write the timeslice to output file
100 0 : output_file->write(ts);
101 0 : }
102 0 : }
103 :
104 : /**
105 : * @brief Group and order input TimeSlice files per TPStream writer application.
106 : *
107 : * Opens each input file, validates that it is of `TimeSlice` type, and groups
108 : * files by their `application_name` attribute. Within each application group,
109 : * files are sorted by their `file_index` attribute so consecutive segments from
110 : * the same writer are processed in order.
111 : *
112 : * @param _files Vector of input HDF5 file paths.
113 : * @return Map keyed by `application_name`, each value a vector of file handles
114 : * ordered by `file_index`.
115 : * @throws std::runtime_error If any input file is not of type `TimeSlice`.
116 : */
117 : std::map<std::string, std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>>>
118 0 : sort_files_per_writer(const std::vector<std::string>& _files)
119 : {
120 0 : std::map<std::string, std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>>> files_sorted;
121 :
122 : // Put files into per-writer app groups
123 0 : for (const std::string& file : _files) {
124 0 : std::shared_ptr<hdf5libs::HDF5RawDataFile> fl = std::make_shared<hdf5libs::HDF5RawDataFile>(file);
125 0 : if (!fl->is_timeslice_type()) {
126 0 : throw std::runtime_error(fmt::format("ERROR: input file '{}' not of type 'TimeSlice'", file));
127 : }
128 :
129 0 : std::string application_name = fl->get_attribute<std::string>("application_name");
130 0 : files_sorted[application_name].push_back(fl);
131 0 : }
132 :
133 : // Sort files for each writer application individually
134 0 : for (auto& [app_name, vec_files] : files_sorted) {
135 : // Don't sort if we have 0 or 1 files in the application...
136 0 : if (vec_files.size() <= 1) {
137 0 : continue;
138 : }
139 :
140 : // Sort w.r.t. file index attribute
141 0 : std::sort(
142 : vec_files.begin(),
143 : vec_files.end(),
144 0 : [](const std::shared_ptr<hdf5libs::HDF5RawDataFile>& a, const std::shared_ptr<hdf5libs::HDF5RawDataFile>& b) {
145 0 : return a->get_attribute<size_t>("file_index") < b->get_attribute<size_t>("file_index");
146 : });
147 : }
148 :
149 0 : return files_sorted;
150 0 : };
151 :
152 : /**
153 : * @brief Compute the common SliceID interval shared by all writer groups.
154 : *
155 : * For each writer application, this function scans all of its files and finds
156 : * the minimum and maximum available record IDs. It then computes the global
157 : * intersection across applications:
158 : * - global_start = max(all per-application starts)
159 : * - global_end = min(all per-application ends)
160 : *
161 : * The returned range is therefore the SliceID window for which data is
162 : * expected to be available from every application.
163 : *
164 : * @todo: Rather than returning the SliceID range, should try to return a time range and have processors use that.
165 : *
166 : * @param _files Map of writer application names to vectors of input HDF5 files.
167 : * @param _quiet If false, print per-application and global range diagnostics.
168 : * @return Inclusive pair `{global_start, global_end}` of overlapping SliceIDs.
169 : * @throws std::runtime_error If `_files` is empty, if any file has no records,
170 : * or if no overlapping SliceID interval exists.
171 : */
172 : std::pair<uint64_t, uint64_t>
173 0 : get_available_slice_id_range(
174 : const std::map<std::string, std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>>>& _files,
175 : bool _quiet)
176 : {
177 0 : if (_files.empty()) {
178 0 : throw std::runtime_error("No files provided");
179 : }
180 :
181 0 : uint64_t global_start = std::numeric_limits<uint64_t>::min();
182 0 : uint64_t global_end = std::numeric_limits<uint64_t>::max();
183 :
184 : // Get the min & max record id per application, and the global
185 0 : for (auto& [appid, vec_files] : _files) {
186 0 : uint64_t app_start = std::numeric_limits<uint64_t>::max();
187 0 : uint64_t app_end = std::numeric_limits<uint64_t>::min();
188 :
189 : // Find min / max record id for this application
190 0 : for (auto& file : vec_files) {
191 0 : auto record_ids = file->get_all_record_ids();
192 0 : if (record_ids.empty()) {
193 0 : throw std::runtime_error(fmt::format("File from application {} contains no records.", appid));
194 : }
195 0 : app_start = std::min(app_start, record_ids.begin()->first);
196 0 : app_end = std::max(app_end, record_ids.rbegin()->first);
197 0 : }
198 :
199 : // Update the global min / max record id
200 0 : global_start = std::max(global_start, app_start);
201 0 : global_end = std::min(global_end, app_end);
202 :
203 0 : if (!_quiet) {
204 0 : std::cout << "Application: " << appid << " " << " TimeSliceID start: " << app_start << " end: " << app_end
205 0 : << std::endl;
206 : }
207 : }
208 :
209 0 : if (!_quiet) {
210 0 : std::cout << "Global start: " << global_start << " Global end: " << global_end << std::endl;
211 : }
212 0 : if (global_start > global_end) {
213 0 : throw std::runtime_error("One of the provided files' id range did not overlap with the rest. Please select files "
214 0 : "with overlapping TimeSlice IDs");
215 : }
216 :
217 : // Extra validation / error handling
218 0 : for (auto& [appid, vec_files] : _files) {
219 0 : for (auto& file : vec_files) {
220 0 : auto record_ids = file->get_all_record_ids();
221 :
222 0 : uint64_t file_start = record_ids.begin()->first;
223 0 : uint64_t file_end = record_ids.rbegin()->first;
224 0 : if ((file_start > global_end || file_end < global_start)) {
225 0 : uint64_t file_index = file->get_attribute<size_t>("file_index");
226 0 : throw std::runtime_error(fmt::format("File from TPStreamWrite application '{}' (index '{}') has record id "
227 : "range [{}, {}], which does not overlap with global range [{}, {}].",
228 : appid,
229 : file_index,
230 : file_start,
231 : file_end,
232 : global_start,
233 0 : global_end));
234 : }
235 0 : }
236 : }
237 :
238 0 : return { global_start, global_end };
239 : }
240 :
241 : /**
242 : * @brief Struct with available cli application options
243 : */
244 : struct Options
245 : {
246 : /// @brief vector of input filenames
247 : std::vector<std::string> input_files;
248 : /// @brief output filename
249 : std::string output_filename;
250 : /// @brief the configuration filename
251 : std::string config_name;
252 : /// @brief do we want to quiet down the cout? Default: no
253 : bool quiet = false;
254 : /// @brief do we want to measure latencies? Default: no
255 : /// @todo: Latencies currently not supported!
256 : bool latencies = false;
257 : /// @brief runs each TAMaker on a separate thread
258 : bool run_parallel = false;
259 : /// @brief optional lower bound (inclusive) for TimeSlice IDs to process
260 : std::optional<uint64_t> slice_start = std::nullopt;
261 : /// @brief optional number of TimeSlices to process
262 : std::optional<uint64_t> num_slices = std::nullopt;
263 : };
264 :
265 : /**
266 : * @brief Adds options to our CLI application
267 : *
268 : * @param _app CLI application
269 : * @param _opts Struct with the available options
270 : */
271 : void
272 0 : parse_app(CLI::App& _app, Options& _opts)
273 : {
274 0 : _app.add_option("-i,--input-files", _opts.input_files, "List of input files (required)")
275 : ->required()
276 0 : ->check(CLI::ExistingFile); // Validate that each file exists
277 :
278 0 : _app.add_option("-o,--output-file", _opts.output_filename, "Output file (required)")
279 0 : ->required(); // make the argument required
280 :
281 0 : _app
282 0 : .add_option("-j,--json-config", _opts.config_name, "Trigger Activity and Candidate config JSON to use (required)")
283 : ->required()
284 0 : ->check(CLI::ExistingFile);
285 :
286 0 : _app.add_flag("--parallel", _opts.run_parallel, "Run the TAMakers in parallel");
287 :
288 0 : _app.add_flag("--quiet", _opts.quiet, "Quiet outputs.");
289 :
290 0 : _app.add_flag("--latencies", _opts.latencies, "Saves latencies per TP into csv");
291 :
292 0 : _app.add_option("-s,--slice-start", _opts.slice_start, "Inclusive lower bound for TimeSlice ID to process");
293 0 : _app.add_option("-n,--num-slices", _opts.num_slices, "Number of TimeSlices to process");
294 0 : }
295 :
296 : int
297 0 : main(int argc, char const* argv[])
298 : {
299 : // Do all the CLI processing first
300 0 : CLI::App app{ "Offline trigger TriggerActivity & TriggerCandidate emulatior" };
301 0 : Options opts{};
302 :
303 0 : parse_app(app, opts);
304 :
305 0 : try {
306 0 : app.parse(argc, argv);
307 0 : } catch (const CLI::ParseError& e) {
308 0 : return app.exit(e);
309 0 : }
310 :
311 : // Get the configuration file
312 0 : std::ifstream config_stream(opts.config_name);
313 0 : nlohmann::json config = nlohmann::json::parse(config_stream);
314 :
315 0 : if (!opts.quiet) {
316 0 : std::cout << "Files to process:\n";
317 0 : for (const std::string& file : opts.input_files) {
318 0 : std::cout << "- " << file << "\n";
319 : }
320 : }
321 :
322 : // Sort the files into a map writer_id::vector<HDF5>
323 0 : std::map<std::string, std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>>> sorted_files =
324 0 : sort_files_per_writer(opts.input_files);
325 :
326 : // Get the available record_id range
327 0 : std::pair<uint64_t, uint64_t> recordid_range = get_available_slice_id_range(sorted_files, opts.quiet);
328 0 : std::pair<uint64_t, uint64_t> processing_range = recordid_range;
329 :
330 0 : if (opts.slice_start.has_value()) {
331 0 : processing_range.first = std::max(processing_range.first, opts.slice_start.value());
332 : }
333 :
334 0 : if (opts.num_slices.has_value()) {
335 0 : if (opts.num_slices.value() == 0) {
336 0 : throw std::runtime_error("Invalid --num-slices value: must be greater than 0");
337 : }
338 :
339 0 : uint64_t requested_end = processing_range.first + (opts.num_slices.value() - 1);
340 0 : if (requested_end < processing_range.first) {
341 0 : requested_end = std::numeric_limits<uint64_t>::max();
342 : }
343 :
344 0 : processing_range.second = std::min(processing_range.second, requested_end);
345 : }
346 :
347 0 : if (processing_range.first > processing_range.second) {
348 0 : throw std::runtime_error(fmt::format(
349 : "Requested timeslice subset (slice-start={}, num-slices={}) does not overlap available range [{}, {}]",
350 0 : opts.slice_start.value_or(0),
351 0 : opts.num_slices.value_or(0),
352 : recordid_range.first,
353 0 : recordid_range.second));
354 : }
355 :
356 0 : if (!opts.quiet) {
357 0 : std::cout << "Processing TimeSliceID range: [" << processing_range.first << ", " << processing_range.second << "]"
358 0 : << std::endl;
359 : }
360 :
361 : // Create the file handlers
362 0 : std::vector<std::unique_ptr<TAEmulationWorker>> ta_emu_workers;
363 0 : for (auto [name, files] : sorted_files) {
364 0 : ta_emu_workers.push_back(
365 0 : std::make_unique<TAEmulationWorker>(files, config, processing_range, opts.run_parallel, opts.quiet));
366 0 : }
367 :
368 : // Start each file handler
369 0 : for (const auto& handler : ta_emu_workers) {
370 0 : handler->start_processing();
371 : }
372 :
373 : // Output map of TA vectors & function that appends TAs to that vector
374 0 : std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>> tas;
375 0 : auto append_tas = [&tas](std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>>&& _tas) {
376 0 : for (auto& [sliceid, src_vec] : _tas) {
377 0 : auto& dest_vec = tas[sliceid];
378 0 : dest_vec.reserve(dest_vec.size() + src_vec.size());
379 :
380 0 : dest_vec.insert(dest_vec.end(), std::make_move_iterator(src_vec.begin()), std::make_move_iterator(src_vec.end()));
381 0 : src_vec.clear();
382 : }
383 0 : };
384 :
385 : // Output map of Fragment vectors & function that appends TAs to that vector
386 0 : std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>> frags;
387 0 : auto append_frags = [&frags](std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>>&& _frags) {
388 0 : for (auto& [sliceid, src_vec] : _frags) {
389 0 : auto& dest_vec = frags[sliceid];
390 :
391 0 : dest_vec.reserve(dest_vec.size() + src_vec.size());
392 :
393 0 : dest_vec.insert(dest_vec.end(), std::make_move_iterator(src_vec.begin()), std::make_move_iterator(src_vec.end()));
394 0 : src_vec.clear();
395 : }
396 0 : };
397 :
398 : // Iterate over the handlers, wait for them to complete their job & append
399 : // their TAs to our vector when ready.
400 0 : hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t sourceid_geoid_map;
401 0 : for (const auto& handler : ta_emu_workers) {
402 : // Wait for all TAs to be made
403 0 : handler->wait_to_complete_work();
404 :
405 : // Append output TA/fragments
406 0 : append_tas(std::move(handler->get_tas()));
407 0 : append_frags(std::move(handler->get_frags()));
408 :
409 : // Get and merge the source ID map
410 0 : hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t map = handler->get_sourceid_geoid_map();
411 0 : sourceid_geoid_map.insert(map.begin(), map.end());
412 0 : }
413 :
414 : // Sort the TAs in each slice before pushing them into the TCMaker
415 0 : size_t n_tas = 0;
416 0 : for (auto& [sliceid, vec_tas] : tas) {
417 0 : std::sort(
418 0 : vec_tas.begin(), vec_tas.end(), [](const triggeralgs::TriggerActivity& a, const triggeralgs::TriggerActivity& b) {
419 0 : return std::tie(a.time_start, a.channel_start, a.time_end) <
420 0 : std::tie(b.time_start, b.channel_start, b.time_end);
421 : });
422 0 : n_tas += vec_tas.size();
423 : }
424 :
425 0 : if (!opts.quiet) {
426 0 : std::cout << "Total number of TAs made: " << n_tas << std::endl;
427 0 : std::cout << "Creating a TCMaker..." << std::endl;
428 : }
429 : // Create the TC emulator
430 0 : std::string algo_name = config["trigger_candidate_plugin"][0];
431 0 : nlohmann::json algo_config = config["trigger_candidate_config"][0];
432 :
433 0 : std::unique_ptr<triggeralgs::TriggerCandidateMaker> tc_maker =
434 0 : triggeralgs::TriggerCandidateFactory::get_instance()->build_maker(algo_name);
435 0 : tc_maker->configure(algo_config);
436 :
437 0 : trgtools::TCEmulationUnit tc_emulator;
438 0 : tc_emulator.set_maker(tc_maker);
439 :
440 : // Emulate the TriggerCandidates
441 0 : std::vector<triggeralgs::TriggerCandidate> tcs;
442 0 : for (auto& [sliceid, vec_tas] : tas) {
443 0 : std::unique_ptr<daqdataformats::Fragment> tc_frag = tc_emulator.emulate_vector(vec_tas);
444 0 : if (!tc_frag) {
445 0 : continue;
446 : }
447 :
448 : // Manipulate the fragment header
449 0 : daqdataformats::FragmentHeader frag_hdr = tc_frag->get_header();
450 0 : frag_hdr.element_id =
451 0 : daqdataformats::SourceID{ daqdataformats::SourceID::Subsystem::kTrigger, tc_frag->get_element_id().id + 10000 };
452 :
453 0 : tc_frag->set_header_fields(frag_hdr);
454 0 : tc_frag->set_type(daqdataformats::FragmentType::kTriggerCandidate);
455 0 : tc_frag->set_trigger_number(sliceid);
456 0 : tc_frag->set_window_begin(frags[sliceid][0]->get_window_begin());
457 0 : tc_frag->set_window_end(frags[sliceid][0]->get_window_end());
458 :
459 : // Push the fragment to our list
460 0 : frags[sliceid].push_back(std::move(tc_frag));
461 :
462 0 : std::vector<triggeralgs::TriggerCandidate> tmp_tcs = tc_emulator.get_last_output_buffer();
463 0 : tcs.reserve(tcs.size() + tmp_tcs.size());
464 0 : tcs.insert(tcs.end(), std::make_move_iterator(tmp_tcs.begin()), std::make_move_iterator(tmp_tcs.end()));
465 0 : }
466 0 : if (!opts.quiet) {
467 0 : std::cout << "Total number of TCs made: " << tcs.size() << std::endl;
468 : }
469 :
470 0 : if (!frags.empty()) {
471 0 : save_fragments(opts.output_filename, sourceid_geoid_map, std::move(frags), opts.quiet);
472 : } else {
473 0 : std::cout << "No TA/TC fragments generated. Output file will not be generated" << std::endl;
474 : }
475 :
476 0 : return 0;
477 0 : }
|
