249 lines
11 KiB
Python
249 lines
11 KiB
Python
"""Headless smoke: replay the recorded VNA_AlGo dataset through the full pipeline.
|
|
|
|
Loads the replay profile, creates neutral preprocess sets, launches the whole
|
|
pipeline (replay_raw_producer -> data_preprocessor -> data_processor) via the
|
|
ProcessSupervisor with unique ring names, forces the sar3d processor through
|
|
processing_live.json, and asserts that:
|
|
|
|
- sar3d projection payloads arrive with a positive peak;
|
|
- collection ids increase monotonically (including across a dataset loop);
|
|
- the preprocessor is transparent: a preprocessed_tap trace equals the
|
|
recorded snapshot's s21 within float32 rounding.
|
|
|
|
Usage:
|
|
python -m python_app.scripts.replay_smoke_run \
|
|
[--config run_config_examples/run_config_replay_vna.example.json] \
|
|
[--duration 20.0]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
from python_app.models.run_config_model import RunConfigModel
|
|
from python_app.orchestration.config_writer import ConfigWriter
|
|
from python_app.orchestration.live_processing_config import (
|
|
ProcessingLiveConfig,
|
|
ProcessingLiveConfigWriter,
|
|
)
|
|
from python_app.orchestration.process_supervisor import ProcessSupervisor
|
|
from python_app.orchestration.shm import ShmRingReader
|
|
from python_app.orchestration.shm.decoder import PREPROC_MAGIC, decode_trace_collection
|
|
from python_app.storage.npz_store import NpzStore, radar_key_from_config
|
|
from python_app.workflows.kamil_adc_neutral_preprocess import build_neutral_s21_sets
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def _make_unique_ring_name(prefix: str) -> str:
|
|
"""Build unique POSIX SHM ring name."""
|
|
return f"/{prefix}_{os.getpid()}_{time.monotonic_ns()}"
|
|
|
|
|
|
def _sar3d_live_config(profile_path: Path) -> ProcessingLiveConfig:
|
|
"""Build a sar3d live config from the profile's gui.processing.sar3d section.
|
|
|
|
RunConfigModel does not carry the GUI section (it belongs to the GUI profile
|
|
layer), so the sar3d defaults are read straight from the profile JSON.
|
|
"""
|
|
payload = json.loads(profile_path.read_text(encoding="utf-8"))
|
|
sar3d = (
|
|
payload.get("gui", {}).get("processing", {}).get("sar3d", {})
|
|
if isinstance(payload, dict)
|
|
else {}
|
|
)
|
|
|
|
def _positions(text: str) -> list[int]:
|
|
return [int(value) for value in str(text).split(",") if value.strip()]
|
|
|
|
live = ProcessingLiveConfig(processor_mode="sar3d", ignore_socket_speed=True)
|
|
if not isinstance(sar3d, dict) or not sar3d:
|
|
return live
|
|
live.gpr_input_positions = _positions(sar3d.get("input_positions", ""))
|
|
live.gpr_output_positions = _positions(sar3d.get("output_positions", ""))
|
|
live.gpr_min_depth_m = float(sar3d.get("min_depth_m", live.gpr_min_depth_m))
|
|
live.gpr_max_depth_m = float(sar3d.get("max_depth_m", live.gpr_max_depth_m))
|
|
live.gpr_start_freq_mhz = float(sar3d.get("start_freq_mhz", live.gpr_start_freq_mhz))
|
|
live.gpr_stop_freq_mhz = float(sar3d.get("stop_freq_mhz", live.gpr_stop_freq_mhz))
|
|
live.gpr_speed_m_s = float(sar3d.get("speed_m_s", live.gpr_speed_m_s))
|
|
live.gpr_look_angle_deg = float(sar3d.get("look_angle_deg", live.gpr_look_angle_deg))
|
|
live.gpr_direction_sign = float(sar3d.get("direction_sign", live.gpr_direction_sign))
|
|
live.gpr_apply_freq_phase_correction = bool(
|
|
sar3d.get("apply_freq_phase_correction", live.gpr_apply_freq_phase_correction)
|
|
)
|
|
live.sar3d_frame_radius = int(sar3d.get("frame_radius", live.sar3d_frame_radius))
|
|
live.sar3d_edge_frame_weight = float(
|
|
sar3d.get("edge_frame_weight", live.sar3d_edge_frame_weight)
|
|
)
|
|
live.sar3d_pitch_deg = float(sar3d.get("pitch_deg", live.sar3d_pitch_deg))
|
|
live.sar3d_center_depth_m = float(sar3d.get("center_depth_m", live.sar3d_center_depth_m))
|
|
live.sar3d_half_width_x_m = float(sar3d.get("half_width_x_m", live.sar3d_half_width_x_m))
|
|
live.sar3d_half_width_y_m = float(sar3d.get("half_width_y_m", live.sar3d_half_width_y_m))
|
|
live.sar3d_half_width_z_m = float(sar3d.get("half_width_z_m", live.sar3d_half_width_z_m))
|
|
live.sar3d_grid_nx = int(sar3d.get("grid_nx", live.sar3d_grid_nx))
|
|
live.sar3d_grid_ny = int(sar3d.get("grid_ny", live.sar3d_grid_ny))
|
|
live.sar3d_grid_nz = int(sar3d.get("grid_nz", live.sar3d_grid_nz))
|
|
live.sar3d_use_compensation = bool(sar3d.get("use_compensation", live.sar3d_use_compensation))
|
|
live.sar3d_normalize_pairs = bool(sar3d.get("normalize_pairs", live.sar3d_normalize_pairs))
|
|
live.sar3d_bottom_view_deg = float(sar3d.get("bottom_view_deg", live.sar3d_bottom_view_deg))
|
|
return live
|
|
|
|
|
|
def _check_preprocessor_transparency(
|
|
tap_reader: ShmRingReader, dataset_dir: Path
|
|
) -> bool:
|
|
"""Compare one preprocessed trace against the recorded snapshot's s21."""
|
|
payload = tap_reader.peek_latest_payload()
|
|
if payload is None:
|
|
logger.warning("No preprocessed_tap payload available for transparency check")
|
|
return False
|
|
preprocessed = decode_trace_collection(payload, PREPROC_MAGIC)
|
|
|
|
# Any snapshot works: neutral preprocessing is time-invariant, so match the
|
|
# trace against every recorded snapshot and accept the best one.
|
|
trace = preprocessed.traces[0]
|
|
combo_file = f"i{trace.combo.input}_o{trace.combo.output}_s21.npy"
|
|
best_delta = np.inf
|
|
for folder in sorted(p for p in dataset_dir.iterdir() if p.name[:1].isdigit()):
|
|
recorded = np.load(folder / combo_file).astype(np.complex64)
|
|
if recorded.size != np.asarray(trace.s21).size:
|
|
continue
|
|
delta = float(np.max(np.abs(np.asarray(trace.s21) - recorded)))
|
|
best_delta = min(best_delta, delta)
|
|
if best_delta == 0.0:
|
|
break
|
|
logger.info("Preprocessor transparency: max|delta| vs best snapshot = %.3e", best_delta)
|
|
return best_delta < 1e-6
|
|
|
|
|
|
def main() -> int:
|
|
"""Run the replay pipeline smoke and report pass/fail."""
|
|
parser = argparse.ArgumentParser(description="Headless replay pipeline smoke")
|
|
parser.add_argument(
|
|
"--config",
|
|
type=Path,
|
|
default=PROJECT_ROOT / "run_config_examples/run_config_replay_vna.example.json",
|
|
)
|
|
parser.add_argument("--duration", type=float, default=20.0)
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
|
|
|
|
config = RunConfigModel.load_from_path(args.config)
|
|
config.ensure_combos()
|
|
# expanduser: "~/..." dataset paths from machine-portable profiles.
|
|
dataset_dir = Path(config.radar.replay.dataset_dir).expanduser()
|
|
if not dataset_dir.is_dir():
|
|
raise RuntimeError(f"Replay dataset not found: {dataset_dir}")
|
|
|
|
config.rings.raw.name = _make_unique_ring_name("radar_raw_replay_smoke")
|
|
config.rings.raw_tap.name = _make_unique_ring_name("radar_raw_tap_replay_smoke")
|
|
config.rings.preprocessed.name = _make_unique_ring_name("radar_preprocessed_replay_smoke")
|
|
config.rings.preprocessed_tap.name = _make_unique_ring_name("radar_preprocessed_tap_replay_smoke")
|
|
config.rings.results.name = _make_unique_ring_name("radar_results_replay_smoke")
|
|
# Private live-config path so the smoke never clobbers the GUI's file.
|
|
live_path = PROJECT_ROOT / "python_app/runtime/processing_live_replay_smoke.json"
|
|
config.runtime.processing_live_config_path = str(
|
|
live_path.relative_to(PROJECT_ROOT)
|
|
)
|
|
|
|
radar_key = radar_key_from_config(
|
|
model_name=config.radar.model,
|
|
serial=config.radar.serial,
|
|
sweep_start_hz=config.radar.sweep.start_hz,
|
|
sweep_stop_hz=config.radar.sweep.stop_hz,
|
|
sweep_points=config.radar.sweep.points,
|
|
ifbw_hz=config.radar.sweep.if_bandwidth_hz,
|
|
power_dbm=config.radar.sweep.power_dbm,
|
|
extra_serials=config.radar_key_extra_parts() or None,
|
|
)
|
|
store = NpzStore(PROJECT_ROOT / "python_app/data")
|
|
set_name = config.preprocess.s21.calibration.set_name or "neutral_replay"
|
|
if set_name not in store.list_sets("s21_calibration", radar_key):
|
|
calibration, reference = build_neutral_s21_sets(config)
|
|
store.save_set("s21_calibration", radar_key, set_name, calibration)
|
|
store.save_set("s21_reference", radar_key, set_name, reference)
|
|
logger.info("Created neutral sets '%s' (radar_key=%s)", set_name, radar_key)
|
|
config.preprocess.s21.calibration.set_name = set_name
|
|
config.preprocess.s21.reference.set_name = set_name
|
|
|
|
ProcessingLiveConfigWriter(live_path).write(_sar3d_live_config(args.config))
|
|
|
|
config_writer = ConfigWriter(PROJECT_ROOT / "python_app/runtime")
|
|
config_writer.prepare_preprocess_bundles(store, radar_key, config)
|
|
config_path = config_writer.write(
|
|
config, PROJECT_ROOT / "python_app/runtime/run_config_replay_smoke.json"
|
|
)
|
|
|
|
supervisor = ProcessSupervisor(PROJECT_ROOT)
|
|
result_reader: ShmRingReader | None = None
|
|
tap_reader: ShmRingReader | None = None
|
|
failures: list[str] = []
|
|
try:
|
|
supervisor.start(config_path)
|
|
result_reader = ShmRingReader(config.rings.results.name, open_timeout_s=15.0)
|
|
tap_reader = ShmRingReader(config.rings.preprocessed_tap.name, open_timeout_s=15.0)
|
|
|
|
deadline = time.monotonic() + args.duration
|
|
received = 0
|
|
sar3d_results = 0
|
|
last_peak = 0.0
|
|
last_id = 0
|
|
ids_monotonic = True
|
|
while time.monotonic() < deadline:
|
|
result = result_reader.pop_result_collection()
|
|
if result is None:
|
|
time.sleep(0.02)
|
|
continue
|
|
received += 1
|
|
if last_id and result.collection_id <= last_id:
|
|
ids_monotonic = False
|
|
last_id = result.collection_id
|
|
names = {p.processing_name: p for p in result.collection_payloads}
|
|
if "sar3d_proj_xz" in names and "sar3d_proj_bottom" in names:
|
|
sar3d_results += 1
|
|
peak_payload = names.get("sar3d_peak")
|
|
if peak_payload is not None:
|
|
last_peak = float(peak_payload.scalar_value)
|
|
|
|
logger.info(
|
|
"Results: total=%d, with sar3d projections=%d, last collection_id=%d, last peak=%.4e",
|
|
received, sar3d_results, last_id, last_peak,
|
|
)
|
|
if received == 0:
|
|
failures.append("no result collections received")
|
|
if sar3d_results == 0:
|
|
failures.append("no sar3d projection payloads received")
|
|
if not (last_peak > 0.0):
|
|
failures.append(f"sar3d peak is not positive: {last_peak}")
|
|
if not ids_monotonic:
|
|
failures.append("collection ids were not monotonically increasing")
|
|
if not _check_preprocessor_transparency(tap_reader, dataset_dir):
|
|
failures.append("preprocessor is not transparent vs recorded snapshots")
|
|
finally:
|
|
supervisor.stop_all()
|
|
if result_reader is not None:
|
|
result_reader.close()
|
|
if tap_reader is not None:
|
|
tap_reader.close()
|
|
|
|
if failures:
|
|
for failure in failures:
|
|
logger.error("FAIL: %s", failure)
|
|
return 1
|
|
logger.info("REPLAY SMOKE OK")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|