"""Calibrate the Kamil ADC reference-phase frequency law from live sweeps. Captures many raw sweeps from the collector and reports — and optionally writes back — the calibration the runtime uses: * ``phase0_rad`` / ``phase1_rad`` — the unwrapped reference phase at the sweep start and stop, taken as the MEDIAN over all captured sweeps. ``freq0_hz`` / ``freq1_hz`` (the known sweep endpoints) come from config and are kept as-is. * ``band.points`` — recommended as the median number of usable points landing inside ``[band.start_hz, band.stop_hz]``, so the fixed output grid matches the native density rather than inflating it. It also reports how reliably the configured band is covered (sweeps that do not span it are rejected at runtime) and how many points the crop discards. The collector is launched with the ``do8_freq_ref`` arguments regardless of what the on-disk config says, so the reference channel is always present. With ``--apply`` the config is migrated to that collector/args and the calibrated anchors + recommended point count are written back. Run on the Pi, e.g.:: .venv/bin/python -m python_app.scripts.kamil_adc_calibrate \ --config run_config_kamil_adc.pi.json --sweeps 200 --apply """ from __future__ import annotations import argparse from contextlib import suppress import json import logging from pathlib import Path import statistics import time import numpy as np from python_app.hardware_full.kamil_adc import KamilAdcService, apply_kamil_adc_laser_control from python_app.models.run_config_model import RunConfigModel logger = logging.getLogger("kamil_adc_calibrate") # The collector arguments that enable the DI8 reference overlay (mirrors the # kamil example config / run_do8_freq_ref.sh). Forced on so calibration always # sees the reference channel even if the on-disk config predates it. DO8_FREQ_REF_ARGS = [ "profile:phase", "clock:internal", "internal_ref_hz:2000000", "start:di_syn2_rise", "stop:di_syn2_fall", "sample_clock_hz:max", "range:2", "duration_ms:100", "packet_limit:0", "do1_toggle_per_frame", "do1_pair_subtract_avg", "do8_freq_ref", "do8_cycle_period:8", ] COLLECTOR_PATH = "build/bin/kamil_adc_collector" def _open_with_retry(service: KamilAdcService, *, attempts: int = 4, delay_s: float = 8.0) -> None: """Open the collector, retrying the transient E-502 device-busy after a close. The L-Card device is not always reacquirable immediately after a previous collector released it; a short backoff lets it settle before the next try. """ for attempt in range(1, attempts + 1): try: service.open() return except Exception as exc: # noqa: BLE001 — device-busy is expected and retried logger.warning("Collector open failed (attempt %d/%d): %s", attempt, attempts, exc) with suppress(Exception): service.close() if attempt < attempts: time.sleep(delay_s) raise RuntimeError(f"Collector did not open after {attempts} attempts") def _capture_reference_phases(service: KamilAdcService, *, warmup: int, sweeps: int) -> list[np.ndarray]: """Capture `sweeps` reference-phase arrays after discarding `warmup` sweeps.""" for index in range(warmup): service.read_raw_sweep() if index == 0: logger.info("Warming up (%d sweeps) while the sweep settles...", warmup) phases: list[np.ndarray] = [] for index in range(sweeps): raw = service.read_raw_sweep() if raw.reference.size >= 2: phases.append(np.unwrap(np.angle(raw.reference.astype(np.complex128)))) if (index + 1) % 50 == 0: logger.info("Captured %d/%d sweeps", index + 1, sweeps) return phases def _summarize(values: np.ndarray) -> str: """Compact min / median / max summary for a 1-D array.""" return f"min={np.min(values):.6g} median={np.median(values):.6g} max={np.max(values):.6g}" def _analyze(phases: list[np.ndarray], config: RunConfigModel) -> dict: """Derive the calibration and band diagnostics from captured phase arrays.""" kamil = config.radar.kamil_adc freq0, freq1 = kamil.phase_calibration.freq0_hz, kamil.phase_calibration.freq1_hz band_start, band_stop = kamil.band.start_hz, kamil.band.stop_hz phase0 = float(statistics.median(float(phase[0]) for phase in phases)) phase1 = float(statistics.median(float(phase[-1]) for phase in phases)) if phase1 == phase0: raise RuntimeError("Degenerate calibration: median start and stop phases are equal") slope = (freq1 - freq0) / (phase1 - phase0) total_points = np.array([phase.size for phase in phases], dtype=np.float64) f_starts, f_stops, in_band_counts, covers = [], [], [], [] for phase in phases: freqs = freq0 + (phase - phase0) * slope lo, hi = float(np.min(freqs)), float(np.max(freqs)) f_starts.append(lo) f_stops.append(hi) in_band_counts.append(int(np.count_nonzero((freqs >= band_start) & (freqs <= band_stop)))) covers.append(lo <= band_start and hi >= band_stop) in_band = np.array(in_band_counts, dtype=np.float64) f_start = np.array(f_starts) f_stop = np.array(f_stops) recommended_points = int(round(float(np.median(in_band)))) # A band that ~95% of sweeps satisfy on each edge: start at the 95th percentile # of per-sweep start frequencies, stop at the 5th percentile of stop frequencies. rec_band_start = float(np.quantile(f_start, 0.95)) rec_band_stop = float(np.quantile(f_stop, 0.05)) rec_coverage = float(np.mean((f_start <= rec_band_start) & (f_stop >= rec_band_stop))) return { "sweeps": len(phases), "phase0_rad": phase0, "phase1_rad": phase1, "phase_span_rad": phase1 - phase0, "phase0_mad_rad": float(np.median(np.abs([float(p[0]) - phase0 for p in phases]))), "phase1_mad_rad": float(np.median(np.abs([float(p[-1]) - phase1 for p in phases]))), "freq0_hz": freq0, "freq1_hz": freq1, "band_start_hz": band_start, "band_stop_hz": band_stop, "total_points_median": float(np.median(total_points)), "in_band_points_median": float(np.median(in_band)), "recommended_points": recommended_points, "cropped_fraction": 1.0 - float(np.median(in_band)) / float(np.median(total_points)), "coverage_fraction": float(np.mean(covers)), "lower_ok_fraction": float(np.mean(f_start <= band_start)), "upper_ok_fraction": float(np.mean(f_stop >= band_stop)), "rec_band_start_hz": rec_band_start, "rec_band_stop_hz": rec_band_stop, "rec_coverage_fraction": rec_coverage, "f_start": f_start, "f_stop": f_stop, } def _report(result: dict) -> None: """Print a human-readable calibration report.""" print("\n" + "=" * 72) print(f"Kamil ADC calibration over {result['sweeps']} sweeps") print("=" * 72) print( f"phase0_rad = {result['phase0_rad']:.6f} (median start phase, MAD " f"{result['phase0_mad_rad']:.4f}) -> {result['freq0_hz'] / 1e9:.4f} GHz" ) print( f"phase1_rad = {result['phase1_rad']:.6f} (median stop phase, MAD " f"{result['phase1_mad_rad']:.4f}) -> {result['freq1_hz'] / 1e9:.4f} GHz" ) print(f"phase span = {result['phase_span_rad']:.4f} rad") print( f"points/sweep: total median={result['total_points_median']:.0f}, " f"in-band median={result['in_band_points_median']:.0f}" ) print(f"recommended band.points = {result['recommended_points']}") print( f"band [{result['band_start_hz'] / 1e9:.3f}, {result['band_stop_hz'] / 1e9:.3f}] GHz: " f"covered by {result['coverage_fraction'] * 100:.1f}% of sweeps " f"(start<=lo: {result['lower_ok_fraction'] * 100:.1f}%, stop>=hi: {result['upper_ok_fraction'] * 100:.1f}%), " f"{result['cropped_fraction'] * 100:.1f}% of points cropped" ) print(f"per-sweep start freq (GHz): {_summarize(result['f_start'] / 1e9)}") print(f"per-sweep stop freq (GHz): {_summarize(result['f_stop'] / 1e9)}") print( f"suggested band for ~95%/edge: [{result['rec_band_start_hz'] / 1e9:.3f}, " f"{result['rec_band_stop_hz'] / 1e9:.3f}] GHz -> covers " f"{result['rec_coverage_fraction'] * 100:.1f}% of sweeps" ) if result["coverage_fraction"] < 0.95: print( "WARNING: many sweeps do not cover the configured band and would be rejected; " "consider the suggested band above (or longer laser settling if sweeps are partial)." ) print("=" * 72 + "\n") def _apply(config_path: Path, result: dict) -> None: """Write the migrated collector args + calibrated anchors + points to the config.""" payload = json.loads(config_path.read_text(encoding="utf-8")) kamil = payload.setdefault("radar", {}).setdefault("kamil_adc", {}) kamil["executable_path"] = COLLECTOR_PATH kamil["args"] = list(DO8_FREQ_REF_ARGS) kamil["phase_calibration"] = { "phase0_rad": result["phase0_rad"], "freq0_hz": result["freq0_hz"], "phase1_rad": result["phase1_rad"], "freq1_hz": result["freq1_hz"], } kamil["band"] = { "start_hz": result["band_start_hz"], "stop_hz": result["band_stop_hz"], "points": result["recommended_points"], } config_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") print(f"Applied calibration to {config_path}") def main() -> int: parser = argparse.ArgumentParser(description="Calibrate Kamil ADC reference phase -> frequency") parser.add_argument("--config", required=True, type=Path, help="Path to the kamil_adc run config") parser.add_argument("--sweeps", type=int, default=200, help="Sweeps to average (default 200)") parser.add_argument("--warmup", type=int, default=10, help="Sweeps to discard first (default 10)") parser.add_argument("--apply", action="store_true", help="Write the calibration back to --config") parser.add_argument("--no-laser", action="store_true", help="Skip laser setup (already running)") args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") config = RunConfigModel.load_from_path(args.config) if not config.is_kamil_adc: raise SystemExit("Config is not a kamil_adc profile") # Force the reference-producing collector regardless of the on-disk config. config.radar.kamil_adc.executable_path = COLLECTOR_PATH config.radar.kamil_adc.args = list(DO8_FREQ_REF_ARGS) if not args.no_laser: logger.info("Applying laser control...") apply_kamil_adc_laser_control(config) service = KamilAdcService(config) logger.info("Opening collector: %s", " ".join(service.command)) _open_with_retry(service) try: phases = _capture_reference_phases(service, warmup=args.warmup, sweeps=args.sweeps) finally: service.close() if len(phases) < max(2, args.sweeps // 2): raise SystemExit(f"Only {len(phases)} usable sweeps captured; check the reference signal") result = _analyze(phases, config) _report(result) if args.apply: _apply(args.config, result) return 0 if __name__ == "__main__": raise SystemExit(main())