174 lines
8.3 KiB
Python
174 lines
8.3 KiB
Python
#!/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())
|