CbmRecoUnpack.h

Go to the documentation of this file.

/* Copyright (C) 2021 GSI Helmholtzzentrum fuer Schwerionenforschung, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Volker Friese [committer], Pierre-Alain Loizeau, Pascal Raisig  */

 
 
#ifndef CBMRECOUNPACK_H
#define CBMRECOUNPACK_H 1
 
#include "AccDataSis18.h"
#include "CbmBmonUnpackConfig.h"
#include "CbmMuchUnpackConfig.h"
#include "CbmPsdUnpackConfig.h"
#include "CbmRichUnpackConfig.h"
#include "CbmStsUnpackConfig.h"
#include "CbmTofUnpackConfig.h"
#include "CbmTrdUnpackConfig.h"
#include "CbmTrdUnpackFaspConfig.h"
#include "CbmTsEventHeader.h"
 
#include <MicrosliceDescriptor.hpp>
#include <Timeslice.hpp>
 
#include <FairParAsciiFileIo.h>
#include <FairRootManager.h>
 
#include <RtypesCore.h>
#include <THnSparse.h>
#include <TObject.h>
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <ctime>
#include <memory>
#include <type_traits>  // this is std::lib used for template is_member_function_pointer
#include <utility>
#include <vector>
 
class CbmTimeSlice;
class TH1;
class TStopwatch;
namespace cbm
{
  class Monitor;
}

class CbmRecoUnpack : public TObject {
 
 public:
  CbmRecoUnpack();
 

  ~CbmRecoUnpack();

  CbmRecoUnpack(const CbmRecoUnpack&) = delete;

  CbmRecoUnpack& operator=(const CbmRecoUnpack&) = delete;

  void Finish();

  Bool_t Init();

  void Reset();

  void SetDebugPrintout(bool value = true) { fDoDebugPrints = value; }

  void SetDoPerfProfiling(bool value = true) { fDoPerfProf = value; }

  void SetDoPerfProfilingPerTs(bool value = true)
  {
    if (value) fDoPerfProf = value;
    fDoPerfProfPerTs = value;
  }

  void SetPublishProfMoni(bool value = true, std::string sUriMonitor = "", bool useProcTime = false,
                          std::string sJobId = "local")
  {
    if (value) {
      fDoPerfProf      = value;
      fDoPerfProfPerTs = value;
    }
    fPublishProfMoni = value;
    if ("" != sUriMonitor) {  //
      fUriPublishProfMoni = sUriMonitor;
    }
    fPubMoniProcTime = useProcTime;
    fMoniJobId       = sJobId;
  }

  void SetOutputFilename(std::string value)
  {
    if (!value.empty()) fOutfilename = value;
  }

  void SetTimeSorting(bool bIn = true) { bOutputFullTimeSorting = bIn; }

  void AddSpillFilter(std::string sAccEvtTimingFileIn, double_t dWinStart = -1000.0, double_t dWinStop = 4500.0,
                      int32_t iSpillStartDelay = 28400, uint64_t ulTsDurationNs = 128000000, bool bMoniFindPeak = false)
  {
    bSpillFilter              = true;
    sAccEvtTimingFile         = sAccEvtTimingFileIn;
    dSpillFilterWinStart      = dWinStart;
    dSpillFilterWinStop       = dWinStop;
    iSpillFilterStartDelay    = iSpillStartDelay;
    ulSpillFilterTsDurationNs = ulTsDurationNs;
    bOutputFullTimeSorting    = true;
    bSpillFilterMoniFindPeak  = bMoniFindPeak;
  }

  void SetMonitoringOnly(bool bIn = true) { bMonitoringOnly = bIn; }

  void SetUnpackConfig(std::shared_ptr<CbmMuchUnpackConfig> config) { fMuchConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmPsdUnpackConfig> config) { fPsdConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmRichUnpackConfig> config) { fRichConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmStsUnpackConfig> config) { fStsConfig = config; }
 
  // /** @brief Set the Tof Unpack Config @param config */
  void SetUnpackConfig(std::shared_ptr<CbmTofUnpackConfig> config) { fTofConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmTrdUnpackConfig> config) { fTrd1DConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmTrdUnpackFaspConfig> config) { fTrd2DConfig = config; }

