Add CTP lumi information to TPC SCD calib

Author: martenole

Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/include/TPCInterpolationWorkflow/TPCResidualAggregatorSpec.h

@@ -30,6 +30,7 @@
 #include <chrono>
 
 using namespace o2::framework;
+using GID = o2::dataformats::GlobalTrackID;
 
 namespace o2
 {
@@ -39,7 +40,7 @@ namespace calibration
 class ResidualAggregatorDevice : public o2::framework::Task
 {
  public:
-  ResidualAggregatorDevice(std::shared_ptr<o2::base::GRPGeomRequest> req, bool trackInput, bool writeOutput, bool writeUnbinnedResiduals, bool writeBinnedResiduals, bool writeTrackData) : mCCDBRequest(req), mTrackInput(trackInput), mWriteOutput(writeOutput), mWriteUnbinnedResiduals(writeUnbinnedResiduals), mWriteBinnedResiduals(writeBinnedResiduals), mWriteTrackData(writeTrackData) {}
+  ResidualAggregatorDevice(std::shared_ptr<o2::base::GRPGeomRequest> req, bool trackInput, bool ctpInput, bool writeOutput, bool writeUnbinnedResiduals, bool writeBinnedResiduals, bool writeTrackData, std::shared_ptr<o2::globaltracking::DataRequest> dataRequest) : mCCDBRequest(req), mTrackInput(trackInput), mCTPInput(ctpInput), mWriteOutput(writeOutput), mWriteUnbinnedResiduals(writeUnbinnedResiduals), mWriteBinnedResiduals(writeBinnedResiduals), mWriteTrackData(writeTrackData), mDataRequest(dataRequest) {}
 
   void init(o2::framework::InitContext& ic) final
   {
@@ -92,11 +93,12 @@ class ResidualAggregatorDevice : public o2::framework::Task
   void run(o2::framework::ProcessingContext& pc) final
   {
     auto runStartTime = std::chrono::high_resolution_clock::now();
+    o2::globaltracking::RecoContainer recoCont;
+    recoCont.collectData(pc, *mDataRequest);
     updateTimeDependentParams(pc);
     std::chrono::duration<double, std::milli> ccdbUpdateTime = std::chrono::high_resolution_clock::now() - runStartTime;
 
     auto residualsData = pc.inputs().get<gsl::span<o2::tpc::UnbinnedResid>>("unbinnedRes");
-
     // track data input is optional
     const gsl::span<const o2::tpc::TrackData>* trkDataPtr = nullptr;
     using trkDataType = std::decay_t<decltype(pc.inputs().get<gsl::span<o2::tpc::TrackData>>(""))>;
@@ -105,11 +107,20 @@ class ResidualAggregatorDevice : public o2::framework::Task
       trkData.emplace(pc.inputs().get<gsl::span<o2::tpc::TrackData>>("trkData"));
       trkDataPtr = &trkData.value();
     }
+    // CTP lumi input (optional)
+    const o2::ctp::LumiInfo* lumi = nullptr;
+    using lumiDataType = std::decay_t<decltype(pc.inputs().get<o2::ctp::LumiInfo>(""))>;
+    std::optional<lumiDataType> lumiInput;
+    if (mCTPInput) {
+      recoCont.getCTPLumi();
+      lumiInput = recoCont.getCTPLumi();
+      lumi = &lumiInput.value();
+    }
 
     auto data = std::make_pair<gsl::span<const o2::tpc::TrackData>, gsl::span<const o2::tpc::UnbinnedResid>>(std::move(*trkData), std::move(residualsData));
     o2::base::TFIDInfoHelper::fillTFIDInfo(pc, mAggregator->getCurrentTFInfo());
     LOG(info) << "Processing TF " << mAggregator->getCurrentTFInfo().tfCounter << " with " << trkData->size() << " tracks and " << residualsData.size() << " unbinned residuals associated to them";
-    mAggregator->process(data);
+    mAggregator->process(data, lumi);
     std::chrono::duration<double, std::milli> runDuration = std::chrono::high_resolution_clock::now() - runStartTime;
     LOGP(info, "Duration for run method: {} ms. From this taken for time dependent param update: {} ms",
          std::chrono::duration_cast<std::chrono::milliseconds>(runDuration).count(),
@@ -135,7 +146,9 @@ class ResidualAggregatorDevice : public o2::framework::Task
   }
   std::unique_ptr<o2::tpc::ResidualAggregator> mAggregator; ///< the TimeSlotCalibration device
   std::shared_ptr<o2::base::GRPGeomRequest> mCCDBRequest;
+  std::shared_ptr<o2::globaltracking::DataRequest> mDataRequest; ///< optional CTP input
   bool mTrackInput{false};             ///< flag whether to expect track data as input
+  bool mCTPInput{false};               ///< flag whether to expect luminosity input from CTP
   bool mWriteOutput{true};             ///< if false, no output file will be written
   bool mWriteBinnedResiduals{false};   ///< flag, whether to write binned residuals to output file
   bool mWriteUnbinnedResiduals{false}; ///< flag, whether to write unbinned residuals to output file
@@ -147,9 +160,13 @@ class ResidualAggregatorDevice : public o2::framework::Task
 namespace framework
 {
 
-DataProcessorSpec getTPCResidualAggregatorSpec(bool trackInput, bool writeOutput, bool writeUnbinnedResiduals, bool writeBinnedResiduals, bool writeTrackData)
+DataProcessorSpec getTPCResidualAggregatorSpec(bool trackInput, bool ctpInput, bool writeOutput, bool writeUnbinnedResiduals, bool writeBinnedResiduals, bool writeTrackData)
 {
-  std::vector<InputSpec> inputs;
+  std::shared_ptr<o2::globaltracking::DataRequest> dataRequest = std::make_shared<o2::globaltracking::DataRequest>();
+  if (ctpInput) {
+    dataRequest->requestClusters(GID::getSourcesMask("CTP"), false);
+  }
+  auto& inputs = dataRequest->inputs;
   inputs.emplace_back("unbinnedRes", "GLO", "UNBINNEDRES");
   if (trackInput) {
     inputs.emplace_back("trkData", "GLO", "TRKDATA");
@@ -165,7 +182,7 @@ DataProcessorSpec getTPCResidualAggregatorSpec(bool trackInput, bool writeOutput
     "residual-aggregator",
     inputs,
     Outputs{},
-    AlgorithmSpec{adaptFromTask<o2::calibration::ResidualAggregatorDevice>(ccdbRequest, trackInput, writeOutput, writeUnbinnedResiduals, writeBinnedResiduals, writeTrackData)},
+    AlgorithmSpec{adaptFromTask<o2::calibration::ResidualAggregatorDevice>(ccdbRequest, trackInput, ctpInput, writeOutput, writeUnbinnedResiduals, writeBinnedResiduals, writeTrackData, dataRequest)},
     Options{
       {"tf-per-slot", VariantType::UInt32, 6'000u, {"number of TFs per calibration time slot (put 0 for infinite slot length)"}},
       {"updateInterval", VariantType::UInt32, 6'000u, {"update interval in number of TFs in case slot length is infinite"}},

Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/TPCResidualWriterSpec.cxx

@@ -16,6 +16,7 @@
 #include "SpacePoints/TrackInterpolation.h"
 #include "SpacePoints/TrackResiduals.h"
 #include "TPCInterpolationWorkflow/TPCResidualWriterSpec.h"
+#include "DataFormatsCTP/LumiInfo.h"
 #include "DPLUtils/MakeRootTreeWriterSpec.h"
 
 using namespace o2::framework;

Detectors/GlobalTrackingWorkflow/tpcinterpolationworkflow/src/tpc-residual-aggregator.cxx

@@ -13,15 +13,18 @@
 #include "SpacePoints/ResidualAggregator.h"
 #include "TPCInterpolationWorkflow/TPCResidualAggregatorSpec.h"
 #include "TPCInterpolationWorkflow/TPCUnbinnedResidualReaderSpec.h"
+#include "GlobalTrackingWorkflowHelpers/InputHelper.h"
 
 using namespace o2::framework;
+using GID = o2::dataformats::GlobalTrackID;
 
 // we need to add workflow options before including Framework/runDataProcessing
 void customize(std::vector<o2::framework::ConfigParamSpec>& workflowOptions)
 {
   std::vector<o2::framework::ConfigParamSpec> options{
     {"output-type", VariantType::String, "binnedResid", {"Comma separated list of outputs (without spaces). Valid strings: unbinnedResid, binnedResid, trackParams"}},
     {"enable-track-input", VariantType::Bool, false, {"Whether to expect track data from interpolation workflow"}},
+    {"enable-ctp", VariantType::Bool, false, {"Subscribe to lumi info from CTP"}},
     {"disable-root-input", VariantType::Bool, false, {"disable root-files input readers"}},
     {"disable-root-output", VariantType::Bool, false, {"Disables ROOT file writing"}},
     {"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings ..."}}};
@@ -36,6 +39,7 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
 {
   o2::conf::ConfigurableParam::updateFromString(configcontext.options().get<std::string>("configKeyValues"));
   auto trkInput = configcontext.options().get<bool>("enable-track-input");
+  auto ctpInput = configcontext.options().get<bool>("enable-ctp");
 
   bool writeUnbinnedResiduals = false;
   bool writeBinnedResiduals = false;
@@ -69,6 +73,14 @@ WorkflowSpec defineDataProcessing(ConfigContext const& configcontext)
   if (!configcontext.options().get<bool>("disable-root-input")) {
     specs.emplace_back(o2::tpc::getUnbinnedTPCResidualsReaderSpec(trkInput));
   }
-  specs.emplace_back(getTPCResidualAggregatorSpec(trkInput, fileOutput, writeUnbinnedResiduals, writeBinnedResiduals, writeTrackData));
+  specs.emplace_back(getTPCResidualAggregatorSpec(trkInput, ctpInput, fileOutput, writeUnbinnedResiduals, writeBinnedResiduals, writeTrackData));
+
+  // CTP input
+  if (ctpInput) {
+    auto maskClusters = GID::getSourcesMask("CTP");
+    auto maskNone = GID::getSourcesMask(GID::NONE);
+    o2::globaltracking::InputHelper::addInputSpecs(configcontext, specs, maskClusters, maskNone, maskNone, false);
+  }
+
   return specs;
 }

Detectors/TPC/calibration/SpacePoints/include/SpacePoints/ResidualAggregator.h

@@ -18,6 +18,7 @@
 
 #include "DetectorsCalibration/TimeSlotCalibration.h"
 #include "DetectorsCalibration/TimeSlot.h"
+#include "DataFormatsCTP/LumiInfo.h"
 #include "DataFormatsTPC/Defs.h"
 #include "SpacePoints/TrackResiduals.h"
 #include "CommonUtils/StringUtils.h"
@@ -46,7 +47,7 @@ struct ResidualsContainer {
   void fillStatisticsBranches();
   uint64_t getNEntries() const { return nResidualsTotal; }
 
-  void fill(const o2::dataformats::TFIDInfo& ti, const std::pair<gsl::span<const o2::tpc::TrackData>, gsl::span<const UnbinnedResid>> data);
+  void fill(const o2::dataformats::TFIDInfo& ti, const std::pair<gsl::span<const o2::tpc::TrackData>, gsl::span<const UnbinnedResid>> data, const o2::ctp::LumiInfo* lumiInput);
   void merge(ResidualsContainer* prev);
   void print();
   void writeToFile(bool closeFileAfterwards);
@@ -59,6 +60,7 @@ struct ResidualsContainer {
   uint32_t runNumber;                                                        ///< run number (required for meta data file)
   std::vector<uint32_t> tfOrbits, *tfOrbitsPtr{&tfOrbits};                   ///< first TF orbit
   std::vector<uint32_t> sumOfResiduals, *sumOfResidualsPtr{&sumOfResiduals}; ///< sum of residuals for each TF
+  std::vector<o2::ctp::LumiInfo> lumi, *lumiPtr{&lumi};                      ///< luminosity information from CTP per TF
   std::vector<UnbinnedResid> unbinnedRes, *unbinnedResPtr{&unbinnedRes};     // unbinned residuals
   std::vector<TrackData> trkData, *trkDataPtr{&trkData};                                 // track data and cluster ranges
 

Detectors/TPC/calibration/SpacePoints/src/ResidualAggregator.cxx

@@ -66,6 +66,7 @@ ResidualsContainer::ResidualsContainer(ResidualsContainer&& rhs)
   runNumber = rhs.runNumber;
   tfOrbits = std::move(rhs.tfOrbits);
   sumOfResiduals = std::move(rhs.sumOfResiduals);
+  lumi = std::move(rhs.lumi);
 }
 
 void ResidualsContainer::init(const TrackResiduals* residualsEngine, std::string outputDir, bool wFile, bool wBinnedResid, bool wUnbinnedResid, bool wTrackData, int autosave, int compression)
@@ -113,6 +114,7 @@ void ResidualsContainer::init(const TrackResiduals* residualsEngine, std::string
     }
     treeOutRecords->Branch("firstTForbit", &tfOrbitsPtr);
     treeOutRecords->Branch("sumOfResiduals", &sumOfResidualsPtr);
+    treeOutRecords->Branch("lumi", &lumiPtr);
   }
 }
 
@@ -127,7 +129,7 @@ void ResidualsContainer::fillStatisticsBranches()
   }
 }
 
-void ResidualsContainer::fill(const o2::dataformats::TFIDInfo& ti, const std::pair<gsl::span<const o2::tpc::TrackData>, gsl::span<const UnbinnedResid>> data)
+void ResidualsContainer::fill(const o2::dataformats::TFIDInfo& ti, const std::pair<gsl::span<const o2::tpc::TrackData>, gsl::span<const UnbinnedResid>> data, const o2::ctp::LumiInfo* lumiInput)
 {
   // receives large vector of unbinned residuals and fills the sector-wise vectors
   // with binned residuals and statistics
@@ -190,6 +192,9 @@ void ResidualsContainer::fill(const o2::dataformats::TFIDInfo& ti, const std::pa
   }
   runNumber = ti.runNumber;
   tfOrbits.push_back(ti.firstTForbit);
+  if (lumiInput) {
+    lumi.push_back(*lumiInput);
+  }
 
   if (autosaveInterval > 0 && (tfOrbits.size() % autosaveInterval) == 0 && writeToRootFile) {
     writeToFile(false);
@@ -292,6 +297,8 @@ void ResidualsContainer::merge(ResidualsContainer* prev)
   std::swap(prev->tfOrbits, tfOrbits);
   prev->sumOfResiduals.insert(prev->sumOfResiduals.end(), sumOfResiduals.begin(), sumOfResiduals.end());
   std::swap(prev->sumOfResiduals, sumOfResiduals);
+  prev->lumi.insert(prev->lumi.end(), lumi.begin(), lumi.end());
+  std::swap(prev->lumi, lumi);
 }
 
 void ResidualsContainer::print()

prodtests/full-system-test/aggregator-workflow.sh

@@ -176,7 +176,7 @@ if [[ $AGGREGATOR_TASKS == BARREL_TF ]] || [[ $AGGREGATOR_TASKS == ALL ]]; then
   if [[ $CALIB_TPC_SCDCALIB == 1 ]]; then
     # TODO: the residual aggregator should have --output-dir and --meta-output-dir defined
     # without that the residuals will be stored in the local working directory (and deleted after a week)
-    add_W o2-calibration-residual-aggregator "--disable-root-input $ENABLE_TRACK_INPUT --output-type trackParams,unbinnedResid,binnedResid --autosave-interval $RESIDUAL_AGGREGATOR_AUTOSAVE"
+    add_W o2-calibration-residual-aggregator "--disable-root-input $ENABLE_TRACK_INPUT $CALIB_TPC_SCDCALIB_CTP_INPUT --output-type trackParams,unbinnedResid,binnedResid --autosave-interval $RESIDUAL_AGGREGATOR_AUTOSAVE"
   fi
   if [[ $CALIB_TPC_VDRIFTTGL == 1 ]]; then
     # options available via ARGS_EXTRA_PROCESS_o2_tpc_vdrift_tgl_calibration_workflow="--nbins-tgl 20 --nbins-dtgl 50 --max-tgl-its 2. --max-dtgl-itstpc 0.15 --min-entries-per-slot 1000 --time-slot-seconds 600 <--vdtgl-histos-file-name name> "