Add standalone AOD producer

Detectors/PHOS/workflow/CMakeLists.txt

@@ -20,6 +20,8 @@ o2_add_library(PHOSWorkflow
                        src/EntropyEncoderSpec.cxx
                        src/EntropyDecoderSpec.cxx
                        src/WriterSpec.cxx
+                       src/EventBuilderSpec.cxx
+                       src/StandaloneAODProducerSpec.cxx
                PUBLIC_LINK_LIBRARIES O2::Framework O2::DataFormatsPHOS
                                      O2::DPLUtils
                                      O2::PHOSBase
@@ -41,3 +43,13 @@ o2_add_executable(cell-writer-workflow
                   COMPONENT_NAME phos
                   SOURCES src/cell-writer-workflow.cxx
                   PUBLIC_LINK_LIBRARIES O2::PHOSWorkflow)
+
+o2_add_executable(event-builder
+                  COMPONENT_NAME phos
+                  SOURCES src/event-builder-workflow.cxx
+                  PUBLIC_LINK_LIBRARIES O2::PHOSWorkflow)
+
+o2_add_executable(standalone-aod-producer-workflow
+                  COMPONENT_NAME phos
+                  SOURCES src/phos-standalone-aod-producer-workflow.cxx
+                  PUBLIC_LINK_LIBRARIES O2::PHOSWorkflow)

Detectors/PHOS/workflow/include/PHOSWorkflow/EventBuilderSpec.h

@@ -0,0 +1,64 @@
+// 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.
+
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/Task.h"
+
+namespace o2
+{
+
+namespace phos
+{
+class TriggerRecord;
+class Cell;
+
+/// \class EventBuilderSpec
+/// \brief Class merges subevents from two FLPs
+/// \ingroup PHOSworkflow
+/// \author Dmitri Peresunko, NRC "Kurchatov institute"
+/// \since March, 2022
+///
+/// Merge subevents from two FLPs. FLPs send messages with non-zero subspecifications
+/// take TrigRecs with same BC stamps and copy corresponding cells
+/// sells are sorted, so find subspec with smaller absId and copy it first (exclude/handle trigger cells!)
+/// check if all halves of events were found, otherwise send warning
+class EventBuilderSpec : public framework::Task
+{
+ public:
+  /// \brief Constructor
+  EventBuilderSpec() = default;
+
+  /// \brief Destructor
+  ~EventBuilderSpec() override = default;
+
+  void init(framework::InitContext& ctx) final;
+  void run(framework::ProcessingContext& ctx) final;
+
+ private:
+  class SubspecSet
+  {
+   public:
+    SubspecSet(gsl::span<const o2::phos::TriggerRecord> r, gsl::span<const o2::phos::Cell> c)
+    {
+      trSpan = r;
+      cellSpan = c;
+    }
+    ~SubspecSet() = default;
+    gsl::span<const o2::phos::TriggerRecord> trSpan;
+    gsl::span<const o2::phos::Cell> cellSpan;
+  };
+};
+
+o2::framework::DataProcessorSpec getEventBuilderSpec();
+
+} // namespace phos
+
+} // namespace o2

Detectors/PHOS/workflow/include/PHOSWorkflow/RawToCellConverterSpec.h

@@ -40,7 +40,7 @@ class RawToCellConverterSpec : public framework::Task
  public:
   /// \brief Constructor
   /// \param propagateMC If true the MCTruthContainer is propagated to the output
-  RawToCellConverterSpec() : framework::Task(){};
+  RawToCellConverterSpec(unsigned int flpId) : mflpId(flpId), framework::Task(){};
 
   /// \brief Destructor
   ~RawToCellConverterSpec() override = default;
@@ -65,6 +65,7 @@ class RawToCellConverterSpec : public framework::Task
   bool mCombineGHLG = true;                                   ///< Combine or not HG and LG channels (def: combine, LED runs: not combine)
   bool mPedestalRun = false;                                  ///< Analyze pedestal run (calculate pedestal mean and RMS)
   int mLastSize = 0;                                          ///< size of last send list of cells to reserve same in next bunch
+  unsigned int mflpId = 0;                                    ///< subspec of output stream
   std::unique_ptr<AltroDecoder> mDecoder;                     ///!<! Raw decoder
   std::unique_ptr<CaloRawFitter> mRawFitter;                  ///!<! Raw fitter
   std::array<std::vector<Cell>, 14> mTmpCells;                ///< Temporary cells storage to all 14 DLL
@@ -78,7 +79,7 @@ class RawToCellConverterSpec : public framework::Task
 /// \brief Creating DataProcessorSpec for the PHOS Cell Converter Spec
 ///
 /// Refer to RawToCellConverterSpec::run for input and output specs
-framework::DataProcessorSpec getRawToCellConverterSpec(int flpId);
+framework::DataProcessorSpec getRawToCellConverterSpec(unsigned int flpId);
 
 } // namespace reco_workflow
 

