frank/ruvnet--RuView
1
0
Fork 0
mirror of https://github.com/ruvnet/RuView synced 2026-06-23 12:33:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(firmware): QEMU ESP32-S3 testing platform (ADR-061)

Browse Source 
Implement full QEMU emulation framework for firmware testing without
physical hardware:

Mock CSI Generator (mock_csi.c):
- 10 test scenarios: empty room, static/walking person, fall, multi-person,
  channel sweep, MAC filter, ring overflow, boundary RSSI, zero-length
- Physics-based signal model with breathing modulation and Doppler
- LFSR pseudo-random noise, CONFIG_CSI_MOCK_ENABLED Kconfig guard
- Scenario 255 runs all sequentially

QEMU Runner & CI:
- qemu-esp32s3-test.sh: build, merge flash image, run QEMU, validate
- validate_qemu_output.py: 14 automated checks (boot, NVS, edge, vitals,
  crash detection) with colored output and severity-based exit codes
- generate_nvs_matrix.py: 14 NVS provisioning configs for matrix testing
- firmware-qemu.yml: GitHub Actions CI with 4-scenario matrix

Fuzz Testing:
- 3 libFuzzer targets: CSI serialize, NVS config validation, ring buffer
- Host-compilable ESP-IDF stubs (no ESP-IDF dependency for fuzzing)
- 6 seed corpus files for guided fuzzing
- Makefile with ASAN + UBSAN sanitizers

Documentation:
- firmware/esp32-csi-node/README.md: comprehensive QEMU testing guide
- Root README.md: collapsed QEMU testing section

Build verified: normal firmware build (RC=0) with mock_csi excluded.

