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,279 @@
+// 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/ConfigParamRegistry.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 "CommonUtils/TreeStreamRedirector.h"
+#include "TRandom.h"
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+#include <vector>
+#include <chrono>
+#include <thread>
+#include <cmath>
+#include <memory>
+#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
+// ─────────────────────────────────────────────────────────────────────────────
+class CMVGeneratorDevice : public o2::framework::Task
+{
+ public:
+  static constexpr uint32_t sOrbitsPerPacket = 8; ///< each CMV packet covers 8 heartbeat orbits
+
+  CMVGeneratorDevice(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)
+    : mCRUs(crus), mDelayCRUs(delayCRUs), mMaxTFs(maxTFs), mDelay(delay), mDelayTime(delayTime), mDelayTimeCRUs(delayTimeCRUs), mDropTFsRandom(dropTFsRandom), mRangeTFsDrop(rangeTFsDrop), mAmplitude(amplitude), mNoise(noise) {}
+
+  void init(o2::framework::InitContext& ic) final
+  {
+    LOGP(info, "crus: {}", fmt::join(mCRUs, ", "));
+    const std::vector<uint32_t> delayCRUsSorted(mDelayCRUs.begin(), mDelayCRUs.end());
+    LOGP(info, "delayCRUs: {}", fmt::join(delayCRUsSorted, ", "));
+
+    mWriteDebug = ic.options().get<bool>("write-debug");
+    if (mWriteDebug) {
+      mDebugStreamFileName = ic.options().get<std::string>("debug-file-name");
+      LOGP(info, "Creating debug stream {}", mDebugStreamFileName);
+      mDebugStream = std::make_unique<o2::utils::TreeStreamRedirector>(mDebugStreamFileName.data(), "recreate");
+    }
+  }
+
+  void run(o2::framework::ProcessingContext& ctx) final
+  {
+    using timer = std::chrono::high_resolution_clock;
+    const auto tf = o2::tpc::processing_helpers::getCurrentTF(ctx);
+
+    // ── TF dropping ──────────────────────────────────────────────────────────
+    if (!mRangeTFsDrop.empty() && tf >= (uint32_t)mRangeTFsDrop.front() && tf <= (uint32_t)mRangeTFsDrop.back()) {
+      LOGP(info, "Dropping TF {} (range drop)", tf);
+      return;
+    }
+    if (mDropTFsRandom > 0 && !gRandom->Integer(mDropTFsRandom)) {
+      LOGP(info, "Dropping TF {} (random drop)", tf);
+      return;
+    }
+
+    auto start = timer::now();
+
+    // Slow sinusoidal baseline that drifts across TFs
+    const float signal = mAmplitude * std::sin(tf * 0.05f);
+
+    for (const auto cru : mCRUs) {
+      const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
+
+      // ── per-CRU delay ────────────────────────────────────────────────────
+      if (mDelayCRUs.count(cru)) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTimeCRUs));
+      }
+
+      // ── 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 + mNoise * 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.
+      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);
+
+      if (mWriteDebug) {
+        auto& stream = (*mDebugStream) << "cmvs";
+        stream << "cru=" << cru
+               << "tfCounter=" << tf
+               << "nCMVs=" << cmvVec.size()
+               << "cmvs=" << cmvVec
+               << "\n";
+      }
+    }
+
+    // ── global delay ─────────────────────────────────────────────────────────
+    if (mDelay) {
+      auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - start).count();
+      if (elapsed < mDelayTime) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTime - elapsed));
+      }
+    }
+
+    if (!(tf % 100)) {
+      LOGP(info, "Generated CMV data for TF {}", tf);
+    }
+
+    if (tf >= mMaxTFs - 1) {
+      ctx.services().get<ControlService>().endOfStream();
+      ctx.services().get<ControlService>().readyToQuit(QuitRequest::Me);
+    }
+  }
+
+  void endOfStream(o2::framework::EndOfStreamContext&) final { closeFiles(); }
+  void stop() final { closeFiles(); }
+
+ private:
+  void closeFiles()
+  {
+    if (mDebugStream) {
+      auto& stream = (*mDebugStream) << "cmvs";
+      auto& tree = stream.getTree();
+      tree.SetAlias("sector", "int(cru/10)");
+      mDebugStream->Close();
+      mDebugStream.reset(nullptr);
+    }
+  }
+
+  const std::vector<uint32_t> mCRUs{};
+  const std::unordered_set<uint32_t> mDelayCRUs{};
+  const unsigned int mMaxTFs{};
+  const bool mDelay{false};
+  const int mDelayTime{1};
+  const int mDelayTimeCRUs{1};
+  const int mDropTFsRandom{0};
+  const std::vector<int> mRangeTFsDrop{};
+  const float mAmplitude{5.f};
+  const float mNoise{1.f};
+  bool mWriteDebug{false};
+  std::string mDebugStreamFileName{};
+  std::unique_ptr<o2::utils::TreeStreamRedirector> mDebugStream{};
+};
+
+// ─────────────────────────────────────────────────────────────────────────────
+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)
+{
+  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{adaptFromTask<CMVGeneratorDevice>(crus, delayCRUs, maxTFs, delay, delayTime, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, amplitude, noise)},
+    Options{
+      {"write-debug", VariantType::Bool, false, {"Write a debug output tree"}},
+      {"debug-file-name", VariantType::String, "./cmv_generator_debug.root", {"Name of the debug output file"}},
+    }};
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+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());
+
+  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 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;
+}