CbmDeviceMcbmUnpack.cxx

Go to the documentation of this file.

/* Copyright (C) 2020 Facility for Antiproton and Ion Research in Europe, Darmstadt
   SPDX-License-Identifier: GPL-3.0-only
   Authors: Pierre-Alain Loizeau [committer] */
 
#include "CbmDeviceMcbmUnpack.h"
 
#include "CbmFlesCanvasTools.h"
#include "CbmMQDefs.h"
#include "CbmMcbm2018UnpackerAlgoMuch.h"
#include "CbmMcbm2018UnpackerAlgoPsd.h"
#include "CbmMcbm2018UnpackerAlgoRich.h"
#include "CbmMcbm2018UnpackerAlgoSts.h"
#include "CbmMcbm2018UnpackerAlgoTof.h"
#include "CbmMcbm2018UnpackerAlgoTrdR.h"
 
#include "StorableTimeslice.hpp"
#include "TimesliceMetaData.h"
 
#include "FairMQLogger.h"
#include "FairMQProgOptions.h"  // device->fConfig
#include "FairParGenericSet.h"
 
#include "TCanvas.h"
#include "TFile.h"
#include "TH1.h"
#include "TList.h"
#include "TNamed.h"
 
#include "BoostSerializer.h"
#include <boost/archive/binary_iarchive.hpp>
#include <boost/serialization/utility.hpp>
 
#include <array>
#include <iomanip>
#include <stdexcept>
#include <string>
 
#include "RootSerializer.h"
struct InitTaskError : std::runtime_error {
  using std::runtime_error::runtime_error;
};
 
using namespace std;
 
Bool_t bMcbm2018MonitorTaskBmonResetHistos = kFALSE;
 
CbmDeviceMcbmUnpack::CbmDeviceMcbmUnpack()
{
  fUnpAlgoSts  = new CbmMcbm2018UnpackerAlgoSts();
  fUnpAlgoMuch = new CbmMcbm2018UnpackerAlgoMuch();
  fUnpAlgoTrd  = new CbmMcbm2018UnpackerAlgoTrdR();
  fUnpAlgoTof  = new CbmMcbm2018UnpackerAlgoTof();
  fUnpAlgoRich = new CbmMcbm2018UnpackerAlgoRich();
  fUnpAlgoPsd  = new CbmMcbm2018UnpackerAlgoPsd();
}
 
void CbmDeviceMcbmUnpack::InitTask()
try {
  LOG(info) << "Init options for CbmDeviceMcbmUnpack.";
  fbIgnoreOverlapMs       = fConfig->GetValue<bool>("IgnOverMs");
  fvsSetTimeOffs          = fConfig->GetValue<std::vector<std::string>>("SetTrigWin");
  fsChannelNameDataInput  = fConfig->GetValue<std::string>("TsNameIn");
  fsChannelNameDataOutput = fConfig->GetValue<std::string>("TsNameOut");
  fsAllowedChannels[0] = fsChannelNameDataInput;
 
  // 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, &CbmDeviceMcbmUnpack::HandleData);
    }  // if( entry.first.find( "ts" )
  }    // for( auto const &entry : fChannels )
  InitContainers();
}
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);
}
 
bool CbmDeviceMcbmUnpack::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;
    }  // if (pos1!=std::string::npos)
  }    // for(auto const &entry : fsAllowedChannels)
  LOG(info) << "Channel name " << channelName << " not found in list of allowed channel names.";
  LOG(error) << "Stop device.";
  return false;
}
 