  void SetUnpackConfig(std::shared_ptr<CbmBmonUnpackConfig> config) { fBmonConfig = config; }

  void Unpack(std::unique_ptr<fles::Timeslice> ts);
 
 private:
  using Subsystem = fles::Subsystem;

  bool fDoDebugPrints = false;  

  bool fDoPerfProf = false;  

  bool fDoPerfProfPerTs = false;  

  bool fPublishProfMoni = false;  

  std::string fUriPublishProfMoni = "influx1:localhost:8086:monitoring_tests";  

  bool fPubMoniProcTime = true;

  std::map<Subsystem, std::pair<std::string, size_t>> fNameMap = {};  

  std::map<Subsystem, std::pair<double, double>> fTimeMap = {};  

  std::map<Subsystem, std::pair<double, double>> fDataSizeMap = {};  

  std::map<Subsystem, std::pair<std::string, size_t>> fNameMapPerTs = {};  

  std::map<Subsystem, std::pair<double, double>> fTimeMapPerTs = {};  

  std::map<Subsystem, std::pair<double, double>> fDataSizeMapPerTs = {};  
 
  TH1* hProducedDigis                        = nullptr;
  TH1* hSpeedPerf                            = nullptr;
  TH1* hDataPerf                             = nullptr;
  TStopwatch* fTimerTs                       = nullptr;
  TH1* fhCpuTimePerTs                        = nullptr;  
  TH1* fhRealTimePerTs                       = nullptr;  
  TH1* fhCpuTimePerTsHist                    = nullptr;  
  TH1* fhRealTimePerTsHist                   = nullptr;  
  std::map<Subsystem, TH1*> fvhInpRatioPerTs = {};       
  std::map<Subsystem, TH1*> fvhOutRatioPerTs = {};       
  std::map<Subsystem, TH1*> fvhUnpRatioPerTs = {};       
  TH1* fhUnpackingRatioPerTs                 = nullptr;  
 
  std::unique_ptr<cbm::Monitor> fMonitor;  
  std::string fMoniCurrrentHostname                                  = "";
  std::string fMoniJobId                                             = "";
  std::chrono::system_clock::time_point fMonitorSecCurrentTs         = std::chrono::system_clock::time_point();  
  std::map<Subsystem, std::pair<double, double>> fDataSizeMapCurrSec = {};                                       

  void WriteHistograms();

  void performanceProfiling();

  void performanceProfilingPerTs();

  void initPerformanceMaps(Subsystem subsystem, std::string name);
 
 
  template<class TConfig>
  void RegisterOutputs(FairRootManager* ioman, std::shared_ptr<TConfig> config)
  {
    /*
 * FIXME: Compiling but leading to a segfault at runtime
    auto pOutVect = config->GetOutputVec();
    if (pOutVect) {
      ioman->RegisterAny(config->GetOutputBranchName().data(), pOutVect, kTRUE);
      LOG(info) << "CbmRecoUnpack::RegisterOutputs() " << config->GetOutputBranchName() << " at " << pOutVect;
    }
 
    auto pOutVectOptA = config->GetOptOutAVec();
    if (pOutVectOptA) {
      ioman->RegisterAny(config->GetOutputBranchNameOptA().data(), pOutVectOptA, kTRUE);
      LOG(info) << "CbmRecoUnpack::RegisterOutputs() " << config->GetOutputBranchNameOptA() << " at " << pOutVectOptA;
    }
 
    auto pOutVectOptB = config->GetOptOutBVec();
    if (pOutVectOptB) {
      ioman->RegisterAny(config->GetOutputBranchNameOptB().data(), pOutVectOptB, kTRUE);
      LOG(info) << "CbmRecoUnpack::RegisterOutputs() " << config->GetOutputBranchNameOptB() << " at " << pOutVectOptB;
    }
*/
    config->RegisterOutput(ioman);
  }

