Skip to content

Commit d2574d2

Browse files
ehellbarehellbar
committed
TPC: add test workflow to create dummy CMV data
1 parent d961ddf commit d2574d2

2 files changed

Lines changed: 311 additions & 0 deletions

File tree

‎Detectors/TPC/workflow/CMakeLists.txt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,11 @@ o2_add_executable(idc-test-ft
198198
SOURCES test/test_ft_EPN_Aggregator.cxx
199199
PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
200200

201+
o2_add_executable(cmv-test-generator
202+
COMPONENT_NAME tpc
203+
SOURCES test/test_cmv_generator.cxx
204+
PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
205+
201206
o2_add_executable(miptrack-filter
202207
COMPONENT_NAME tpc
203208
SOURCES src/tpc-miptrack-filter.cxx

‎Detectors/TPC/workflow/test/test_cmv_generator.cxx‎

Lines changed: 306 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,306 @@
1+
// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2+
// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3+
// All rights not expressly granted are reserved.
4+
//
5+
// This software is distributed under the terms of the GNU General Public
6+
// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7+
//
8+
// In applying this license CERN does not waive the privileges and immunities
9+
// granted to it by virtue of its status as an Intergovernmental Organization
10+
// or submit itself to any jurisdiction.
11+
12+
/// \file test_cmv_generator.cxx
13+
/// \brief DPL source workflow that generates dummy CMV data for testing the CMV FLP pipeline.
14+
///
15+
/// Replaces o2-tpc-cmv-to-vector in tests; directly emits CMVVECTOR and CMVORBITS
16+
/// messages per CRU per TF so the workflow can be piped straight into o2-tpc-cmv-flp:
17+
///
18+
/// o2-tpc-cmv-test-generator --crus 0-359 --timeframes 100 \
19+
/// | o2-tpc-cmv-flp --crus 0-359 --n-TFs-buffer 10 \
20+
/// | o2-dpl-output-proxy --dataspec "downstream:TPC/CMVGROUP;downstream:TPC/CMVORBITINFO" ...
21+
///
22+
/// \author Ernst Hellbar <ernst.hellbar@cern.ch>
23+
24+
#include "Framework/DataProcessorSpec.h"
25+
#include "Framework/Task.h"
26+
#include "Framework/ControlService.h"
27+
#include "Framework/ConfigParamRegistry.h"
28+
#include "Framework/ConfigParamSpec.h"
29+
#include "Framework/Logger.h"
30+
#include "Headers/DataHeader.h"
31+
#include "Algorithm/RangeTokenizer.h"
32+
#include "TPCBase/CRU.h"
33+
#include "DataFormatsTPC/CMV.h"
34+
#include "TPCWorkflow/ProcessingHelpers.h"
35+
#include "CommonUtils/TreeStreamRedirector.h"
36+
#include "CommonUtils/ConfigurableParam.h"
37+
#include "DetectorsRaw/HBFUtilsInitializer.h"
38+
#include "DetectorsRaw/HBFUtils.h"
39+
#include <fmt/format.h>
40+
#include <fmt/ranges.h>
41+
42+
#include <vector>
43+
#include <chrono>
44+
#include <thread>
45+
#include <cmath>
46+
#include <memory>
47+
#include <limits>
48+
#include <random>
49+
#include <unordered_set>
50+
51+
using namespace o2::framework;
52+
using o2::header::gDataOriginTPC;
53+
54+
// ─────────────────────────────────────────────────────────────────────────────
55+
// workflow options
56+
// ─────────────────────────────────────────────────────────────────────────────
57+
void customize(std::vector<ConfigParamSpec>& workflowOptions)
58+
{
59+
const std::string cruDefault = "0-" + std::to_string(o2::tpc::CRU::MaxCRU - 1);
60+
std::vector<ConfigParamSpec> options{
61+
{"crus", VariantType::String, cruDefault.c_str(), {"List of CRUs, comma-separated ranges, e.g. 0-3,7,9-15"}},
62+
{"timeframes", VariantType::Int, 100, {"Number of TFs to generate; use -1 to run indefinitely"}},
63+
{"delay", VariantType::Bool, false, {"Add delay after sending all CRUs"}},
64+
{"delayTime", VariantType::Int, 1, {"Duration of the global per-TF delay in ms (requires --delay true)"}},
65+
{"delayCRUs", VariantType::String, "", {"CRUs for which to add an extra per-CRU delay before sending, comma-separated ranges"}},
66+
{"delayTimeCRUs", VariantType::Int, 1, {"Duration of the per-CRU delay in ms (requires --delayCRUs)"}},
67+
{"dropTFsRandom", VariantType::Int, 0, {"Drop a whole TF randomly: on average one every N TFs (0 = disabled)"}},
68+
{"dropTFsRange", VariantType::String, "", {"Drop all TFs in this range, e.g. 10-12"}},
69+
{"seed", VariantType::Int, 42, {"RNG seed for CMV value generation"}},
70+
{"amplitude", VariantType::Float, 5.0f, {"Amplitude of the sinusoidal CMV signal (ADC units)"}},
71+
{"noise", VariantType::Float, 1.0f, {"Gaussian noise std-dev added per time bin (ADC units)"}},
72+
{"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings"}}};
73+
o2::raw::HBFUtilsInitializer::addConfigOption(options);
74+
std::swap(workflowOptions, options);
75+
}
76+
77+
#include "Framework/runDataProcessing.h"
78+
79+
// ─────────────────────────────────────────────────────────────────────────────
80+
// generator device
81+
// ─────────────────────────────────────────────────────────────────────────────
82+
class CMVGeneratorDevice : public o2::framework::Task
83+
{
84+
public:
85+
static constexpr uint32_t sOrbitsPerPacket = 8; ///< each CMV packet covers 8 heartbeat orbits
86+
87+
CMVGeneratorDevice(const std::vector<uint32_t>& crus,
88+
const std::unordered_set<uint32_t>& delayCRUs,
89+
unsigned int maxTFs,
90+
bool delay,
91+
int delayTime,
92+
int delayTimeCRUs,
93+
int dropTFsRandom,
94+
const std::vector<int>& rangeTFsDrop,
95+
float amplitude,
96+
float noise,
97+
int seed)
98+
: mCRUs(crus), mDelayCRUs(delayCRUs), mMaxTFs(maxTFs), mDelay(delay), mDelayTime(delayTime), mDelayTimeCRUs(delayTimeCRUs), mDropTFsRandom(dropTFsRandom), mRangeTFsDrop(rangeTFsDrop), mAmplitude(amplitude), mNoise(noise), mRng(static_cast<std::mt19937::result_type>(seed)) {}
99+
100+
void init(o2::framework::InitContext& ic) final
101+
{
102+
mTimer100TFs = std::chrono::high_resolution_clock::now();
103+
104+
if (!mCRUs.empty()) {
105+
LOGP(info, "crus: {}", fmt::join(mCRUs, ", "));
106+
}
107+
if (!mDelayCRUs.empty()) {
108+
const std::vector<uint32_t> delayCRUsSorted(mDelayCRUs.begin(), mDelayCRUs.end());
109+
LOGP(info, "delayCRUs: {}", fmt::join(delayCRUsSorted, ", "));
110+
}
111+
112+
mWriteDebug = ic.options().get<bool>("write-debug");
113+
if (mWriteDebug) {
114+
mDebugStreamFileName = ic.options().get<std::string>("debug-file-name");
115+
LOGP(info, "Creating debug stream {}", mDebugStreamFileName);
116+
mDebugStream = std::make_unique<o2::utils::TreeStreamRedirector>(mDebugStreamFileName.data(), "recreate");
117+
}
118+
}
119+
120+
void run(o2::framework::ProcessingContext& ctx) final
121+
{
122+
using timer = std::chrono::high_resolution_clock;
123+
const auto tf = o2::tpc::processing_helpers::getCurrentTF(ctx);
124+
125+
// ── TF dropping ──────────────────────────────────────────────────────────
126+
// Note: RangeTokenizer guarantees sorted output, so front()/back() are min/max.
127+
if (!mRangeTFsDrop.empty() && tf >= static_cast<uint32_t>(mRangeTFsDrop.front()) && tf <= static_cast<uint32_t>(mRangeTFsDrop.back())) {
128+
LOGP(info, "Dropping TF {} (range drop)", tf);
129+
return;
130+
}
131+
if (mDropTFsRandom > 0 && std::uniform_int_distribution<int>{0, mDropTFsRandom - 1}(mRng) == 0) {
132+
LOGP(info, "Dropping TF {} (random drop)", tf);
133+
return;
134+
}
135+
136+
auto start = timer::now();
137+
138+
// ── CMV values (generated once per TF, reused for all CRUs) ─────────────
139+
// NTimeBinsPerTF = NPacketsPerTFPerCRU (4) * NTimeBinsPerPacket (3564) = 14256
140+
// Using std::mt19937 + std::normal_distribution for fast noise generation.
141+
// All CRUs intentionally share the same noise realization to avoid 360x RNG overhead.
142+
const float signal = mAmplitude * std::sin(tf * 0.05f);
143+
std::normal_distribution<float> noiseDist{0.f, mNoise};
144+
std::vector<uint16_t> cmvVec(o2::tpc::cmv::NTimeBinsPerTF);
145+
for (auto& v : cmvVec) {
146+
o2::tpc::cmv::Data d;
147+
d.setCMVFloat(signal + noiseDist(mRng));
148+
v = d.getCMV();
149+
}
150+
151+
// ── Orbit / BC info (same for all CRUs) ──────────────────────────────────
152+
// One packed (orbit<<32|bc) entry per CMV packet (4 per TF).
153+
// Each packet covers 8 heartbeat orbits (NTimeBinsPerPacket = 3564 = 8 LHC orbits),
154+
// so the orbit advances by 8 per packet and by NPacketsPerTFPerCRU*8 = 32 per TF.
155+
std::vector<uint64_t> orbitBCVec(o2::tpc::cmv::NPacketsPerTFPerCRU);
156+
for (uint32_t pkt = 0; pkt < o2::tpc::cmv::NPacketsPerTFPerCRU; ++pkt) {
157+
const uint32_t orbit = static_cast<uint32_t>(tf * o2::tpc::cmv::NPacketsPerTFPerCRU * sOrbitsPerPacket + pkt * sOrbitsPerPacket);
158+
orbitBCVec[pkt] = uint64_t(orbit) << 32; // bc = 0
159+
}
160+
161+
for (const auto cru : mCRUs) {
162+
const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
163+
164+
// ── per-CRU delay ────────────────────────────────────────────────────
165+
if (mDelayCRUs.count(cru)) {
166+
std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTimeCRUs));
167+
}
168+
169+
ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVVECTOR", subSpec}, cmvVec);
170+
ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVORBITS", subSpec}, orbitBCVec);
171+
172+
if (mWriteDebug) {
173+
auto& stream = (*mDebugStream) << "cmvs";
174+
stream << "cru=" << cru
175+
<< "tfCounter=" << tf
176+
<< "nCMVs=" << cmvVec.size()
177+
<< "cmvs=" << cmvVec
178+
<< "\n";
179+
}
180+
}
181+
182+
if (!(tf % 100)) {
183+
const auto elapsed100 = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - mTimer100TFs).count();
184+
LOGP(info, "Generated CMV data for TF {} ({} ms for last 100 TFs)", tf, elapsed100);
185+
mTimer100TFs = timer::now();
186+
}
187+
188+
// ── global delay ─────────────────────────────────────────────────────────
189+
if (mDelay) {
190+
auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - start).count();
191+
if (elapsed < mDelayTime) {
192+
std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTime - elapsed));
193+
}
194+
}
195+
196+
// endOfStream() propagates the EoS signal to downstream devices (required for source devices).
197+
if (mMaxTFs != std::numeric_limits<unsigned int>::max() && tf >= mMaxTFs - 1) {
198+
ctx.services().get<ControlService>().endOfStream();
199+
ctx.services().get<ControlService>().readyToQuit(QuitRequest::Me);
200+
}
201+
}
202+
203+
void endOfStream(o2::framework::EndOfStreamContext&) final { closeFiles(); }
204+
void stop() final { closeFiles(); }
205+
206+
private:
207+
void closeFiles()
208+
{
209+
if (mDebugStream) {
210+
auto& stream = (*mDebugStream) << "cmvs";
211+
auto& tree = stream.getTree();
212+
tree.SetAlias("sector", "int(cru/10)");
213+
mDebugStream->Close();
214+
mDebugStream.reset(nullptr);
215+
}
216+
}
217+
218+
const std::vector<uint32_t> mCRUs{};
219+
const std::unordered_set<uint32_t> mDelayCRUs{};
220+
const unsigned int mMaxTFs{};
221+
const bool mDelay{false};
222+
const int mDelayTime{1};
223+
const int mDelayTimeCRUs{1};
224+
const int mDropTFsRandom{0};
225+
const std::vector<int> mRangeTFsDrop{};
226+
const float mAmplitude{5.f};
227+
const float mNoise{1.f};
228+
std::mt19937 mRng{};
229+
std::chrono::high_resolution_clock::time_point mTimer100TFs{};
230+
bool mWriteDebug{false};
231+
std::string mDebugStreamFileName{};
232+
std::unique_ptr<o2::utils::TreeStreamRedirector> mDebugStream{};
233+
};
234+
235+
// ─────────────────────────────────────────────────────────────────────────────
236+
DataProcessorSpec generateCMVsCRU(const std::vector<uint32_t>& crus,
237+
const std::unordered_set<uint32_t>& delayCRUs,
238+
unsigned int maxTFs,
239+
bool delay,
240+
int delayTime,
241+
int delayTimeCRUs,
242+
int dropTFsRandom,
243+
const std::vector<int>& rangeTFsDrop,
244+
float amplitude,
245+
float noise,
246+
int seed)
247+
{
248+
std::vector<OutputSpec> outputSpecs;
249+
outputSpecs.reserve(crus.size() * 2);
250+
for (const auto cru : crus) {
251+
const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
252+
outputSpecs.emplace_back(gDataOriginTPC, "CMVVECTOR", subSpec, Lifetime::Timeframe);
253+
outputSpecs.emplace_back(gDataOriginTPC, "CMVORBITS", subSpec, Lifetime::Timeframe);
254+
}
255+
256+
return DataProcessorSpec{
257+
"tpc-cmv-generator",
258+
Inputs{},
259+
outputSpecs,
260+
AlgorithmSpec{adaptFromTask<CMVGeneratorDevice>(crus, delayCRUs, maxTFs, delay, delayTime, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, amplitude, noise, seed)},
261+
Options{
262+
{"write-debug", VariantType::Bool, false, {"Write a debug output tree"}},
263+
{"debug-file-name", VariantType::String, "./cmv_generator_debug.root", {"Name of the debug output file"}},
264+
}};
265+
}
266+
267+
// ─────────────────────────────────────────────────────────────────────────────
268+
WorkflowSpec defineDataProcessing(ConfigContext const& config)
269+
{
270+
const auto tpcCRUs = o2::RangeTokenizer::tokenize<int>(config.options().get<std::string>("crus"));
271+
const std::vector<uint32_t> crus(tpcCRUs.begin(), tpcCRUs.end());
272+
273+
const auto delayCRUsStr = config.options().get<std::string>("delayCRUs");
274+
std::unordered_set<uint32_t> delayCRUs;
275+
if (!delayCRUsStr.empty()) {
276+
for (const auto cru : o2::RangeTokenizer::tokenize<int>(delayCRUsStr)) {
277+
delayCRUs.insert(static_cast<uint32_t>(cru));
278+
}
279+
}
280+
281+
const auto dropTFsRangeStr = config.options().get<std::string>("dropTFsRange");
282+
const auto rangeTFsDrop = dropTFsRangeStr.empty() ? std::vector<int>{} : o2::RangeTokenizer::tokenize<int>(dropTFsRangeStr);
283+
const int timeframesInt = config.options().get<int>("timeframes");
284+
// -1 means run indefinitely; map to UINT_MAX so the termination check never fires.
285+
const auto timeframes = (timeframesInt < 0) ? std::numeric_limits<unsigned int>::max() : static_cast<unsigned int>(timeframesInt);
286+
const auto delay = config.options().get<bool>("delay");
287+
const auto delayTime = config.options().get<int>("delayTime");
288+
const auto delayTimeCRUs = config.options().get<int>("delayTimeCRUs");
289+
const auto dropTFsRandom = config.options().get<int>("dropTFsRandom");
290+
const auto seed = config.options().get<int>("seed");
291+
const auto amplitude = config.options().get<float>("amplitude");
292+
const auto noise = config.options().get<float>("noise");
293+
294+
o2::conf::ConfigurableParam::updateFromString(config.options().get<std::string>("configKeyValues"));
295+
296+
WorkflowSpec workflow;
297+
workflow.emplace_back(generateCMVsCRU(crus, delayCRUs, timeframes, delay, delayTime, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, amplitude, noise, seed));
298+
299+
auto& hbfu = o2::raw::HBFUtils::Instance();
300+
long startTime = hbfu.startTime > 0 ? hbfu.startTime : std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now()).time_since_epoch().count();
301+
o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.startTime={}", startTime).data());
302+
o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.nHBFPerTF={}", hbfu.nHBFPerTF).data());
303+
o2::raw::HBFUtilsInitializer hbfIni(config, workflow);
304+
305+
return workflow;
306+
}

0 commit comments

Comments
 (0)