TAFileHandler.hpp

Go to the documentation of this file.

#ifndef TRGTOOLS_TAFILEHANDLER_HPP_
#define TRGTOOLS_TAFILEHANDLER_HPP_
 
#include "trgtools/EmulateTAUnit.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 "hdf5libs/HDF5RawDataFile.hpp"
#include "trgdataformats/TriggerPrimitive.hpp"
#include "triggeralgs/TriggerActivityFactory.hpp"
#include "triggeralgs/TriggerObjectOverlay.hpp"
 
namespace dunedaq::trgtools 
{
 
class TAFileHandler
{
  public:
    TAFileHandler(std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>> _input_files,
                  nlohmann::json _config,
                  std::pair<uint64_t, uint64_t> _sliceid_range,
                  bool _run_parallel,
                  bool _quiet);
 
    ~TAFileHandler() = default;
 
    std::vector<daqdataformats::SourceID>
    get_valid_sourceids(daqdataformats::TimeSlice& _timeslice);
  
    void start_processing();
 
    void wait_to_complete_work();
 
    std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>>
    get_tas();
 
    std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>>
    get_frags();
 
    hdf5libs::HDF5SourceIDHandler::source_id_geo_id_map_t
    get_sourceid_geoid_map();
 
 
  private:
    void process_tasks();
 
    void process_task(daqdataformats::SourceID _source_id,
                      uint64_t _rec,
                      daqdataformats::FragmentHeader _header,
                      std::vector<trgdataformats::TriggerPrimitive>&& _tps);
 
    void worker_thread();
 
    void enqueue_task(std::function<void()> task);
 
    void wait_to_complete_tasks();
 
  private:
    std::vector<std::shared_ptr<hdf5libs::HDF5RawDataFile>> m_input_files;
 
    nlohmann::json m_configuration;
 
    std::vector<std::string> m_input_paths;
 
    std::map<daqdataformats::SourceID, std::unique_ptr<trgtools::EmulateTAUnit>> m_ta_emulators;
 
    std::pair<uint64_t, uint64_t> m_sliceid_range;
 
    const bool m_run_parallel;
 
    const bool m_quiet;
 
    /*
     * Threading objects for the main file handler
     */
 
    std::thread m_main_thread;
 
    std::atomic<bool> m_stop{false};
 
    /*
     * Optional threading objects for the tasks
     * i.e. one thread per TAMaker
     */
 
    std::mutex m_savetps_mutex;
    std::vector<std::thread> m_thread_pool;
    std::condition_variable m_condition;
    std::condition_variable m_task_complete_condition;
    std::queue<std::function<void()>> m_task_queue;
    std::mutex m_queue_mutex;
    std::atomic<size_t> m_active_tasks;
 
    std::map<uint64_t, std::vector<triggeralgs::TriggerActivity>> m_tas;
 
    std::map<uint64_t, std::vector<std::unique_ptr<daqdataformats::Fragment>>> m_ta_fragments;
 
    uint16_t m_id;
    static uint16_t m_id_next;
    static const size_t SIZE_TP  = sizeof(trgdataformats::TriggerPrimitive);
};
 
};
 
#endif