CbmDevBuildEvents.cxx

Go to the documentation of this file.

/* Copyright (C) 2022 Facility for Antiproton and Ion Research in Europe, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Pierre-Alain Loizeau[committer], Dominik Smith */
 
#include "CbmDevBuildEvents.h"
 
#include "CbmMQDefs.h"
 
#include "BoostSerializer.h"
#include "EventBuilderConfig.h"
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"
#include "FairRootFileSink.h"
#include "FairRootManager.h"
#include "FairRunOnline.h"
#include "RootSerializer.h"
 
#include "TimesliceMetaData.h"
 
#include <boost/archive/binary_iarchive.hpp>
#include <boost/serialization/utility.hpp>
 
#include <array>
#include <iomanip>
#include <stdexcept>
#include <string>
struct InitTaskError : std::runtime_error {
  using std::runtime_error::runtime_error;
};
 
using namespace std;
 
CbmDevBuildEvents::CbmDevBuildEvents() {}
 
void CbmDevBuildEvents::InitTask()
try {
  LOG(info) << "Init options for CbmDevBuildEvents.";
 
  fsOutputFileName = fConfig->GetValue<std::string>("OutFileName");  //For storage of events
 
  // Event builder algorithm params
  const std::vector<std::string> vsSetEvbuildWin = fConfig->GetValue<std::vector<std::string>>("SetEvbuildWin");
 
  fsChannelNameDataInput  = fConfig->GetValue<std::string>("TrigNameIn");
  fsChannelNameDataOutput = fConfig->GetValue<std::string>("EvtNameOut");
  fsAllowedChannels[0]    = fsChannelNameDataInput;
 
  if ("" != fsOutputFileName) {
    fpRun         = new FairRunOnline();
    fpFairRootMgr = FairRootManager::Instance();
    fpFairRootMgr->SetSink(new FairRootFileSink(fsOutputFileName));
    if (nullptr == fpFairRootMgr->GetOutFile()) {
      throw InitTaskError("Could not open root file");
    }
    LOG(info) << "Init Root Output to " << fsOutputFileName;
    fpFairRootMgr->InitSink();
 
    fEventsSelOut = new std::vector<CbmDigiEvent>();
    fpFairRootMgr->RegisterAny("DigiEvent", fEventsSelOut, kTRUE);
 
    fTimeSliceMetaDataArrayOut = new TClonesArray("TimesliceMetaData", 1);
    fpFairRootMgr->Register("TimesliceMetaData", "TS Meta Data", fTimeSliceMetaDataArrayOut, kTRUE);
 
    fpFairRootMgr->WriteFolder();
  }  // if( "" != fsOutputFileName )
 
  // Get the information about created channels from the device
  // Check if the defined channels from the topology (by name)
  // are in the list of channels which are possible/allowed
  // for the device
  // The idea is to check at initilization if the devices are
  // properly connected. For the time beeing this is done with a
  // nameing convention. It is not avoided that someone sends other
  // data on this channel.
  //logger::SetLogLevel("INFO");
  int noChannel = fChannels.size();
  LOG(info) << "Number of defined channels: " << noChannel;
  for (auto const& entry : fChannels) {
    LOG(info) << "Channel name: " << entry.first;
    if (std::string::npos != entry.first.find(fsChannelNameDataInput)) {
      if (!IsChannelNameAllowed(entry.first)) throw InitTaskError("Channel name does not match.");
      OnData(entry.first, &CbmDevBuildEvents::HandleData);
    }
  }
 
  for (std::vector<std::string>::const_iterator itStrEvbuildWin = vsSetEvbuildWin.begin();
       itStrEvbuildWin != vsSetEvbuildWin.end(); ++itStrEvbuildWin) {
    size_t charPosDel = (*itStrEvbuildWin).find(',');
    if (std::string::npos == charPosDel) {
      LOG(info) << "CbmDevBuildEvents::InitTask => "
                << "Trying to set event builder window with invalid option pattern, ignored! "
                << " (Should be ECbmModuleId,dWinBeg,dWinEnd but instead found " << (*itStrEvbuildWin) << " )";
      continue;
    }
 
    std::string sSelDet       = (*itStrEvbuildWin).substr(0, charPosDel);
    const ECbmModuleId selDet = GetDetectorId(sSelDet);
 
    if (ECbmModuleId::kNotExist == selDet) {
      LOG(info) << "CbmDevBuildEvents::InitTask => "
                << "Trying to set trigger window for unsupported detector, ignored! " << sSelDet;
      continue;
    }
 
    charPosDel++;
    std::string sNext = (*itStrEvbuildWin).substr(charPosDel);
    charPosDel        = sNext.find(',');
    if (std::string::npos == charPosDel) {
      LOG(info) << "CbmDevBuildEvents::InitTask => "
                << "Trying to set event builder window with invalid option pattern, ignored! "
                << " (Should be ECbmModuleId,dWinBeg,dWinEnd but instead found " << (*itStrEvbuildWin) << " )";
      continue;
    }
    double dWinBeg = std::stod(sNext.substr(0, charPosDel));
 
    charPosDel++;
    double dWinEnd = std::stod(sNext.substr(charPosDel));
    cbm::algo::evbuild::EventBuilderConfig config;
    fEvbuildAlgo = std::make_unique<cbm::algo::evbuild::EventBuilder>(config);
  }
}
catch (InitTaskError& e) {
  LOG(error) << e.what();
  // Wrapper defined in CbmMQDefs.h to support different FairMQ versions
  cbm::mq::ChangeState(this, cbm::mq::Transition::ErrorFound);
}
 
