TPC: add test workflow to create dummy CMV data

Author: ehellbar

Detectors/TPC/workflow/CMakeLists.txt

@@ -198,6 +198,11 @@ o2_add_executable(idc-test-ft
                 SOURCES test/test_ft_EPN_Aggregator.cxx
                 PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
 
+o2_add_executable(cmv-test-generator
+                COMPONENT_NAME tpc
+                SOURCES test/test_cmv_generator.cxx
+                PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
+
 o2_add_executable(miptrack-filter
                   COMPONENT_NAME tpc
                   SOURCES src/tpc-miptrack-filter.cxx

Detectors/TPC/workflow/test/test_cmv_generator.cxx

@@ -0,0 +1,204 @@
+// 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  test_cmv_generator.cxx
+/// \brief DPL source workflow that generates dummy CMV data for testing the CMV FLP pipeline.
+///
+/// Replaces o2-tpc-cmv-to-vector in tests; directly emits CMVVECTOR and CMVORBITS
+/// messages per CRU per TF so the workflow can be piped straight into o2-tpc-cmv-flp:
+///
+///   o2-tpc-cmv-test-generator --crus 0-359 --timeframes 100 \
+///   | o2-tpc-cmv-flp --crus 0-359 --n-TFs-buffer 10        \
+///   | o2-dpl-output-proxy --dataspec "downstream:TPC/CMVGROUP;downstream:TPC/CMVORBITINFO" ...
+///
+/// \author Ernst Hellbar <ernst.hellbar@cern.ch>
+
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/Task.h"
+#include "Framework/ControlService.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/Logger.h"
+#include "Headers/DataHeader.h"
+#include "Algorithm/RangeTokenizer.h"
+#include "TPCBase/CRU.h"
+#include "DataFormatsTPC/CMV.h"
+#include "TPCWorkflow/ProcessingHelpers.h"
+#include "TRandom.h"
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+#include <vector>
+#include <chrono>
+#include <thread>
+#include <cmath>
+#include <unordered_set>
+
+using namespace o2::framework;
+using o2::header::gDataOriginTPC;
+
+// ─────────────────────────────────────────────────────────────────────────────
+// workflow options
+// ─────────────────────────────────────────────────────────────────────────────
+void customize(std::vector<ConfigParamSpec>& workflowOptions)
+{
+  const std::string cruDefault = "0-" + std::to_string(o2::tpc::CRU::MaxCRU - 1);
+  std::vector<ConfigParamSpec> options{
+    {"crus", VariantType::String, cruDefault.c_str(), {"List of CRUs, comma-separated ranges, e.g. 0-3,7,9-15"}},
+    {"timeframes", VariantType::Int, 100, {"Number of TFs to generate"}},
+    {"delay", VariantType::Bool, false, {"Add delay after sending all CRUs"}},
+    {"delayTime", VariantType::Int, 1, {"Duration of the global per-TF delay in ms (requires --delay true)"}},
+    {"delayCRUs", VariantType::String, "", {"CRUs for which to add an extra per-CRU delay before sending, comma-separated ranges"}},
+    {"delayTimeCRUs", VariantType::Int, 1, {"Duration of the per-CRU delay in ms (requires --delayCRUs)"}},
+    {"dropTFsRandom", VariantType::Int, 0, {"Drop a whole TF randomly: on average one every N TFs (0 = disabled)"}},
+    {"dropTFsRange", VariantType::String, "", {"Drop all TFs in this range, e.g. 10-12"}},
+    {"seed", VariantType::Int, 42, {"RNG seed for CMV value generation"}},
+    {"amplitude", VariantType::Float, 5.0f, {"Amplitude of the sinusoidal CMV signal (ADC units)"}},
+    {"noise", VariantType::Float, 1.0f, {"Gaussian noise std-dev added per time bin (ADC units)"}},
+  };
+  std::swap(workflowOptions, options);
+}
+
+#include "Framework/runDataProcessing.h"
+
+// ─────────────────────────────────────────────────────────────────────────────
+// generator device
+// ─────────────────────────────────────────────────────────────────────────────
+DataProcessorSpec generateCMVsCRU(const std::vector<uint32_t>& crus,
+                                   const std::unordered_set<uint32_t>& delayCRUs,
+                                   unsigned int maxTFs,
+                                   bool delay,
+                                   int delayTime,
+                                   int delayTimeCRUs,
+                                   int dropTFsRandom,
+                                   const std::vector<int>& rangeTFsDrop,
+                                   float amplitude,
+                                   float noise)
+{
+  using timer = std::chrono::high_resolution_clock;
+
+  std::vector<OutputSpec> outputSpecs;
+  outputSpecs.reserve(crus.size() * 2);
+  for (const auto cru : crus) {
+    const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
+    outputSpecs.emplace_back(ConcreteDataMatcher{gDataOriginTPC, "CMVVECTOR", subSpec}, Lifetime::Timeframe);
+    outputSpecs.emplace_back(ConcreteDataMatcher{gDataOriginTPC, "CMVORBITS", subSpec}, Lifetime::Timeframe);
+  }
+
+  return DataProcessorSpec{
+    "tpc-cmv-generator",
+    Inputs{},
+    outputSpecs,
+    AlgorithmSpec{
+      [maxTFs, delay, delayTime, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, amplitude, noise, crus, delayCRUs](ProcessingContext& ctx) {
+        const auto tf = o2::tpc::processing_helpers::getCurrentTF(ctx);
+
+        // ── TF dropping ──────────────────────────────────────────────────────
+        if (!rangeTFsDrop.empty() && tf >= (uint32_t)rangeTFsDrop.front() && tf <= (uint32_t)rangeTFsDrop.back()) {
+          LOGP(info, "Dropping TF {} (range drop)", tf);
+          return;
+        }
+        if (dropTFsRandom > 0 && !gRandom->Integer(dropTFsRandom)) {
+          LOGP(info, "Dropping TF {} (random drop)", tf);
+          return;
+        }
+
+        auto start = timer::now();
+
+        // FIXME: Slow sinusoidal baseline that drifts across TFs
+        // this does NOT mimick a realistic common mode value distribution
+        const float signal = amplitude * std::sin(tf * 0.05f);
+
+        for (const auto cru : crus) {
+          const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
+
+          // ── per-CRU delay ────────────────────────────────────────────────
+          if (delayCRUs.count(cru)) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(delayTimeCRUs));
+          }
+
+          // ── CMV values ──────────────────────────────────────────────────
+          // NTimeBinsPerTF = NPacketsPerTFPerCRU (4) * NTimeBinsPerPacket (3564) = 14256
+          std::vector<uint16_t> cmvVec;
+          cmvVec.reserve(o2::tpc::cmv::NTimeBinsPerTF);
+          for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) {
+            const float val = signal + noise * gRandom->Gaus(0, 1);
+            o2::tpc::cmv::Data d;
+            d.setCMVFloat(val);
+            cmvVec.push_back(d.getCMV());
+          }
+
+          // ── Orbit / BC info ─────────────────────────────────────────────
+          // One packed (orbit<<32|bc) entry per CMV packet (4 per TF).
+          // Each packet covers 8 heartbeat orbits (NTimeBinsPerPacket = 3564 = 8 LHC orbits),
+          // so the orbit advances by 8 per packet and by NPacketsPerTFPerCRU*8 = 32 per TF.
+          static constexpr uint32_t sOrbitsPerPacket = 8;
+          std::vector<uint64_t> orbitBCVec;
+          orbitBCVec.reserve(o2::tpc::cmv::NPacketsPerTFPerCRU);
+          for (uint32_t pkt = 0; pkt < o2::tpc::cmv::NPacketsPerTFPerCRU; ++pkt) {
+            const uint32_t orbit = static_cast<uint32_t>(tf * o2::tpc::cmv::NPacketsPerTFPerCRU * sOrbitsPerPacket + pkt * sOrbitsPerPacket);
+            orbitBCVec.push_back(uint64_t(orbit) << 32); // bc = 0
+          }
+
+          ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVVECTOR", subSpec}, cmvVec);
+          ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVORBITS", subSpec}, orbitBCVec);
+        }
+
+        // ── global delay ─────────────────────────────────────────────────────
+        if (delay) {
+          auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - start).count();
+          if (elapsed < delayTime) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(delayTime - elapsed));
+          }
+        }
+
+        if (!(tf % 100)) {
+          LOGP(info, "Generated CMV data for TF {}", tf);
+        }
+
+        if (tf >= maxTFs - 1) {
+          ctx.services().get<ControlService>().readyToQuit(QuitRequest::Me);
+        }
+      }}};
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+WorkflowSpec defineDataProcessing(ConfigContext const& config)
+{
+  const auto tpcCRUs = o2::RangeTokenizer::tokenize<int>(config.options().get<std::string>("crus"));
+  const std::vector<uint32_t> crus(tpcCRUs.begin(), tpcCRUs.end());
+  LOGP(info, "crus: {}", fmt::join(crus, ", "));
+
+  const auto delayCRUsStr = config.options().get<std::string>("delayCRUs");
+  std::unordered_set<uint32_t> delayCRUs;
+  if (!delayCRUsStr.empty()) {
+    for (const auto cru : o2::RangeTokenizer::tokenize<int>(delayCRUsStr)) {
+      delayCRUs.insert(static_cast<uint32_t>(cru));
+    }
+  }
+  const std::vector<uint32_t> delayCRUsSorted(delayCRUs.begin(), delayCRUs.end());
+  LOGP(info, "delayCRUs: {}", fmt::join(delayCRUsSorted, ", "));
+
+  const auto rangeTFsDrop = o2::RangeTokenizer::tokenize<int>(config.options().get<std::string>("dropTFsRange"));
+  const auto timeframes = static_cast<unsigned int>(config.options().get<int>("timeframes"));
+  const auto delay = config.options().get<bool>("delay");
+  const auto delayTime = config.options().get<int>("delayTime");
+  const auto delayTimeCRUs = config.options().get<int>("delayTimeCRUs");
+  const auto dropTFsRandom = config.options().get<int>("dropTFsRandom");
+  const auto seed = config.options().get<int>("seed");
+  const auto amplitude = config.options().get<float>("amplitude");
+  const auto noise = config.options().get<float>("noise");
+
+  gRandom->SetSeed(seed);
+
+  WorkflowSpec workflow;
+  workflow.emplace_back(generateCMVsCRU(crus, delayCRUs, timeframes, delay, delayTime, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, amplitude, noise));
+  return workflow;
+}