PdspHDFFileReader.hh

Go to the documentation of this file.

#ifndef dune_ArtModules_PdspHDFFileReader_hh
#define dune_ArtModules_PdspHDFFileReader_hh

#include "artdaq-core/Data/ContainerFragment.hh"
#include "artdaq-core/Data/Fragment.hh"
#include "artdaq-core/Data/RawEvent.hh"
#include "artdaq-core/Utilities/ExceptionHandler.hh"
#include "artdaq-core/Utilities/TimeUtils.hh"

#include "art/Framework/Core/Frameworkfwd.h"

#include "art/Framework/Core/FileBlock.h"
#include "art/Framework/Core/ProductRegistryHelper.h"
#include "art/Framework/IO/Sources/SourceHelper.h"
#include "art/Framework/IO/Sources/put_product_in_principal.h"
#include "art/Framework/Principal/EventPrincipal.h"
#include "art/Framework/Principal/RunPrincipal.h"
#include "art/Framework/Principal/SubRunPrincipal.h"
#include "art/Framework/Services/Registry/ServiceHandle.h"
#include "canvas/Persistency/Provenance/FileFormatVersion.h"
#include "fhiclcpp/ParameterSet.h"

#include "dune-raw-data/HDFUtils/HDFFileReader.hh"

#include <sys/time.h>
#include <map>
#include <string>

namespace dune
{

    /**
     * \brief The PdspHDFFileReader is a class which implements the methods needed by art::Source
     */
struct PdspHDFFileReader
{
  /**
   * \brief Copy Constructor is deleted
   */
  PdspHDFFileReader(PdspHDFFileReader const&) = delete;

  /**
   * \brief Copy Assignment operator is deleted
   * \return PdspHDFFileReader copy
   */
  PdspHDFFileReader& operator=(PdspHDFFileReader const&) = delete;

  art::SourceHelper const& pmaker;                            ///< An art::SourceHelper instance
  std::string pretend_module_name;                            ///< The module name to store data under
  size_t bytesRead;                                           ///< running total of number of bytes received
  unsigned readNext_calls_;                                   ///< The number of times readNext has been called
  //std::unique_ptr<artdaq::hdf5::FragmentDataset> inputFile_;  ///< The Dataset plugin which this input source will be reading from
  std::unique_ptr<dune::HDFFileReader::HDFFileInfo> hdf_file_;
  std::list<std::string> unprocessedEventList_;

  /**
   * \brief PdspHDFFileReader Constructor
   * \param ps ParameterSet used for configuring PdspHDFFileReader
   * \param help art::ProductRegistryHelper which is used to inform art about different Fragment types
   * \param pm art::SourceHelper used to initalize the SourceHelper member
   *
   * \verbatim
   * PdspHDFFileReader accepts the following Parameters:
   * "raw_data_label" (Default: "daq"): The label to use for all raw data
   * "shared_memory_key" (Default: 0xBEE7): The key for the shared memory segment
   * \endverbatim
   */
  PdspHDFFileReader(fhicl::ParameterSet const& ps,
                    art::ProductRegistryHelper& help,
                    art::SourceHelper const& pm)
  : pmaker(pm)
  , pretend_module_name(ps.get<std::string>("raw_data_label", "daq"))
    , bytesRead(0)
    , readNext_calls_(0)
  {
    TLOG_INFO("PdspHDFFileReader") << "PdspHDFFileReader initializing with ParameterSet: " << ps.to_string();

    std::set<std::string> instance_names = {"FELIX", "PHOTON"};
    for (const auto& set_iter : instance_names)
    {
      help.reconstitutes<artdaq::Fragments, art::InEvent>(pretend_module_name, set_iter);
    }
    hdf_file_ = dune::HDFFileReader::openFile(ps.get<std::string>("fileName"));
    unprocessedEventList_ = dune::HDFFileReader::getTopLevelGroupNames(hdf_file_);

    TLOG_INFO("PdspHDFFileReader") << "PdspHDFFileReader opened HDF file with run number " << hdf_file_->runNumber
                                   << " and " << unprocessedEventList_.size() << " events";
  }

  /**
   * \brief PdspHDFFileReader destructor
   */
  virtual ~PdspHDFFileReader()
  {
    dune::HDFFileReader::closeFile(std::move(hdf_file_));
  }

  /**
   * \brief Emulate closing a file. No-Op.
   */
  void closeCurrentFile() {}

  /**
   * \brief Emulate opening a file
   * \param[out] fb art::FileBlock object
   */
  void readFile(std::string const&, art::FileBlock*& fb)
  {
    TLOG_ARB(5, "PdspHDFFileReader") << "readFile enter/start";
    fb = new art::FileBlock(art::FileFormatVersion(1, "RawEvent2011"), "nothing");
  }

