Enable ROOT file output in TPC chunked-digit merger (#14608)

We can now regenerate a TPC digit file using the TPC chunked-digit merger.
This is useful when we have the TPC "drift-time" digits (as in O2DPG) and want
to obtain the full timeframe of TPC digits in ROOT file tpcdigits.root.

There is a new special mode for this and the command line to use is
```
${O2_ROOT}/bin/o2-tpc-chunkeddigit-merger --writer-mode --tpc-lanes 8
```
for the case in which we had 8 separate lanes of drift-time digits.

This commit includes:
* small modularity and library refactoring
* do not fatal in the TPCDigitRootWriter if no triggers are sent
* merging/treatment of the TPC common modes in the merger (so far not done)

Co-author: @ChSonnabend.

Author: sawenzel

Detectors/TPC/CMakeLists.txt

@@ -13,6 +13,7 @@ add_subdirectory(base)
 add_subdirectory(reconstruction)
 add_subdirectory(calibration)
 add_subdirectory(simulation)
+add_subdirectory(simworkflow)
 add_subdirectory(monitor)
 add_subdirectory(workflow)
 add_subdirectory(qc)

Detectors/TPC/simworkflow/CMakeLists.txt

@@ -0,0 +1,28 @@
+# Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+# See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+# All rights not expressly granted are reserved.
+#
+# This software is distributed under the terms of the GNU General Public
+# License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+#
+# In applying this license CERN does not waive the privileges and immunities
+# granted to it by virtue of its status as an Intergovernmental Organization
+# or submit itself to any jurisdiction.
+
+o2_add_library(TPCSimWorkflow
+               SOURCES
+                       src/ChunkedDigitPublisher.cxx
+                       src/TPCDigitRootWriterSpec.cxx
+               PUBLIC_LINK_LIBRARIES O2::TPCSimulation O2::Framework)
+
+o2_add_executable(chunkeddigit-merger
+        COMPONENT_NAME tpc
+        TARGETVARNAME mergertargetName
+        SOURCES src/ChunkedDigitPublisher.cxx
+        PUBLIC_LINK_LIBRARIES O2::TPCSimWorkflow)
+
+if(OpenMP_CXX_FOUND)
+   # Must be private, depending libraries might be compiled by compiler not understanding -fopenmp
+   target_compile_definitions(${mergertargetName} PRIVATE WITH_OPENMP)
+   target_link_libraries(${mergertargetName} PRIVATE OpenMP::OpenMP_CXX)
+endif()

…zerWorkflow/src/TPCDigitRootWriterSpec.h …/TPCSimWorkflow/TPCDigitRootWriterSpec.hSteer/DigitizerWorkflow/src/TPCDigitRootWriterSpec.h renamed to Detectors/TPC/simworkflow/include/TPCSimWorkflow/TPCDigitRootWriterSpec.h Steer/DigitizerWorkflow/src/TPCDigitRootWriterSpec.h renamed to Detectors/TPC/simworkflow/include/TPCSimWorkflow/TPCDigitRootWriterSpec.h

@@ -19,6 +19,7 @@
 #include "Framework/DataAllocator.h"
 #include "Framework/ControlService.h"
 #include "DataFormatsTPC/Digit.h"
+#include "TPCSimWorkflow/TPCDigitRootWriterSpec.h"
 #include "CommonUtils/ConfigurableParam.h"
 #include "DetectorsRaw/HBFUtilsInitializer.h"
 #include "TPCSimulation/CommonMode.h"
@@ -46,6 +47,9 @@
 #include <omp.h>
 #endif
 #include <TStopwatch.h>
+#include "CommonDataFormat/RangeReference.h"
+
+using DigiGroupRef = o2::dataformats::RangeReference<int, int>;
 
 using SubSpecificationType = o2::framework::DataAllocator::SubSpecificationType;
 
@@ -71,6 +75,9 @@ void customize(std::vector<o2::framework::ConfigParamSpec>& workflowOptions)
   workflowOptions.push_back(
     ConfigParamSpec{"tpc-sectors", VariantType::String, sectorDefault.c_str(), {sectorshelp}});
 
+  // option to write merged data to file
+  workflowOptions.push_back(ConfigParamSpec{"writer-mode", o2::framework::VariantType::Bool, false, {"enable ROOT file output"}});
+
   // option to disable MC truth
   workflowOptions.push_back(ConfigParamSpec{"disable-mc", o2::framework::VariantType::Bool, false, {"disable  mc-truth"}});
   workflowOptions.push_back(ConfigParamSpec{"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings ..."}});
@@ -104,14 +111,30 @@ void copyHelper<MCTruthContainer>(MCTruthContainer const& origin, MCTruthContain
   target.mergeAtBack(origin);
 }
 
