emulate_from_tpstream.cxx File Reference

#include "trgtools/TCEmulationUnit.hpp"
#include "trgtools/TAEmulationWorker.hpp"
#include "CLI/App.hpp"
#include "CLI/Config.hpp"
#include "CLI/Formatter.hpp"
#include <fmt/core.h>
#include <fmt/format.h>
#include <fmt/chrono.h>
#include <filesystem>
#include <optional>
#include "hdf5libs/HDF5RawDataFile.hpp"
#include "hdf5libs/HDF5SourceIDHandler.hpp"
#include "triggeralgs/TriggerCandidateFactory.hpp"

Include dependency graph for emulate_from_tpstream.cxx:

Go to the source code of this file.

Classes
struct	Options
	Struct with available cli application options. More...

Functions
void	save_fragments (const std::string &_outputfilename, const hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t &_sourceid_geoid_map, std::map< uint64_t, std::vector< std::unique_ptr< daqdataformats::Fragment > > > _frags, bool _quiet)
	Saves fragments in timeslice format into output HDF5 file.
std::map< std::string, std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > > >	sort_files_per_writer (const std::vector< std::string > &_files)
	Group and order input TimeSlice files per TPStream writer application.
std::pair< uint64_t, uint64_t >	get_available_slice_id_range (const std::map< std::string, std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > > > &_files, bool _quiet)
	Compute the common SliceID interval shared by all writer groups.
void	parse_app (CLI::App &_app, Options &_opts)
	Adds options to our CLI application.
int	main (int argc, char const *argv[])

Function Documentation

get_available_slice_id_range()

std::pair< uint64_t, uint64_t > get_available_slice_id_range	(	const std::map< std::string, std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > > > &	_files,
		bool	_quiet )

Compute the common SliceID interval shared by all writer groups.

For each writer application, this function scans all of its files and finds the minimum and maximum available record IDs. It then computes the global intersection across applications:

• global_start = max(all per-application starts)
• global_end = min(all per-application ends)

The returned range is therefore the SliceID window for which data is expected to be available from every application.

Todo

: Rather than returning the SliceID range, should try to return a time range and have processors use that.

Parameters

_files	Map of writer application names to vectors of input HDF5 files.
_quiet	If false, print per-application and global range diagnostics.

Returns

Inclusive pair {global_start, global_end} of overlapping SliceIDs.

Exceptions

std::runtime_error

If _files is empty, if any file has no records, or if no overlapping SliceID interval exists.

Definition at line 168 of file emulate_from_tpstream.cxx.

main()

int main	(	int	argc,
		char const *	argv[] )

Definition at line 282 of file emulate_from_tpstream.cxx.

parse_app()

void parse_app	(	CLI::App &	_app,
		Options &	_opts )

Adds options to our CLI application.

Parameters

_app	CLI application
_opts	Struct with the available options

Definition at line 259 of file emulate_from_tpstream.cxx.

save_fragments()

void save_fragments	(	const std::string &	_outputfilename,
		const hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t &	_sourceid_geoid_map,
		std::map< uint64_t, std::vector< std::unique_ptr< daqdataformats::Fragment > > >	_frags,
		bool	_quiet )

Saves fragments in timeslice format into output HDF5 file.

Parameters

_outputfilename	Name of the hdf5 file to save the output into
_sourceid_geoid_map	sourceid–geoid map required to create a HDF5 file
_frags	Map of fragments, with a vector of fragment pointers for each slice id
_quiet	Do we want to quiet down the cout?

Todo

Maybe in the future we will want to emulate PDS TPs.

