Optionally aggregate CMV triggers per FLP + Optionally exclude certain specs from the trigger

Detectors/Raw/README.md

@@ -564,6 +564,11 @@ Selection string for the external trigger to dump particular TF. Must be contain
 ```
 Include data with DPL-generated 0xdeadbeef subspecs (for data missing in the original TF).
 ```
+--exclude-trigger-specs (="")
+
+```
+Ignore trigger seen in these inputs of triggerspec (e.g. to suppress noisy trigger inputs)
+```
 --max-dump-rate arg (=0)
 ```
 Fraction in (`%`) of TFs to dump. W/o external trigger: random(>0) or periodic(<0) rejection. With external trigger: throttle dumping to have the lowest estimated acceptance rate compatible with this rate.

Detectors/Raw/TFReaderDD/src/RawTFDumpSpec.cxx

@@ -29,6 +29,7 @@
 #include <TMath.h>
 #include <filesystem>
 #include <random>
+#include <set>
 
 namespace o2::rawdd
 {
@@ -64,6 +65,7 @@ class RawTFDump : public Task
   std::map<EquipmentIdentifier, std::tuple<size_t, size_t, size_t>> mDataMap;
   std::vector<InputSpec> mFilter{};
   std::vector<InputSpec> mTriggerFilter{};
+  std::vector<InputSpec> mExclTriggerFilter{};
 
   size_t mTFSize = 0;
   size_t mMinFileSize = 0; // if > 0, accumulate TFs in the same file until the total size exceeds this minimum
@@ -101,6 +103,7 @@ class RawTFDump : public Task
   o2::framework::TimingInfo mTimingInfo{};
 
   std::string mTrigger{}; // external trigger input
+  std::string mExclTriggerSpecs{}; // trigger specs to ignore
   std::string mHostName{};
   std::string mTFDir{};
   std::string mTFMetaFileDir = "/dev/null";
@@ -182,7 +185,6 @@ void RawTFDump::init(InitContext& ic)
     mWriteTF = false;
     mStoreMetaFile = false;
   }
-
   mRejectDEADBEEF = !ic.options().get<bool>("include-deadbeef");
   mCreateRunEnvDir = !ic.options().get<bool>("ignore-partition-run-dir");
   mMinFileSize = ic.options().get<int64_t>("min-file-size");
@@ -199,6 +201,10 @@ void RawTFDump::init(InitContext& ic)
   mWarnThrottleTF = ic.options().get<int>("mute-warn-period");
 
   mVerbose = ic.options().get<int>("verbosity-level");
+  mExclTriggerSpecs = ic.options().get<std::string>("exclude-trigger-specs");
+  if (!mExclTriggerSpecs.empty()) {
+    mExclTriggerFilter = select(mExclTriggerSpecs.c_str());
+  }
   if (mTrigger.empty()) {
     if (mMaxAccRate >= 0.f) {
       LOGP(info, "Will accept randomly {}% of TFs", mMaxAccRate);
@@ -208,6 +214,9 @@ void RawTFDump::init(InitContext& ic)
   } else {
     mMaxAccRate = std::abs(mMaxAccRate);
     LOGP(info, "Will limit TFs triggered with {} by {}% at most", mTrigger, mMaxAccRate);
+    if (!mExclTriggerFilter.empty()) {
+      LOGP(info, "Inputs excluded from the trigger: {}", mExclTriggerSpecs);
+    }
   }
 
   if (mWriteTF) {
@@ -450,6 +459,20 @@ bool RawTFDump::triggerTF(ProcessingContext& pc)
              dh->splitPayloadIndex, dh->splitPayloadParts, dh->payloadSize, dh->firstTForbit, dh->tfCounter, extTrig.size(), extTrig.size() > 0 ? extTrig[0] : false);
       }
       if (extTrig.size() && extTrig[0]) {
+        // is the input with this trigger vetoed?
+        bool veto = false;
+        for (const auto& excl : mExclTriggerFilter) {
+          if (DataRefUtils::match(ref, excl)) {
+            if (mVerbose > 0) {
+              LOGP(info, "ignoring trigger from black-listed {}", DataSpecUtils::describe(OutputSpec{dh->dataOrigin, dh->dataDescription, dh->subSpecification}));
+            }
+            veto = true;
+            break;
+          }
+        }
+        if (veto) {
+          continue;
+        }
         trig = true;
         break;
       }
@@ -554,6 +577,7 @@ DataProcessorSpec getRawTFDumpSpec(const std::string& inpconfig, const std::stri
     AlgorithmSpec{adaptFromTask<RawTFDump>(trigger)},
     Options{
       {"include-deadbeef", VariantType::Bool, false, {"Include DPL-generated 0xdeadbeef subspecs for missing data"}},
+      {"exclude-trigger-specs", VariantType::String, "", {"Ignore trigger seen in these inputs of triggerspec"}},
       {"max-dump-rate", VariantType::Float, 0.f, {"%-age of TFs to dump. W/o external trigger: random(>0) or periodic(<0) rejection, with: max limit"}},
       {"rate-est-conf-limit", VariantType::Float, 0.05f, {"quantile for the lowest rate estimate confidence limit"}},
       {"max-warn", VariantType::Int, 5, {"max allowed warnings on throttling"}},

Detectors/TPC/workflow/include/TPCWorkflow/TPCFLPCMVSpec.h

@@ -44,6 +44,7 @@ class TPCFLPCMVDevice : public o2::framework::Task
   {
     mDumpCMVs = ic.options().get<bool>("dump-cmvs-flp");
     mEnableTrigger = ic.options().get<bool>("trigger");
+    mTriggerPerFLP = ic.options().get<bool>("trigger-per-flp");
     mTriggerThresholdCMV = ic.options().get<float>("trigger-threshold-cmv");
     mTriggerThresholdMeanMax = ic.options().get<float>("trigger-threshold-cmvMeanMax");
     mTriggerThresholdMeanMin = ic.options().get<float>("trigger-threshold-cmvMeanMin");
@@ -79,10 +80,16 @@ class TPCFLPCMVDevice : public o2::framework::Task
       mCMVs[cru].insert(mCMVs[cru].end(), vecCMVs.begin(), vecCMVs.end());
 
       const bool cruTriggered = mEnableTrigger && evaluateTrigger(vecCMVs);
-      triggered |= cruTriggered;
+      if (!mTriggerPerFLP) {
+        pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, getDataDescriptionCMVTrigger(), tpcCRUHeader->subSpecification}, cruTriggered);
+      } else {
+        triggered |= cruTriggered;
+      }
+    }
+    if (mTriggerPerFLP) {
+      const header::DataHeader::SubSpecificationType trigSubSpec{mCRUs.front() << 7};
+      pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, getDataDescriptionCMVTrigger(), trigSubSpec}, triggered);
     }
-    const header::DataHeader::SubSpecificationType trigSubSpec{mCRUs.front() << 7};
-    pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, getDataDescriptionCMVTrigger(), trigSubSpec}, triggered);
 
     if (mCountTFsForBuffer >= mNTFsBuffer) {
       mCountTFsForBuffer = 0;
@@ -128,6 +135,7 @@ class TPCFLPCMVDevice : public o2::framework::Task
   const std::vector<uint32_t> mCRUs{};                                 ///< CRUs to process in this instance
   int mNTFsBuffer{1};                                                  ///< number of TFs to buffer before sending
   bool mDumpCMVs{};                                                    ///< dump CMVs to file for debugging
+  bool mTriggerPerFLP{false};                                          ///< send per-FLP trigger decision aggregated over CRUs
   int mCountTFsForBuffer{0};                                           ///< counts TFs to track when to send output
   std::unordered_map<unsigned int, o2::pmr::vector<uint16_t>> mCMVs{}; ///< buffered raw 16-bit CMV values per CRU
   std::unordered_map<uint32_t, uint64_t> mFirstOrbitBC{};              ///< first packed orbit/BC per CRU for the current buffer window
@@ -224,6 +232,7 @@ o2::framework::DataProcessorSpec getTPCFLPCMVSpec(const int ilane, const std::ve
     o2::framework::Options{
       {"dump-cmvs-flp", o2::framework::VariantType::Bool, false, {"Dump CMVs to file"}},
       {"trigger", o2::framework::VariantType::Bool, false, {"Enable CMV trigger evaluation"}},
+      {"trigger-per-flp", o2::framework::VariantType::Bool, false, {"Aggregate triggers of CRUs on FLP to a single trigger"}},
       {"trigger-threshold-cmv", o2::framework::VariantType::Float, -10.f, {"CMV threshold: sequence starts when value drops below this (ADC units)"}},
       {"trigger-threshold-cmvMeanMax", o2::framework::VariantType::Float, -40.f, {"Upper bound on trigger-sequence mean CMV value"}},
       {"trigger-threshold-cmvMeanMin", o2::framework::VariantType::Float, -80.f, {"Lower bound on trigger-sequence mean CMV value"}},