+// a trait to map TPC data types to a DPL channel name
+template <typename T>
+struct OutputChannelName;
+template <>
+struct OutputChannelName<std::vector<o2::tpc::Digit>> {
+  static constexpr char value[] = "DIGITS";
+};
+template <>
+struct OutputChannelName<std::vector<o2::tpc::CommonMode>> {
+  static constexpr char value[] = "COMMONMODE";
+};
+template <>
+struct OutputChannelName<std::vector<DigiGroupRef>> {
+  static constexpr char value[] = "DIGTRIGGERS";
+};
+
 template <typename T>
 auto makePublishBuffer(framework::ProcessingContext& pc, int sector, uint64_t activeSectors)
 {
-  LOG(info) << "PUBLISHING SECTOR " << sector;
+  LOG(info) << "PUBLISHING SECTOR " << sector << " FOR CHANNEL " << OutputChannelName<T>::value;
 
   o2::tpc::TPCSectorHeader header{sector};
   header.activeSectors = activeSectors;
-  return &pc.outputs().make<T>(Output{"TPC", "DIGITS", static_cast<SubSpecificationType>(sector), header});
+  return &pc.outputs().make<T>(Output{"TPC", OutputChannelName<T>::value, static_cast<SubSpecificationType>(sector), header});
 }
 
 template <>
@@ -187,6 +210,30 @@ void mergeHelper(const char* brprefix, std::vector<int> const& tpcsectors, uint6
   }
 }
 
