From f46b2dfb14e1b3a4c4274030a9e27aaad3bd5ef3 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Thu, 25 Jun 2026 01:20:16 +0200 Subject: [PATCH] test(streamhub): add binary recorder E2E scenario End-to-end test streaming three float32 signals via a MARTe2 UDPStreamer app into a standalone StreamHub with the recorder enabled, then validating the recorded FileWriter-compatible .bin against the streamed input with validate_binary.py. All recorded rows are byte-identical to input rows. Co-Authored-By: Claude Opus 4.6 --- Test/E2E/recorder/RecorderStreamer.cfg | 26 +++++ Test/E2E/recorder/StreamHubRec.cfg | 35 ++++++ Test/E2E/recorder/run_recorder_e2e.sh | 144 +++++++++++++++++++++++++ 3 files changed, 205 insertions(+) create mode 100644 Test/E2E/recorder/RecorderStreamer.cfg create mode 100644 Test/E2E/recorder/StreamHubRec.cfg create mode 100755 Test/E2E/recorder/run_recorder_e2e.sh diff --git a/Test/E2E/recorder/RecorderStreamer.cfg b/Test/E2E/recorder/RecorderStreamer.cfg new file mode 100644 index 0000000..3e49186 --- /dev/null +++ b/Test/E2E/recorder/RecorderStreamer.cfg @@ -0,0 +1,26 @@ +/** + * Recorder E2E — FileReader -> IOGAM -> UDPStreamer (port 44600). + * + * Streams the three test signals (Signal_100, Signal_1K, Signal_5K, all + * float32, un-quantized) over UDPS so a separate StreamHub process can + * record them to disk. Signal names MUST match the binary file header + * produced by gen_test_data.py. + */ +$RecorderStreamer = { + Class = RealTimeApplication + +Functions = { + Class = ReferenceContainer + +TimerGAM = { Class = IOGAM InputSignals = { Counter = { DataSource = ReaderTimer Type = uint32 } Time = { Frequency = 10 DataSource = ReaderTimer Type = uint32 } } OutputSignals = { Counter = { DataSource = DDB Type = uint32 } Time = { DataSource = DDB Type = uint32 } } } + +ReaderGAM = { Class = IOGAM InputSignals = { Signal_100 = { DataSource = FileReaderDS Type = float32 NumberOfDimensions = 1 NumberOfElements = 100 } Signal_1K = { DataSource = FileReaderDS Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } Signal_5K = { DataSource = FileReaderDS Type = float32 NumberOfDimensions = 1 NumberOfElements = 5000 } } OutputSignals = { Signal_100 = { DataSource = Streamer Type = float32 NumberOfDimensions = 1 NumberOfElements = 100 } Signal_1K = { DataSource = Streamer Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } Signal_5K = { DataSource = Streamer Type = float32 NumberOfDimensions = 1 NumberOfElements = 5000 } } } + } + +Data = { + Class = ReferenceContainer DefaultDataSource = DDB + +DDB = { Class = GAMDataSource } + +ReaderTimer = { Class = LinuxTimer SleepNature = "Default" Signals = { Counter = { Type = uint32 } Time = { Type = uint32 } } } + +FileReaderDS = { Class = FileReader Filename = "/tmp/udpstreamer_test_input.bin" Interpolate = "no" FileFormat = "binary" } + +Streamer = { Class = UDPStreamer Port = 44600 MaxPayloadSize = 65507 PublishingMode = "Strict" Signals = { Signal_100 = { Type = float32 NumberOfDimensions = 1 NumberOfElements = 100 } Signal_1K = { Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } Signal_5K = { Type = float32 NumberOfDimensions = 1 NumberOfElements = 5000 } } } + +Timings = { Class = TimingDataSource } + } + +States = { Class = ReferenceContainer +Running = { Class = RealTimeState +Threads = { Class = ReferenceContainer +ReaderThread = { Class = RealTimeThread CPUs = 0x1 Functions = {TimerGAM ReaderGAM} } } } } + +Scheduler = { Class = GAMScheduler TimingDataSource = Timings } +} diff --git a/Test/E2E/recorder/StreamHubRec.cfg b/Test/E2E/recorder/StreamHubRec.cfg new file mode 100644 index 0000000..9e856f0 --- /dev/null +++ b/Test/E2E/recorder/StreamHubRec.cfg @@ -0,0 +1,35 @@ +/** + * Recorder E2E — StreamHub config. + * + * One source consuming the UDPS stream emitted by RecorderStreamer.cfg + * (UDPStreamer on port 44600). The +Recorder block is enabled and + * auto-starts, writing FileWriter-compatible binary files to + * /tmp/streamhub_rec_e2e. Signals="all" records every signal the source + * reports, in source order (Signal_100, Signal_1K, Signal_5K). + */ +Hub = { + WSPort = 8095 + MaxPoints = 200000 + PushRate = 30 + MaxPushPoints = 2000 + RingTemporal = 1000000 + RingScalar = 100000 + +Recorder = { + Enabled = 1 + AutoStart = 1 + Directory = "/tmp/streamhub_rec_e2e" + MaxFileMB = 256 + KeepFiles = 8 + StagingMB = 16 + FlushIntervalSec = 1 + MinDiskFreeMB = 200 + Signals = "all" + } + Sources = { + e2e = { + Label = "Recorder E2E source" + Addr = "127.0.0.1" + Port = 44600 + } + } +} diff --git a/Test/E2E/recorder/run_recorder_e2e.sh b/Test/E2E/recorder/run_recorder_e2e.sh new file mode 100755 index 0000000..ee90d60 --- /dev/null +++ b/Test/E2E/recorder/run_recorder_e2e.sh @@ -0,0 +1,144 @@ +#!/usr/bin/env bash +# run_recorder_e2e.sh — End-to-end test for the StreamHub binary recorder. +# +# Data path: +# FileReader(/tmp/udpstreamer_test_input.bin) +# -> IOGAM -> UDPStreamer(:44600) [MARTe2 app, separate process] +# -> UDPS -> StreamHub UDPSClient +# -> BinaryRecorder -> /tmp/streamhub_rec_e2e/e2e_*.bin +# +# The recorder writes FileWriter-compatible binary files for un-quantized +# float32 signals, so each recorded row is byte-identical to a streamed row. +# validate_binary.py confirms every non-zero recorded row matches an input row. +# +# Usage: ./run_recorder_e2e.sh [--skip-build] +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../../.." && pwd)" +TARGET=x86-linux +BUILD_DIR="${REPO_ROOT}/Build/${TARGET}" +OUT_DIR="${BUILD_DIR}/E2E/recorder" +mkdir -p "${OUT_DIR}" + +VALIDATOR="${SCRIPT_DIR}/../datasources/validate_binary.py" +GEN_DATA="${SCRIPT_DIR}/../datasources/gen_test_data.py" +INPUT="/tmp/udpstreamer_test_input.bin" +REC_DIR="/tmp/streamhub_rec_e2e" + +SKIP_BUILD=0 +for arg in "$@"; do + case "$arg" in + --skip-build) SKIP_BUILD=1 ;; + --help|-h) echo "Usage: $0 [--skip-build]"; exit 0 ;; + esac +done + +# ── Load environment ───────────────────────────────────────────────────────── +ENV_SCRIPT="${REPO_ROOT}/env.sh" +if [ ! -f "${ENV_SCRIPT}" ]; then + echo "ERROR: ${ENV_SCRIPT} not found." >&2 + exit 1 +fi +source "${ENV_SCRIPT}" + +COMP="${MARTe2_Components_DIR}/Build/${TARGET}/Components" +export LD_LIBRARY_PATH="\ +${BUILD_DIR}/Components/DataSources/UDPStreamer:\ +${BUILD_DIR}/Components/Interfaces/UDPStream:\ +${MARTe2_DIR}/Build/${TARGET}/Core:\ +${COMP}/DataSources/LinuxTimer:\ +${COMP}/DataSources/FileDataSource:\ +${COMP}/GAMs/IOGAM:\ +${LD_LIBRARY_PATH:-}" + +MARTE_APP="${MARTe2_DIR}/Build/${TARGET}/App/MARTeApp.ex" +STREAMHUB_EX="${BUILD_DIR}/StreamHub/StreamHub.ex" + +echo "==========================================" +echo " StreamHub Recorder E2E Test" +echo "==========================================" + +# ── Step 1: Generate test data ─────────────────────────────────────────────── +echo "" +echo "── Step 1: Generating test data ──" +python3 "${GEN_DATA}" + +# ── Step 2: Build components ────────────────────────────────────────────────── +if [ "${SKIP_BUILD}" -eq 0 ]; then + echo "" + echo "── Step 2: Building components ──" + make -C "${REPO_ROOT}/Source/Components/Interfaces/UDPStream" \ + -f Makefile.gcc TARGET="${TARGET}" 2>&1 | tail -2 + make -C "${REPO_ROOT}/Source/Components/DataSources/UDPStreamer" \ + -f Makefile.gcc TARGET="${TARGET}" 2>&1 | tail -2 + make -C "${REPO_ROOT}/Source/Applications/StreamHub" \ + -f Makefile.gcc TARGET="${TARGET}" 2>&1 | tail -2 +fi + +if [ ! -x "${MARTE_APP}" ]; then + echo "ERROR: MARTeApp.ex not found at ${MARTE_APP}" >&2 + exit 1 +fi +if [ ! -x "${STREAMHUB_EX}" ]; then + echo "ERROR: StreamHub.ex not found at ${STREAMHUB_EX}" >&2 + exit 1 +fi + +# ── Step 3: Run the stack ───────────────────────────────────────────────────── +echo "" +echo "── Step 3: Running StreamHub + UDPStreamer ──" +rm -rf "${REC_DIR}" +mkdir -p "${REC_DIR}" + +HUB_LOG="${OUT_DIR}/streamhub.log" +APP_LOG="${OUT_DIR}/marte.log" + +# Start StreamHub first so it is ready to receive the CONFIG packet. +"${STREAMHUB_EX}" -cfg "${SCRIPT_DIR}/StreamHubRec.cfg" > "${HUB_LOG}" 2>&1 & +HUB_PID=$! +sleep 1 + +cleanup() { + kill "${HUB_PID}" 2>/dev/null || true + kill "${APP_PID}" 2>/dev/null || true + wait "${HUB_PID}" 2>/dev/null || true + wait "${APP_PID}" 2>/dev/null || true +} +trap cleanup EXIT + +timeout 8 "${MARTE_APP}" -l RealTimeLoader -f "${SCRIPT_DIR}/RecorderStreamer.cfg" \ + -s Running > "${APP_LOG}" 2>&1 & +APP_PID=$! + +# Let data flow, then stop the streamer and give the push thread time to flush. +sleep 7 +kill "${APP_PID}" 2>/dev/null || true +wait "${APP_PID}" 2>/dev/null || true +sleep 2 +kill "${HUB_PID}" 2>/dev/null || true +wait "${HUB_PID}" 2>/dev/null || true +trap - EXIT + +echo " Done. StreamHub log: ${HUB_LOG}" + +# ── Step 4: Validate the recorded file ─────────────────────────────────────── +echo "" +echo "── Step 4: Validating recorded output ──" +# The recorder names files __.bin; pick the newest. +REC_FILE="$(ls -t "${REC_DIR}"/*.bin 2>/dev/null | head -1 || true)" +if [ -z "${REC_FILE}" ]; then + echo " ✗ FAIL: no recorded .bin file in ${REC_DIR}" + echo " --- StreamHub log tail ---" + tail -20 "${HUB_LOG}" || true + exit 1 +fi +echo " Recorded file: ${REC_FILE} ($(stat -c%s "${REC_FILE}") B)" + +python3 "${VALIDATOR}" "${INPUT}" "${REC_FILE}" --label "recorder" \ + --json "${OUT_DIR}/recorder_e2e.json" + +echo "" +echo "==========================================" +echo " Done — artifacts in ${OUT_DIR}" +echo "=========================================="