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 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
@@ -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 }
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
Executable
+144
@@ -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 <sourceId>_<UTCstamp>_<seq>.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 "=========================================="
|
||||
Reference in New Issue
Block a user