Added CbmDeviceEventSink (MQ device which writes events for CbmDeviceEventBuilder).

MQ/mcbm/CMakeLists.txt

@@ -8,6 +8,7 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/startEventBuilder.sh.in ${CMAKE_BINAR
 set(INCLUDE_DIRECTORIES
     ${CMAKE_CURRENT_SOURCE_DIR}
     ${CMAKE_SOURCE_DIR}/MQ/base
+    ${CBMROOT_SOURCE_DIR}/algo
     ${CBMROOT_SOURCE_DIR}/algo/evbuild
     ${CBMROOT_SOURCE_DIR}/algo/trigger
     ${CBMROOT_SOURCE_DIR}/fles/mcbm2018/unpacker
@@ -310,3 +311,24 @@ set(DEPENDENCIES
 )
 #GENERATE_LIBRARY()
 GENERATE_EXECUTABLE()
+
+set(EXE_NAME EventSink)
+set(SRCS CbmDeviceEventSink.cxx runEventSink.cxx)
+
+set(DEPENDENCIES
+  ${DEPENDENCIES}
+  ${FAIR_LIBS}
+  ${BOOST_LIBS}
+  fles_ipc
+  CbmFlibMcbm2018
+  CbmFlibFlesTools
+  CbmBase
+  CbmData
+  Core
+  RIO
+  Tree
+  Net
+  Hist
+  RHTTP
+)
+GENERATE_EXECUTABLE()

MQ/mcbm/CbmDeviceEventSink.h

+/* Copyright (C) 2020-2022 Facility for Antiproton and Ion Research in Europe, Darmstadt
+   SPDX-License-Identifier: GPL-3.0-only
+   Authors: Pierre-Alain Loizeau [committer], Dominik Smith */
+
+#ifndef CBMDEVICEEVTSINK_H_
+#define CBMDEVICEEVTSINK_H_
+
+/// CBM headers
+#include "CbmDigiEvent.h"
+#include "CbmEvent.h"
+#include "CbmMqTMessage.h"
+#include "CbmMuchDigi.h"
+#include "CbmPsdDigi.h"
+#include "CbmRichDigi.h"
+#include "CbmStsDigi.h"
+#include "CbmTofDigi.h"
+#include "CbmTrdDigi.h"
+#include "CbmTsEventHeader.h"
+
+#include "TimesliceMetaData.h"
+
+/// FAIRROOT headers
+#include "FairMQDevice.h"
+
+/// FAIRSOFT headers (geant, boost, ...)
+#include "Rtypes.h"
+#include "TClonesArray.h"
+#include "TObjArray.h"
+
+/// C/C++ headers
+#include <chrono>
+#include <map>
+#include <vector>
+
+class TFile;
+class TList;
+class TClonesArray;
+//class TimesliceMetaData;
+class FairRunOnline;
+class FairRootManager;
+
+class CbmEventTimeslice {
+  /// TODO: rename to CbmTsWithEvents
+public:
+  CbmEventTimeslice(FairMQParts& parts);
+  ~CbmEventTimeslice();
+
+  std::vector<CbmDigiEvent> GetSelectedData();
+
+  /// TS information in header
+  CbmTsEventHeader fCbmTsEventHeader;
+  TimesliceMetaData fTsMetaData;
+  std::vector<CbmDigiEvent> fvEvents;
+};
+
+class CbmDeviceEventSink : public FairMQDevice {
+public:
+  CbmDeviceEventSink();
+  virtual ~CbmDeviceEventSink();
+
+protected:
+  virtual void InitTask();
+  bool HandleMissTsData(FairMQMessagePtr&, int);
+  bool HandleData(FairMQParts&, int);
+  bool HandleCommand(FairMQMessagePtr&, int);
+
+private:
+  /// Constants
+
+  /// Control flags
+  Bool_t fbFillHistos = false;  //! Switch ON/OFF filling of histograms
+  Bool_t fbFinishDone = false;  //! Keep track of whether the Finish was already called
+
+  /// User settings parameters
+  /// Algo enum settings
+  std::string fsOutputFileName = "mcbm_digis_events.root";
+  /// message queues
+  std::string fsChannelNameMissedTs    = "missedts";
+  std::string fsChannelNameDataInput   = "events";
+  std::string fsChannelNameCommands    = "commands";
+  std::string fsChannelNameHistosInput = "histogram-in";
+  /// Histograms management
+  uint32_t fuPublishFreqTs  = 100;
+  double_t fdMinPublishTime = 0.5;
+  double_t fdMaxPublishTime = 5.0;
+
+  /// List of MQ channels names
+  std::vector<std::string> fsAllowedChannels = {fsChannelNameDataInput};
+
+  /// Parameters management
+  //      TList* fParCList = nullptr;
+  //      Bool_t InitParameters( TList* fParCList );
+
+  /// Statistics & missed TS detection
+  uint64_t fuPrevTsIndex                                 = 0;
+  uint64_t fulNumMessages                                = 0;
+  uint64_t fulTsCounter                                  = 0;
+  uint64_t fulMissedTsCounter                            = 0;
+  std::chrono::system_clock::time_point fLastPublishTime = std::chrono::system_clock::now();
+
+  /// Control Commands reception
+  bool fbReceivedEof      = false;
+  uint64_t fuLastTsIndex  = 0;
+  uint64_t fuTotalTsCount = 0;
+
+  /// Data reception
+  /// Event (TS) header
+  CbmTsEventHeader* fEvtHeader = nullptr;
+  /// TS MetaData storage
+  TClonesArray* fTimeSliceMetaDataArray = nullptr;  //!
+  TimesliceMetaData* fTsMetaData        = nullptr;
+  /// CbmEvents
+  std::vector<CbmDigiEvent>* fEventsSel = nullptr;  //! output container of CbmEvents
+
+  /// Storage for re-ordering
+  /// Missed TS vector
+  std::vector<uint64_t> fvulMissedTsIndices = {};
+  /// Buffered TS
+  std::map<uint64_t, CbmEventTimeslice> fmFullTsStorage = {};
+
+  /// Data storage
+  FairRunOnline* fpRun           = nullptr;
+  FairRootManager* fpFairRootMgr = nullptr;
+
+  /// Array of histograms to send to the histogram server
+  TObjArray fArrayHisto = {};
+  /// Vector of string pairs with ( HistoName, FolderPath ) to send to the histogram server
+  std::vector<std::pair<std::string, std::string>> fvpsHistosFolder = {};
+  /// Vector of string pairs with ( CanvasName, CanvasConfig ) to send to the histogram server
+  /// Format of Can config is "NbPadX(U);NbPadY(U);ConfigPad1(s);....;ConfigPadXY(s)"
+  /// Format of Pad config is "GrixX(b),GridY(b),LogX(b),LogY(b),LogZ(b),HistoName(s),DrawOptions(s)"
+  std::vector<std::pair<std::string, std::string>> fvpsCanvasConfig = {};
+  /// Flag indicating whether the histograms and canvases configurations were already published
+  bool fbConfigSent = false;
+
+  /// Internal methods
+  bool IsChannelNameAllowed(std::string channelName);
+  bool InitHistograms();
+  void CheckTsQueues();
+  void PrepareTreeEntry(CbmEventTimeslice unpTs);
+  void DumpTreeEntry();
+  bool SendHistoConfAndData();
+  bool SendHistograms();
+  void Finish();
+};
+
+#endif /* CBMDEVICEEVTSINK_H_ */

MQ/mcbm/runEventSink.cxx

+/* Copyright (C) 2021 Facility for Antiproton and Ion Research in Europe, Darmstadt
+   SPDX-License-Identifier: GPL-3.0-only
+   Authors: Pierre-Alain Loizeau [committer] */
+
+#include "CbmDeviceEventSink.h"
+
+#include <iomanip>
+#include <string>
+
+#include "runFairMQDevice.h"
+
+namespace bpo = boost::program_options;
+using namespace std;
+
+void addCustomOptions(bpo::options_description& options)
+{
+  options.add_options()("StoreFullTs", bpo::value<bool>()->default_value(false),
+                        "Store digis vectors with full TS in addition to selected events if true");
+  options.add_options()("OutFileName", bpo::value<std::string>()->default_value("mcbm_digis_events.root"),
+                        "Name (full or relative path) of the output .root file ");
+  options.add_options()("EvtNameIn", bpo::value<std::string>()->default_value("events"),
+                        "MQ channel name for built events");
+  options.add_options()("FillHistos", bpo::value<bool>()->default_value(false),
+                        "Fill histograms and send them to histo server if true");
+
+  options.add_options()("PubFreqTs", bpo::value<uint32_t>()->default_value(100), "Histo publishing frequency in TS");
+  options.add_options()("PubTimeMin", bpo::value<double_t>()->default_value(1.0),
+                        "Minimal time between two publishing");
+  options.add_options()("PubTimeMax", bpo::value<double_t>()->default_value(10.0),
+                        "Maximal time between two publishing");
+  options.add_options()("ChNameIn", bpo::value<std::string>()->default_value("histogram-in"),
+                        "MQ channel name for histos");
+}
+
+FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/) { return new CbmDeviceEventSink(); }