Bool_t CbmDeviceMcbmUnpack::InitContainers()
{
  LOG(info) << "Init parameter containers for CbmDeviceMcbmUnpack.";
 
  if (kFALSE == InitParameters(fUnpAlgoSts->GetParList())) return kFALSE;
  if (kFALSE == InitParameters(fUnpAlgoMuch->GetParList())) return kFALSE;
  if (kFALSE == InitParameters(fUnpAlgoTrd->GetParList())) return kFALSE;
  if (kFALSE == InitParameters(fUnpAlgoTof->GetParList())) return kFALSE;
  if (kFALSE == InitParameters(fUnpAlgoRich->GetParList())) return kFALSE;
  if (kFALSE == InitParameters(fUnpAlgoPsd->GetParList())) return kFALSE;
 
  fUnpAlgoSts->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
  fUnpAlgoMuch->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
  fUnpAlgoTrd->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
  fUnpAlgoTof->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
  fUnpAlgoRich->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
  fUnpAlgoPsd->SetIgnoreOverlapMs(fbIgnoreOverlapMs);
 
  for (std::vector<std::string>::iterator itStrOffs = fvsSetTimeOffs.begin(); itStrOffs != fvsSetTimeOffs.end();
       ++itStrOffs) {
    size_t charPosDel = (*itStrOffs).find(',');
    if (std::string::npos == charPosDel) {
      LOG(info) << "CbmDeviceMcbmUnpack::InitContainers => "
                << "Trying to set trigger window with invalid option pattern, ignored! "
                << " (Should be ECbmModuleId,dWinBeg,dWinEnd but instead found " << (*itStrOffs) << " )";
    }  // if( std::string::npos == charPosDel )
 
    std::string sSelDet = (*itStrOffs).substr(0, charPosDel);
    charPosDel++;
    Double_t dOffset = std::stod((*itStrOffs).substr(charPosDel));
 
    if ("kSTS" == sSelDet) { fUnpAlgoSts->SetTimeOffsetNs(dOffset); }  // if( "kSTS"  == sSelDet )
    else if ("kMUCH" == sSelDet) {
      fUnpAlgoMuch->SetTimeOffsetNs(dOffset);
    }  // else if( "kMUCH" == sSelDet )
    else if ("kTRD" == sSelDet) {
      fUnpAlgoTrd->SetTimeOffsetNs(dOffset);
    }  // else if( "kTRD"  == sSelDet  )
    else if ("kTOF" == sSelDet) {
      fUnpAlgoTof->SetTimeOffsetNs(dOffset);
    }  // else if( "kTOF"  == sSelDet )
    else if ("kRICH" == sSelDet) {
      fUnpAlgoRich->SetTimeOffsetNs(dOffset);
    }  // else if( "kRICH" == sSelDet )
    else if ("kPSD" == sSelDet) {
      fUnpAlgoPsd->SetTimeOffsetNs(dOffset);
    }  // else if( "kPSD"  == sSelDet )
    else {
      LOG(info) << "CbmDeviceMcbmUnpack::InitContainers => Trying to set time "
                   "offset for unsupported detector, ignored! "
                << (sSelDet);
      continue;
    }  // else of detector enum detection
  }  // for( std::vector< std::string >::iterator itStrAdd = fvsAddDet.begin(); itStrAdd != fvsAddDet.end(); ++itStrAdd )
 
 
  fUnpAlgoSts->SetBinningFwFlag(kTRUE);
  fUnpAlgoMuch->SetBinningFwFlag(kTRUE);
 
  Bool_t initOK = fUnpAlgoSts->InitContainers();
  initOK &= fUnpAlgoMuch->InitContainers();
  initOK &= fUnpAlgoTrd->InitContainers();
  initOK &= fUnpAlgoTof->InitContainers();
  initOK &= fUnpAlgoRich->InitContainers();
  initOK &= fUnpAlgoPsd->InitContainers();
 
  initOK &= fUnpAlgoTrd->SetDigiOutputPointer(&(fUnpAlgoTrd->GetVector()));
 
  //   Bool_t initOK = fMonitorAlgo->ReInitContainers();
 
  return initOK;
}
 
