sar3d: GPU port of the notebook with four views
The sar3d processor is rewritten as a checked port of Multi_antenna_3D_clean_v1.ipynb: NVRTC kernels, timing from the trace stamps, notebook parameter names and one cubic SAR_VOXEL_M. It adds front, side, top and 3D views, drops gpr_cpp and the parity tools, and the GUI speed now applies only while the socket speed is ignored.
This commit is contained in:
Ayzen
@@ -1,262 +0,0 @@
|
||||
"""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())
|
||||
@@ -1,389 +0,0 @@
|
||||
"""Reference implementation of the sar3d volume from Multi_antenna_3D_clean_v1.ipynb.
|
||||
|
||||
Executable specification for the C++ port: recomputes a single-frame
|
||||
(aperture radius 0) 3D SAR volume from a preprocessed snapshot exactly the
|
||||
way the notebook does, then compares it against a reference *_volume.npz
|
||||
exported by the notebook.
|
||||
|
||||
Usage:
|
||||
python sar3d_reference_check.py <dataset_root> <reference_volume.npz>
|
||||
|
||||
<dataset_root> is an extracted replay dataset (contains config_profile.json
|
||||
and preprocessed/<snapshot>/i*_o*_{s21,freq}.npy + meta.json). The snapshot
|
||||
name and all physics parameters are taken from the npz meta.
|
||||
|
||||
Exits non-zero when the recomputed volume differs from the reference by more
|
||||
than the tolerance. The only expected difference is the Gaussian smoothing
|
||||
(manual separable filter vs scipy) at the 1e-6 level.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
TOLERANCE = 2e-6
|
||||
|
||||
C_LIGHT = 3e8
|
||||
|
||||
# Notebook constants not recorded in the npz meta (cell 0 / cell 4).
|
||||
BP_OVERSAMPLE = 8
|
||||
BP_WINDOW = True
|
||||
PAIR_NORM_PERCENTILE = 50.0
|
||||
PAIR_NORM_EPS = 1e-15
|
||||
COMP_RANGE_POWER = 0.3
|
||||
COMP_RANGE_WEIGHT_MAX = 5.0
|
||||
COMP_REF_DEPTH = 5.0
|
||||
BP_SMOOTH_SIGMA = 1.5
|
||||
SAR_Y_RANGE_M = (-0.5, 2.0)
|
||||
SAR_NORM_EDGE_MARGIN_M = 0.3
|
||||
X_MARGIN_M = 2.0
|
||||
MOTION_PHASE_SIGN = -1.0
|
||||
DIRECTION_SIGN = +1.0
|
||||
|
||||
|
||||
def gaussian_kernel1d(sigma: float, radius: int) -> np.ndarray:
|
||||
x = np.arange(-radius, radius + 1, dtype=float)
|
||||
phi = np.exp(-0.5 / (sigma * sigma) * x * x)
|
||||
return phi / phi.sum()
|
||||
|
||||
|
||||
def correlate1d_reflect(arr: np.ndarray, weights: np.ndarray, axis: int) -> np.ndarray:
|
||||
r = (len(weights) - 1) // 2
|
||||
a = np.moveaxis(arr, axis, -1)
|
||||
pad = np.concatenate([a[..., r - 1::-1], a, a[..., :-r - 1:-1]], axis=-1)
|
||||
out = np.zeros_like(a)
|
||||
for k, w in enumerate(weights):
|
||||
out += w * pad[..., k:k + a.shape[-1]]
|
||||
return np.moveaxis(out, -1, axis)
|
||||
|
||||
|
||||
def gaussian_filter(arr: np.ndarray, sigma: tuple[float, ...], truncate: float = 4.0) -> np.ndarray:
|
||||
"""scipy.ndimage.gaussian_filter equivalent (mode='reflect')."""
|
||||
out = np.asarray(arr, dtype=float).copy()
|
||||
for axis, s in enumerate(sigma):
|
||||
if s <= 0:
|
||||
continue
|
||||
radius = int(truncate * float(s) + 0.5)
|
||||
out = correlate1d_reflect(out, gaussian_kernel1d(float(s), radius), axis)
|
||||
return out
|
||||
|
||||
|
||||
def load_geometry(profile_path: Path) -> dict:
|
||||
profile = json.loads(profile_path.read_text())
|
||||
gpr = profile["gpr"]
|
||||
tx_geom = {int(e["output_pos"]): (e["x_m"], e.get("y_m", 0.0), e.get("z_m", 0.0))
|
||||
for e in gpr["tx_geometry"]}
|
||||
rx_geom = {int(e["input_pos"]): (e["x_m"], e.get("y_m", 0.0), e.get("z_m", 0.0))
|
||||
for e in gpr["rx_geometry"]}
|
||||
combos = profile["run"]["combos"]
|
||||
pair_order = [(int(c["output"]), int(c["input"])) for c in combos]
|
||||
tx_order_full: list[int] = []
|
||||
for tx, _ in pair_order:
|
||||
if tx not in tx_order_full:
|
||||
tx_order_full.append(tx)
|
||||
output_idx = sorted(tx_geom)
|
||||
input_idx = sorted(rx_geom)
|
||||
tx_xyz = np.array([tx_geom[i] for i in output_idx], dtype=float)
|
||||
rx_xyz = np.array([rx_geom[i] for i in input_idx], dtype=float)
|
||||
return {
|
||||
"output_idx": output_idx, "input_idx": input_idx,
|
||||
"tx_xyz": tx_xyz, "rx_xyz": rx_xyz,
|
||||
"pair_order": pair_order, "tx_order_full": tx_order_full,
|
||||
}
|
||||
|
||||
|
||||
def snapshot_timing(snapshot_dir: Path) -> dict | None:
|
||||
"""Per-Tx capture windows from the snapshot's meta.json trace labels."""
|
||||
try:
|
||||
meta = json.loads((snapshot_dir / "meta.json").read_text())
|
||||
except OSError:
|
||||
return None
|
||||
traces = [t for t in meta.get("traces", [])
|
||||
if "capture_start_ns" in t and "capture_end_ns" in t and "output" in t]
|
||||
if not traces:
|
||||
return None
|
||||
first = min(int(t["capture_start_ns"]) for t in traces)
|
||||
last = max(int(t["capture_end_ns"]) for t in traces)
|
||||
by_tx: dict[int, tuple[int, int]] = {}
|
||||
for t in traces:
|
||||
by_tx.setdefault(int(t["output"]), (int(t["capture_start_ns"]), int(t["capture_end_ns"])))
|
||||
return {
|
||||
"acq_s": (last - first) * 1e-9,
|
||||
"center_abs_s": 0.5 * (first + last) * 1e-9,
|
||||
"tx_window_s": {tx: ((a - first) * 1e-9, (b - first) * 1e-9) for tx, (a, b) in by_tx.items()},
|
||||
}
|
||||
|
||||
|
||||
def pair_timestamps(geometry: dict, timing: dict, speed_m_s: float, look_angle_deg: float,
|
||||
wave_speed: float) -> dict:
|
||||
"""Segment/slot assignment and coarse pair dtau (notebook compute_pair_timestamps_for_snapshot)."""
|
||||
windows = timing["tx_window_s"]
|
||||
hw_order = [tx for tx in geometry["tx_order_full"] if tx in windows]
|
||||
hw_order += [tx for tx in sorted(windows) if tx not in hw_order]
|
||||
groups: dict[tuple[float, float], list[int]] = {}
|
||||
for tx in hw_order:
|
||||
groups.setdefault(windows[tx], []).append(tx)
|
||||
seg_of = {}
|
||||
for seg_idx, win in enumerate(sorted(groups)):
|
||||
for slot, tx in enumerate(groups[win]):
|
||||
seg_of[tx] = (seg_idx, slot, len(groups[win]))
|
||||
cos_theta = np.cos(np.radians(look_angle_deg))
|
||||
t_ref = 0.5 * timing["acq_s"]
|
||||
tx_map = {p: k for k, p in enumerate(geometry["output_idx"])}
|
||||
rx_map = {p: k for k, p in enumerate(geometry["input_idx"])}
|
||||
ts = {}
|
||||
for tx_phys, rx_phys in geometry["pair_order"]:
|
||||
_, slot, n_slots = seg_of[tx_phys]
|
||||
t0, t1 = windows[tx_phys]
|
||||
t_center = 0.5 * (t0 + t1)
|
||||
dz_motion = DIRECTION_SIGN * speed_m_s * (t_center - t_ref) * cos_theta
|
||||
ts[(tx_map[tx_phys], rx_map[rx_phys])] = {
|
||||
"slot": slot, "n_slots": n_slots,
|
||||
"segment_start_s": t0, "segment_time_s": t1 - t0,
|
||||
"t_center_s": t_center,
|
||||
"dtau_motion_s": 2.0 * dz_motion / wave_speed,
|
||||
}
|
||||
return ts
|
||||
|
||||
|
||||
def timing_abs_center(snapshot_dir: Path) -> float:
|
||||
timing = snapshot_timing(snapshot_dir)
|
||||
if timing is None:
|
||||
raise RuntimeError(f"meta.json trace labels required in {snapshot_dir}")
|
||||
return timing["center_abs_s"]
|
||||
|
||||
|
||||
def load_mimo(snapshot_dir: Path, geometry: dict) -> tuple[dict, dict]:
|
||||
s21_data, freq_data = {}, {}
|
||||
for f in snapshot_dir.glob("i*_o*_s21.npy"):
|
||||
parts = f.stem.split("_")
|
||||
rx_phys = int(parts[0][1:])
|
||||
tx_phys = int(parts[1][1:])
|
||||
if tx_phys not in geometry["output_idx"] or rx_phys not in geometry["input_idx"]:
|
||||
continue
|
||||
key = (geometry["output_idx"].index(tx_phys), geometry["input_idx"].index(rx_phys))
|
||||
s21_data[key] = np.load(f)
|
||||
freq_data[key] = np.asarray(np.load(f.parent / f"i{rx_phys}_o{tx_phys}_freq.npy"), dtype=float)
|
||||
return s21_data, freq_data
|
||||
|
||||
|
||||
def apply_phase_correction(s21: np.ndarray, freq_full: np.ndarray, info: dict,
|
||||
f_start: float, f_stop: float, speed_m_s: float,
|
||||
look_angle_deg: float, wave_speed: float) -> np.ndarray:
|
||||
"""Coarse pair + intra-sweep motion correction as one phase ramp before the IFFT."""
|
||||
idx = np.flatnonzero((freq_full >= f_start) & (freq_full <= f_stop))
|
||||
n_full = len(freq_full)
|
||||
dt_event = info["segment_time_s"] / (info["n_slots"] * n_full)
|
||||
t_abs = info["segment_start_s"] + (info["n_slots"] * idx + info["slot"]) * dt_event
|
||||
dt_intra = t_abs - info["t_center_s"]
|
||||
dtau_intra = 2.0 * DIRECTION_SIGN * speed_m_s * np.cos(np.radians(look_angle_deg)) * dt_intra / wave_speed
|
||||
total_dtau = dtau_intra + info["dtau_motion_s"]
|
||||
phi = 2.0 * np.pi * freq_full[idx] * total_dtau
|
||||
out = np.array(s21, dtype=np.complex128, copy=True)
|
||||
out[idx] *= np.exp(1j * MOTION_PHASE_SIGN * phi)
|
||||
return out
|
||||
|
||||
|
||||
def compute_ascan(s21: np.ndarray, freq: np.ndarray, f_start: float, f_stop: float):
|
||||
"""Oversampled complex A-scan with the notebook's frequency-shifted IFFT."""
|
||||
mask = (freq >= f_start) & (freq <= f_stop)
|
||||
freq_cut = freq[mask]
|
||||
s21_cut = s21[mask]
|
||||
df = float(np.median(np.diff(freq_cut)))
|
||||
n = len(freq_cut)
|
||||
k0 = int(round(freq_cut[0] / df))
|
||||
n_fft = (1 << int(np.ceil(np.log2(2 * (k0 + n - 1))))) * BP_OVERSAMPLE
|
||||
dt = 1.0 / (n_fft * df)
|
||||
s = s21_cut * np.hanning(n) if BP_WINDOW else s21_cut.copy()
|
||||
spectrum = np.zeros(n_fft, dtype=np.complex128)
|
||||
spectrum[k0:k0 + n] = s
|
||||
return np.arange(n_fft, dtype=float) * dt, np.fft.ifft(spectrum)
|
||||
|
||||
|
||||
def interp_complex(tau: np.ndarray, t_axis: np.ndarray, h_axis: np.ndarray) -> np.ndarray:
|
||||
n = t_axis.size
|
||||
dt = (float(t_axis[-1]) - float(t_axis[0])) / (n - 1)
|
||||
pos = (tau - float(t_axis[0])) / dt
|
||||
inside = (pos >= 0.0) & (pos <= n - 1)
|
||||
k = np.clip(np.floor(pos), 0, n - 2).astype(np.intp)
|
||||
frac = pos - k
|
||||
out = h_axis[k] + frac * (h_axis[k + 1] - h_axis[k])
|
||||
return np.where(inside, out, 0.0 + 0.0j)
|
||||
|
||||
|
||||
def axis_linspace(lo: float, hi: float, step: float) -> np.ndarray:
|
||||
return np.linspace(lo, hi, max(int(round((hi - lo) / step)) + 1, 2))
|
||||
|
||||
|
||||
def compute_volume(dataset_root: Path, meta: dict) -> dict:
|
||||
"""Single-frame volume via the notebook's ANIMATION path.
|
||||
|
||||
The reference npz files were exported from animation frames: the base grid
|
||||
is shifted by the radar displacement between the frame and the scene-origin
|
||||
snapshot (meta reference_snapshot), antennas get the same offset, and the
|
||||
depth window follows the radar (depth_shift). Physically it all cancels for
|
||||
an aperture of radius 0, but replicating the exact float arithmetic is what
|
||||
makes the comparison bit-tight at the depth-mask knife edge.
|
||||
"""
|
||||
geometry = load_geometry(dataset_root / "config_profile.json")
|
||||
snapshot_dir = dataset_root / "preprocessed" / meta["snapshot"]
|
||||
|
||||
eps_r = 1.0
|
||||
wave_speed = C_LIGHT / np.sqrt(eps_r)
|
||||
f_start = float(meta["f_start_hz"])
|
||||
f_stop = float(meta["f_stop_hz"])
|
||||
min_depth = float(meta["min_depth_m"])
|
||||
max_depth = float(meta["max_depth_m"])
|
||||
speed = float(meta["speed_m_s"])
|
||||
look_angle = float(meta["look_angle_deg"])
|
||||
ground_depth = float(meta["ground_depth_m"])
|
||||
voxel = [float(a) for a in meta["voxel_m"]]
|
||||
|
||||
x_tx, y_tx, z_tx = geometry["tx_xyz"].T
|
||||
x_rx, y_rx, z_rx = geometry["rx_xyz"].T
|
||||
n_tx, n_rx = len(x_tx), len(x_rx)
|
||||
|
||||
timing = snapshot_timing(snapshot_dir)
|
||||
if timing is None:
|
||||
raise RuntimeError(f"meta.json trace labels required in {snapshot_dir}")
|
||||
ts = pair_timestamps(geometry, timing, speed, look_angle, wave_speed)
|
||||
|
||||
s21_data, freq_data = load_mimo(snapshot_dir, geometry)
|
||||
ascan_h, ascan_t = {}, {}
|
||||
for key, s21 in s21_data.items():
|
||||
corrected = apply_phase_correction(
|
||||
s21, freq_data[key], ts[key], f_start, f_stop, speed, look_angle, wave_speed)
|
||||
t_axis, h = compute_ascan(corrected, freq_data[key], f_start, f_stop)
|
||||
z_axis = t_axis * wave_speed / 2
|
||||
m = (z_axis >= min_depth) & (z_axis <= max_depth)
|
||||
scale = float(np.percentile(np.abs(h)[m], PAIR_NORM_PERCENTILE))
|
||||
if not np.isfinite(scale) or scale <= PAIR_NORM_EPS:
|
||||
scale = 1.0
|
||||
ascan_h[key] = h / (scale + PAIR_NORM_EPS)
|
||||
ascan_t[key] = t_axis
|
||||
|
||||
pitch = np.radians(look_angle)
|
||||
cos_p, sin_p = np.cos(pitch), np.sin(pitch)
|
||||
|
||||
def scene_to_radar_yz(y_scene, z_scene):
|
||||
y0 = y_scene - ground_depth
|
||||
return y0 * cos_p + z_scene * sin_p, -y0 * sin_p + z_scene * cos_p
|
||||
|
||||
x_ant = np.concatenate([x_tx, x_rx])
|
||||
y_lo, y_hi = SAR_Y_RANGE_M
|
||||
z_span = [(d - float(scene_to_radar_yz(y, 0.0)[1])) / cos_p
|
||||
for y in (y_lo, y_hi) for d in (min_depth, max_depth)]
|
||||
gx = axis_linspace(float(x_ant.min() - X_MARGIN_M), float(x_ant.max() + X_MARGIN_M), voxel[0])
|
||||
gy = axis_linspace(y_lo, y_hi, voxel[1])
|
||||
gz = axis_linspace(min(z_span), max(z_span), voxel[2])
|
||||
|
||||
# Displacement of this frame relative to the scene-origin snapshot
|
||||
# (notebook sar_frame_offsets + anim_frame_displacement).
|
||||
off_dy = off_dz = 0.0
|
||||
disp_y = disp_z = 0.0
|
||||
reference_name = meta.get("reference_snapshot", meta["snapshot"])
|
||||
if reference_name != meta["snapshot"]:
|
||||
ref_timing = snapshot_timing(dataset_root / "preprocessed" / reference_name)
|
||||
if ref_timing is None:
|
||||
raise RuntimeError(f"meta.json trace labels required in {reference_name}")
|
||||
dt = timing_abs_center(snapshot_dir) - ref_timing["center_abs_s"]
|
||||
horizontal = DIRECTION_SIGN * speed * dt
|
||||
off_dy = float(horizontal * np.sin(pitch))
|
||||
off_dz = float(horizontal * np.cos(pitch))
|
||||
# Scene displacement = difference of two transformed points; the ground
|
||||
# lift cancels, leaving essentially (0, ~0, horizontal).
|
||||
disp_y = (off_dy * cos_p - off_dz * sin_p + ground_depth) - ground_depth
|
||||
disp_z = off_dy * sin_p + off_dz * cos_p
|
||||
gy_s = gy + disp_y
|
||||
gz_s = gz + disp_z
|
||||
depth_shift = off_dz
|
||||
grid_xx, grid_zz = np.meshgrid(gx, gz_s)
|
||||
|
||||
ground_plane_y = (-ground_depth + COMP_REF_DEPTH * sin_p) / cos_p
|
||||
|
||||
def pair_geo_ref(i, j):
|
||||
xc = (x_tx[i] + x_rx[j]) / 2.0
|
||||
r_tx = np.sqrt((xc - x_tx[i]) ** 2 + (ground_plane_y - y_tx[i]) ** 2 + (COMP_REF_DEPTH - z_tx[i]) ** 2)
|
||||
r_rx = np.sqrt((xc - x_rx[j]) ** 2 + (ground_plane_y - y_rx[j]) ** 2 + (COMP_REF_DEPTH - z_rx[j]) ** 2)
|
||||
return 1.0 / (r_tx * r_rx + 1e-12) + 1e-30
|
||||
|
||||
geo_ref = {(i, j): pair_geo_ref(i, j) for i in range(n_tx) for j in range(n_rx)}
|
||||
dx2_tx = [(grid_xx - x_tx[i]) ** 2 for i in range(n_tx)]
|
||||
dx2_rx = [(grid_xx - x_rx[j]) ** 2 for j in range(n_rx)]
|
||||
|
||||
acc_vol = np.zeros((gy.size, gz.size, gx.size), dtype=np.complex128)
|
||||
cnt_vol = np.zeros(acc_vol.shape, dtype=float)
|
||||
for iy, y_plane in enumerate(gy_s):
|
||||
y_rad, z_rad = scene_to_radar_yz(y_plane, grid_zz)
|
||||
depth_ok = (z_rad - depth_shift >= min_depth) & (z_rad - depth_shift <= max_depth)
|
||||
r_tx_all = [np.sqrt(dx2_tx[i] + (y_rad - (y_tx[i] + off_dy)) ** 2 + (z_rad - (z_tx[i] + off_dz)) ** 2)
|
||||
for i in range(n_tx)]
|
||||
r_rx_all = [np.sqrt(dx2_rx[j] + (y_rad - (y_rx[j] + off_dy)) ** 2 + (z_rad - (z_rx[j] + off_dz)) ** 2)
|
||||
for j in range(n_rx)]
|
||||
acc = np.zeros_like(grid_xx, dtype=np.complex128)
|
||||
cnt = np.zeros_like(grid_xx, dtype=float)
|
||||
for i in range(n_tx):
|
||||
for j in range(n_rx):
|
||||
key = (i, j)
|
||||
if key not in ascan_h:
|
||||
continue
|
||||
r_tx, r_rx = r_tx_all[i], r_rx_all[j]
|
||||
tau = (r_tx + r_rx) / wave_speed
|
||||
t_axis = ascan_t[key]
|
||||
valid = depth_ok & (tau >= t_axis[0]) & (tau <= t_axis[-1])
|
||||
h = interp_complex(tau, t_axis, ascan_h[key])
|
||||
ratio = (1.0 / (r_tx * r_rx + 1e-12) + 1e-30) / geo_ref[key]
|
||||
weight = np.clip(1.0 / (ratio ** COMP_RANGE_POWER + 1e-12), 0.0, COMP_RANGE_WEIGHT_MAX)
|
||||
acc += np.where(valid, h * weight, 0.0 + 0.0j)
|
||||
cnt += valid
|
||||
acc_vol[iy] = acc
|
||||
cnt_vol[iy] = cnt
|
||||
|
||||
volume = np.abs(acc_vol / (cnt_vol + 1e-12))
|
||||
z_rad_planes = np.asarray([scene_to_radar_yz(yp, grid_zz)[1] - depth_shift for yp in gy_s])
|
||||
volume = np.where((z_rad_planes >= min_depth) & (z_rad_planes <= max_depth), volume, 0.0)
|
||||
ref_planes = (gy_s >= gy_s[0] + SAR_NORM_EDGE_MARGIN_M) & (gy_s <= gy_s[-1] - SAR_NORM_EDGE_MARGIN_M)
|
||||
level_raw = volume[ref_planes].max() + 1e-12
|
||||
volume = gaussian_filter(volume / level_raw, sigma=(0.0, BP_SMOOTH_SIGMA, BP_SMOOTH_SIGMA))
|
||||
level_smooth = volume[ref_planes].max() + 1e-12
|
||||
volume = np.clip(volume / level_smooth, 0.0, None).astype(np.float32)
|
||||
return {"volume": volume, "level": float(level_raw * level_smooth), "x": gx, "y": gy, "z": gz}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("dataset_root", type=Path,
|
||||
help="extracted dataset dir (config_profile.json + preprocessed/)")
|
||||
parser.add_argument("reference_npz", type=Path, help="reference *_volume.npz from the notebook")
|
||||
args = parser.parse_args()
|
||||
|
||||
ref = np.load(args.reference_npz)
|
||||
meta = json.loads(str(ref["meta"]))
|
||||
print(f"snapshot: {meta['snapshot']}")
|
||||
result = compute_volume(args.dataset_root, meta)
|
||||
|
||||
volume, ref_volume = result["volume"], ref["volume"]
|
||||
axes_ok = all(np.allclose(result[k], ref[f"{k}_m"]) for k in ("x", "y", "z"))
|
||||
level_rel = abs(result["level"] - meta["level"]) / abs(meta["level"])
|
||||
print(f"axes match: {axes_ok}; level rel diff: {level_rel:.3e} "
|
||||
f"(mine {result['level']:.9f}, ref {meta['level']:.9f})")
|
||||
if volume.shape != ref_volume.shape:
|
||||
print(f"FAIL: shape mismatch {volume.shape} vs {ref_volume.shape}")
|
||||
return 1
|
||||
diff = np.abs(volume.astype(np.float64) - ref_volume.astype(np.float64))
|
||||
argmax = np.unravel_index(int(np.argmax(volume)), volume.shape)
|
||||
ref_argmax = np.unravel_index(int(np.argmax(ref_volume)), ref_volume.shape)
|
||||
print(f"max |diff| = {diff.max():.3e}, mean = {diff.mean():.3e}")
|
||||
print(f"argmax: mine {argmax}, ref {ref_argmax}")
|
||||
ok = axes_ok and diff.max() <= TOLERANCE and argmax == ref_argmax and level_rel <= TOLERANCE
|
||||
print("OK" if ok else "FAIL")
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,89 +0,0 @@
|
||||
"""Compare a sar3d volume dumped by the C++ processor with a notebook npz export.
|
||||
|
||||
Usage:
|
||||
python sar3d_volume_compare.py <dump_base> <reference_volume.npz>
|
||||
|
||||
<dump_base> is the path prefix written by `sar3d_parity --dump-dir` (or by the
|
||||
processor with `sar3d_volume_dump_dir` set): <dump_base>.f32 holds the raw
|
||||
float32 volume [iy][iz][ix], <dump_base>.json its axes and level.
|
||||
|
||||
The C++ path accumulates the backprojection in float32 (CUDA kernel), so the
|
||||
comparison is tolerance-based rather than bitwise; geometry is computed in
|
||||
double, keeping the expected difference at the 1e-5 level. Voxels within the
|
||||
smoothing radius of the depth-mask boundary are reported separately: there the
|
||||
float32 knife-edge can flip whole boundary voxels (the same effect exists
|
||||
between two float64 runs with shifted grids).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
MAX_DIFF_TOLERANCE = 1e-4
|
||||
LEVEL_REL_TOLERANCE = 1e-4
|
||||
# Gaussian smoothing radius (sigma 1.5 -> radius 6) around the depth-mask edge.
|
||||
EDGE_EXCLUSION_VOXELS = 7
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||||
parser.add_argument("dump_base", type=Path, help="path prefix of the .f32/.json dump")
|
||||
parser.add_argument("reference_npz", type=Path, help="reference *_volume.npz from the notebook")
|
||||
args = parser.parse_args()
|
||||
|
||||
header = json.loads(Path(str(args.dump_base) + ".json").read_text())
|
||||
shape = (header["ny"], header["nz"], header["nx"])
|
||||
volume = np.fromfile(str(args.dump_base) + ".f32", dtype=np.float32).reshape(shape)
|
||||
|
||||
ref = np.load(args.reference_npz)
|
||||
ref_volume = ref["volume"]
|
||||
meta = json.loads(str(ref["meta"]))
|
||||
print(f"snapshot: {meta['snapshot']}")
|
||||
if volume.shape != ref_volume.shape:
|
||||
print(f"FAIL: shape mismatch {volume.shape} vs {ref_volume.shape}")
|
||||
return 1
|
||||
|
||||
axes_ok = all(
|
||||
np.allclose(np.asarray(header[key]), ref[key])
|
||||
for key in ("x_m", "y_m", "z_m")
|
||||
)
|
||||
level_rel = abs(header["level"] - meta["level"]) / abs(meta["level"])
|
||||
print(f"axes match: {axes_ok}; level rel diff: {level_rel:.3e} "
|
||||
f"(dump {header['level']:.9f}, ref {meta['level']:.9f})")
|
||||
|
||||
diff = np.abs(volume.astype(np.float64) - ref_volume.astype(np.float64))
|
||||
|
||||
# Depth-mask boundary band: voxels whose (iy, iz) sit within the smoothing
|
||||
# radius of a masked/unmasked transition. Reconstructed from the reference
|
||||
# parameters exactly like the processor computes the mask.
|
||||
pitch = np.radians(float(meta.get("ground_pitch_deg") or meta["look_angle_deg"]))
|
||||
y0 = np.asarray(ref["y_m"]) - float(meta["ground_depth_m"])
|
||||
z_rad = -y0[:, None] * np.sin(pitch) + np.asarray(ref["z_m"])[None, :] * np.cos(pitch)
|
||||
mask = (z_rad >= float(meta["min_depth_m"])) & (z_rad <= float(meta["max_depth_m"]))
|
||||
edge = np.zeros_like(mask)
|
||||
r = EDGE_EXCLUSION_VOXELS
|
||||
for shift in range(-r, r + 1):
|
||||
edge |= np.roll(mask, shift, axis=1) != mask
|
||||
interior = ~edge
|
||||
|
||||
argmax = np.unravel_index(int(np.argmax(volume)), volume.shape)
|
||||
ref_argmax = np.unravel_index(int(np.argmax(ref_volume)), ref_volume.shape)
|
||||
interior_diff = diff[interior[:, :, None] & np.ones(volume.shape, dtype=bool)]
|
||||
print(f"max |diff| (all voxels) = {diff.max():.3e}")
|
||||
print(f"max |diff| (off depth-mask edge) = {interior_diff.max():.3e}, "
|
||||
f"mean = {interior_diff.mean():.3e}")
|
||||
print(f"argmax: dump {argmax}, ref {ref_argmax}")
|
||||
|
||||
ok = (axes_ok and level_rel <= LEVEL_REL_TOLERANCE and argmax == ref_argmax
|
||||
and interior_diff.max() <= MAX_DIFF_TOLERANCE)
|
||||
print("OK" if ok else "FAIL")
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user