ECbmModuleId CbmDevBuildEvents::GetDetectorId(std::string detName)
{
  /* clang-format off */
  ECbmModuleId detId = ("kBmon"   == detName ? ECbmModuleId::kBmon
                       : ("kSts"   == detName ? ECbmModuleId::kSts
                       : ("kMuch"  == detName ? ECbmModuleId::kMuch
                       : ("kTrd"   == detName ? ECbmModuleId::kTrd
                       : ("kTrd2d" == detName ? ECbmModuleId::kTrd2d
                       : ("kTof"   == detName ? ECbmModuleId::kTof
                       : ("kRich"  == detName ? ECbmModuleId::kRich
                       : ("kPsd"   == detName ? ECbmModuleId::kPsd
                       : ("kFsd"   == detName ? ECbmModuleId::kFsd
                                             : ECbmModuleId::kNotExist)))))))));
  return detId; 
  /* clang-format on */
}
 
bool CbmDevBuildEvents::IsChannelNameAllowed(std::string channelName)
{
  for (auto const& entry : fsAllowedChannels) {
    std::size_t pos1 = channelName.find(entry);
    if (pos1 != std::string::npos) {
      const vector<std::string>::const_iterator pos =
        std::find(fsAllowedChannels.begin(), fsAllowedChannels.end(), entry);
      const vector<std::string>::size_type idx = pos - fsAllowedChannels.begin();
      LOG(info) << "Found " << entry << " in " << channelName;
      LOG(info) << "Channel name " << channelName << " found in list of allowed channel names at position " << idx;
      return true;
    }
  }
  LOG(info) << "Channel name " << channelName << " not found in list of allowed channel names.";
  LOG(error) << "Stop device.";
  return false;
}
 
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool CbmDevBuildEvents::HandleData(FairMQParts& parts, int /*index*/)
{
  fulNumMessages++;
  LOG(info) << "Received message number " << fulNumMessages << " with " << parts.Size() << " parts"
            << ", size0: " << parts.At(0)->GetSize();
 
  if (0 == fulNumMessages % 10000) LOG(info) << "Received " << fulNumMessages << " messages";
 
  uint32_t uPartIdx = 0;
 
  CbmDigiTimeslice ts;
  std::string msgStrTS(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issTS(msgStrTS);
  boost::archive::binary_iarchive inputArchiveTS(issTS);
  inputArchiveTS >> ts;
  ++uPartIdx;
 
  TimesliceMetaData* tsMetaData = new TimesliceMetaData();
  RootSerializer().Deserialize(*parts.At(uPartIdx), tsMetaData);
  ++uPartIdx;
 
  std::vector<double> triggers;
  std::string msgStrTrig(static_cast<char*>(parts.At(uPartIdx)->GetData()), (parts.At(uPartIdx))->GetSize());
  std::istringstream issTrig(msgStrTrig);
  boost::archive::binary_iarchive inputArchiveTrig(issTrig);
  inputArchiveTrig >> triggers;
  ++uPartIdx;
 
  //if (1 == fulNumMessages) {
  //fpAlgo->SetTsParameters(0, fTsMetaDataOut->GetDuration(), fTsMetaDataOut->GetOverlapDuration());
  //}
 
  LOG(debug) << "Bmon Vector size: " << ts.fData.fBmon.fDigis.size();
  LOG(debug) << "STS Vector size: " << ts.fData.fSts.fDigis.size();
  LOG(debug) << "MUCH Vector size: " << ts.fData.fMuch.fDigis.size();
  LOG(debug) << "TRD Vector size: " << ts.fData.fTrd.fDigis.size();
  LOG(debug) << "TOF Vector size: " << ts.fData.fTof.fDigis.size();
  LOG(debug) << "RICH Vector size: " << ts.fData.fRich.fDigis.size();
  LOG(debug) << "PSD Vector size: " << ts.fData.fPsd.fDigis.size();
  LOG(debug) << "FSD Vector size: " << ts.fData.fFsd.fDigis.size();
  LOG(debug) << "triggers: " << triggers.size();
 
  std::vector<CbmDigiEvent> vEvents = (*fEvbuildAlgo)(ts, triggers).first;
  LOG(debug) << "vEvents size: " << vEvents.size();
 
  if (!SendEvents(vEvents, tsMetaData)) {
    return false;
  }
 
  // FIXME: poor man solution with lots of data copy until we undertand how to properly deal
 
  if ("" != fsOutputFileName) {
    (*fEventsSelOut) = std::move(vEvents);
    LOG(debug) << "fEventSel size: " << fEventsSelOut->size();
 
    new ((*fTimeSliceMetaDataArrayOut)[fTimeSliceMetaDataArrayOut->GetEntriesFast()])
      TimesliceMetaData(std::move(*tsMetaData));
 
    DumpTreeEntry();
 
    fTimeSliceMetaDataArrayOut->Clear();
    fEventsSelOut->clear();
  }
 
  return true;
}
 
void CbmDevBuildEvents::DumpTreeEntry()
{
  // Unpacked digis + CbmEvent output to root file
 
  fpFairRootMgr->StoreWriteoutBufferData(fpFairRootMgr->GetEventTime());
  fpFairRootMgr->Fill();
  fpFairRootMgr->DeleteOldWriteoutBufferData();
}
 
bool CbmDevBuildEvents::SendEvents(const std::vector<CbmDigiEvent>& vEvents, const TimesliceMetaData* tsMetaData)
{
  LOG(debug) << "Vector size: " << vEvents.size();
 
  FairMQParts partsOut;
 
  // Prepare TS meta data
  FairMQMessagePtr messTsMeta(NewMessage());
  RootSerializer().Serialize(*messTsMeta, tsMetaData);
  partsOut.AddPart(std::move(messTsMeta));
 
  // Prepare event vector.
  std::stringstream ossEvt;
  boost::archive::binary_oarchive oaEvt(ossEvt);
  oaEvt << vEvents;
  std::string* strMsgEvt = new std::string(ossEvt.str());
 
  partsOut.AddPart(NewMessage(
    const_cast<char*>(strMsgEvt->c_str()),  // data
    strMsgEvt->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgEvt));  // object that manages the data
 
  if (Send(partsOut, fsChannelNameDataOutput) < 0) {
    LOG(error) << "Problem sending data to " << fsChannelNameDataOutput;
    return false;
  }
  return true;
}
 
void CbmDevBuildEvents::Finish()
{
  if ("" != fsOutputFileName) {
    // Clean closure of output to root file
    fpFairRootMgr->Write();
    fpFairRootMgr->CloseSink();
  }
  fbFinishDone = kTRUE;
}
 
CbmDevBuildEvents::~CbmDevBuildEvents()
{
  if (!fbFinishDone) Finish();
  if (fEventsSelOut) {
    delete fEventsSelOut;
  }
  if (fpRun) {
    delete fpRun;
  }
  if (fTimeSliceMetaDataArrayOut) {
    delete fTimeSliceMetaDataArrayOut;
  }
}