Bool_t CbmDeviceMcbmUnpack::InitParameters(TList* fParCList)
{
  for (int iparC = 0; iparC < fParCList->GetEntries(); iparC++) {
    FairParGenericSet* tempObj = (FairParGenericSet*) (fParCList->At(iparC));
    fParCList->Remove(tempObj);
    std::string paramName {tempObj->GetName()};
    // NewSimpleMessage creates a copy of the data and takes care of its destruction (after the transfer takes place).
    // Should only be used for small data because of the cost of an additional copy
 
    // Her must come the proper Runid
    std::string message = paramName + ",111";
    LOG(info) << "Requesting parameter container " << paramName << ", sending message: " << message;
 
    FairMQMessagePtr req(NewSimpleMessage(message));
    FairMQMessagePtr rep(NewMessage());
 
    FairParGenericSet* newObj = nullptr;
 
    if (Send(req, "parameters") > 0) {
      if (Receive(rep, "parameters") >= 0) {
        if (0 != rep->GetSize()) {
          CbmMqTMessage tmsg(rep->GetData(), rep->GetSize());
          newObj = static_cast<FairParGenericSet*>(tmsg.ReadObject(tmsg.GetClass()));
          LOG(info) << "Received unpack parameter from the server:";
          newObj->print();
        }  // if( 0 !=  rep->GetSize() )
        else {
          LOG(error) << "Received empty reply. Parameter not available";
          return kFALSE;
        }  // else of if( 0 !=  rep->GetSize() )
      }    // if( Receive( rep, "parameters" ) >= 0)
    }      // if( Send(req, "parameters") > 0 )
    fParCList->AddAt(newObj, iparC);
    delete tempObj;
  }  // for( int iparC = 0; iparC < fParCList->GetEntries(); iparC++ )
 
  return kTRUE;
}
 
// handler is called whenever a message arrives on "data", with a reference to the message and a sub-channel index (here 0)
bool CbmDeviceMcbmUnpack::HandleData(FairMQMessagePtr& msg, int /*index*/)
{
  fulNumMessages++;
  LOG(debug) << "Received message number " << fulNumMessages << " with size " << msg->GetSize();
 
  if (0 == fulNumMessages % 10000) LOG(info) << "Received " << fulNumMessages << " messages";
 
  std::string msgStr(static_cast<char*>(msg->GetData()), msg->GetSize());
  std::istringstream iss(msgStr);
  boost::archive::binary_iarchive inputArchive(iss);
 
  fles::StorableTimeslice ts {0};
  inputArchive >> ts;
 
  if (-1.0 == fdTsCoreSizeInNs) {
    fuNbCoreMsPerTs  = ts.num_core_microslices();
    fuNbOverMsPerTs  = ts.num_microslices(0) - ts.num_core_microslices();
    fdTsCoreSizeInNs = fdMsSizeInNs * (fuNbCoreMsPerTs);
    fdTsOverSizeInNs = fdMsSizeInNs * (fuNbOverMsPerTs);
    fdTsFullSizeInNs = fdTsCoreSizeInNs + fdTsOverSizeInNs;
    LOG(info) << "Timeslice parameters: each TS has " << fuNbCoreMsPerTs << " Core MS and " << fuNbOverMsPerTs
              << " Overlap MS, for a core duration of " << fdTsCoreSizeInNs << " ns and a full duration of "
              << fdTsFullSizeInNs << " ns";
  }  // if( -1.0 == fdTsCoreSizeInNs )
 
  fTsMetaData = new TimesliceMetaData(ts.descriptor(0, 0).idx, fdTsCoreSizeInNs, fdTsOverSizeInNs, ts.index());
 
  DoUnpack(ts, 0);
 
  if (!SendUnpData()) return false;
 
  delete fTsMetaData;
 
  fUnpAlgoSts->ClearVector();
  fUnpAlgoMuch->ClearVector();
  fUnpAlgoTrd->ClearVector();
  fUnpAlgoTof->ClearVector();
  fUnpAlgoRich->ClearVector();
  fUnpAlgoPsd->ClearVector();
 
  fUnpAlgoSts->ClearErrorVector();
  fUnpAlgoMuch->ClearErrorVector();
  fUnpAlgoTrd->ClearErrorVector();
  fUnpAlgoTof->ClearErrorVector();
  fUnpAlgoRich->ClearErrorVector();
  fUnpAlgoPsd->ClearErrorVector();
 
  return true;
}
 