+template <>
+void mergeHelper<std::vector<DigiGroupRef>>(const char* brprefix, std::vector<int> const& tpcsectors, uint64_t activeSectors,
+                                            TFile& originfile, framework::ProcessingContext& pc)
+{
+  // specialization for TPC Trigger
+  auto keyslist = originfile.GetListOfKeys();
+  for (int i = 0; i < keyslist->GetEntries(); ++i) {
+    auto key = keyslist->At(i);
+    int sector = atoi(key->GetName());
+    if (std::find(tpcsectors.begin(), tpcsectors.end(), sector) == tpcsectors.end()) {
+      // do nothing if sector not wanted
+      continue;
+    }
+
+    using AccumType = std::decay_t<decltype(makePublishBuffer<std::vector<DigiGroupRef>>(pc, sector, activeSectors))>;
+    AccumType accum;
+#pragma omp critical
+    accum = makePublishBuffer<std::vector<DigiGroupRef>>(pc, sector, activeSectors);
+    // no actual data sent. Continuous mode.
+
+    publishBuffer(pc, sector, activeSectors, accum);
+  }
+}
+
 void publishMergedTimeframes(std::vector<int> const& lanes, std::vector<int> const& tpcsectors, bool domctruth, framework::ProcessingContext& pc)
 {
   uint64_t activeSectors = 0;
@@ -208,13 +255,21 @@ void publishMergedTimeframes(std::vector<int> const& lanes, std::vector<int> con
     auto originfile = new TFile(filename.c_str(), "OPEN");
     assert(originfile);
 
-    //data definitions
+    // data definitions
     using DigitsType = std::vector<o2::tpc::Digit>;
     using LabelType = o2::dataformats::MCTruthContainer<o2::MCCompLabel>;
     mergeHelper<DigitsType>("TPCDigit_", tpcsectors, activeSectors, *originfile, pc);
     if (domctruth) {
       mergeHelper<LabelType>("TPCDigitMCTruth_", tpcsectors, activeSectors, *originfile, pc);
     }
+
+    // we also merge common modes and publish a (fake) trigger entry
+    using CommonModeType = std::vector<o2::tpc::CommonMode>;
+    mergeHelper<CommonModeType>("TPCCommonMode_", tpcsectors, activeSectors, *originfile, pc);
+
+    using TriggerType = std::vector<DigiGroupRef>;
+    mergeHelper<TriggerType>("TPCCommonMode_", tpcsectors, activeSectors, *originfile, pc);
+
     originfile->Close();
     delete originfile;
   }
@@ -257,7 +312,7 @@ class Task
 /// MC truth information is also aggregated and written out
 DataProcessorSpec getSpec(std::vector<int> const& laneConfiguration, std::vector<int> const& tpcsectors, bool mctruth, bool publish = true)
 {
-  //data definitions
+  // data definitions
   using DigitsOutputType = std::vector<o2::tpc::Digit>;
   using CommonModeOutputType = std::vector<o2::tpc::CommonMode>;
 
@@ -266,10 +321,14 @@ DataProcessorSpec getSpec(std::vector<int> const& laneConfiguration, std::vector
     // effectively the input expects one sector per subspecification
     for (int s = 0; s < 36; ++s) {
       OutputLabel binding{std::to_string(s)};
-      outputs.emplace_back(/*binding,*/ "TPC", "DIGITS", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
+      outputs.emplace_back("TPC", "DIGITS", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
       if (mctruth) {
-        outputs.emplace_back(/*binding,*/ "TPC", "DIGITSMCTR", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
+        outputs.emplace_back("TPC", "DIGITSMCTR", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
       }
+      // common mode
+      outputs.emplace_back("TPC", "COMMONMODE", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
+      // trigger records
+      outputs.emplace_back("TPC", "DIGTRIGGERS", static_cast<SubSpecificationType>(s), Lifetime::Timeframe);
     }
   }
 
@@ -287,12 +346,25 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
 
   auto numlanes = configcontext.options().get<int>("tpc-lanes");
   bool mctruth = !configcontext.options().get<bool>("disable-mc");
+  bool writeout = configcontext.options().get<bool>("writer-mode");
   auto tpcsectors = o2::RangeTokenizer::tokenize<int>(configcontext.options().get<std::string>("tpc-sectors"));
 
   std::vector<int> lanes(numlanes);
   std::iota(lanes.begin(), lanes.end(), 0);
   specs.emplace_back(o2::tpc::getSpec(lanes, tpcsectors, mctruth));
 
+  if (writeout) {
+    // for now writeout to a ROOT file only works if all sectors
+    // are included
+    if (tpcsectors.size() != 36) {
+      LOG(error) << "You currently need to include all TPC sectors in the ROOT writer-mode";
+    } else {
+      std::vector<int> writerlanes(tpcsectors.size());
+      std::iota(writerlanes.begin(), writerlanes.end(), 0);
+      specs.emplace_back(o2::tpc::getTPCDigitRootWriterSpec(writerlanes, mctruth));
+    }
+  }
+
   // configure dpl timer to inject correct firstTForbit: start from the 1st orbit of TF containing 1st sampled orbit
   o2::raw::HBFUtilsInitializer hbfIni(configcontext, specs);
   return specs;

…C/workflow/src/ChunkedDigitPublisher.cxx …imworkflow/src/ChunkedDigitPublisher.cxxDetectors/TPC/workflow/src/ChunkedDigitPublisher.cxx renamed to Detectors/TPC/simworkflow/src/ChunkedDigitPublisher.cxx Detectors/TPC/workflow/src/ChunkedDigitPublisher.cxx renamed to Detectors/TPC/simworkflow/src/ChunkedDigitPublisher.cxx

@@ -14,7 +14,7 @@
 /// @since  2018-04-19
 /// @brief  Processor spec for a ROOT file writer for TPC digits
 
-#include "TPCDigitRootWriterSpec.h"
+#include "TPCSimWorkflow/TPCDigitRootWriterSpec.h"
 #include "DataFormatsTPC/TPCSectorHeader.h"
 #include "CommonDataFormat/RangeReference.h"
 #include "Framework/InputRecord.h"
@@ -77,7 +77,7 @@ DataProcessorSpec getTPCDigitRootWriterSpec(std::vector<int> const& laneConfigur
     }
   };
 
-  //branch definitions for RootTreeWriter spec
+  // branch definitions for RootTreeWriter spec
   using DigitsOutputType = std::vector<o2::tpc::Digit>;
   using CommonModeOutputType = std::vector<o2::tpc::CommonMode>;
 
@@ -156,8 +156,8 @@ DataProcessorSpec getTPCDigitRootWriterSpec(std::vector<int> const& laneConfigur
       LOG(info) << "DIGIT SIZE " << digiData.size();
       const auto& trigS = (*trigP2Sect.get())[sector];
       int entries = 0;
-      if (!trigS.size()) {
-        std::runtime_error("Digits for sector " + std::to_string(sector) + " are received w/o info on grouping in triggers");
+      if (trigS.size() == 0) {
+        LOG(warn) << "Digits for sector " + std::to_string(sector) + " are received w/o trigger info. Will assume continuous mode";
       } else { // check consistency of Ndigits with that of expected from the trigger
         int nExp = trigS.back().getFirstEntry() + trigS.back().getEntries() - trigS.front().getFirstEntry();
         if (nExp != digiData.size()) {
@@ -167,7 +167,7 @@ DataProcessorSpec getTPCDigitRootWriterSpec(std::vector<int> const& laneConfigur
       }
 
       {
-        if (trigS.size() == 1) { // just 1 entry (continous mode?), use digits directly
+        if (trigS.size() <= 1) { // just 1 entry (continous mode?), use digits directly
           auto ptr = &digiData;
           branch.SetAddress(&ptr);
           branch.Fill();
@@ -214,8 +214,8 @@ DataProcessorSpec getTPCDigitRootWriterSpec(std::vector<int> const& laneConfigur
       LOG(info) << "MCTRUTH ELEMENTS " << labeldata.getIndexedSize()
                 << " WITH " << labeldata.getNElements() << " LABELS";
       const auto& trigS = (*trigP2Sect.get())[sector];
-      if (!trigS.size()) {
-        throw std::runtime_error("MCTruth for sector " + std::to_string(sector) + " are received w/o info on grouping in triggers");
+      if (trigS.size() == 0) {
+        LOG(warn) << "MCTruth for sector " + std::to_string(sector) + " received w/o trigger info. Will assume continuous mode";
       } else {
         int nExp = trigS.back().getFirstEntry() + trigS.back().getEntries() - trigS.front().getFirstEntry();
         if (nExp != labeldata.getIndexedSize()) {
@@ -225,7 +225,7 @@ DataProcessorSpec getTPCDigitRootWriterSpec(std::vector<int> const& laneConfigur
         }
       }
       {
-        if (trigS.size() == 1) { // just 1 entry (continous mode?), use labels directly
+        if (trigS.size() <= 1) { // just 0 or 1 entry (continous mode?), use labels directly
           outputcontainer.adopt(labelbuffer);
           br->Fill();
           br->ResetAddress();

…rWorkflow/src/TPCDigitRootWriterSpec.cxx …mworkflow/src/TPCDigitRootWriterSpec.cxxSteer/DigitizerWorkflow/src/TPCDigitRootWriterSpec.cxx renamed to Detectors/TPC/simworkflow/src/TPCDigitRootWriterSpec.cxx Steer/DigitizerWorkflow/src/TPCDigitRootWriterSpec.cxx renamed to Detectors/TPC/simworkflow/src/TPCDigitRootWriterSpec.cxx

@@ -70,17 +70,6 @@ o2_add_library(TPCWorkflowStudies
                                      O2::GlobalTrackingWorkflow
            )
 
-o2_add_executable(chunkeddigit-merger
-        COMPONENT_NAME tpc
-        TARGETVARNAME mergertargetName
-        SOURCES src/ChunkedDigitPublisher.cxx
-        PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
-
-if(OpenMP_CXX_FOUND)
-  # Must be private, depending libraries might be compiled by compiler not understanding -fopenmp
-  target_compile_definitions(${mergertargetName} PRIVATE WITH_OPENMP)
-  target_link_libraries(${mergertargetName} PRIVATE OpenMP::OpenMP_CXX)
-endif()
 
 
 o2_add_executable(reco-workflow

Detectors/TPC/workflow/CMakeLists.txt

@@ -21,9 +21,3 @@ o2_add_library(TPCReaderWorkflow
                                      O2::DPLUtils
                                      O2::TPCBase
            )
-
-if(OpenMP_CXX_FOUND)
-  # Must be private, depending libraries might be compiled by compiler not understanding -fopenmp
-  target_compile_definitions(${mergertargetName} PRIVATE WITH_OPENMP)
-  target_link_libraries(${mergertargetName} PRIVATE OpenMP::OpenMP_CXX)
-endif()

Detectors/TPC/workflow/readers/CMakeLists.txt

@@ -24,7 +24,6 @@ o2_add_executable(digitizer-workflow
                           src/CPVDigitizerSpec.cxx
                           src/SimReaderSpec.cxx
                           src/SimpleDigitizerWorkflow.cxx
-                          src/TPCDigitRootWriterSpec.cxx
                           src/TPCDigitizerSpec.cxx
                           src/ZDCDigitizerSpec.cxx
                           src/TOFDigitizerSpec.cxx
@@ -59,6 +58,7 @@ o2_add_executable(digitizer-workflow
                                         O2::TOFReconstruction
                                         O2::TOFWorkflowIO
                                         O2::TPCSimulation
+                                        O2::TPCSimWorkflow
                                         O2::TRDSimulation
                                         O2::TRDWorkflow
                                         O2::TRDWorkflowIO

Steer/DigitizerWorkflow/CMakeLists.txt

@@ -30,7 +30,7 @@
 
 // for TPC
 #include "TPCDigitizerSpec.h"
-#include "TPCDigitRootWriterSpec.h"
+#include "TPCSimWorkflow/TPCDigitRootWriterSpec.h"
 #include "TPCBase/Sector.h"
 #include "TPCBase/CDBInterface.h"
 // needed in order to init the **SHARED** polyadist file (to be done before the digitizers initialize)