ctf-reader can optionally inject only selected CTF IDs from input (multi-ctf) files

Author: shahor02

Detectors/CTF/README.md

@@ -111,6 +111,13 @@ There is a possibility to read remote root files directly, w/o caching them loca
 2) provide proper regex to define remote files, e.g. for the example above: `--remote-regex "^root://.+/eos/aliceo2/.+"`.
 3) pass an option `--copy-cmd no-copy`.
 
+```
+--select-ctf-ids <id's of CTFs to select>
+```
+This is a `ctf-reader` device local option allowing selective reading of particular CTFs. It is useful when dealing with CTF files containing multiple TFs. The comma-separated list of increasing CTFs indices must be provided in the format parsed by the `RangeTokenizer<int>`, e.g. `1,4-6,...`.
+Note that the index corresponds not to the entry of the TF in the CTF tree but to the reader own counter incremented throught all input files (e.g. if the 10 CTF files with 20 TFs each are provided for the input and the selection of TFs
+`0,2,22,66` is provided, the reader will inject to the DPL the TFs at entries 0 and 2 from the 1st CTF file, entry 5 of the second file, entry 6 of the 3d and will finish the job.
+
 For the ITS and MFT entropy decoding one can request either to decompose clusters to digits and send them instead of clusters (via `o2-ctf-reader-workflow` global options `--its-digits` and `--mft-digits` respectively)
 or to apply the noise mask to decoded clusters (or decoded digits). If the masking (e.g. via option `--its-entropy-decoder " --mask-noise "`) is requested, user should provide to the entropy decoder the noise mask file (eventually will be loaded from CCDB) and cluster patterns decoding dictionary (if the clusters were encoded with patterns IDs).
 For example,

Detectors/CTF/workflow/include/CTFWorkflow/CTFReaderSpec.h

@@ -29,6 +29,7 @@ struct CTFReaderInp {
   std::string copyCmd{};
   std::string tffileRegex{};
   std::string remoteRegex{};
+  std::vector<int> ctfIDs{};
   int maxFileCache = 1;
   int64_t delay_us = 0;
   int maxLoops = 0;

Detectors/CTF/workflow/src/CTFReaderSpec.cxx

@@ -77,6 +77,7 @@ class CTFReaderSpec : public o2::framework::Task
  private:
   void openCTFFile(const std::string& flname);
   void processTF(ProcessingContext& pc);
+  void checkTreeEntries();
   void stop();
   CTFReaderInp mInput{};
   std::unique_ptr<o2::utils::FileFetcher> mFileFetcher;
@@ -86,6 +87,7 @@ class CTFReaderSpec : public o2::framework::Task
   int mCTFCounter = 0;
   long mLastSendTime = 0L;
   long mCurrTreeEntry = 0;
+  size_t mSelIDEntry = 0; // next CTFID to select from the mInput.ctfIDs (if non-empty)
   TStopwatch mTimer;
 };
 
@@ -124,6 +126,7 @@ void CTFReaderSpec::stop()
 ///_______________________________________
 void CTFReaderSpec::init(InitContext& ic)
 {
+  mInput.ctfIDs = o2::RangeTokenizer::tokenize<int>(ic.options().get<std::string>("select-ctf-ids"));
   mRunning = true;
   mFileFetcher = std::make_unique<o2::utils::FileFetcher>(mInput.inpdata, mInput.tffileRegex, mInput.remoteRegex, mInput.copyCmd);
   mFileFetcher->setMaxFilesInQueue(mInput.maxFileCache);
@@ -150,15 +153,24 @@ void CTFReaderSpec::openCTFFile(const std::string& flname)
 void CTFReaderSpec::run(ProcessingContext& pc)
 {
   std::string tfFileName;
-  if (mCTFCounter >= mInput.maxTFs) { // done
+  if (mCTFCounter >= mInput.maxTFs || (!mInput.ctfIDs.empty() && mSelIDEntry >= mInput.ctfIDs.size())) { // done
+    LOG(INFO) << "All CTFs from selected range were injected, stopping";
     mRunning = false;
   }
 
   while (mRunning) {
     if (mCTFTree) { // there is a tree open with multiple CTF
-      LOG(INFO) << "TF " << mCTFCounter << " of " << mInput.maxTFs << " loop " << mFileFetcher->getNLoops();
-      processTF(pc);
-      break;
+      if (mInput.ctfIDs.empty() || mInput.ctfIDs[mSelIDEntry] == mCTFCounter) { // no selection requested or matching CTF ID is found
+        LOG(DEBUG) << "TF " << mCTFCounter << " of " << mInput.maxTFs << " loop " << mFileFetcher->getNLoops();
+        mSelIDEntry++;
+        processTF(pc);
+        break;
+      } else { // explict CTF ID selection list was provided and current entry is not selected
+        LOGP(INFO, "Skipping CTF${} ({} of {} in {})", mCTFCounter, mCurrTreeEntry, mCTFTree->GetEntries(), mCTFFile->GetName());
+        checkTreeEntries();
+        mCTFCounter++;
+        continue;
+      }
     }
     //
     tfFileName = mFileFetcher->getNextFileInQueue();
@@ -322,14 +334,7 @@ void CTFReaderSpec::processTF(ProcessingContext& pc)
   }
 
   auto entryStr = fmt::format("({} of {} in {})", mCurrTreeEntry, mCTFTree->GetEntries(), mCTFFile->GetName());
-  if (++mCurrTreeEntry >= mCTFTree->GetEntries()) { // this file is done, check if there are other files
-    mCTFTree.reset();
-    mCTFFile->Close();
-    mCTFFile.reset();
-    if (mFileFetcher) {
-      mFileFetcher->popFromQueue(mFileFetcher->getNLoops() >= mInput.maxLoops);
-    }
-  }
+  checkTreeEntries();
   mTimer.Stop();
   // do we need to way to respect the delay ?
   long tNow = std::chrono::time_point_cast<std::chrono::microseconds>(std::chrono::system_clock::now()).time_since_epoch().count();
@@ -347,6 +352,20 @@ void CTFReaderSpec::processTF(ProcessingContext& pc)
   mCTFCounter++;
 }
 
+///_______________________________________
+void CTFReaderSpec::checkTreeEntries()
+{
+  // check if the tree has entries left, if needed, close current tree/file
+  if (++mCurrTreeEntry >= mCTFTree->GetEntries()) { // this file is done, check if there are other files
+    mCTFTree.reset();
+    mCTFFile->Close();
+    mCTFFile.reset();
+    if (mFileFetcher) {
+      mFileFetcher->popFromQueue(mFileFetcher->getNLoops() >= mInput.maxLoops);
+    }
+  }
+}
+
 ///_______________________________________
 DataProcessorSpec getCTFReaderSpec(const CTFReaderInp& inp)
 {
@@ -364,7 +383,7 @@ DataProcessorSpec getCTFReaderSpec(const CTFReaderInp& inp)
     Inputs{},
     outputs,
     AlgorithmSpec{adaptFromTask<CTFReaderSpec>(inp)},
-    Options{}};
+    Options{{"select-ctf-ids", VariantType::String, "", {"comma-separated list CTF IDs to inject (from cumulative counter of CTFs seen)"}}}};
 }
 
 } // namespace ctf

Detectors/CTF/workflow/src/ctf-reader-workflow.cxx

@@ -20,6 +20,7 @@
 #include "DataFormatsParameters/GRPObject.h"
 #include "DetectorsCommonDataFormats/DetID.h"
 #include "CommonUtils/ConfigurableParam.h"
+#include "Algorithm/RangeTokenizer.h"
 
 // Specific detectors specs
 #include "ITSMFTWorkflow/EntropyDecoderSpec.h"