Definition at line 32 of file emulate_from_tpstream.cxx.

  params_trigger.detector_group_name = "TPC";
  params_trigger.element_name_prefix = "Link";
  params_trigger.digits_for_element_number = 5;
 
  // Fill the HDF5 layout
  std::vector<hdf5libs::HDF5PathParameters> params;
  params.push_back(params_trigger);
  layout_params.record_name_prefix = "TimeSlice";
  layout_params.digits_for_record_number = 6;
  layout_params.digits_for_sequence_number = 0;
  layout_params.record_header_dataset_name = "TimeSliceHeader";
  layout_params.raw_data_group_name = "RawData";
  layout_params.view_group_name = "Views";
  layout_params.path_params_list = {params};
 
  // Create pointer to a new output HDF5 file
  std::unique_ptr<hdf5libs::HDF5RawDataFile> output_file = std::make_unique<hdf5libs::HDF5RawDataFile>(
      output_filename,
      _frags.begin()->second[0]->get_run_number(),
      0,
      "emulate_from_tpstream",
      layout_params,
      _sourceid_geoid_map);
 
  // Iterate over the time slices & save all the fragments
  for (auto& [slice_id, vec_frags]: _frags) {
    // Create a new timeslice header
    daqdataformats::TimeSliceHeader tsh;
    tsh.timeslice_number = slice_id;
    tsh.run_number = vec_frags[0]->get_run_number();
    tsh.element_id = dunedaq::daqdataformats::SourceID(dunedaq::daqdataformats::SourceID::Subsystem::kTRBuilder, 0);
 
    // Create a new timeslice
    dunedaq::daqdataformats::TimeSlice ts(tsh);
    if (!_quiet) {
      std::cout << "Time slice number: " << slice_id << std::endl;
    }
    // Add the fragments to the timeslices
    for (std::unique_ptr<daqdataformats::Fragment>& frag_ptr: vec_frags) {
      if (!_quiet) {
        std::cout << "  Writing elementid: " << frag_ptr->get_element_id()  << " trigger number: " << frag_ptr->get_trigger_number() << " trigger_timestamp: " << frag_ptr->get_trigger_timestamp() << " window_begin: " << frag_ptr->get_window_begin() << " sequence_no: " << frag_ptr->get_sequence_number() << std::endl;
      }
      ts.add_fragment(std::move(frag_ptr));
    }
 
    // Write the timeslice to output file
    output_file->write(ts);
  }
}

sort_files_per_writer()

std::map< std::string, std::vector< std::shared_ptr< hdf5libs::HDF5RawDataFile > > > sort_files_per_writer

(

const std::vector< std::string > &

_files

)

Group and order input TimeSlice files per TPStream writer application.

Opens each input file, validates that it is of TimeSlice type, and groups files by their application_name attribute. Within each application group, files are sorted by their file_index attribute so consecutive segments from the same writer are processed in order.

Parameters

_files

Vector of input HDF5 file paths.

Returns

Map keyed by application_name, each value a vector of file handles ordered by file_index.

Exceptions

std::runtime_error

If any input file is not of type TimeSlice.

Definition at line 114 of file emulate_from_tpstream.cxx.

{
  std::map<std::string, std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>>> files_sorted;
 
  // Put files into per-writer app groups
  for (const std::string& file : _files) {
    std::shared_ptr<hdf5libs::HDF5RawDataFile> fl = std::make_shared<hdf5libs::HDF5RawDataFile>(file);
    if (!fl->is_timeslice_type()) {
      throw std::runtime_error(fmt::format("ERROR: input file '{}' not of type 'TimeSlice'", file));
    }
 
    std::string application_name = fl->get_attribute<std::string>("application_name");
    files_sorted[application_name].push_back(fl);
  }
 
  // Sort files for each writer application individually
  for (auto& [app_name, vec_files]: files_sorted) {
    // Don't sort if we have 0 or 1 files in the application...
    if (vec_files.size() <= 1) {
      continue;
    }
 
    // Sort w.r.t. file index attribute
    std::sort(vec_files.begin(), vec_files.end(),
        [](const std::shared_ptr<hdf5libs::HDF5RawDataFile>& a, const std::shared_ptr<hdf5libs::HDF5RawDataFile>& b) {
        return a->get_attribute<size_t>("file_index") <
               b->get_attribute<size_t>("file_index");
        });
  }
 
  return files_sorted;
};