#!/usr/bin/env python3 """Replay SignalSim IQ through E1B-only and E1C-aided receiver modes.""" from __future__ import annotations import argparse from dataclasses import asdict import json from pathlib import Path import sys import numpy as np PROJECT_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(PROJECT_ROOT / "src")) from galileo_rx.diagnostics.boc_lock_validation import ( boc11_autocorrelation, boc11_five_tap_shape, calibration_points, plot_boc11_lock_calibration, ) from galileo_rx.navigation.collection import collect_navigation from galileo_rx.signal_profiles import galileo_e1_profile from galileo_rx.signals.signalsim_codes import SignalSimCodeProvider def _satellite_row(item: dict) -> dict: lock = item["lock_analysis"] stable = lock["post_pull_in"] return { "prn": int(item["prn"]), "legacy_code_lock_falling_edges": int(lock["legacy_code_lock_falling_edges"]), "boc_aware_code_lock_falling_edges": int(lock["boc_aware_code_lock_falling_edges"]), "longest_boc_aware_code_lock_s": float(lock["boc_aware_longest_code_lock_s"]), "pll_error_rms_rad": float(stable["pll_error_rms_rad"]), "dll_error_rms_chips": float(stable["dll_error_rms_chips"]), "cn0_median_db_hz": float(stable["cn0_median_db_hz"]), "cycle_slips": int(item["cycle_slips"]), "sample_discontinuities": int(item["sample_discontinuities"]), "raw_navigation_symbols": int(item["raw_navigation_symbol_count"]), "crc_valid_pages": int(item["crc_passes"]), "crc_invalid_pages": int(item["crc_failures"]), "complete_ephemeris": bool(item["complete_ephemeris"]), "secondary_code_sync": item["secondary_code_sync"], } def _plot_comparison(path: Path, comparison: dict) -> None: import matplotlib.pyplot as plt modes = list(comparison["modes"]) prns = [int(value) for value in comparison["prns"]] x = np.arange(len(prns), dtype=np.float64) width = 0.36 figure, axes = plt.subplots(3, 2, figsize=(13, 11)) metrics = ( ("boc_aware_code_lock_falling_edges", "corrected lock falling edges"), ("longest_boc_aware_code_lock_s", "longest code lock (s)"), ("pll_error_rms_rad", "PLL RMS (rad)"), ("dll_error_rms_chips", "DLL RMS (chips)"), ("cn0_median_db_hz", "median C/N0 (dB-Hz)"), ("crc_valid_pages", "CRC-valid I/NAV pages"), ) for axis, (field, label) in zip(axes.flat, metrics, strict=True): for mode_index, mode in enumerate(modes): rows = comparison["modes"][mode]["satellites"] values = [rows[str(prn)][field] for prn in prns] axis.bar(x + (mode_index - 0.5) * width, values, width, label=mode) axis.set_xticks(x, [str(prn) for prn in prns]) axis.set_xlabel("PRN") axis.set_ylabel(label) axis.grid(True, axis="y", alpha=0.25) axes[0, 0].legend(fontsize=8) figure.tight_layout() figure.savefig(path, dpi=160) plt.close(figure) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--config", default="configs/navigation_collection.yaml") parser.add_argument("--output-dir", default="reports/e1_tracking_mode_comparison") parser.add_argument("--signalsim-path") parser.add_argument( "--report-only", action="store_true", help="regenerate plots/Markdown from the existing comparison JSON", ) args = parser.parse_args() output = Path(args.output_dir).resolve() output.mkdir(parents=True, exist_ok=True) comparison_path = output / "e1_tracking_mode_comparison.json" if args.report_only: if not comparison_path.is_file(): raise FileNotFoundError(f"comparison JSON is unavailable: {comparison_path}") comparison = json.loads(comparison_path.read_text(encoding="utf-8")) passed = bool(comparison["all_navigation_ready"]) else: comparison = {"modes": {}, "prns": []} passed = True for mode in ("e1b_only", "e1c_pilot_aided_e1b"): report, ready = collect_navigation( args.config, output / mode, signalsim_path=args.signalsim_path, tracking_mode=mode, ) mode_rows = { str(prn): _satellite_row(item) for prn, item in report["satellites"].items() } comparison["modes"][mode] = { "navigation_ready": bool(ready), "satellites": mode_rows, "total_crc_valid_pages": sum(row["crc_valid_pages"] for row in mode_rows.values()), "complete_ephemerides": sum(row["complete_ephemeris"] for row in mode_rows.values()), } comparison["prns"] = sorted(int(prn) for prn in mode_rows) passed = passed and ready input_report_path = output / "e1b_only" / "inav_collection_report.json" input_report = json.loads(input_report_path.read_text(encoding="utf-8")) comparison["input"] = { "sample_file": input_report["sample_file"], "sample_sha256": input_report["sample_sha256"], "complex_sample_count": input_report["sample_count"], "sampling_frequency_hz": input_report["sampling_frequency_hz"], "stored_duration_s": input_report["sample_count"] / input_report["sampling_frequency_hz"], "configured_duration_s": 32.2, } source = Path(args.signalsim_path or PROJECT_ROOT / "signalsim" / "SignalSim").resolve() provider = SignalSimCodeProvider(source) profile = galileo_e1_profile() shape = boc11_autocorrelation(provider.primary_code(profile, "E1B", comparison["prns"][0])) errors = np.linspace(-1.0, 1.0, 513, dtype=np.float64) taps = boc11_five_tap_shape(shape, errors) comparison["boc11_calibration_points"] = calibration_points(taps) comparison["all_navigation_ready"] = passed plot_boc11_lock_calibration(output / "boc11_correlation_and_lock_metrics.png", shape, taps) _plot_comparison(output / "e1_tracking_mode_comparison.png", comparison) comparison_path.write_text( json.dumps(comparison, indent=2) + "\n", encoding="utf-8" ) rows = [ "# Corrected BOC lock and E1 tracking architecture comparison", "", f"Both modes consume the same SignalSim IQ8 vector configured for 32.2 s. The file contains {comparison['input']['complex_sample_count']} complex samples ({comparison['input']['stored_duration_s']:.3f} s at {comparison['input']['sampling_frequency_hz']:.0f} Hz), so the stored vector is exactly 1 ms shorter than the requested duration. The E1C mode uses E1C acquisition, secondary-code synchronization/wipeoff, and pilot loop states; only its separate E1B Prompt branch enters I/NAV decoding.", "", "| Mode | PRN | Legacy falls | Corrected falls | Longest lock (s) | PLL RMS (rad) | DLL RMS (chips) | C/N0 (dB-Hz) | Slip candidates | Discontinuities | Secondary corr. | Symbols | CRC pages | Ephemeris |", "| --- | --: | --: | --: | --: | --: | --: | --: | --: | --: | --: | --: | --: | ---: |", ] for mode, mode_result in comparison["modes"].items(): for prn in comparison["prns"]: item = mode_result["satellites"][str(prn)] secondary_correlation = ( "-" if item["secondary_code_sync"] is None else f"{item['secondary_code_sync']['normalized_correlation']:.4f}" ) rows.append( f"| {mode} | {prn} | {item['legacy_code_lock_falling_edges']} | " f"{item['boc_aware_code_lock_falling_edges']} | {item['longest_boc_aware_code_lock_s']:.3f} | " f"{item['pll_error_rms_rad']:.6f} | {item['dll_error_rms_chips']:.6f} | " f"{item['cn0_median_db_hz']:.3f} | {item['cycle_slips']} | " f"{item['sample_discontinuities']} | {secondary_correlation} | {item['raw_navigation_symbols']} | " f"{item['crc_valid_pages']} | {item['complete_ephemeris']} |" ) rows.extend(["", f"All navigation readiness gates passed: **{passed}**."]) (output / "e1_tracking_mode_comparison.md").write_text("\n".join(rows) + "\n", encoding="utf-8") print(json.dumps({"passed": passed, "report": str(output / "e1_tracking_mode_comparison.json")}, indent=2)) return 0 if passed else 1 if __name__ == "__main__": raise SystemExit(main())