"""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_override_enabled = bool( sar3d.get("pitch_override_enabled", live.sar3d_pitch_override_enabled) ) live.sar3d_pitch_deg = float(sar3d.get("pitch_deg", live.sar3d_pitch_deg)) live.sar3d_ground_depth_m = float(sar3d.get("ground_depth_m", live.sar3d_ground_depth_m)) live.sar3d_x_min_m = float(sar3d.get("x_min_m", live.sar3d_x_min_m)) live.sar3d_x_max_m = float(sar3d.get("x_max_m", live.sar3d_x_max_m)) live.sar3d_y_min_m = float(sar3d.get("y_min_m", live.sar3d_y_min_m)) live.sar3d_y_max_m = float(sar3d.get("y_max_m", live.sar3d_y_max_m)) live.sar3d_z_min_m = float(sar3d.get("z_min_m", live.sar3d_z_min_m)) live.sar3d_z_max_m = float(sar3d.get("z_max_m", live.sar3d_z_max_m)) live.sar3d_voxel_x_m = float(sar3d.get("voxel_x_m", live.sar3d_voxel_x_m)) live.sar3d_voxel_y_m = float(sar3d.get("voxel_y_m", live.sar3d_voxel_y_m)) live.sar3d_voxel_z_m = float(sar3d.get("voxel_z_m", live.sar3d_voxel_z_m)) live.sar3d_norm_edge_margin_m = float( sar3d.get("norm_edge_margin_m", live.sar3d_norm_edge_margin_m) ) live.sar3d_smooth_sigma = float(sar3d.get("smooth_sigma", live.sar3d_smooth_sigma)) live.sar3d_comp_range_power = float( sar3d.get("comp_range_power", live.sar3d_comp_range_power) ) live.sar3d_tx_per_segment = int(sar3d.get("tx_per_segment", live.sar3d_tx_per_segment)) 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())