From f146fc1016f71863723f91b4e4715aa822712bc7 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 09:01:20 +0200 Subject: [PATCH] test(e2e-chain): MARTe + StreamHub config generator Co-Authored-By: Claude Opus 4.6 --- Test/E2E/chain/gen_cfg.py | 202 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 Test/E2E/chain/gen_cfg.py diff --git a/Test/E2E/chain/gen_cfg.py b/Test/E2E/chain/gen_cfg.py new file mode 100644 index 0000000..f192121 --- /dev/null +++ b/Test/E2E/chain/gen_cfg.py @@ -0,0 +1,202 @@ +#!/usr/bin/env python3 +""" +gen_cfg.py — MARTe2 app + StreamHub config generator for the streaming-chain E2E. + +Given a scenario (scenarios.py) it produces two config files: + + * MARTe app cfg: ``LinuxTimer + FileReader(input) -> IOGAM -> UDPStreamer`` per + source. When ``oracle in {fed,both}`` a second IOGAM branch taps the same fed + signals into a ``FileWriter`` (the "fed reference"). + * StreamHub cfg: one ``Source`` per UDPStreamer (unicast or multicast) on the + scenario's ``ws_port``. + +The producer rate (LinuxTimer Frequency) defaults to 1 kHz; the FileReader uses +``EOF = "Rewind"`` so the NUM_ROWS-row input loops for the whole run. The +validator reconstructs truth as the cyclic ground-truth sequence, so the wrap is +expected, not an error. +""" +import argparse +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import scenarios as S # noqa: E402 + +PRODUCER_HZ = 1000 # LinuxTimer frequency (Hz) + + +def _ndims(elements): + return 0 if elements == 1 else 1 + + +def _gam_sig(sig, datasource): + """A GAM signal entry referencing a DataSource (no UDPStreamer extras).""" + return (f"{sig['name']} = {{ Type = {sig['type']} " + f"NumberOfDimensions = {_ndims(sig['elements'])} " + f"NumberOfElements = {sig['elements']} DataSource = {datasource} }}") + + +def _streamer_sig(sig): + """A UDPStreamer DataSource signal entry (carries time/quant/unit options).""" + parts = [f"Type = {sig['type']}", + f"NumberOfDimensions = {_ndims(sig['elements'])}", + f"NumberOfElements = {sig['elements']}"] + if sig["time_mode"] and sig["time_mode"] != "PacketTime": + parts.append(f'TimeMode = "{sig["time_mode"]}"') + if sig["time_signal"]: + parts.append(f'TimeSignal = "{sig["time_signal"]}"') + if sig["sampling_rate"]: + parts.append(f"SamplingRate = {sig['sampling_rate']}") + if sig["quant"] and sig["quant"] != "none": + parts.append(f'QuantizedType = "{sig["quant"]}"') + parts.append(f"RangeMin = {sig['range_min']}") + parts.append(f"RangeMax = {sig['range_max']}") + if sig["unit"]: + parts.append(f'Unit = "{sig["unit"]}"') + return f"{sig['name']} = {{ {' '.join(parts)} }}" + + +def _streamer_block(src, scenario): + s = src["signals"] + parts = [f"Port = {src['udp_port']}", + f"MaxPayloadSize = {scenario['max_payload']}", + f'PublishingMode = "{scenario["publishing"]}"'] + if scenario["publishing"] == "Accumulate": + parts.append(f"MinRefreshRate = {scenario['min_refresh_hz']}") + if scenario["publishing"] == "Decimate": + parts.append(f"Ratio = {scenario['ratio']}") + if scenario["network"] == "multicast": + parts.append(f'MulticastGroup = "{src["multicast_group"]}"') + parts.append(f"DataPort = {src['data_port']}") + sigs = " ".join(_streamer_sig(sig) for sig in s) + return (f" +Streamer_{src['id']} = {{ Class = UDPStreamer " + f"{' '.join(parts)} Signals = {{ {sigs} }} }}") + + +def write_marte_cfg(scenario, path, input_bin, tap_bin=None): + """Write the MARTe app cfg. ``input_bin`` is the src[0] file; additional + sources read ``.`` (matching gen_data.write_input).""" + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + srcs = scenario["sources"] + want_tap = scenario["oracle"] in ("fed", "both") and tap_bin is not None + + gams = [] # +Functions entries + datas = [] # +Data entries + thread_funcs = ["TimerGAM"] + + # Timer GAM (drives the schedule) + gams.append( + " +TimerGAM = { Class = IOGAM " + "InputSignals = { Counter = { DataSource = ReaderTimer Type = uint32 } " + f"Time = {{ Frequency = {PRODUCER_HZ} DataSource = ReaderTimer Type = uint32 }} }} " + "OutputSignals = { Counter = { DataSource = DDB Type = uint32 } " + "Time = { DataSource = DDB Type = uint32 } } }") + + for i, src in enumerate(srcs): + sid = src["id"] + fpath = input_bin if i == 0 else f"{input_bin}.{sid}" + rds = f"FileReaderDS_{sid}" + in_sigs = " ".join(_gam_sig(sig, rds) for sig in src["signals"]) + out_sigs = " ".join(_gam_sig(sig, f"Streamer_{sid}") for sig in src["signals"]) + gams.append( + f" +ReaderGAM_{sid} = {{ Class = IOGAM " + f"InputSignals = {{ {in_sigs} }} OutputSignals = {{ {out_sigs} }} }}") + thread_funcs.append(f"ReaderGAM_{sid}") + datas.append( + f' +{rds} = {{ Class = FileReader Filename = "{fpath}" ' + f'Interpolate = "no" FileFormat = "binary" EOF = "Rewind" }}') + datas.append(_streamer_block(src, scenario)) + + if want_tap: + # tap the first source's signals to a FileWriter (fed reference); + # read again from the FileReader DS (a second consumer is allowed) + src = srcs[0] + tap_in = " ".join(_gam_sig(sig, f"FileReaderDS_{src['id']}") + for sig in src["signals"]) + tap_out = " ".join(_gam_sig(sig, "TapWriterDS") for sig in src["signals"]) + gams.append( + f" +TapGAM = {{ Class = IOGAM " + f"InputSignals = {{ {tap_in} }} OutputSignals = {{ {tap_out} }} }}") + thread_funcs.append("TapGAM") + tap_sigs = " ".join(_gam_sig(sig, "TapWriterDS").replace( + " DataSource = TapWriterDS", "") for sig in src["signals"]) + datas.append( + f' +TapWriterDS = {{ Class = FileWriter NumberOfBuffers = 10 ' + f'CPUMask = 0x4 StackSize = 10000000 Filename = "{tap_bin}" ' + f'Overwrite = "yes" StoreOnTrigger = 0 FileFormat = "binary" ' + f"Signals = {{ {tap_sigs} }} }}") + + funcs_block = "\n".join(gams) + data_block = "\n".join(datas) + thread_list = " ".join(thread_funcs) + + cfg = f"""$ChainE2E = {{ + Class = RealTimeApplication + +Functions = {{ + Class = ReferenceContainer +{funcs_block} + }} + +Data = {{ + Class = ReferenceContainer DefaultDataSource = DDB + +DDB = {{ Class = GAMDataSource }} + +ReaderTimer = {{ Class = LinuxTimer SleepNature = "Default" Signals = {{ Counter = {{ Type = uint32 }} Time = {{ Type = uint32 }} }} }} +{data_block} + +Timings = {{ Class = TimingDataSource }} + }} + +States = {{ Class = ReferenceContainer +Running = {{ Class = RealTimeState +Threads = {{ Class = ReferenceContainer +ReaderThread = {{ Class = RealTimeThread CPUs = 0x1 Functions = {{{thread_list}}} }} }} }} }} + +Scheduler = {{ Class = GAMScheduler TimingDataSource = Timings }} +}} +""" + with open(path, "w") as f: + f.write(cfg) + return path + + +def write_hub_cfg(scenario, path): + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + src_blocks = [] + for src in scenario["sources"]: + parts = [f'Label = "chain {src["id"]}"', 'Addr = "127.0.0.1"', + f"Port = {src['udp_port']}"] + if scenario["network"] == "multicast": + parts.append(f'MulticastGroup = "{src["multicast_group"]}"') + parts.append(f"DataPort = {src['data_port']}") + src_blocks.append(f" {src['id']} = {{ {' '.join(parts)} }}") + sources = "\n".join(src_blocks) + cfg = f"""Hub = {{ + WSPort = {scenario['ws_port']} + MaxPoints = 200000 + PushRate = 30 + MaxPushPoints = 2000 + RingTemporal = 1000000 + RingScalar = 100000 + Sources = {{ +{sources} + }} +}} +""" + with open(path, "w") as f: + f.write(cfg) + return path + + +def main(): + p = argparse.ArgumentParser(description="Generate MARTe + StreamHub cfgs") + p.add_argument("--scenario", required=True) + p.add_argument("--input", required=True, help="input_.bin path") + p.add_argument("--marte-out", required=True) + p.add_argument("--hub-out", required=True) + p.add_argument("--tap", default=None) + args = p.parse_args() + sc = next((s for s in S.SCENARIOS if s["id"] == args.scenario), None) + if sc is None: + print(f"unknown scenario {args.scenario}", file=sys.stderr) + sys.exit(2) + write_marte_cfg(sc, args.marte_out, args.input, args.tap) + write_hub_cfg(sc, args.hub_out) + print(f"marte: {args.marte_out}") + print(f"hub: {args.hub_out}") + + +if __name__ == "__main__": + main()