Detectors/PHOS/workflow/include/PHOSWorkflow/StandaloneAODProducerSpec.h

@@ -0,0 +1,52 @@
+// 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.
+
+/// @file   StandaloneAODProducerSpec.h
+/// @brief  Convert CTF (EncodedBlocks) to AO2D PHOS standalone
+
+#ifndef O2_PHOS_STANDALONEAODPRODUCERSPEC_SPEC
+#define O2_PHOS_STANDALONEAODPRODUCERSPEC_SPEC
+
+#include "Framework/AnalysisDataModel.h"
+#include "Framework/AnalysisHelpers.h"
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/Task.h"
+#include <TStopwatch.h>
+
+namespace o2
+{
+namespace phos
+{
+
+class StandaloneAODProducerSpec : public o2::framework::Task
+{
+ public:
+  StandaloneAODProducerSpec();
+  ~StandaloneAODProducerSpec() override = default;
+  void run(o2::framework::ProcessingContext& pc) final;
+  void init(o2::framework::InitContext& ic) final;
+  void endOfStream(o2::framework::EndOfStreamContext& ec) final;
+
+ private:
+  int64_t mTFNumber = -1;          // Timeframe ID
+  int mRunNumber = -1;             // Run number
+  uint32_t mCaloAmp = 0xFFFFFF00;  // 15 bits
+  uint32_t mCaloTime = 0xFFFFFF00; // 15 bits
+  TStopwatch mTimer;
+};
+
+/// create a processor spec
+framework::DataProcessorSpec getPHOSStandaloneAODProducerSpec();
+
+} // namespace phos
+} // namespace o2
+
+#endif

Detectors/PHOS/workflow/src/EventBuilderSpec.cxx