  virtual Bool_t initParContainers(std::vector<std::pair<std::string, std::shared_ptr<FairParGenericSet>>>* reqparvec)
  {
    LOG(debug) << GetName() << "::Init - initParContainers";
    if (!reqparvec) {
      LOG(info) << GetName() << "::Init - initParContainers - empty requirements vector no parameters initialized.";
      return kTRUE;
    }
 
    // Now get the actual ascii files and init the containers with the asciiIo
    for (auto& pair : *reqparvec) {
      auto filepath = pair.first;
      auto parset   = pair.second;
      FairParAsciiFileIo asciiInput;
      if (!filepath.empty()) {
        if (asciiInput.open(filepath.data())) {
          parset->init(&asciiInput);
        }
      }
    }
    return kTRUE;
  }

  template<typename TVecobj>
  typename std::enable_if<std::is_same<TVecobj, std::nullptr_t>::value == true, void>::type
  timesort(std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmRecoUnpack::timesort() got an object that has no member function GetTime(). Hence, we can and "
                  "will not timesort it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<!std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  timesort(std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmRecoUnpack::timesort() " << TVecobj::Class_Name()
               << "is  an object that has no member function GetTime(). Hence, we can and "
                  "will not timesort it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  timesort(std::vector<TVecobj>* vec = nullptr)
  {
    if (vec == nullptr) return;
    std::sort(vec->begin(), vec->end(),
              [](const TVecobj& a, const TVecobj& b) -> bool { return a.GetTime() < b.GetTime(); });
  }
 

  template<class TConfig, class TOptOutA = std::nullptr_t, class TOptOutB = std::nullptr_t>
  size_t unpack(const Subsystem subsystem, const fles::Timeslice* ts, uint16_t icomp, TConfig config,
                std::vector<TOptOutA>* optouttargetvecA = nullptr, std::vector<TOptOutB>* optouttargetvecB = nullptr)
  {
 
    auto wallstarttime        = std::chrono::high_resolution_clock::now();
    std::clock_t cpustarttime = std::clock();
 
    auto algo                        = config->GetUnpacker();
    std::vector<TOptOutA> optoutAvec = {};
    std::vector<TOptOutB> optoutBvec = {};
    if (optouttargetvecA) {
      algo->SetOptOutAVec(&optoutAvec);
    }
    if (optouttargetvecB) {
      algo->SetOptOutBVec(&optoutBvec);
    }
 
    // Set the start time of the current TS for this algorithm
    algo->SetTsStartTime(ts->start_time());
 
    // Run the actual unpacking
    auto digivec = algo->Unpack(ts, icomp);
 
    // Check if we want to write the output to somewhere (in pure online monitoring mode for example this can/would/should be skipped)
    if (!bMonitoringOnly) {
      if (config->GetOutputVec()) {
        // Lets do some time-sorting if we are not doing it later
        if (!bOutputFullTimeSorting) timesort(&digivec);
 
        // Transfer the data from the timeslice vector to the target branch vector
        // Digis/default output retrieved as offered by the algorithm
        for (auto digi : digivec)
          config->GetOutputVec()->emplace_back(digi);
      }
      if (optouttargetvecA) {
        // Lets do some timesorting
        if (!bOutputFullTimeSorting) timesort(&optoutAvec);
        // Transfer the data from the timeslice vector to the target branch vector
        for (auto optoutA : optoutAvec)
          optouttargetvecA->emplace_back(optoutA);
      }
      if (optouttargetvecB) {
        // Second opt output is not time sorted to allow non GetTime data container.
        // Lets do some timesorting
        timesort(&optoutAvec);
        // Transfer the data from the timeslice vector to the target branch vector
        for (auto optoutB : optoutBvec)
          optouttargetvecB->emplace_back(optoutB);
      }
    }
 
    std::clock_t cpuendtime = std::clock();
    auto wallendtime        = std::chrono::high_resolution_clock::now();
 
    // Cpu time in [mus]
    auto cputime = 1e6 * (cpuendtime - cpustarttime) / CLOCKS_PER_SEC;
    algo->AddCpuTime(cputime);
    // Real time in [mus]
    auto walltime = std::chrono::duration<double, std::micro>(wallendtime - wallstarttime).count();
    algo->AddWallTime(walltime);
 
 
    // Check some numbers from this timeslice
    size_t nDigis = digivec.size();
    LOG(debug) << "Component " << icomp << " connected to config " << config->GetName() << "   n-Digis " << nDigis
               << " processed in walltime(cputime) = " << walltime << "(" << cputime << cputime << ") micro s"
               << "this timeslice.";
 
    if (fDoPerfProf) {
      auto timeit = fTimeMap.find(subsystem);
      timeit->second.first += cputime;
      timeit->second.second += walltime;
 
      auto datait = fDataSizeMap.find(subsystem);
      datait->second.first += ts->size_component(icomp) / 1.0e6;
      datait->second.second += nDigis * algo->GetOutputObjSize() / 1.0e6;
 
      fNameMap.find(subsystem)->second.second += nDigis;
    }
 
    if (fDoPerfProfPerTs) {
      auto timeit = fTimeMapPerTs.find(subsystem);
      timeit->second.first += cputime;
      timeit->second.second += walltime;
 
      auto datait = fDataSizeMapPerTs.find(subsystem);
      datait->second.first += ts->size_component(icomp) / 1.0e6;
      datait->second.second += nDigis * algo->GetOutputObjSize() / 1.0e6;
 
      fNameMapPerTs.find(subsystem)->second.second += nDigis;
    }
 
    return nDigis;
  }
  // ----------------------------------------------------------------------------

  CbmTimeSlice* fTimeSlice;  

  std::shared_ptr<CbmMuchUnpackConfig> fMuchConfig = nullptr;  

  std::shared_ptr<CbmPsdUnpackConfig> fPsdConfig = nullptr;  

  std::shared_ptr<CbmRichUnpackConfig> fRichConfig = nullptr;  

  std::shared_ptr<CbmStsUnpackConfig> fStsConfig = nullptr;  

  std::shared_ptr<CbmTofUnpackConfig> fTofConfig = nullptr;  

  std::shared_ptr<CbmTrdUnpackConfig> fTrd1DConfig = nullptr;  

  std::shared_ptr<CbmTrdUnpackFaspConfig> fTrd2DConfig = nullptr;  

  std::shared_ptr<CbmBmonUnpackConfig> fBmonConfig = nullptr;  

  CbmTsEventHeader* fCbmTsEventHeader = nullptr;

  std::string fOutfilename = "CbmRecoUnpack.perf.root";

  bool bOutputFullTimeSorting = false;

  bool bMonitoringOnly = false;

  bool bSpillFilter = false;

  std::string sAccEvtTimingFile = "";

  double_t dSpillFilterWinStart = 0.0;

  double_t dSpillFilterWinStop = 0.0;

  int32_t iSpillFilterStartDelay = 0;

  uint64_t ulSpillFilterTsDurationNs = 128000000;

  bool bSpillFilterMoniFindPeak = false;

  std::vector<AccTimingEvent> vAccEvtsSpillStart = {};

  std::vector<AccTimingEvent> vAccEvtsSpillEnd = {};

  std::vector<AccTimingEvent>::iterator itSpillStart = vAccEvtsSpillStart.end();

  std::vector<AccTimingEvent>::iterator itSpillStop = vAccEvtsSpillEnd.end();

  uint32_t uSpillIndex = 0;

  TH1* fhDigiSpillStartDist                              = nullptr;  
  TH1* fhDigiSpillStopDist                               = nullptr;  
  TH2* fhDigiSpillFilterStart                            = nullptr;  
  TH2* fhDigiSpillFilterStop                             = nullptr;  
  std::map<Subsystem, TH1*> fhDigiSpillStartDistPerDet   = {};       
  std::map<Subsystem, TH1*> fhDigiSpillStopDistPerDet    = {};       
  std::map<Subsystem, TH2*> fhDigiSpillFilterStartPerDet = {};       
  std::map<Subsystem, TH2*> fhDigiSpillFilterStopPerDet  = {};       

  void InitSpillFilter();

  void initSpillFilterMonitoring(Subsystem subsystem, std::string name);

  template<typename TVecobj>
  typename std::enable_if<std::is_same<TVecobj, std::nullptr_t>::value == true, void>::type
  SpillFilter(uint64_t ulTsStartNs, const Subsystem subsystem, std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmRecoUnpack::SpillFilter() got an object that has no member function GetTime(). Hence, we can and "
                  "will not filter it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<!std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  SpillFilter(uint64_t ulTsStartNs, const Subsystem subsystem, std::vector<TVecobj>* /*vec = nullptr*/)
  {
    LOG(debug) << "CbmRecoUnpack::SpillFilter() " << TVecobj::Class_Name()
               << "is  an object that has no member function GetTime(). Hence, we can and "
                  "will not filter it!";
  }
 
  template<typename TVecobj>
  typename std::enable_if<std::is_member_function_pointer<decltype(&TVecobj::GetTime)>::value, void>::type
  SpillFilter(uint64_t ulTsStartNs, const Subsystem subsystem, std::vector<TVecobj>* vec = nullptr)
  {
    size_t originalSize = vec->size();
 
    ulTsStartNs -= iSpillFilterStartDelay;  
    uint64_t ulTsStopNs     = ulTsStartNs + ulSpillFilterTsDurationNs;
    bool bStartInTs         = false;
    double_t dStartTimeInTs = 0.0;
    if (ulTsStartNs <= itSpillStart->GetTime() && itSpillStart->GetTime() <= ulTsStopNs) {
      bStartInTs     = true;
      dStartTimeInTs = itSpillStart->GetTime() - ulTsStartNs;
    }
    bool bStopInTs         = false;
    double_t dStopTimeInTs = 0.0;
    if (ulTsStartNs <= itSpillStop->GetTime() && itSpillStop->GetTime() <= ulTsStopNs) {
      bStopInTs     = true;
      dStopTimeInTs = itSpillStop->GetTime() - ulTsStartNs;
    }
    if (bStartInTs || bStopInTs) {
      auto itWinStart  = vec->begin();
      auto itWinStop   = vec->end();
      bool bFoundStart = false;
      bool bFoundStop  = false;
      for (auto itDigi = vec->begin(); itDigi != vec->end(); ++itDigi) {
        double_t dTimeToStart = itDigi->GetTime() - dStartTimeInTs;
        double_t dTimeToStop  = itDigi->GetTime() - dStopTimeInTs;
 
        if (!bFoundStart && dSpillFilterWinStart <= dTimeToStart) {
          itWinStart  = itDigi;
          bFoundStart = true;
        }
        if (!bFoundStop && dSpillFilterWinStop < dTimeToStart) {
          itWinStop  = itDigi;
          bFoundStop = true;
        }
 
        fhDigiSpillStartDist->Fill(dTimeToStart);
        fhDigiSpillStopDist->Fill(dTimeToStop);
        fhDigiSpillStartDistPerDet[subsystem]->Fill(dTimeToStart);
        fhDigiSpillStopDistPerDet[subsystem]->Fill(dTimeToStop);
        if (dSpillFilterWinStart <= dTimeToStart && dTimeToStart <= dSpillFilterWinStop) {
          fhDigiSpillFilterStart->Fill(dTimeToStart, uSpillIndex);
          fhDigiSpillFilterStartPerDet[subsystem]->Fill(dTimeToStart, uSpillIndex);
        }
        if (dSpillFilterWinStart <= dTimeToStop && dTimeToStop <= dSpillFilterWinStop) {
          fhDigiSpillFilterStop->Fill(dTimeToStop, uSpillIndex);
          fhDigiSpillFilterStopPerDet[subsystem]->Fill(dTimeToStop, uSpillIndex);
        }
 
        if (!bSpillFilterMoniFindPeak && dSpillFilterWinStop < dTimeToStart && dSpillFilterWinStop < dTimeToStop) {
          // Stop once both SpillStart and SpillStop passed, except if long range monitoring enabled
          break;
        }
      }
      if (bFoundStop) {
        vec->erase(itWinStop, vec->end());
      }

      if (bFoundStart) {
        vec->erase(vec->begin(), itWinStart);
      }
    }
    else {
      vec->clear();
    }
 
    LOG(debug2) << "Filtered " << fles::to_string(subsystem) << " digi vector from " << originalSize << " to "
                << vec->size();
  }
 
  ClassDef(CbmRecoUnpack, 4);
};
 
 
#endif