From 10dec4c5d203c00f1c4df7817403f6b159507019 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 08:58:49 +0200 Subject: [PATCH] test(e2e-chain): deterministic typed/shaped data generator Co-Authored-By: Claude Opus 4.6 --- Test/E2E/chain/gen_data.py | 174 +++++++++++++++++++++++++++++++++++++ 1 file changed, 174 insertions(+) create mode 100644 Test/E2E/chain/gen_data.py diff --git a/Test/E2E/chain/gen_data.py b/Test/E2E/chain/gen_data.py new file mode 100644 index 0000000..b82c04a --- /dev/null +++ b/Test/E2E/chain/gen_data.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +""" +gen_data.py — Deterministic typed/shaped data generator for the streaming-chain +E2E suite. + +For a scenario (see scenarios.py) it writes a MARTe2 FileReader-compatible binary +``input_.bin`` and returns a *ground-truth* dict the waveform validator uses +to reconstruct the expected stream without re-deriving the layout. + +MARTe2 binary format +-------------------- + [u32 numSigs] + per signal: [u16 TypeDescriptor.all][32B name (null-padded)][u32 numElements] + then NUM_ROWS rows, each row = all signals' elements concatenated, every value + little-endian at its native width. + +Ground-truth dict schema (keyed ":") +------------------------------------------------------ + { + "t": np.ndarray[float64] # intended sample time (s), flattened + "v": np.ndarray # native-dtype values, flattened + "dt": float # per-sample spacing (s) + "formula": str + "freq": float | None + "elements": int + "rows": int + "type": str + "quant": str + "range_min": float | None + "range_max": float | None + "is_time": bool + } + +Values are the *raw native* values fed to the FileReader. Wire-side quantisation +is performed by the UDPStreamer, not here — the validator applies the quant +tolerance using the recorded quant/range fields. +""" +import argparse +import os +import struct +import sys + +import numpy as np + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import scenarios as S # noqa: E402 (TYPE_CODES / NP_DTYPE / SCENARIOS) + +NUM_ROWS = 200 # producer cycles written to the FileReader input +ROW_DT = 1.0e-3 # seconds per producer cycle (row); 1 kHz producer + + +def _sample_dt(sig): + """Per-element time spacing (s) for a signal.""" + if sig["sampling_rate"]: + return 1.0 / sig["sampling_rate"] + e = sig["elements"] + return ROW_DT / e if e > 1 else ROW_DT + + +def _sample_times(sig): + """Flattened intended sample times (s): NUM_ROWS*elements values.""" + e = sig["elements"] + sdt = _sample_dt(sig) + rows = np.arange(NUM_ROWS, dtype=np.float64).reshape(-1, 1) * ROW_DT + cols = np.arange(e, dtype=np.float64).reshape(1, -1) * sdt + return (rows + cols).reshape(-1) + + +def _values(sig, t): + """Compute float64 values for flattened sample times ``t``.""" + f = sig["formula"] + e = sig["elements"] + idx = np.arange(t.size, dtype=np.float64) + if f == "sine": + freq = sig["freq"] if sig["freq"] else 1.0 + return np.sin(2.0 * np.pi * freq * t) + if f == "ramp": + # linear in the global element index, normalised to a modest range + return (idx % 1000.0) + if f == "counter": + return idx + if f == "time_ns": + return np.round(t * 1.0e9) + if f == "time_us": + return np.round(t * 1.0e6) + raise ValueError(f"unknown formula {f!r} for {sig['name']}") + + +def _native(sig, vals): + """Cast float64 values to the signal's native numpy dtype.""" + dt = S.NP_DTYPE[sig["type"]] + if sig["type"] in S.FLOAT_TYPES: + return vals.astype(dt) + # integer: round then cast (deterministic, no banker's rounding surprises) + return np.rint(vals).astype(dt) + + +def build_ground_truth(scenario): + """Return {":": gt_dict} for every signal in the scenario.""" + gt = {} + for src in scenario["sources"]: + for sig in src["signals"]: + t = _sample_times(sig) + v = _native(sig, _values(sig, t)) + gt[f"{src['id']}:{sig['name']}"] = { + "t": t, "v": v, "dt": _sample_dt(sig), + "formula": sig["formula"], "freq": sig["freq"], + "elements": sig["elements"], "rows": NUM_ROWS, + "type": sig["type"], "quant": sig["quant"], + "range_min": sig["range_min"], "range_max": sig["range_max"], + "is_time": sig["is_time"], + } + return gt + + +def write_input(scenario, path): + """Write the MARTe binary for *one* source and return its ground-truth dict. + + The MARTe FileReader reads a single flat row layout, so one input file maps + to one source. Multi-source scenarios call this once per source with distinct + paths (the orchestrator handles the per-source filename); here we write the + first source by default but accept an explicit ``src`` via the scenario when + a single source is present. + """ + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + gt = build_ground_truth(scenario) + + # write each source to its own file: for src[0], . else. + written = {} + for i, src in enumerate(scenario["sources"]): + p = path if i == 0 else f"{path}.{src['id']}" + _write_source_bin(src, p) + written[src["id"]] = p + gt["_files"] = written + return gt + + +def _write_source_bin(src, path): + sigs = src["signals"] + # per-signal native 2D arrays [NUM_ROWS, elements] + cols = {} + for sig in sigs: + t = _sample_times(sig) + v = _native(sig, _values(sig, t)) + cols[sig["name"]] = v.reshape(NUM_ROWS, sig["elements"]) + + with open(path, "wb") as f: + f.write(struct.pack("