bool CbmDeviceMcbmUnpack::SendUnpData()
{
 
  /*
   std::stringstream ossTsMeta;
   boost::archive::binary_oarchive oaTsMeta(ossTsMeta);
   oaTsMeta << *(fTsMetaData);
   std::string* strMsgTsMetaE = new std::string(ossTsMeta.str());
*/
  FairMQMessagePtr messTsMeta(NewMessage());
  //  Serialize<RootSerializer>(*messTsMeta, fTsMetaData);
  RootSerializer().Serialize(*messTsMeta, fTsMetaData);
 
  std::stringstream ossSts;
  boost::archive::binary_oarchive oaSts(ossSts);
  oaSts << fUnpAlgoSts->GetVector();
  std::string* strMsgSts = new std::string(ossSts.str());
 
  std::stringstream ossMuch;
  boost::archive::binary_oarchive oaMuch(ossMuch);
  oaMuch << fUnpAlgoMuch->GetVector();
  std::string* strMsgMuch = new std::string(ossMuch.str());
 
  std::stringstream ossTrd;
  boost::archive::binary_oarchive oaTrd(ossTrd);
  oaTrd << fUnpAlgoTrd->GetVector();
  std::string* strMsgTrd = new std::string(ossTrd.str());
 
  std::vector<CbmTofDigi>& vDigiTofBmon = fUnpAlgoTof->GetVector();
  std::vector<CbmTofDigi> vDigiTof    = {};
  std::vector<CbmTofDigi> vDigiBmon     = {};
 
  for (auto digi : vDigiTofBmon) {
    if (fuDigiMaskedIdBmon == (digi.GetAddress() & fuDigiMaskId)) {
      vDigiBmon.emplace_back(digi);
    }  // if( fuDigiMaskedIdBmon == ( digi.GetAddress() & fuDigiMaskId ) )
    else {
      vDigiTof.emplace_back(digi);
    }  // else of if( fuDigiMaskedIdBmon == ( digi.GetAddress() & fuDigiMaskId ) )
  }    // for( auto digi: vDigi )
 
  std::stringstream ossTof;
  boost::archive::binary_oarchive oaTof(ossTof);
  oaTof << vDigiTof;
  std::string* strMsgTof = new std::string(ossTof.str());
 
  std::stringstream ossBmon;
  boost::archive::binary_oarchive oaBmon(ossBmon);
  oaBmon << vDigiBmon;
  std::string* strMsgBmon = new std::string(ossBmon.str());
 
  std::stringstream ossRich;
  boost::archive::binary_oarchive oaRich(ossRich);
  oaRich << fUnpAlgoRich->GetVector();
  std::string* strMsgRich = new std::string(ossRich.str());
 
  std::stringstream ossPsd;
  boost::archive::binary_oarchive oaPsd(ossPsd);
  oaPsd << fUnpAlgoPsd->GetVector();
  std::string* strMsgPsd = new std::string(ossPsd.str());
 
  FairMQParts parts;
 
  parts.AddPart(std::move(messTsMeta));
  /*
   parts.AddPart( NewMessage( const_cast< char * >( strMsgTsMetaE->c_str() ), // data
                              strMsgTsMetaE->length(), // size
                              []( void* , void* object ){ delete static_cast< std::string * >( object ); },
                              strMsgTsMetaE
                            )
                ); // object that manages the data
*/
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgBmon->c_str()),  // data
    strMsgBmon->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgBmon));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgSts->c_str()),  // data
    strMsgSts->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgSts));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgMuch->c_str()),  // data
    strMsgMuch->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgMuch));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgTrd->c_str()),  // data
    strMsgTrd->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgTrd));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgTof->c_str()),  // data
    strMsgTof->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgTof));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgRich->c_str()),  // data
    strMsgRich->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgRich));  // object that manages the data
 
  parts.AddPart(NewMessage(
    const_cast<char*>(strMsgPsd->c_str()),  // data
    strMsgPsd->length(),                    // size
    [](void*, void* object) { delete static_cast<std::string*>(object); },
    strMsgPsd));  // object that manages the data
 
  if (Send(parts, fsChannelNameDataOutput) < 0) {
    LOG(error) << "Problem sending data to " << fsChannelNameDataOutput;
    return false;
  }
 
  return true;
}
 
 
CbmDeviceMcbmUnpack::~CbmDeviceMcbmUnpack()
{
  if (nullptr != fUnpAlgoSts) delete fUnpAlgoSts;
  if (nullptr != fUnpAlgoMuch) delete fUnpAlgoMuch;
  if (nullptr != fUnpAlgoTrd) delete fUnpAlgoTrd;
  if (nullptr != fUnpAlgoTof) delete fUnpAlgoTof;
  if (nullptr != fUnpAlgoRich) delete fUnpAlgoRich;
  if (nullptr != fUnpAlgoPsd) delete fUnpAlgoPsd;
}
 
 
Bool_t CbmDeviceMcbmUnpack::DoUnpack(const fles::Timeslice& ts, size_t /*component*/)
{
  fulTsCounter++;
 
  if (kFALSE == fbComponentsAddedToList) {
    for (uint32_t uCompIdx = 0; uCompIdx < ts.num_components(); ++uCompIdx) {
      switch (ts.descriptor(uCompIdx, 0).sys_id) {
        case kusSysIdSts: {
          fUnpAlgoSts->AddMsComponentToList(uCompIdx, kusSysIdSts);
          break;
        }  // case kusSysIdSts
        case kusSysIdMuch: {
          fUnpAlgoMuch->AddMsComponentToList(uCompIdx, kusSysIdMuch);
          break;
        }  // case kusSysIdMuch
        case kusSysIdTrd: {
          fUnpAlgoTrd->AddMsComponentToList(uCompIdx, kusSysIdTrd);
          break;
        }  // case kusSysIdTrd
        case kusSysIdTof: {
          fUnpAlgoTof->AddMsComponentToList(uCompIdx, kusSysIdTof);
          break;
        }  // case kusSysIdTof
        case kusSysIdBmon: {
          fUnpAlgoTof->AddMsComponentToList(uCompIdx, kusSysIdBmon);
          break;
        }  // case kusSysIdBmon
        case kusSysIdRich: {
          fUnpAlgoRich->AddMsComponentToList(uCompIdx, kusSysIdRich);
          break;
        }  // case kusSysIdRich
        case kusSysIdPsd: {
          fUnpAlgoPsd->AddMsComponentToList(uCompIdx, kusSysIdPsd);
          break;
        }  // case kusSysIdPsd
        default: break;
      }  // switch( ts.descriptor( uCompIdx, 0 ).sys_id )
    }    // for( uint32_t uComp = 0; uComp < ts.num_components(); ++uComp )
    fbComponentsAddedToList = kTRUE;
  }  // if( kFALSE == fbComponentsAddedToList )
 
  if (kFALSE == fUnpAlgoSts->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in STS unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoSts->ProcessTs( ts ) )
 
  if (kFALSE == fUnpAlgoMuch->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in MUCH unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoMuch->ProcessTs( ts ) )
 
  if (kFALSE == fUnpAlgoTrd->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in TRD unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoTrd->ProcessTs( ts ) )
 
  if (kFALSE == fUnpAlgoTof->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in TOF unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoTof->ProcessTs( ts ) )
 
  if (kFALSE == fUnpAlgoRich->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in RICH unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoRich->ProcessTs( ts ) )
 
  if (kFALSE == fUnpAlgoPsd->ProcessTs(ts)) {
    LOG(error) << "Failed processing TS " << ts.index() << " in PSD unpacker algorithm class";
    return kFALSE;
  }  // if( kFALSE == fUnpAlgoPsd->ProcessTs( ts ) )
 
 
  if (0 == fulTsCounter % 10000) LOG(info) << "Processed " << fulTsCounter << " time slices";
 
  return kTRUE;
}
 
void CbmDeviceMcbmUnpack::Finish() {}