@@ -0,0 +1,202 @@
+// 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.
+#include "FairLogger.h"
+
+#include "DataFormatsPHOS/Cell.h"
+#include "DataFormatsPHOS/TriggerRecord.h"
+#include "PHOSWorkflow/EventBuilderSpec.h"
+#include "Framework/ControlService.h"
+#include "CommonDataFormat/InteractionRecord.h"
+#include <vector>
+
+using namespace o2::phos;
+
+void EventBuilderSpec::init(framework::InitContext& ctx)
+{
+}
+
+void EventBuilderSpec::run(framework::ProcessingContext& ctx)
+{
+  // Merge subevents from two FLPs. FLPs send messages with non-zero subspecifications
+  // take TrigRecs with same BC stamps and copy corresponding cells
+  // sells are sorted, so find subspec with smaller absId and copy it first (exclude/handle trigger cells!)
+  // check if all halves of events were found, otherwise send warning
+
+  std::vector<o2::phos::Cell> outputCells;
+  std::vector<o2::phos::TriggerRecord> outputTriggers;
+  int nOutCells = 0;
+
+  int posCells = ctx.inputs().getPos("inputcells");
+  int posTriggerRecords = ctx.inputs().getPos("inputtriggers");
+  int numSlotsCells = ctx.inputs().getNofParts(posCells);
+  int numSlotsTriggerRecords = ctx.inputs().getNofParts(posTriggerRecords);
+
+  std::vector<SubspecSet> subspecs;
+
+  // Combine pairs (trigRec,cells) from different subspecs
+  for (int islot = 0; islot < numSlotsTriggerRecords; islot++) {
+    auto trgrecorddata = ctx.inputs().getByPos(posTriggerRecords, islot);
+    auto subspecification = framework::DataRefUtils::getHeader<header::DataHeader*>(trgrecorddata)->subSpecification;
+    // Discard message if it is a deadbeaf message (empty timeframe)
+    if (subspecification == 0xDEADBEEF) {
+      continue;
+    }
+    auto celldata = ctx.inputs().getByPos(posCells, islot);
+    // check if cells from the same source?
+    // if not try to find proper slot
+    if (framework::DataRefUtils::getHeader<header::DataHeader*>(celldata)->subSpecification != subspecification) {
+      for (int islotCell = 0; islotCell < numSlotsCells; islotCell++) {
+        celldata = ctx.inputs().getByPos(posCells, islotCell);
+        if (framework::DataRefUtils::getHeader<header::DataHeader*>(celldata)->subSpecification == subspecification) {
+          break;
+        }
+      }
+      // if still not found pair
+      if (framework::DataRefUtils::getHeader<header::DataHeader*>(celldata)->subSpecification != subspecification) {
+        LOG(error) << "Can not find cells for TrigRecords with subspecification " << subspecification;
+        continue;
+      }
+    }
+    subspecs.emplace_back(ctx.inputs().get<gsl::span<o2::phos::TriggerRecord>>(trgrecorddata),
+                          ctx.inputs().get<gsl::span<o2::phos::Cell>>(celldata));
+  }
+
+  // Cells in vectors are sorted. Copy in correct order to preserve sorting
+  int first1 = 0;             // which subspec should go first: 0: undefined yet, 1:[0],2:[1],
+  if (subspecs.size() == 1) { // one input, just copy to output
+    for (const Cell& c : subspecs[0].cellSpan) {
+      outputCells.emplace_back(c);
+    }
+    for (const TriggerRecord& r : subspecs[0].trSpan) {
+      outputTriggers.emplace_back(r);
+    }
+
+  } else {
+    if (subspecs.size() > 2) { // error
+      LOG(error) << " Too many subspecs for event building:" << subspecs.size();
+    } else { // combine two subspecs
+      auto tr1 = subspecs[0].trSpan.begin();
+      auto tr2 = subspecs[1].trSpan.begin();
+      while (tr1 != subspecs[0].trSpan.end() && tr2 != subspecs[1].trSpan.end()) {
+        if (tr1->getBCData() == tr2->getBCData()) { // OK, copy
+          if (first1 == 0) {                        // order of subspecs not yet defined
+            if (subspecs[0].cellSpan.size() > 0) {
+              short absId = subspecs[0].cellSpan[0].getAbsId();
+              if (absId > 0) { // this is readout cell
+                first1 = 1 + (absId - 1) / (2 * 64 * 56);
+              } else { // TRU cell
+                first1 = 1 + (subspecs[0].cellSpan[0].getTRUId() - 1) / (8 * 28 * 7);
+              }
+            } else {
+              if (subspecs[1].cellSpan.size() > 0) {
+                short absId = subspecs[1].cellSpan[0].getAbsId();
+                if (absId > 0) { // this is readout cell
+                  first1 = 2 - (absId - 1) / (2 * 64 * 56);
+                } else { // TRU cell
+                  first1 = 2 - (subspecs[1].cellSpan[0].getTRUId() - 1) / (8 * 28 * 7);
+                }
+              }
+              // if both lists are empty, keep first1 zero and try next time
+            }
+          }
+
+          gsl::details::span_iterator<const o2::phos::Cell> itC1, end1, itC2, end2;
+          if (first1 > 1) { // copy first set1 then set2
+            itC1 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry();
+            end1 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry() + tr1->getNumberOfObjects();
+            itC2 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry();
+            end2 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry() + tr2->getNumberOfObjects();
+          } else {
+            itC1 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry();
+            end1 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry() + tr2->getNumberOfObjects();
+            itC2 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry();
+            end2 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry() + tr1->getNumberOfObjects();
+          }
+          // first copy readout Cells from both events, then trigger
+          while (itC1 != end1) {
+            if (itC1->getAbsId() == 0) { // TRU cells further
+              break;
+            }
+            outputCells.emplace_back(*itC1);
+            ++itC1;
+          }
+          while (itC2 != end2) {
+            if (itC2->getAbsId() == 0) {
+              break;
+            }
+            outputCells.emplace_back(*itC2);
+            ++itC2;
+          }
+          // Copy trigger
+          while (itC1 != end1) {
+            outputCells.emplace_back(*itC1);
+            ++itC1;
+          }
+          while (itC2 != end2) {
+            outputCells.emplace_back(*itC2);
+            ++itC2;
+          }
+
+          outputTriggers.emplace_back(tr1->getBCData(), nOutCells, outputCells.size() - nOutCells);
+          nOutCells = outputCells.size();
+          ++tr1;
+          ++tr2;
+        } else { // inconsistent BCs
+          // find smaller BC and copy one part and mark as missing second part
+          if (tr1->getBCData() < tr2->getBCData()) {
+            auto itC1 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry();
+            auto end1 = subspecs[0].cellSpan.begin() + tr1->getFirstEntry() + tr1->getNumberOfObjects();
+            while (itC1 != end1) {
+              outputCells.emplace_back(*itC1);
+            }
+            outputTriggers.emplace_back(tr1->getBCData(), nOutCells, outputCells.size() - nOutCells);
+            nOutCells = outputCells.size();
+            ++tr1;
+          } else {
+            auto itC1 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry();
+            auto end1 = subspecs[1].cellSpan.begin() + tr2->getFirstEntry() + tr2->getNumberOfObjects();
+            while (itC1 != end1) {
+              outputCells.emplace_back(*itC1);
+            }
+            outputTriggers.emplace_back(tr2->getBCData(), nOutCells, outputCells.size() - nOutCells);
+            nOutCells = outputCells.size();
+            ++tr2;
+          }
+        }
+      }
+      if (tr1 != subspecs[0].trSpan.end() || tr2 != subspecs[1].trSpan.end()) {
+        LOG(error) << "Inconsistent number of TriggerRecords in subsec 1 and 2: " << subspecs[0].trSpan.size() << " and " << subspecs[1].trSpan.size();
+      }
+    }
+  }
+
+  ctx.outputs().snapshot(framework::Output{o2::header::gDataOriginPHS, "CELLS", 0, framework::Lifetime::Timeframe}, outputCells);
+  ctx.outputs().snapshot(framework::Output{o2::header::gDataOriginPHS, "CELLTRIGREC", 0, framework::Lifetime::Timeframe}, outputTriggers);
+}
+
+o2::framework::DataProcessorSpec o2::phos::getEventBuilderSpec()
+{
+  std::vector<o2::framework::InputSpec> inputs;
+  inputs.emplace_back("inputcells", "PHS", "CELLS", 1, o2::framework::Lifetime::Optional);          // Input subspec 1, output subspec 0
+  inputs.emplace_back("inputtriggers", "PHS", "CELLTRIGREC", 1, o2::framework::Lifetime::Optional); // Input subspec 1, output subspec 0
+  // output should be anyway generated by raw converter. Or not?
+  //  inputs.emplace_back("STFDist", "FLP", "DISTSUBTIMEFRAME", 0, o2::framework::Lifetime::Timeframe);
+
+  o2::framework::Outputs outputs{
+    {o2::header::gDataOriginPHS, "CELLS", 0, o2::framework::Lifetime::Timeframe}, // output is zero subspec
+    {o2::header::gDataOriginPHS, "CELLTRIGREC", 0, o2::framework::Lifetime::Timeframe}};
+
+  return o2::framework::DataProcessorSpec{"PHOSEventBuilder",
+                                          inputs,
+                                          outputs,
+                                          o2::framework::adaptFromTask<o2::phos::EventBuilderSpec>(),
+                                          o2::framework::Options{}};
+}