Closes #259

Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv
2026-03-13 09:17:07 -04:00
parent a467dfed9f
commit ffeaa46bc6
33 changed files with 3274 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_nvs_matrix.py
+410
View File
@@ -0,0 +1,410 @@
#!/usr/bin/env python3
"""
NVS Test Matrix Generator (ADR-061)
Generates NVS partition binaries for 14 test configurations using the
provision.py script's CSV builder and NVS binary generator. Each binary
can be injected into a QEMU flash image at offset 0x9000 for automated
firmware testing under different NVS configurations.
Usage:
python3 generate_nvs_matrix.py --output-dir build/nvs_matrix
# Generate only specific configs:
python3 generate_nvs_matrix.py --output-dir build/nvs_matrix --only default,full-adr060
Requirements:
- esp_idf_nvs_partition_gen (pip install) or ESP-IDF nvs_partition_gen.py
- Python 3.8+
"""
import argparse
import csv
import io
import os
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple
# NVS partition size must match partitions_display.csv: 0x6000 = 24576 bytes
NVS_PARTITION_SIZE = 0x6000
@dataclass
class NvsEntry:
"""A single NVS key-value entry."""
key: str
type: str # "data" or "namespace"
encoding: str # "string", "u8", "u16", "u32", "hex2bin", ""
value: str
@dataclass
class NvsConfig:
"""A named NVS configuration with a list of entries."""
name: str
description: str
entries: List[NvsEntry] = field(default_factory=list)
def to_csv(self) -> str:
"""Generate NVS CSV content."""
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["key", "type", "encoding", "value"])
writer.writerow(["csi_cfg", "namespace", "", ""])
for entry in self.entries:
writer.writerow([entry.key, entry.type, entry.encoding, entry.value])
return buf.getvalue()
def define_configs() -> List[NvsConfig]:
"""Define all 14 NVS test configurations."""
configs = []
# 1. default - no NVS entries (firmware uses Kconfig defaults)
configs.append(NvsConfig(
name="default",
description="No NVS entries; firmware uses Kconfig defaults",
entries=[],
))
# 2. wifi-only - just WiFi credentials
configs.append(NvsConfig(
name="wifi-only",
description="WiFi SSID and password only",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
],
))
# 3. full-adr060 - channel override + MAC filter
configs.append(NvsConfig(
name="full-adr060",
description="ADR-060: channel override + MAC filter + full config",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("target_port", "data", "u16", "5005"),
NvsEntry("node_id", "data", "u8", "1"),
NvsEntry("csi_channel", "data", "u8", "6"),
NvsEntry("filter_mac", "data", "hex2bin", "aabbccddeeff"),
],
))
# 4. edge-tier0 - raw passthrough (no DSP)
configs.append(NvsConfig(
name="edge-tier0",
description="Edge tier 0: raw CSI passthrough, no on-device DSP",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "0"),
],
))
# 5. edge-tier1 - basic presence/motion detection
configs.append(NvsConfig(
name="edge-tier1",
description="Edge tier 1: basic presence and motion detection",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "1"),
NvsEntry("pres_thresh", "data", "u16", "50"),
],
))
# 6. edge-tier2-custom - full pipeline with custom thresholds
configs.append(NvsConfig(
name="edge-tier2-custom",
description="Edge tier 2: full pipeline with custom thresholds",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "2"),
NvsEntry("pres_thresh", "data", "u16", "100"),
NvsEntry("fall_thresh", "data", "u16", "3000"),
NvsEntry("vital_win", "data", "u16", "512"),
NvsEntry("vital_int", "data", "u16", "500"),
NvsEntry("subk_count", "data", "u8", "16"),
],
))
# 7. tdm-3node - TDM mesh with 3 nodes (slot 0)
configs.append(NvsConfig(
name="tdm-3node",
description="TDM mesh: 3-node schedule, this node is slot 0",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("node_id", "data", "u8", "0"),
NvsEntry("tdm_slot", "data", "u8", "0"),
NvsEntry("tdm_nodes", "data", "u8", "3"),
],
))
# 8. wasm-signed - WASM runtime with signature verification
configs.append(NvsConfig(
name="wasm-signed",
description="WASM runtime enabled with Ed25519 signature verification",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "2"),
],
))
# 9. wasm-unsigned - WASM runtime without signature verification
configs.append(NvsConfig(
name="wasm-unsigned",
description="WASM runtime with signature verification disabled",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "2"),
],
))
# 10. 5ghz-channel - 5 GHz channel override
configs.append(NvsConfig(
name="5ghz-channel",
description="ADR-060: 5 GHz channel 36 override",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork5G"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("csi_channel", "data", "u8", "36"),
],
))
# 11. boundary-max - maximum values for all numeric fields
configs.append(NvsConfig(
name="boundary-max",
description="Boundary test: maximum values for all numeric NVS fields",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("target_port", "data", "u16", "65535"),
NvsEntry("node_id", "data", "u8", "255"),
NvsEntry("edge_tier", "data", "u8", "2"),
NvsEntry("pres_thresh", "data", "u16", "65535"),
NvsEntry("fall_thresh", "data", "u16", "65535"),
NvsEntry("vital_win", "data", "u16", "65535"),
NvsEntry("vital_int", "data", "u16", "10000"),
NvsEntry("subk_count", "data", "u8", "32"),
],
))
# 12. boundary-min - minimum values for all numeric fields
configs.append(NvsConfig(
name="boundary-min",
description="Boundary test: minimum values for all numeric NVS fields",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("target_port", "data", "u16", "1024"),
NvsEntry("node_id", "data", "u8", "0"),
NvsEntry("edge_tier", "data", "u8", "0"),
NvsEntry("pres_thresh", "data", "u16", "1"),
NvsEntry("fall_thresh", "data", "u16", "1"),
NvsEntry("vital_win", "data", "u16", "1"),
NvsEntry("vital_int", "data", "u16", "100"),
NvsEntry("subk_count", "data", "u8", "1"),
],
))
# 13. power-save - low power duty cycle configuration
configs.append(NvsConfig(
name="power-save",
description="Power-save mode: 10% duty cycle for battery-powered nodes",
entries=[
NvsEntry("ssid", "data", "string", "TestNetwork"),
NvsEntry("password", "data", "string", "testpass123"),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
NvsEntry("edge_tier", "data", "u8", "1"),
],
))
# 14. empty-strings - empty SSID/password to test fallback to Kconfig
configs.append(NvsConfig(
name="empty-strings",
description="Empty SSID and password to verify Kconfig fallback",
entries=[
NvsEntry("ssid", "data", "string", ""),
NvsEntry("password", "data", "string", ""),
NvsEntry("target_ip", "data", "string", "10.0.2.2"),
],
))
return configs
def generate_nvs_binary(csv_content: str, size: int) -> bytes:
"""Generate an NVS partition binary from CSV content.
Tries multiple methods to find nvs_partition_gen:
1. esp_idf_nvs_partition_gen pip package
2. Legacy nvs_partition_gen pip package
3. ESP-IDF bundled script (via IDF_PATH)
4. Module invocation
"""
import subprocess
import tempfile
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f_csv:
f_csv.write(csv_content)
csv_path = f_csv.name
bin_path = csv_path.replace(".csv", ".bin")
try:
# Try pip-installed version first
try:
from esp_idf_nvs_partition_gen import nvs_partition_gen
nvs_partition_gen.generate(csv_path, bin_path, size)
with open(bin_path, "rb") as f:
return f.read()
except ImportError:
pass
# Try legacy import
try:
import nvs_partition_gen
nvs_partition_gen.generate(csv_path, bin_path, size)
with open(bin_path, "rb") as f:
return f.read()
except ImportError:
pass
# Try ESP-IDF bundled script
idf_path = os.environ.get("IDF_PATH", "")
gen_script = os.path.join(
idf_path, "components", "nvs_flash",
"nvs_partition_generator", "nvs_partition_gen.py"
)
if os.path.isfile(gen_script):
subprocess.check_call([
sys.executable, gen_script, "generate",
csv_path, bin_path, hex(size)
])
with open(bin_path, "rb") as f:
return f.read()
# Last resort: try as a module
subprocess.check_call([
sys.executable, "-m", "nvs_partition_gen", "generate",
csv_path, bin_path, hex(size)
])
with open(bin_path, "rb") as f:
return f.read()
finally:
for p in (csv_path, bin_path):
if os.path.isfile(p):
os.unlink(p)
def main():
parser = argparse.ArgumentParser(
description="Generate NVS partition binaries for QEMU firmware test matrix (ADR-061)",
)
parser.add_argument(
"--output-dir", required=True,
help="Directory to write NVS binary files",
)
parser.add_argument(
"--only", type=str, default=None,
help="Comma-separated list of config names to generate (default: all)",
)
parser.add_argument(
"--csv-only", action="store_true",
help="Only generate CSV files, skip binary generation",
)
parser.add_argument(
"--list", action="store_true", dest="list_configs",
help="List all available configurations and exit",
)
args = parser.parse_args()
all_configs = define_configs()
if args.list_configs:
print(f"{'Name':<20} {'Description'}")
print("-" * 70)
for cfg in all_configs:
print(f"{cfg.name:<20} {cfg.description}")
sys.exit(0)
# Filter configs if --only specified
if args.only:
selected = set(args.only.split(","))
configs = [c for c in all_configs if c.name in selected]
missing = selected - {c.name for c in configs}
if missing:
print(f"WARNING: Unknown config names: {', '.join(sorted(missing))}",
file=sys.stderr)
else:
configs = all_configs
output_dir = Path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
print(f"Generating {len(configs)} NVS configurations in {output_dir}/")
print()
success = 0
errors = 0
for cfg in configs:
csv_content = cfg.to_csv()
# Always write the CSV for reference
csv_path = output_dir / f"nvs_{cfg.name}.csv"
csv_path.write_text(csv_content)
if cfg.name == "default" and not cfg.entries:
# "default" means no NVS — just produce an empty marker
print(f" [{cfg.name}] No NVS entries (uses Kconfig defaults)")
# Write a zero-filled NVS partition (erased state = 0xFF)
bin_path = output_dir / f"nvs_{cfg.name}.bin"
bin_path.write_bytes(b"\xff" * NVS_PARTITION_SIZE)
success += 1
continue
if args.csv_only:
print(f" [{cfg.name}] CSV only: {csv_path}")
success += 1
continue
try:
nvs_bin = generate_nvs_binary(csv_content, NVS_PARTITION_SIZE)
bin_path = output_dir / f"nvs_{cfg.name}.bin"
bin_path.write_bytes(nvs_bin)
print(f" [{cfg.name}] {len(nvs_bin)} bytes -> {bin_path}")
success += 1
except Exception as e:
print(f" [{cfg.name}] ERROR: {e}", file=sys.stderr)
errors += 1
print()
print(f"Done: {success} succeeded, {errors} failed")
if errors > 0:
sys.exit(1)
if __name__ == "__main__":
main()
scripts/qemu-esp32s3-test.sh
Executable
+150
View File
@@ -0,0 +1,150 @@
#!/bin/bash
# QEMU ESP32-S3 Firmware Test Runner (ADR-061)
#
# Builds the firmware with mock CSI enabled, merges binaries into a single
# flash image, optionally injects a pre-provisioned NVS partition, runs the
# image under QEMU with a timeout, and validates the UART output.
#
# Environment variables:
# QEMU_PATH - Path to qemu-system-xtensa (default: qemu-system-xtensa)
# QEMU_TIMEOUT - Timeout in seconds (default: 60)
# SKIP_BUILD - Set to "1" to skip the idf.py build step
# NVS_BIN - Path to a pre-built NVS binary to inject (optional)
#
# Exit codes:
# 0 All checks passed
# 1 Warnings (non-critical checks failed)
# 2 Errors (critical checks failed)
# 3 Fatal (crash detected or build failure)
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
FIRMWARE_DIR="$PROJECT_ROOT/firmware/esp32-csi-node"
BUILD_DIR="$FIRMWARE_DIR/build"
QEMU_BIN="${QEMU_PATH:-qemu-system-xtensa}"
FLASH_IMAGE="$BUILD_DIR/qemu_flash.bin"
LOG_FILE="$BUILD_DIR/qemu_output.log"
TIMEOUT_SEC="${QEMU_TIMEOUT:-60}"
echo "=== QEMU ESP32-S3 Firmware Test (ADR-061) ==="
echo "Firmware dir: $FIRMWARE_DIR"
echo "QEMU binary: $QEMU_BIN"
echo "Timeout: ${TIMEOUT_SEC}s"
echo ""
# Verify QEMU is available
if ! command -v "$QEMU_BIN" &>/dev/null; then
echo "ERROR: QEMU binary not found: $QEMU_BIN"
echo "Set QEMU_PATH to the qemu-system-xtensa binary."
exit 3
fi
# 1. Build with mock CSI enabled (skip if already built)
if [ "${SKIP_BUILD:-}" != "1" ]; then
echo "[1/4] Building firmware (mock CSI mode)..."
idf.py -C "$FIRMWARE_DIR" \
-D SDKCONFIG_DEFAULTS="sdkconfig.defaults;sdkconfig.qemu" \
build
echo ""
else
echo "[1/4] Skipping build (SKIP_BUILD=1)"
echo ""
fi
# Verify build artifacts exist
for artifact in \
"$BUILD_DIR/bootloader/bootloader.bin" \
"$BUILD_DIR/partition_table/partition-table.bin" \
"$BUILD_DIR/esp32-csi-node.bin"; do
if [ ! -f "$artifact" ]; then
echo "ERROR: Build artifact not found: $artifact"
echo "Run without SKIP_BUILD=1 or build the firmware first."
exit 3
fi
done
# 2. Merge binaries into single flash image
echo "[2/4] Creating merged flash image..."
# Check for ota_data_initial.bin; some builds don't produce it
OTA_DATA_ARGS=""
if [ -f "$BUILD_DIR/ota_data_initial.bin" ]; then
OTA_DATA_ARGS="0xf000 $BUILD_DIR/ota_data_initial.bin"
fi
python3 -m esptool --chip esp32s3 merge_bin -o "$FLASH_IMAGE" \
--flash_mode dio --flash_freq 80m --flash_size 8MB \
0x0 "$BUILD_DIR/bootloader/bootloader.bin" \
0x8000 "$BUILD_DIR/partition_table/partition-table.bin" \
$OTA_DATA_ARGS \
0x20000 "$BUILD_DIR/esp32-csi-node.bin"
echo "Flash image: $FLASH_IMAGE ($(stat -c%s "$FLASH_IMAGE" 2>/dev/null || stat -f%z "$FLASH_IMAGE") bytes)"
# 2b. Optionally inject pre-provisioned NVS partition
NVS_FILE="${NVS_BIN:-$BUILD_DIR/nvs_test.bin}"
if [ -f "$NVS_FILE" ]; then
echo "[2b] Injecting NVS partition from: $NVS_FILE"
# NVS partition offset = 0x9000 = 36864
dd if="$NVS_FILE" of="$FLASH_IMAGE" \
bs=1 seek=$((0x9000)) conv=notrunc 2>/dev/null
echo "NVS injected ($(stat -c%s "$NVS_FILE" 2>/dev/null || stat -f%z "$NVS_FILE") bytes at 0x9000)"
fi
echo ""
# 3. Run in QEMU with timeout, capture UART output
echo "[3/4] Running QEMU (timeout: ${TIMEOUT_SEC}s)..."
echo "------- QEMU UART output -------"
# Use timeout command; fall back to gtimeout on macOS
TIMEOUT_CMD="timeout"
if ! command -v timeout &>/dev/null; then
if command -v gtimeout &>/dev/null; then
TIMEOUT_CMD="gtimeout"
else
echo "WARNING: 'timeout' command not found. QEMU may run indefinitely."
TIMEOUT_CMD=""
fi
fi
QEMU_EXIT=0
if [ -n "$TIMEOUT_CMD" ]; then
$TIMEOUT_CMD "$TIMEOUT_SEC" "$QEMU_BIN" \
-machine esp32s3 \
-nographic \
-drive file="$FLASH_IMAGE",if=mtd,format=raw \
-serial mon:stdio \
-no-reboot \
2>&1 | tee "$LOG_FILE" || QEMU_EXIT=$?
else
"$QEMU_BIN" \
-machine esp32s3 \
-nographic \
-drive file="$FLASH_IMAGE",if=mtd,format=raw \
-serial mon:stdio \
-no-reboot \
2>&1 | tee "$LOG_FILE" || QEMU_EXIT=$?
fi
echo "------- End QEMU output -------"
echo ""
# timeout returns 124 when the process is killed by timeout — that's expected
if [ "$QEMU_EXIT" -eq 124 ]; then
echo "QEMU exited via timeout (expected for firmware that loops forever)."
elif [ "$QEMU_EXIT" -ne 0 ]; then
echo "WARNING: QEMU exited with code $QEMU_EXIT"
fi
echo ""
# 4. Validate expected output
echo "[4/4] Validating output..."
python3 "$SCRIPT_DIR/validate_qemu_output.py" "$LOG_FILE"
VALIDATE_EXIT=$?
echo ""
echo "=== Test Complete (exit code: $VALIDATE_EXIT) ==="
exit $VALIDATE_EXIT
scripts/validate_qemu_output.py
+366
View File
@@ -0,0 +1,366 @@
#!/usr/bin/env python3
"""
QEMU ESP32-S3 UART Output Validator (ADR-061)
Parses the UART log captured from a QEMU firmware run and validates
14 checks covering boot, NVS, mock CSI, edge processing, vitals,
presence/fall detection, serialization, and crash indicators.
Usage:
python3 validate_qemu_output.py <log_file>
Exit codes:
0 All checks passed (or only INFO-level skips)
1 Warnings (non-critical checks failed)
2 Errors (critical checks failed)
3 Fatal (crash or corruption detected)
"""
import re
import sys
from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path
from typing import List, Optional
class Severity(IntEnum):
PASS = 0
SKIP = 1
WARN = 2
ERROR = 3
FATAL = 4
# ANSI color codes (disabled if not a TTY)
USE_COLOR = sys.stdout.isatty()
def color(text: str, code: str) -> str:
if not USE_COLOR:
return text
return f"\033[{code}m{text}\033[0m"
def green(text: str) -> str:
return color(text, "32")
def yellow(text: str) -> str:
return color(text, "33")
def red(text: str) -> str:
return color(text, "31")
def bold_red(text: str) -> str:
return color(text, "1;31")
@dataclass
class CheckResult:
name: str
severity: Severity
message: str
count: int = 0
@dataclass
class ValidationReport:
checks: List[CheckResult] = field(default_factory=list)
def add(self, name: str, severity: Severity, message: str, count: int = 0):
self.checks.append(CheckResult(name, severity, message, count))
@property
def max_severity(self) -> Severity:
if not self.checks:
return Severity.PASS
return max(c.severity for c in self.checks)
def print_report(self):
print("\n" + "=" * 60)
print(" QEMU Firmware Validation Report (ADR-061)")
print("=" * 60 + "\n")
for check in self.checks:
if check.severity == Severity.PASS:
icon = green("PASS")
elif check.severity == Severity.SKIP:
icon = yellow("SKIP")
elif check.severity == Severity.WARN:
icon = yellow("WARN")
elif check.severity == Severity.ERROR:
icon = red("FAIL")
else:
icon = bold_red("FATAL")
count_str = f" (count={check.count})" if check.count > 0 else ""
print(f" [{icon}] {check.name}: {check.message}{count_str}")
print()
passed = sum(1 for c in self.checks if c.severity <= Severity.SKIP)
total = len(self.checks)
summary = f" {passed}/{total} checks passed"
max_sev = self.max_severity
if max_sev <= Severity.SKIP:
print(green(summary))
elif max_sev == Severity.WARN:
print(yellow(summary + " (with warnings)"))
elif max_sev == Severity.ERROR:
print(red(summary + " (with errors)"))
else:
print(bold_red(summary + " (FATAL issues detected)"))
print()
def validate_log(log_text: str) -> ValidationReport:
"""Run all 14 validation checks against the UART log text."""
report = ValidationReport()
lines = log_text.splitlines()
log_lower = log_text.lower()
# ---- Check 1: Boot ----
# Look for app_main() entry or main_task: tag
boot_patterns = [r"app_main\(\)", r"main_task:", r"main:", r"ESP32-S3 CSI Node"]
boot_found = any(re.search(p, log_text) for p in boot_patterns)
if boot_found:
report.add("Boot", Severity.PASS, "Firmware booted successfully")
else:
report.add("Boot", Severity.ERROR, "No boot indicator found (app_main / main_task)")
# ---- Check 2: NVS load ----
nvs_patterns = [r"nvs_config:", r"nvs_config_load", r"NVS", r"csi_cfg"]
nvs_found = any(re.search(p, log_text) for p in nvs_patterns)
if nvs_found:
report.add("NVS load", Severity.PASS, "NVS configuration loaded")
else:
report.add("NVS load", Severity.WARN, "No NVS load indicator found")
# ---- Check 3: Mock CSI init ----
mock_patterns = [r"mock_csi:", r"mock_csi_init", r"Mock CSI", r"MOCK_CSI"]
mock_found = any(re.search(p, log_text) for p in mock_patterns)
if mock_found:
report.add("Mock CSI init", Severity.PASS, "Mock CSI generator initialized")
else:
# This is only expected when mock is enabled
report.add("Mock CSI init", Severity.SKIP,
"No mock CSI indicator (expected if mock not enabled)")
# ---- Check 4: Frame generation ----
# Count frame-related log lines
frame_patterns = [
r"frame[_ ]count[=: ]+(\d+)",
r"frames?[=: ]+(\d+)",
r"emitted[=: ]+(\d+)",
r"mock_csi:.*frame",
r"csi_collector:.*frame",
r"CSI frame",
]
frame_count = 0
for line in lines:
for pat in frame_patterns:
m = re.search(pat, line, re.IGNORECASE)
if m:
if m.lastindex and m.lastindex >= 1:
try:
frame_count = max(frame_count, int(m.group(1)))
except (ValueError, IndexError):
frame_count = max(frame_count, 1)
else:
frame_count = max(frame_count, 1)
if frame_count > 0:
report.add("Frame generation", Severity.PASS,
f"Frames detected", count=frame_count)
else:
# Also count lines mentioning IQ data or subcarriers
iq_lines = sum(1 for line in lines
if re.search(r"(iq_data|subcarrier|I/Q|enqueue)", line, re.IGNORECASE))
if iq_lines > 0:
report.add("Frame generation", Severity.PASS,
"I/Q data activity detected", count=iq_lines)
else:
report.add("Frame generation", Severity.WARN,
"No frame generation activity detected")
# ---- Check 5: Edge pipeline ----
edge_patterns = [r"edge_processing:", r"DSP task", r"edge_init", r"edge_tier"]
edge_found = any(re.search(p, log_text) for p in edge_patterns)
if edge_found:
report.add("Edge pipeline", Severity.PASS, "Edge processing pipeline active")
else:
report.add("Edge pipeline", Severity.WARN,
"No edge processing indicator found")
# ---- Check 6: Vitals output ----
vitals_patterns = [r"vitals", r"breathing", r"presence", r"heartrate",
r"breathing_bpm", r"heart_rate"]
vitals_count = sum(1 for line in lines
if any(re.search(p, line, re.IGNORECASE) for p in vitals_patterns))
if vitals_count > 0:
report.add("Vitals output", Severity.PASS,
"Vitals/breathing/presence output detected", count=vitals_count)
else:
report.add("Vitals output", Severity.WARN,
"No vitals output lines found")
# ---- Check 7: Presence detection ----
presence_patterns = [
r"presence[=: ]+1",
r"presence_score[=: ]+([0-9.]+)",
r"presence detected",
]
presence_found = False
for line in lines:
for pat in presence_patterns:
m = re.search(pat, line, re.IGNORECASE)
if m:
if m.lastindex and m.lastindex >= 1:
try:
score = float(m.group(1))
if score > 0:
presence_found = True
except (ValueError, IndexError):
presence_found = True
else:
presence_found = True
if presence_found:
report.add("Presence detection", Severity.PASS, "Presence detected in output")
else:
report.add("Presence detection", Severity.WARN,
"No presence=1 or presence_score>0 found")
# ---- Check 8: Fall detection ----
fall_patterns = [r"fall[=: ]+1", r"fall detected", r"fall_event"]
fall_found = any(
re.search(p, line, re.IGNORECASE)
for line in lines for p in fall_patterns
)
if fall_found:
report.add("Fall detection", Severity.PASS, "Fall event detected in output")
else:
report.add("Fall detection", Severity.SKIP,
"No fall event (expected if fall scenario not run)")
# ---- Check 9: MAC filter ----
mac_patterns = [r"MAC filter", r"mac_filter", r"dropped.*MAC",
r"filter_mac", r"filtered"]
mac_found = any(
re.search(p, line, re.IGNORECASE)
for line in lines for p in mac_patterns
)
if mac_found:
report.add("MAC filter", Severity.PASS, "MAC filter activity detected")
else:
report.add("MAC filter", Severity.SKIP,
"No MAC filter activity (expected if filter scenario not run)")
# ---- Check 10: ADR-018 serialize ----
serialize_patterns = [r"[Ss]erializ", r"ADR-018", r"stream_sender",
r"UDP.*send", r"udp.*sent"]
serialize_count = sum(1 for line in lines
if any(re.search(p, line) for p in serialize_patterns))
if serialize_count > 0:
report.add("ADR-018 serialize", Severity.PASS,
"Serialization/streaming activity detected", count=serialize_count)
else:
report.add("ADR-018 serialize", Severity.WARN,
"No serialization activity detected")
# ---- Check 11: No crash ----
crash_patterns = [r"Guru Meditation", r"assert failed", r"abort\(\)",
r"panic", r"LoadProhibited", r"StoreProhibited",
r"InstrFetchProhibited", r"IllegalInstruction"]
crash_found = []
for line in lines:
for pat in crash_patterns:
if re.search(pat, line):
crash_found.append(line.strip()[:120])
if not crash_found:
report.add("No crash", Severity.PASS, "No crash indicators found")
else:
report.add("No crash", Severity.FATAL,
f"Crash detected: {crash_found[0]}",
count=len(crash_found))
# ---- Check 12: Heap OK ----
heap_patterns = [r"HEAP_ERROR", r"out of memory", r"heap_caps_alloc.*failed",
r"malloc.*fail", r"heap corruption"]
heap_errors = [line.strip()[:120] for line in lines
if any(re.search(p, line, re.IGNORECASE) for p in heap_patterns)]
if not heap_errors:
report.add("Heap OK", Severity.PASS, "No heap errors found")
else:
report.add("Heap OK", Severity.ERROR,
f"Heap error: {heap_errors[0]}",
count=len(heap_errors))
# ---- Check 13: Stack OK ----
stack_patterns = [r"[Ss]tack overflow", r"stack_overflow",
r"vApplicationStackOverflowHook"]
stack_errors = [line.strip()[:120] for line in lines
if any(re.search(p, line) for p in stack_patterns)]
if not stack_errors:
report.add("Stack OK", Severity.PASS, "No stack overflow detected")
else:
report.add("Stack OK", Severity.FATAL,
f"Stack overflow: {stack_errors[0]}",
count=len(stack_errors))
# ---- Check 14: Clean exit ----
reboot_patterns = [r"Rebooting\.\.\.", r"rst:0x"]
reboot_found = any(
re.search(p, line)
for line in lines for p in reboot_patterns
)
if not reboot_found:
report.add("Clean exit", Severity.PASS,
"No unexpected reboot detected")
else:
report.add("Clean exit", Severity.WARN,
"Reboot detected (may indicate crash or watchdog)")
return report
def main():
if len(sys.argv) < 2:
print(f"Usage: {sys.argv[0]} <log_file>", file=sys.stderr)
sys.exit(3)
log_path = Path(sys.argv[1])
if not log_path.exists():
print(f"ERROR: Log file not found: {log_path}", file=sys.stderr)
sys.exit(3)
log_text = log_path.read_text(encoding="utf-8", errors="replace")
if not log_text.strip():
print("ERROR: Log file is empty. QEMU may have failed to start.",
file=sys.stderr)
sys.exit(3)
report = validate_log(log_text)
report.print_report()
# Map max severity to exit code
max_sev = report.max_severity
if max_sev <= Severity.SKIP:
sys.exit(0)
elif max_sev == Severity.WARN:
sys.exit(1)
elif max_sev == Severity.ERROR:
sys.exit(2)
else:
sys.exit(3)
if __name__ == "__main__":
main()