  /**
   * \brief Dequeue a RawEvent and declare its Fragment contents to art, creating
   * Run, SubRun, and EventPrincipal objects as necessary
   * \param[in] inR Input art::RunPrincipal
   * \param[in] inSR Input art::SubRunPrincipal
   * \param[out] outR Output art::RunPrincipal
   * \param[out] outSR  Output art::SubRunPrincipal
   * \param[out] outE Output art::EventPrincipal
   * \return Whether an event was returned
   */
  bool readNext(art::RunPrincipal* const& inR, art::SubRunPrincipal* const& inSR, art::RunPrincipal*& outR,
                art::SubRunPrincipal*& outSR, art::EventPrincipal*& outE)
  {
    TLOG_DEBUG("PdspHDFFileReader") << "readNext BEGIN";
    ++readNext_calls_;

    // Establish default 'results'
    outR = 0;
    outSR = 0;
    outE = 0;

    if (unprocessedEventList_.empty()) {return false;}

    std::string nextEventGroupName = unprocessedEventList_.front();
    unprocessedEventList_.pop_front();

    hdf_file_ = dune::HDFFileReader::reopenFile(std::move(hdf_file_));

    dune::HDFFileReader::FragmentListsByType fragmentMap =
      dune::HDFFileReader::getFragmentsForEvent(hdf_file_, nextEventGroupName);

    if (fragmentMap.size() == 0)
    {
      TLOG_ERROR("PdspHDFFileReader") << "No data received, either because of a problem or end of file. Returning false (should exit art)";
      return false;
    }

    artdaq::detail::RawEventHeader evtHeader(hdf_file_->runNumber, 1, readNext_calls_, readNext_calls_, 0);

    for (auto& subDetName_fragmentList_pair : fragmentMap)
    {
      for (auto& frag : *subDetName_fragmentList_pair.second)
      {
        TLOG_DEBUG("PdspHDFFileReader") << subDetName_fragmentList_pair.first << " Fragment, " << " type "
                                        << ((int) frag.type()) << ", timestamp " << frag.timestamp()
                                        << ", sequence_id " << frag.sequenceID()
                                        << " fragment_id " << frag.fragmentID();
        evtHeader.event_id = frag.sequenceID();
        evtHeader.sequence_id = frag.sequenceID();
        evtHeader.timestamp = frag.timestamp();
      }
    }

    art::Timestamp currentTime = 0;
    timespec hi_res_time;
    int retcode = clock_gettime(CLOCK_REALTIME, &hi_res_time);
    TLOG_ARB(15, "PdspHDFFileReader") << "hi_res_time tv_sec = " << hi_res_time.tv_sec
                                  << " tv_nsec = " << hi_res_time.tv_nsec << " (retcode = " << retcode << ")";
    if (retcode == 0)
    {
      currentTime = ((hi_res_time.tv_sec & 0xffffffff) << 32) | (hi_res_time.tv_nsec & 0xffffffff);
    }
    else
    {
      TLOG_ERROR("PdspHDFFileReader")
        << "Unable to fetch a high-resolution time with clock_gettime for art::Event Timestamp. "
        << "The art::Event Timestamp will be zero for event " << evtHeader.event_id;
    }

    // make new run if inR is 0 or if the run has changed
    if (inR == 0 || inR->run() != evtHeader.run_id)
    {
      outR = pmaker.makeRunPrincipal(evtHeader.run_id, currentTime);
    }

    // make new subrun if inSR is 0 or if the subrun has changed
    art::SubRunID subrun_check(evtHeader.run_id, evtHeader.subrun_id);
    if (inSR == 0 || subrun_check != inSR->subRunID())
    {
      outSR = pmaker.makeSubRunPrincipal(evtHeader.run_id, evtHeader.subrun_id, currentTime);
    }

    outE = pmaker.makeEventPrincipal(evtHeader.run_id, evtHeader.subrun_id, evtHeader.event_id, currentTime);

    for (auto& subDetName_fragmentList_pair : fragmentMap)
    {
      put_product_in_principal(std::move(subDetName_fragmentList_pair.second), *outE, pretend_module_name,
                               subDetName_fragmentList_pair.first);
    }


#if 0
    else
    {
      TLOG_TRACE("PdspHDFFileReader") << "EventHeader has Run " << evtHeader.run_id << ", SubRun " << evtHeader.subrun_id << ", Event " << evtHeader.event_id << ", SeqID " << evtHeader.sequence_id << ", IsComplete " << evtHeader.is_complete;
    }


    for (auto& fragmentTypePair : eventMap)
    {
      auto type_code = fragmentTypePair.first;
      TLOG_TRACE("PdspHDFFileReader") << "Before GetFragmentsByType call, type is " << (int)type_code;
      TLOG_TRACE("PdspHDFFileReader") << "After GetFragmentsByType call, number of fragments is " << fragmentTypePair.second->size();

      std::unordered_map<std::string, std::unique_ptr<artdaq::Fragments>> derived_fragments;
      for (auto& frag : *fragmentTypePair.second)
      {
        bytesRead += frag.sizeBytes();

        //      std::pair<bool, std::string> instance_name_result = translator->GetInstanceNameForFragment(frag);
        std::string label = instance_name_result.second;
        if (!instance_name_result.first)
        {
          TLOG_WARNING("PdspHDFFileReader")
            << "UnknownFragmentType: The product instance name mapping for fragment type \"" << ((int)type_code)
            << "\" is not known. Fragments of this "
            << "type will be stored in the event with an instance name of \"" << label << "\".";
        }
        if (!derived_fragments.count(label))
        {
          derived_fragments[label] = std::make_unique<Fragments>();
        }
        auto fragSize = frag.size();
        TLOG_TRACE("PdspHDFFileReader") << "Moving Fragment with size " << fragSize << " from type map (type=" << type_code << ") to derived map label=" << label;
        derived_fragments[label]->emplace_back(std::move(frag));
      }
      TLOG_TRACE("PdspHDFFileReader") << "Placing derived fragments in outE";
      for (auto& type : derived_fragments)
      {
        put_product_in_principal(std::move(type.second), *outE, pretend_module_name, type.first);
      }
    }
#endif
    TLOG_TRACE("PdspHDFFileReader") << "After putting fragments in event";
    TLOG_ARB(10, "PdspHDFFileReader") << "readNext: bytesRead=" << bytesRead;;

    TLOG_TRACE("PdspHDFFileReader") << "Returning from readNext";
    return true;
  }
};
}  // namespace dune

#endif /* dune_ArtModules_PdspHDFFileReader_hh */