test(e2e-chain): MARTe + StreamHub config generator
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
@@ -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 ``<input_bin>.<srcid>`` (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_<id>.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()
|
||||
Reference in New Issue
Block a user