kshana 0.37.0

// SPDX-License-Identifier: AGPL-3.0-only
//! Python bindings (PyO3), built with `maturin` and the `python` feature.
//!
//! ```python
//! import kshana, json
//! result = json.loads(kshana.run(open("scenarios/clock-holdover.toml").read()))
//! print(kshana.version())
//! ```

// The #[pyfunction] macro expands to a `.into()` on the PyErr return, which Clippy
// flags as a useless conversion in macro-generated code we don't control.
#![allow(clippy::useless_conversion)]

use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
use pyo3::types::{PyDict, PyList};

/// Recursively convert a `serde_json::Value` into a native Python object
/// (`dict`/`list`/`str`/`int`/`float`/`bool`/`None`), so callers get structured
/// data instead of a JSON string to re-parse.
fn json_to_py<'py>(py: Python<'py>, v: &serde_json::Value) -> PyResult<pyo3::Bound<'py, PyAny>> {
    use serde_json::Value;
    Ok(match v {
        Value::Null => py.None().into_bound(py),
        Value::Bool(b) => (*b).into_pyobject(py)?.to_owned().into_any(),
        Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                i.into_pyobject(py)?.into_any()
            } else if let Some(u) = n.as_u64() {
                u.into_pyobject(py)?.into_any()
            } else {
                n.as_f64().unwrap_or(f64::NAN).into_pyobject(py)?.into_any()
            }
        }
        Value::String(s) => s.into_pyobject(py)?.into_any(),
        Value::Array(a) => {
            let list = PyList::empty(py);
            for item in a {
                list.append(json_to_py(py, item)?)?;
            }
            list.into_any()
        }
        Value::Object(o) => {
            let dict = PyDict::new(py);
            for (k, val) in o {
                dict.set_item(k, json_to_py(py, val)?)?;
            }
            dict.into_any()
        }
    })
}

/// A scenario run result: the result JSON, the chart SVG, the human summary, and
/// — for the scenario kinds that emit one — the reproducibility table CSV, with a
/// `data()` accessor that returns the parsed result as a Python dict.
// pyo3 0.29 makes the auto `FromPyObject` derive for `Clone` pyclasses opt-in.
// `RunOutput` is only ever returned to Python, never accepted as an argument, so
// we explicitly skip the derive rather than opt in.
#[pyclass(name = "RunOutput", skip_from_py_object)]
#[derive(Clone)]
struct PyRunOutput {
    #[pyo3(get)]
    json: String,
    #[pyo3(get)]
    svg: String,
    #[pyo3(get)]
    summary: String,
    /// The reproducibility table the CLI writes as `<scenario>.table.csv`, for
    /// the kinds that emit one (`realtime-frame-eop`, `lunar-time-budget`,
    /// `lunar-jamming`, and `moonlight-service-volume` only when both
    /// `export_site_lat_deg` and `export_site_lon_deg` are set — the shipped
    /// scenario leaves them commented out, so it emits none); `None` otherwise. It is the
    /// byte-stable, golden-pinned form of the table the papers cite, so a
    /// reviewer reproducing one from the wheel needs it here and not only from
    /// the command line.
    #[pyo3(get)]
    csv: Option<String>,
}

impl PyRunOutput {
    fn parsed(&self) -> PyResult<serde_json::Value> {
        serde_json::from_str(&self.json).map_err(|e| PyValueError::new_err(e.to_string()))
    }
}

/// Import NumPy on demand: the package has no runtime NumPy dependency, so only the
/// array accessors need it.
fn numpy(py: Python<'_>) -> PyResult<pyo3::Bound<'_, pyo3::types::PyModule>> {
    py.import("numpy").map_err(|e| {
        pyo3::exceptions::PyImportError::new_err(format!(
            "RunOutput's array accessors need NumPy (pip install numpy): {e}"
        ))
    })
}

/// Resolve a dotted path (`a.b.0.c`) or a JSON Pointer (`/a/b/0/c`) in `doc`.
fn lookup<'a>(doc: &'a serde_json::Value, path: &str) -> PyResult<&'a serde_json::Value> {
    let missing = || pyo3::exceptions::PyKeyError::new_err(path.to_string());
    if path.starts_with('/') {
        return doc.pointer(path).ok_or_else(missing);
    }
    if path.is_empty() {
        return Ok(doc);
    }
    let mut v = doc;
    for part in path.split('.') {
        v = match v {
            serde_json::Value::Object(o) => o.get(part),
            serde_json::Value::Array(a) => part.parse::<usize>().ok().and_then(|i| a.get(i)),
            _ => None,
        }
        .ok_or_else(missing)?;
    }
    Ok(v)
}

/// A list of numbers, or a non-empty list of equal-length lists of numbers.
fn is_numeric_array(v: &serde_json::Value) -> bool {
    let Some(a) = v.as_array() else { return false };
    if a.iter().all(|x| x.is_number()) {
        return true;
    }
    let rows: Option<Vec<&Vec<serde_json::Value>>> = a.iter().map(|r| r.as_array()).collect();
    match rows {
        Some(rows) => {
            let n = rows[0].len();
            rows.iter()
                .all(|r| r.len() == n && r.iter().all(|x| x.is_number()))
        }
        None => false,
    }
}

/// The fields that are numbers in every record, in first-record order.
fn numeric_fields(records: &[serde_json::Value]) -> Vec<&str> {
    let Some(first) = records.first().and_then(|r| r.as_object()) else {
        return Vec::new();
    };
    first
        .keys()
        .filter(|k| {
            records
                .iter()
                .all(|r| r.get(k.as_str()).is_some_and(|x| x.is_number()))
        })
        .map(String::as_str)
        .collect()
}

fn records_to_columns<'py>(
    py: Python<'py>,
    np: &pyo3::Bound<'py, pyo3::types::PyModule>,
    records: &[serde_json::Value],
) -> PyResult<pyo3::Bound<'py, PyAny>> {
    let out = PyDict::new(py);
    for name in numeric_fields(records) {
        let list = PyList::empty(py);
        for r in records {
            list.append(json_to_py(py, &r[name])?)?;
        }
        out.set_item(name, np.call_method1("asarray", (list,))?)?;
    }
    Ok(out.into_any())
}

/// Collect every non-empty list of records with at least one numeric field, by path.
fn collect_series<'a>(
    v: &'a serde_json::Value,
    path: String,
    out: &mut Vec<(String, &'a [serde_json::Value])>,
) {
    let join = |k: &str| {
        if path.is_empty() {
            k.to_string()
        } else {
            format!("{path}.{k}")
        }
    };
    match v {
        serde_json::Value::Object(o) => {
            for (k, x) in o {
                collect_series(x, join(k), out);
            }
        }
        serde_json::Value::Array(a) => {
            if !a.is_empty() && a.iter().all(|r| r.is_object()) && !numeric_fields(a).is_empty() {
                out.push((path.clone(), a.as_slice()));
            }
            for (i, x) in a.iter().enumerate() {
                collect_series(x, join(&i.to_string()), out);
            }
        }
        _ => {}
    }
}

#[pymethods]
impl PyRunOutput {
    /// The result document parsed into a Python `dict` (figures of merit,
    /// time series, provenance, ...). For NumPy arrays use [`array`](Self::array),
    /// [`columns`](Self::columns) or [`series`](Self::series).
    fn data<'py>(&self, py: Python<'py>) -> PyResult<pyo3::Bound<'py, PyAny>> {
        let v: serde_json::Value =
            serde_json::from_str(&self.json).map_err(|e| PyValueError::new_err(e.to_string()))?;
        json_to_py(py, &v)
    }

    /// The array at `path` in the result document as a NumPy `ndarray`: a list of
    /// numbers gives a 1-D array, a list of equal-length numeric lists a 2-D one.
    /// `path` is dotted (`"classical.adev_curve"`, list indices as integers:
    /// `"points.0.coords"`) or a JSON Pointer (`"/classical/adev_curve"`). The array is
    /// exactly `numpy.asarray(...)` of the same list from [`data`](Self::data), so the
    /// dtype and values are the list path's. NumPy is imported when this is called and is
    /// not a dependency of the package; raises `ImportError` without it, `KeyError` for a
    /// missing path and `ValueError` when the value is not a numeric array.
    fn array<'py>(&self, py: Python<'py>, path: &str) -> PyResult<pyo3::Bound<'py, PyAny>> {
        let doc = self.parsed()?;
        let v = lookup(&doc, path)?;
        if !is_numeric_array(v) {
            return Err(PyValueError::new_err(format!(
                "{path:?} is not a numeric array (a list of numbers, or of equal-length lists of numbers)"
            )));
        }
        let np = numpy(py)?;
        np.call_method1("asarray", (json_to_py(py, v)?,))
    }

    /// The records at `path` (a list of objects, such as a time series) as a `dict` of
    /// column name to NumPy `ndarray`, one entry per field that is a number in every
    /// record; text, nested and sometimes-missing fields are left out. Each column is
    /// `numpy.asarray([r[name] for r in records])` of the [`data`](Self::data) list, so it
    /// keeps the list path's dtype and values. `path` is written as for
    /// [`array`](Self::array); the same errors apply.
    fn columns<'py>(&self, py: Python<'py>, path: &str) -> PyResult<pyo3::Bound<'py, PyAny>> {
        let doc = self.parsed()?;
        let v = lookup(&doc, path)?;
        let records = v
            .as_array()
            .filter(|a| !a.is_empty() && a.iter().all(|r| r.is_object()))
            .ok_or_else(|| {
                PyValueError::new_err(format!("{path:?} is not a non-empty list of records"))
            })?;
        let np = numpy(py)?;
        records_to_columns(py, &np, records)
    }

    /// Every record series in the result document as NumPy columns: a `dict` keyed by the
    /// dotted path of each non-empty list of records with at least one numeric field, whose
    /// value is that list's [`columns`](Self::columns). A series nested in a record of
    /// another is included under its own path (`"points.0.legs"`). Lists of plain numbers
    /// are not series; read those with [`array`](Self::array).
    fn series<'py>(&self, py: Python<'py>) -> PyResult<pyo3::Bound<'py, PyAny>> {
        let doc = self.parsed()?;
        let np = numpy(py)?;
        let out = PyDict::new(py);
        let mut found = Vec::new();
        collect_series(&doc, String::new(), &mut found);
        for (path, records) in found {
            out.set_item(path, records_to_columns(py, &np, records)?)?;
        }
        Ok(out.into_any())
    }

    /// Write the reproducibility table to `path`, returning the number of bytes
    /// written — 0 when this scenario kind emits no table. Mirrors
    /// `kshana::api::RunOutput::write_csv`, which is what the CLI uses.
    fn write_csv(&self, path: &str) -> PyResult<usize> {
        match &self.csv {
            Some(csv) => std::fs::write(path, csv)
                .map(|()| csv.len())
                .map_err(|e| PyValueError::new_err(e.to_string())),
            None => Ok(0),
        }
    }

    fn __repr__(&self) -> String {
        format!(
            "RunOutput(json={} bytes, svg={} bytes, csv={} bytes, summary={:?})",
            self.json.len(),
            self.svg.len(),
            self.csv.as_ref().map_or(0, |c| c.len()),
            self.summary.chars().take(48).collect::<String>()
        )
    }
}

/// Run a scenario from a TOML string and return a typed [`RunOutput`] (with
/// `.json`, `.svg`, `.summary`, `.csv`, `.write_csv()`, `.data()`, and the NumPy
/// accessors `.array()`, `.columns()` and `.series()`). Raises
/// `ValueError` if invalid.
#[pyfunction]
fn run_typed(toml: &str) -> PyResult<PyRunOutput> {
    crate::api::run_toml(toml)
        .map(|o| PyRunOutput {
            json: o.json,
            svg: o.svg,
            summary: o.summary,
            csv: o.csv,
        })
        .map_err(PyValueError::new_err)
}

/// The available scenario kinds and their metadata, as a Python list of dicts
/// (name, description, required/optional fields) — introspectable without source.
/// Assess a real receiver log described by a `receiver-trust` scenario (TOML text). The
/// log must be given inline (`text` or `base64`) or by a path the Python process can
/// read. A `[platform]` table with `kind = "vessel"` selects the maritime monitors and the
/// 0-100 trust score with its reasons (`docs/MARITIME-TRUST.md`); the output is advisory. Advisory software, not type-approved navigation equipment (IEC 61108, IEC 61162): the operator remains responsible for the navigation of the vessel.
/// `receiver-trust live` (a long-running stream process with an optional TCP listener) is
/// command-line only. Returns the result document, the trust-timeline CSV, the chart and a
/// summary.
#[pyfunction]
fn receiver_trust(toml: &str) -> PyResult<PyRunOutput> {
    crate::receiver_trust::scenario::run_toml(toml)
        .map(|o| PyRunOutput {
            json: o.json,
            svg: o.svg,
            summary: o.summary,
            csv: Some(o.csv),
        })
        .map_err(PyValueError::new_err)
}

#[pyfunction]
fn scenario_kinds<'py>(py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
    let json = crate::api::list_scenario_kinds_json();
    let v: serde_json::Value =
        serde_json::from_str(&json).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Validate a scenario TOML string without raising: returns a list of error
/// messages (empty if the scenario is valid). Note: this executes the scenario,
/// so it surfaces both parse/config errors and runtime failures.
#[pyfunction]
fn validate_toml(toml: &str) -> Vec<String> {
    match crate::api::run_scenario(toml) {
        Ok(_) => vec![],
        Err(e) => vec![e.to_string()],
    }
}

/// Run a scenario given as a TOML string; returns the result document as a JSON
/// string. Raises `ValueError` if the scenario is invalid.
#[pyfunction]
fn run(toml: &str) -> PyResult<String> {
    crate::api::run_toml(toml)
        .map(|o| o.json)
        .map_err(PyValueError::new_err)
}

/// Run a scenario; returns `(json, svg, summary)`.
#[pyfunction]
fn run_full(toml: &str) -> PyResult<(String, String, String)> {
    crate::api::run_toml(toml)
        .map(|o| (o.json, o.svg, o.summary))
        .map_err(PyValueError::new_err)
}

/// List the available scenario kinds and their metadata as ONE JSON-array string (name,
/// description, required and optional fields) — a `str`, not a Python list. It predates
/// [`scenario_kinds`], which returns the same metadata already parsed, and keeps its
/// string return so existing callers do not break; new code should call `scenario_kinds`.
#[pyfunction]
fn list_kinds() -> String {
    crate::api::list_scenario_kinds_json()
}

/// Run a scenario; on failure return the structured error *kind* tag
/// (`invalid_input`, `non_convergence`, `unsupported`, `io_error`) instead of
/// raising, so a caller can pattern-match on the failure category. Returns `None`
/// on success.
#[pyfunction]
fn error_kind(toml: &str) -> Option<String> {
    crate::api::run_scenario(toml)
        .err()
        .map(|e| e.kind_tag().to_string())
}

/// Engine version (the crate version).
#[pyfunction]
fn version() -> &'static str {
    env!("CARGO_PKG_VERSION")
}

// --- GNSS IQ layer (src/iq) ---------------------------------------------------------
//
// These expose the signal-level simulation and software-receiver processing of the IQ
// layer with the same conventions as the `kshana iq` CLI, returning native Python data
// (lists and dicts) that `numpy.asarray(...)` wraps directly. The scene generator, the
// acquisition and tracking engines and the loop designs are the crate's own
// (`kshana::iq`), so a Python caller gets the same bits as the CLI and the Rust tests.

use crate::iq::acq::{acquire_peak, AcqConfig};
use crate::iq::cli::{
    build_broadcast_scene, build_chain, build_channel, build_code, build_scene, BroadcastParams,
    ChannelParams, FrontendParams, SceneParams,
};
use crate::iq::scene::TruthRecord;
use crate::iq::signals::SignalCode;
use crate::iq::track::{ChannelInit, EpochOutput};
use crate::iq::{Cf64, SampleSpec, SpreadingCode, VecSink};

/// Build one spreading code per identifier in `prns`.
fn codes_for(signal: &str, prns: &[i64]) -> PyResult<Vec<SignalCode>> {
    if prns.is_empty() {
        return Err(PyValueError::new_err("prns must list at least one PRN"));
    }
    prns.iter()
        .map(|&id| build_code(signal, id).map_err(PyValueError::new_err))
        .collect()
}

/// Interleave two equal-length real sequences into complex samples.
fn samples_from(i: &[f64], q: &[f64]) -> PyResult<Vec<Cf64>> {
    if i.len() != q.len() {
        return Err(PyValueError::new_err(format!(
            "i and q must be the same length ({} vs {})",
            i.len(),
            q.len()
        )));
    }
    Ok(i.iter()
        .zip(q)
        .map(|(&re, &im)| Cf64::new(re, im))
        .collect())
}

/// Generate a multi-satellite GNSS IQ scene in memory. Returns a dict with the sampling
/// (`fs_hz`, `center_hz`, `if_hz`), the complex samples as two float lists (`samples_i`,
/// `samples_q`) and the per-epoch `truth` records. `prns` is the PRN per satellite (the
/// FDMA frequency channel for GLONASS); `dopplers` is one Doppler (Hz) per PRN, a single
/// value applied to all, or omitted for zero. An optional propagation channel is applied to
/// every satellite through the channel knobs: `iono_stec`/`iono_vtec` (TECU) or
/// `iono_klobuchar`, `tropo` (with `tropo_doy`), scintillation `s4`/`scint_tau0`/`sigma_phi`,
/// `multipath_height` (m) with `multipath_ground` (`dry`/`wet`/`sea`), `land_mobile`, and
/// `nlos`. `cn0_profile` is a C/N0 profile file's text ([`crate::iq::channel::cn0_profile`]):
/// time-varying C/N0 offsets per satellite applied over the channel, with the truth's
/// `cn0_dbhz` following them. Raises `ValueError` on an invalid scene, channel or profile.
#[pyfunction]
#[pyo3(signature = (fs_hz, duration_s, signal, prns, dopplers=None, cn0_dbhz=None, center_hz=None, if_hz=0.0, noise=true, noise_figure_db=2.0, seed=1, data=false, threads=1, iono_stec=None, iono_vtec=None, iono_klobuchar=false, tropo=false, tropo_doy=180.0, s4=None, scint_tau0=1.0, sigma_phi=0.0, multipath_height=None, multipath_ground="dry".to_string(), land_mobile=false, nlos=false, cn0_profile=None))]
#[allow(clippy::too_many_arguments)]
fn iq_scene<'py>(
    py: Python<'py>,
    fs_hz: f64,
    duration_s: f64,
    signal: String,
    prns: Vec<i64>,
    dopplers: Option<Vec<f64>>,
    cn0_dbhz: Option<f64>,
    center_hz: Option<f64>,
    if_hz: f64,
    noise: bool,
    noise_figure_db: f64,
    seed: u64,
    data: bool,
    threads: usize,
    iono_stec: Option<f64>,
    iono_vtec: Option<f64>,
    iono_klobuchar: bool,
    tropo: bool,
    tropo_doy: f64,
    s4: Option<f64>,
    scint_tau0: f64,
    sigma_phi: f64,
    multipath_height: Option<f64>,
    multipath_ground: String,
    land_mobile: bool,
    nlos: bool,
    cn0_profile: Option<String>,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::channel::cn0_profile::{Cn0Profile, Cn0ProfileChannel, ProfiledTruth};
    let profile = cn0_profile
        .map(|t| Cn0Profile::parse(&t).map_err(|e| PyValueError::new_err(e.to_string())))
        .transpose()?;
    let params = SceneParams {
        fs_hz,
        duration_s,
        signal,
        ids: prns,
        dopplers: dopplers.unwrap_or_default(),
        center_hz,
        if_hz,
        cn0_dbhz,
        noise_figure_db,
        noise,
        seed,
        data,
        threads,
    };
    let mut scene = build_scene(&params).map_err(PyValueError::new_err)?;
    let spec = scene.config().spec;
    // Optional propagation channel.
    let chan = ChannelParams {
        iono_stec_tecu: iono_stec,
        iono_vtec_tecu: iono_vtec,
        iono_klobuchar,
        tropo,
        tropo_doy,
        s4,
        scint_tau0_s: scint_tau0,
        sigma_phi_rad: sigma_phi,
        multipath_height_m: multipath_height,
        multipath_ground,
        land_mobile,
        nlos,
    };
    let mut inner = None;
    if chan.any() {
        let carrier = scene
            .satellites()
            .first()
            .map(|s| s.code.carrier_hz())
            .unwrap_or(spec.center_hz);
        let start_tow = scene.config().start_tow_s;
        inner =
            build_channel(&chan, carrier, params.seed, start_tow).map_err(PyValueError::new_err)?;
    }
    match (&profile, inner) {
        (Some(p), inner) => scene.set_channel(Box::new(Cn0ProfileChannel::new(p.clone(), inner))),
        (None, Some(ch)) => scene.set_channel(ch),
        (None, None) => {}
    }
    let mut sink = VecSink::default();
    let mut truth: Vec<TruthRecord> = Vec::new();
    match profile {
        // The truth states the profiled C/N0.
        Some(p) => scene.generate(&mut sink, &mut ProfiledTruth::new(p, &mut truth)),
        None => scene.generate(&mut sink, &mut truth),
    }
    .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let i: Vec<f64> = sink.samples.iter().map(|s| s.re).collect();
    let q: Vec<f64> = sink.samples.iter().map(|s| s.im).collect();
    let truth_json: Vec<serde_json::Value> = truth
        .iter()
        .map(|r| serde_json::to_value(r).unwrap_or(serde_json::Value::Null))
        .collect();
    let v = serde_json::json!({
        "fs_hz": spec.fs_hz,
        "center_hz": spec.center_hz,
        "if_hz": spec.if_hz,
        "samples_i": i,
        "samples_q": q,
        "truth": truth_json,
    });
    json_to_py(py, &v)
}

/// Generate a broadcast-ephemeris GNSS IQ scene in memory from RINEX navigation text.
/// Each requested GPS PRN (or every healthy GPS satellite when `prns` is omitted) is placed
/// at its true broadcast geometry for a receiver at `(rx_lat, rx_lon, rx_alt)` (degrees,
/// degrees, metres) over the window starting at GPS time of week `start_tow`. Returns the
/// same dict as [`iq_scene`] (sampling, `samples_i`/`samples_q`, per-epoch `truth`). Raises
/// `ValueError` on an invalid file or scene.
#[pyfunction]
#[pyo3(signature = (fs_hz, window_s, nav_text, rx_lat, rx_lon, rx_alt, prns=None, start_tow=0.0, cn0_dbhz=None, center_hz=None, if_hz=0.0, noise=true, noise_figure_db=2.0, seed=1, mask_deg=5.0, threads=1))]
#[allow(clippy::too_many_arguments)]
fn iq_scene_broadcast<'py>(
    py: Python<'py>,
    fs_hz: f64,
    window_s: f64,
    nav_text: String,
    rx_lat: f64,
    rx_lon: f64,
    rx_alt: f64,
    prns: Option<Vec<i64>>,
    start_tow: f64,
    cn0_dbhz: Option<f64>,
    center_hz: Option<f64>,
    if_hz: f64,
    noise: bool,
    noise_figure_db: f64,
    seed: u64,
    mask_deg: f64,
    threads: usize,
) -> PyResult<Bound<'py, PyAny>> {
    let bp = BroadcastParams {
        fs_hz,
        duration_s: window_s,
        nav_text,
        prns: prns.unwrap_or_default(),
        start_tow_s: start_tow,
        rx_llh_deg: (rx_lat, rx_lon, rx_alt),
        center_hz,
        if_hz,
        cn0_dbhz,
        noise,
        noise_figure_db,
        seed,
        elevation_mask_deg: mask_deg,
        threads,
    };
    let scene = build_broadcast_scene(&bp).map_err(PyValueError::new_err)?;
    let spec = scene.config().spec;
    let mut sink = VecSink::default();
    let mut truth: Vec<TruthRecord> = Vec::new();
    scene
        .generate(&mut sink, &mut truth)
        .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let i: Vec<f64> = sink.samples.iter().map(|s| s.re).collect();
    let q: Vec<f64> = sink.samples.iter().map(|s| s.im).collect();
    let truth_json: Vec<serde_json::Value> = truth
        .iter()
        .map(|r| serde_json::to_value(r).unwrap_or(serde_json::Value::Null))
        .collect();
    let v = serde_json::json!({
        "fs_hz": spec.fs_hz,
        "center_hz": spec.center_hz,
        "if_hz": spec.if_hz,
        "samples_i": i,
        "samples_q": q,
        "truth": truth_json,
    });
    json_to_py(py, &v)
}

/// FFT acquisition of each PRN over complex samples (`i`/`q` lists sampled at `fs_hz`).
/// Returns a list of detection dicts (`code`, `acquired`, `doppler_hz`, `code_phase_chips`,
/// `statistic`, `threshold`, `peak_ratio`, ...). Raises `ValueError` on a bad search.
#[pyfunction]
#[pyo3(signature = (i, q, fs_hz, signal, prns, if_hz=0.0, center_hz=None, coherent=1, noncoherent=1, doppler_max=5000.0, doppler_step=None, pfa=1e-3))]
#[allow(clippy::too_many_arguments)]
fn iq_acquire<'py>(
    py: Python<'py>,
    i: Vec<f64>,
    q: Vec<f64>,
    fs_hz: f64,
    signal: String,
    prns: Vec<i64>,
    if_hz: f64,
    center_hz: Option<f64>,
    coherent: usize,
    noncoherent: usize,
    doppler_max: f64,
    doppler_step: Option<f64>,
    pfa: f64,
) -> PyResult<Bound<'py, PyAny>> {
    let codes = codes_for(&signal, &prns)?;
    let samples = samples_from(&i, &q)?;
    let spec = SampleSpec {
        fs_hz,
        center_hz: center_hz.unwrap_or_else(|| codes[0].carrier_hz()),
        if_hz,
    };
    let coherent = coherent.max(1);
    let cfg = AcqConfig {
        coherent_periods: coherent,
        noncoherent: noncoherent.max(1),
        doppler_max_hz: doppler_max,
        doppler_step_hz: doppler_step
            .unwrap_or(2.0 / (3.0 * coherent as f64 * codes[0].period_s())),
        pfa,
    };
    let mut out = Vec::new();
    for code in &codes {
        let r = acquire_peak(&samples, &spec, code, &cfg).map_err(PyValueError::new_err)?;
        out.push(serde_json::json!({
            "code": r.code_name,
            "acquired": r.acquired,
            "doppler_hz": r.doppler_hz,
            "code_phase_chips": r.code_phase_chips,
            "delay_samples": r.delay_samples,
            "statistic": r.statistic,
            "threshold": r.threshold,
            "peak_ratio": r.peak_ratio,
            "second_peak": r.second_peak,
            "sample_power": r.sample_power,
        }));
    }
    json_to_py(py, &serde_json::Value::Array(out))
}

/// The whole acquisition surface of one PRN over complex samples (`kshana.acq-surface/1`):
/// a dict with `header` (the search, the Doppler bins, the peak, and the `parabolic` and
/// `fine_search` Doppler refinements) and `rows` (`rows[doppler_index][lag]`, the normalised
/// correlation power). Raises `ValueError` on a bad search.
#[pyfunction]
#[pyo3(signature = (i, q, fs_hz, signal, prn, if_hz=0.0, center_hz=None, coherent=1, noncoherent=1, doppler_max=5000.0, doppler_step=None, pfa=1e-3))]
#[allow(clippy::too_many_arguments)]
fn iq_acq_surface<'py>(
    py: Python<'py>,
    i: Vec<f64>,
    q: Vec<f64>,
    fs_hz: f64,
    signal: String,
    prn: i64,
    if_hz: f64,
    center_hz: Option<f64>,
    coherent: usize,
    noncoherent: usize,
    doppler_max: f64,
    doppler_step: Option<f64>,
    pfa: f64,
) -> PyResult<Bound<'py, PyAny>> {
    let codes = codes_for(&signal, &[prn])?;
    let samples = samples_from(&i, &q)?;
    let spec = SampleSpec {
        fs_hz,
        center_hz: center_hz.unwrap_or_else(|| codes[0].carrier_hz()),
        if_hz,
    };
    let coherent = coherent.max(1);
    let cfg = AcqConfig {
        coherent_periods: coherent,
        noncoherent: noncoherent.max(1),
        doppler_max_hz: doppler_max,
        doppler_step_hz: doppler_step
            .unwrap_or(2.0 / (3.0 * coherent as f64 * codes[0].period_s())),
        pfa,
    };
    let surface = crate::iq::acq_surface::Surface::compute(&samples, &spec, &codes[0], &cfg)
        .map_err(PyValueError::new_err)?;
    json_to_py(
        py,
        &serde_json::json!({ "header": surface.header, "rows": surface.grid }),
    )
}

/// Acquire then track each PRN over complex samples. Returns a dict with `fs_hz`, the
/// resolved loop `design` (every field, with its `hash`), the lock-state `events`, any
/// `warnings` (`commensurate_sampling` when `fs_hz` is a multiple of half the chip rate:
/// code-loop jitter and bias are then not representative) and one
/// entry per channel (`code` and a list of per-epoch dicts: `doppler_hz`,
/// `code_phase_chips`, `pli`, `phase_lock`, `cn0_nwpr_dbhz`, the early/prompt/late
/// correlators, the discriminators, `state`, ...). The loop design is `design` (a path to a
/// `kshana.loop-design/1` TOML file, or its text; `design_name` picks one, the first by
/// default) or the GPS-L1-C/A-like built-in default; any of `pll_bw`, `fll_bw`, `dll_bw`,
/// `spacing`, `coherent`, `reacquire` and the acquisition arguments overrides it. The
/// initialising acquisition integrates `acq_coherent` code periods coherently; the default
/// `None` is the design's (auto: ≈4 ms coherent, 4 periods of an untiered 1 ms code such as
/// GPS L1 C/A, 1 period of a code whose full, overlay-included period is 4 ms or longer),
/// and `acq_coherent=1` restores the 0.32 one-period search. Raises `ValueError` if a PRN
/// is not acquired or the design is invalid.
#[pyfunction]
#[pyo3(signature = (i, q, fs_hz, signal, prns, if_hz=0.0, center_hz=None, pll_bw=None, fll_bw=None, dll_bw=None, spacing=None, coherent=None, periods_per_bit=None, acq_coherent=None, acq_noncoherent=None, doppler_max=None, max_seconds=None, design=None, design_name=None, reacquire=None, extra_taps=None, threads=1))]
#[allow(clippy::too_many_arguments)]
fn iq_track<'py>(
    py: Python<'py>,
    i: Vec<f64>,
    q: Vec<f64>,
    fs_hz: f64,
    signal: String,
    prns: Vec<i64>,
    if_hz: f64,
    center_hz: Option<f64>,
    pll_bw: Option<f64>,
    fll_bw: Option<f64>,
    dll_bw: Option<f64>,
    spacing: Option<f64>,
    coherent: Option<usize>,
    periods_per_bit: Option<usize>,
    acq_coherent: Option<usize>,
    acq_noncoherent: Option<usize>,
    doppler_max: Option<f64>,
    max_seconds: Option<f64>,
    design: Option<String>,
    design_name: Option<String>,
    reacquire: Option<bool>,
    extra_taps: Option<Vec<f64>>,
    threads: usize,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::track::design::{Design, DesignFile};
    use crate::iq::track::sink::CollectSink;
    use crate::iq::track::{SessionChannel, TrackSession};
    let codes = codes_for(&signal, &prns)?;
    let samples = samples_from(&i, &q)?;
    let spec = SampleSpec {
        fs_hz,
        center_hz: center_hz.unwrap_or_else(|| codes[0].carrier_hz()),
        if_hz,
    };
    // The design: a file path or TOML text, else the built-in default; arguments override.
    let base = match design {
        None => {
            if design_name.is_some() {
                return Err(PyValueError::new_err("design_name needs design"));
            }
            Design::builtin_default()
        }
        Some(d) => {
            let text = if std::path::Path::new(&d).is_file() {
                std::fs::read_to_string(&d).map_err(|e| PyValueError::new_err(e.to_string()))?
            } else {
                d
            };
            DesignFile::parse(&text)
                .and_then(|f| f.select(design_name.as_deref()).cloned())
                .map_err(PyValueError::new_err)?
        }
    };
    let mut o = String::new();
    let mut section = |name: &str, kv: Vec<(&str, Option<String>)>| {
        let set: Vec<String> = kv
            .into_iter()
            .filter_map(|(k, v)| v.map(|v| format!("{k} = {v}")))
            .collect();
        if !set.is_empty() {
            o.push_str(&format!("[{name}]\n{}\n", set.join("\n")));
        }
    };
    let f = |v: Option<f64>| v.map(|x| format!("{x:?}"));
    let n = |v: Option<usize>| v.map(|x| x.max(1).to_string());
    section(
        "carrier",
        vec![("pll_bw_hz", f(pll_bw)), ("fll_bw_hz", f(fll_bw))],
    );
    section("code", vec![("bw_hz", f(dll_bw))]);
    section(
        "integration",
        vec![
            ("spacing_chips", f(spacing)),
            ("coherent_periods", n(coherent)),
            (
                "extra_taps_chips",
                extra_taps.map(|t| {
                    format!(
                        "[{}]",
                        t.iter()
                            .map(|v| format!("{v:?}"))
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                }),
            ),
        ],
    );
    section(
        "lock",
        vec![("reacquire", reacquire.map(|b| b.to_string()))],
    );
    section(
        "acquisition",
        vec![
            ("coherent_periods", n(acq_coherent)),
            ("noncoherent", n(acq_noncoherent)),
            ("doppler_max_hz", f(doppler_max)),
        ],
    );
    let design = if o.is_empty() {
        base
    } else {
        base.with_overrides(&o).map_err(PyValueError::new_err)?
    };
    // Acquisition to initialise each channel.
    let acq = design.acq_config(codes[0].period_s());
    let mut inits = Vec::new();
    for code in &codes {
        let found = acquire_peak(&samples, &spec, code, &acq).map_err(PyValueError::new_err)?;
        if !found.acquired {
            return Err(PyValueError::new_err(format!(
                "{}: not acquired (statistic {:.2} < threshold {:.2})",
                code.name(),
                found.statistic,
                found.threshold
            )));
        }
        let arc: std::sync::Arc<dyn SpreadingCode + Send + Sync> =
            std::sync::Arc::new(code.clone());
        inits.push(ChannelInit::from_acquisition(
            arc,
            &found,
            &spec,
            0,
            periods_per_bit,
        ));
    }
    let channels = inits
        .into_iter()
        .map(|init| SessionChannel::from_design(init, &design))
        .collect();
    let mut session = TrackSession::new(spec, channels)
        .map_err(PyValueError::new_err)?
        .with_threads(threads);
    let max_samples = max_seconds.map(|s| (s * fs_hz).round() as u64);
    let mut src = crate::iq::VecSource::new(spec, samples);
    let mut sink = CollectSink::default();
    session
        .run(&mut src, max_samples, &mut sink)
        .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let chans: Vec<serde_json::Value> = codes
        .iter()
        .enumerate()
        .map(|(k, code)| {
            let epochs: Vec<serde_json::Value> = sink
                .channels
                .get(k)
                .map(|v| {
                    v.iter()
                        .map(|(e, st)| {
                            let mut j = epoch_value(e);
                            j["state"] = st.as_str().into();
                            j
                        })
                        .collect()
                })
                .unwrap_or_default();
            serde_json::json!({ "code": code.name(), "epochs": epochs })
        })
        .collect();
    let events: Vec<serde_json::Value> = sink
        .events
        .iter()
        .map(|e| serde_json::to_value(e).unwrap_or(serde_json::Value::Null))
        .collect();
    json_to_py(
        py,
        &serde_json::json!({
            "fs_hz": fs_hz,
            "design": design.to_json(),
            "events": events,
            "warnings": crate::iq::cli::sampling_warnings(&spec, &codes),
            "channels": chans,
        }),
    )
}

/// Parse a `kshana.loop-design/1` TOML text (or a path to one) and return its designs,
/// resolved: a list of dicts with every field set, the `name` and the `hash`. Raises
/// `ValueError` on an invalid file.
#[pyfunction]
fn iq_loop_designs<'py>(py: Python<'py>, toml: &str) -> PyResult<Bound<'py, PyAny>> {
    let text = if std::path::Path::new(toml).is_file() {
        std::fs::read_to_string(toml).map_err(|e| PyValueError::new_err(e.to_string()))?
    } else {
        toml.to_string()
    };
    let file = crate::iq::track::design::DesignFile::parse(&text).map_err(PyValueError::new_err)?;
    let v: Vec<serde_json::Value> = file.designs().iter().map(|d| d.to_json()).collect();
    json_to_py(py, &serde_json::Value::Array(v))
}

/// Read a binary tracking-epoch file (`kshana.track-epoch/1`, as `kshana iq track --epochs
/// <path>.bin` writes it). Returns a dict with the `header` (schema, fields, channels with
/// their code, design and design hash, sample rate, engine version) and the `records`, one
/// dict per epoch. Raises `ValueError` on a file that is not one.
#[pyfunction]
fn iq_read_epochs<'py>(py: Python<'py>, path: &str) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::track::sink::BinaryEpochReader;
    let f = std::fs::File::open(path).map_err(|e| PyValueError::new_err(format!("{path}: {e}")))?;
    let reader = BinaryEpochReader::new(std::io::BufReader::new(f))
        .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let header = serde_json::to_value(reader.header()).unwrap_or(serde_json::Value::Null);
    let records = reader
        .map(|r| {
            r.map(|rec| serde_json::to_value(rec).unwrap_or(serde_json::Value::Null))
                .map_err(|e| PyValueError::new_err(e.to_string()))
        })
        .collect::<PyResult<Vec<_>>>()?;
    json_to_py(
        py,
        &serde_json::json!({ "header": header, "records": records }),
    )
}

/// One tracking epoch as a JSON value for the Python surface.
fn epoch_value(e: &EpochOutput) -> serde_json::Value {
    let mut v = serde_json::json!({
        "epoch": e.epoch,
        "sample_index": e.sample_index,
        "code_epoch_s": e.code_epoch_s,
        "t_coh_s": e.t_coh_s,
        "doppler_hz": e.doppler_hz,
        "code_rate_hz": e.code_rate_hz,
        "code_phase_chips": e.code_phase_chips,
        "periods": e.periods,
        "i_early": e.early.re,
        "q_early": e.early.im,
        "i_prompt": e.prompt.re,
        "q_prompt": e.prompt.im,
        "i_late": e.late.re,
        "q_late": e.late.im,
        "carrier_phase_cycles": e.carrier_phase_cycles,
        "bit_edge": e.bit_edge,
        "bit": e.bit,
        "dll_chips": e.disc.dll_chips,
        "pll_rad": e.disc.pll_rad,
        "fll_hz": e.disc.fll_hz,
        "pli": e.pli,
        "phase_lock": e.phase_lock,
        "code_lock": e.code_lock,
        "cn0_nwpr_dbhz": e.cn0_nwpr_dbhz,
        "cn0_beaulieu_dbhz": e.cn0_beaulieu_dbhz,
        "cn0_m2m4_dbhz": e.cn0_m2m4_dbhz,
    });
    if !e.extra.is_empty() {
        v["extra"] = e
            .extra
            .iter()
            .map(|&(offset_chips, c)| {
                serde_json::json!({ "offset_chips": offset_chips, "i": c.re, "q": c.im })
            })
            .collect();
    }
    v
}

/// Run the IQ detection monitors over a recording file (`path`: SigMF, collection, `.sdrx`
/// or a raw file with a sidecar, or raw with `format`/`rate`) in one streaming pass, and
/// return the report as a dict: `series` (name, unit, channel, `t_s`, `value`), `events`
/// (kind, channel, `t_start_s`, `t_alarm_s`, `t_end_s`, peak, threshold), `spectra` and
/// `notes`. Mirrors `kshana iq monitor`: `power` / `spectral` pick the pre-correlation
/// monitors (both when neither is set), `settings` is a TOML or JSON monitor-settings text,
/// and `signal` + `prns` add tracked channels with C/N0, SQM and lock monitors. Raises
/// `ValueError` on a bad argument or an unreadable recording.
#[pyfunction]
#[pyo3(signature = (path, signal=None, prns=None, power=false, spectral=false, settings=None, baseline=None, max_seconds=None, spacing=None, pll_bw=None, fll_bw=None, dll_bw=None, coherent=None, cn0_windows=None, doppler_max=None, periods_per_bit=None, format=None, rate=None, center_hz=None, if_hz=None, header_bytes=None))]
#[allow(clippy::too_many_arguments)]
fn iq_monitor<'py>(
    py: Python<'py>,
    path: String,
    signal: Option<String>,
    prns: Option<Vec<i64>>,
    power: bool,
    spectral: bool,
    settings: Option<String>,
    baseline: Option<f64>,
    max_seconds: Option<f64>,
    spacing: Option<f64>,
    pll_bw: Option<f64>,
    fll_bw: Option<f64>,
    dll_bw: Option<f64>,
    coherent: Option<usize>,
    cn0_windows: Option<usize>,
    doppler_max: Option<f64>,
    periods_per_bit: Option<usize>,
    format: Option<String>,
    rate: Option<f64>,
    center_hz: Option<f64>,
    if_hz: Option<f64>,
    header_bytes: Option<u64>,
) -> PyResult<Bound<'py, PyAny>> {
    let mut args = vec![path];
    let mut opt = |k: &str, v: Option<String>| {
        if let Some(v) = v {
            args.push(k.to_string());
            args.push(v);
        }
    };
    opt("--signal", signal);
    opt(
        "--prn",
        prns.map(|p| {
            p.iter()
                .map(|x| x.to_string())
                .collect::<Vec<_>>()
                .join(",")
        }),
    );
    opt("--baseline", baseline.map(|v| v.to_string()));
    opt("--max-seconds", max_seconds.map(|v| v.to_string()));
    opt("--spacing", spacing.map(|v| v.to_string()));
    opt("--pll-bw", pll_bw.map(|v| v.to_string()));
    opt("--fll-bw", fll_bw.map(|v| v.to_string()));
    opt("--dll-bw", dll_bw.map(|v| v.to_string()));
    opt("--coherent", coherent.map(|v| v.to_string()));
    opt("--cn0-windows", cn0_windows.map(|v| v.to_string()));
    opt("--doppler-max", doppler_max.map(|v| v.to_string()));
    opt("--periods-per-bit", periods_per_bit.map(|v| v.to_string()));
    opt("--format", format);
    opt("--rate", rate.map(|v| v.to_string()));
    opt("--center", center_hz.map(|v| v.to_string()));
    opt("--if", if_hz.map(|v| v.to_string()));
    opt("--header", header_bytes.map(|v| v.to_string()));
    if power {
        args.push("--power".into());
    }
    if spectral {
        args.push("--spectral".into());
    }
    let fail = |f: crate::iq::cli::CliFail| match f {
        crate::iq::cli::CliFail::Usage(m) | crate::iq::cli::CliFail::Run(m) => {
            PyValueError::new_err(m)
        }
    };
    let a = crate::iq::cli::monitor_parse_args(&args).map_err(fail)?;
    let report = crate::iq::cli::monitor_report(&a, settings.as_deref()).map_err(fail)?;
    let v = serde_json::to_value(&report).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Replay one recording across several tracking-loop designs and report the steady-state
/// metrics per design and PRN (`kshana iq sweep`), on complex samples. The designs are every
/// design of `design` (a `kshana.loop-design/1` TOML file's text or path), or the full product
/// of the `pll_bws`, `dll_bws`, `spacings` (chips) and `coherents` (code periods) lists on the
/// built-in default (a list left out keeps the default's value). Each PRN is acquired once
/// (with the first design's acquisition) and every (design, PRN) channel then tracks the same
/// samples, so each result is exactly that of running the design alone. Returns
/// `{"designs": [...], "warnings": [...]}`: per design and PRN `design`, `code`, `epochs`,
/// `phase_jitter_deg`, `code_jitter_chips` (steady state, over the second half of the run),
/// `phase_lock_frac`, `code_lock_frac`, `mean_cn0_dbhz` and `design_hash`. `reacquire`,
/// `periods_per_bit`, `max_seconds` and `threads` are as for `iq_track`. Raises `ValueError`
/// if a PRN is not acquired or a design is invalid.
#[pyfunction]
#[pyo3(signature = (i, q, fs_hz, signal, prns, if_hz=0.0, center_hz=None, pll_bws=None, dll_bws=None, spacings=None, coherents=None, design=None, reacquire=None, periods_per_bit=None, max_seconds=None, threads=1))]
#[allow(clippy::too_many_arguments)]
fn iq_sweep<'py>(
    py: Python<'py>,
    i: Vec<f64>,
    q: Vec<f64>,
    fs_hz: f64,
    signal: String,
    prns: Vec<i64>,
    if_hz: f64,
    center_hz: Option<f64>,
    pll_bws: Option<Vec<f64>>,
    dll_bws: Option<Vec<f64>>,
    spacings: Option<Vec<f64>>,
    coherents: Option<Vec<usize>>,
    design: Option<String>,
    reacquire: Option<bool>,
    periods_per_bit: Option<usize>,
    max_seconds: Option<f64>,
    threads: usize,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::track::design::{Design, DesignFile};
    use crate::iq::track::sink::{Fanout, Summary};
    use crate::iq::track::{SessionChannel, TrackSession};
    let codes = codes_for(&signal, &prns)?;
    let samples = samples_from(&i, &q)?;
    let spec = SampleSpec {
        fs_hz,
        center_hz: center_hz.unwrap_or_else(|| codes[0].carrier_hz()),
        if_hz,
    };
    let bad = |e: String| PyValueError::new_err(e);
    // The designs to sweep.
    let designs: Vec<Design> = match design {
        Some(d) => {
            if pll_bws.is_some() || dll_bws.is_some() || spacings.is_some() || coherents.is_some() {
                return Err(bad(
                    "pll_bws, dll_bws, spacings and coherents cannot be combined with design: \
                     the sweep runs every design in the file"
                        .into(),
                ));
            }
            let text = if std::path::Path::new(&d).is_file() {
                std::fs::read_to_string(&d).map_err(|e| bad(e.to_string()))?
            } else {
                d
            };
            DesignFile::parse(&text).map_err(bad)?.designs().to_vec()
        }
        None => {
            let base = Design::builtin_default();
            let pll = pll_bws
                .filter(|v| !v.is_empty())
                .unwrap_or_else(|| vec![15.0]);
            let dll = dll_bws
                .filter(|v| !v.is_empty())
                .unwrap_or_else(|| vec![base.loop_config().dll_bn_hz]);
            let sp = spacings
                .filter(|v| !v.is_empty())
                .unwrap_or_else(|| vec![base.loop_config().spacing_chips]);
            let coh = coherents
                .filter(|v| !v.is_empty())
                .unwrap_or_else(|| vec![base.loop_config().coherent_periods]);
            let mut out = Vec::new();
            for &p in &pll {
                for &d in &dll {
                    for &s in &sp {
                        for &c in &coh {
                            let c = c.max(1);
                            let dd = base
                                .with_overrides(&format!(
                                    "[carrier]\npll_bw_hz = {p:?}\n[code]\nbw_hz = {d:?}\n\
                                     [integration]\nspacing_chips = {s:?}\ncoherent_periods = {c}\n"
                                ))
                                .map_err(bad)?;
                            out.push(dd.renamed(&format!("pll{p}_dll{d}_sp{s}_coh{c}")));
                        }
                    }
                }
            }
            out
        }
    };
    if designs.is_empty() {
        return Err(bad("no designs to sweep".into()));
    }
    if designs.len() * codes.len() > 256 {
        return Err(bad(format!(
            "{} (design, PRN) channels; at most 256 per sweep",
            designs.len() * codes.len()
        )));
    }
    // Acquire each PRN once, with the first design's acquisition.
    let acq = designs[0].acq_config(codes[0].period_s());
    let mut inits = Vec::new();
    for code in &codes {
        let found = acquire_peak(&samples, &spec, code, &acq).map_err(bad)?;
        if !found.acquired {
            return Err(bad(format!(
                "{}: not acquired (statistic {:.2} < threshold {:.2})",
                code.name(),
                found.statistic,
                found.threshold
            )));
        }
        let arc: std::sync::Arc<dyn SpreadingCode + Send + Sync> =
            std::sync::Arc::new(code.clone());
        inits.push(ChannelInit::from_acquisition(
            arc,
            &found,
            &spec,
            0,
            periods_per_bit,
        ));
    }
    let n_total = samples.len() as u64;
    let max_samples = max_seconds.map(|s| (s * fs_hz).round() as u64);
    let tracked = max_samples.map_or(n_total, |m| m.min(n_total));
    // One channel per (design, PRN), design-major, all on one pass over the samples.
    let mut channels = Vec::new();
    let mut meta = Vec::new();
    for d in &designs {
        let d = if reacquire.unwrap_or(false) {
            d.with_overrides("[lock]\nreacquire = true\n")
                .map_err(bad)?
        } else {
            d.clone()
        };
        for (init, code) in inits.iter().zip(&codes) {
            channels.push(SessionChannel::from_design(init.clone(), &d));
            meta.push((d.name().to_string(), d.hash().to_string(), code.name()));
        }
    }
    let mut session = TrackSession::new(spec, channels)
        .map_err(bad)?
        .with_threads(threads);
    let mut summary = Summary::new(0.5 * tracked as f64 / fs_hz);
    let mut src = crate::iq::VecSource::new(spec, samples);
    {
        let mut fan = Fanout::new();
        fan.push(&mut summary);
        session
            .run(&mut src, max_samples, &mut fan)
            .map_err(|e| PyValueError::new_err(e.to_string()))?;
    }
    let rows: Vec<serde_json::Value> = meta
        .iter()
        .enumerate()
        .map(|(k, (name, hash, code))| {
            let c = summary.channels.get(k).cloned().unwrap_or_default();
            serde_json::json!({
                "design": name,
                "code": code,
                "epochs": c.epochs,
                "phase_jitter_deg": c.phase_jitter_deg,
                "code_jitter_chips": c.code_jitter_chips,
                "phase_lock_frac": c.phase_lock_fraction(),
                "code_lock_frac": c.code_lock_fraction(),
                "mean_cn0_dbhz": c.mean_cn0_dbhz,
                "design_hash": hash,
            })
        })
        .collect();
    json_to_py(
        py,
        &serde_json::json!({
            "designs": rows,
            "warnings": crate::iq::cli::sampling_warnings(&spec, &codes),
        }),
    )
}

/// Describe one IQ recording without processing it (`kshana iq info`): kind (raw with sidecar,
/// SigMF, SigMF collection), sample format, rate, centre frequency, total samples, duration,
/// data files and their sizes, as a dict. `hash=True` adds each data file's SHA-256 (reads the
/// whole file). Reads the file at `path`; nothing is written.
#[pyfunction]
#[pyo3(signature = (path, hash=false))]
fn iq_info<'py>(py: Python<'py>, path: &str, hash: bool) -> PyResult<Bound<'py, PyAny>> {
    let e = crate::iq::io::inventory::inventory_entry(std::path::Path::new(path), hash);
    let v = serde_json::to_value(&e).map_err(|err| PyValueError::new_err(err.to_string()))?;
    json_to_py(py, &v)
}

/// List the IQ recordings under `directory` and describe each (`kshana iq inventory`): a list
/// of the dicts `iq_info` returns, in path order. `recursive=True` descends into
/// sub-folders; `hash=True` adds SHA-256 digests (reads every data file). Reads only.
#[pyfunction]
#[pyo3(signature = (directory, recursive=false, hash=false))]
fn iq_inventory<'py>(
    py: Python<'py>,
    directory: &str,
    recursive: bool,
    hash: bool,
) -> PyResult<Bound<'py, PyAny>> {
    let paths =
        crate::iq::io::inventory::list_recordings(std::path::Path::new(directory), recursive)
            .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let rows: Vec<serde_json::Value> = paths
        .iter()
        .map(|p| {
            serde_json::to_value(crate::iq::io::inventory::inventory_entry(p, hash))
                .unwrap_or(serde_json::Value::Null)
        })
        .collect();
    json_to_py(py, &serde_json::Value::Array(rows))
}

/// Fit the tracking-loop loss-of-lock model to a receiver-trust timeline described by an
/// `iq-labfit` TOML scenario. Returns a dict with the parsed `report`, the `residuals_csv`
/// and `predictions_csv` tables and the `markdown`. Relative log paths are resolved against
/// the working directory. A scenario whose `[input]` names `campaign = "<out-dir>"` fits
/// that `kshana iq campaign` output folder instead (a relative folder is resolved against
/// the working directory) and returns the same keys, with a `kshana.labfit-iq/1` report.
/// Raises `ValueError` on an invalid scenario, unreadable log or unusable campaign.
#[pyfunction]
fn iq_labfit<'py>(py: Python<'py>, toml: &str) -> PyResult<Bound<'py, PyAny>> {
    let [json, residuals_csv, predictions_csv, markdown] =
        crate::iq::labfit::run_toml_views(toml).map_err(PyValueError::new_err)?;
    let report: serde_json::Value =
        serde_json::from_str(&json).map_err(|e| PyValueError::new_err(e.to_string()))?;
    let v = serde_json::json!({
        "report": report,
        "residuals_csv": residuals_csv,
        "predictions_csv": predictions_csv,
        "markdown": markdown,
    });
    json_to_py(py, &v)
}

/// Validate a lab test-condition file (`kshana.test-conditions/1`, TOML or JSON) given as
/// a path or as text. Returns the resolved conditions as a dict, with the condition hash
/// under `hash`. Raises `ValueError` on an invalid file.
#[pyfunction]
fn iq_test_conditions<'py>(py: Python<'py>, conditions: &str) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::campaign::TestConditions;
    let path = std::path::Path::new(conditions);
    let tc = if !conditions.contains('\n') && path.is_file() {
        TestConditions::load(path)
    } else {
        TestConditions::parse(conditions)
    }
    .map_err(PyValueError::new_err)?;
    let mut v = serde_json::to_value(&tc).map_err(|e| PyValueError::new_err(e.to_string()))?;
    if let Some(m) = v.as_object_mut() {
        m.insert("hash".into(), tc.hash().into());
    }
    json_to_py(py, &v)
}

/// Run a lab-replay campaign (`kshana.campaign/1`, a path or TOML text; relative paths in
/// text resolve against the working directory) into `out_dir`, exactly as
/// `kshana iq campaign` does: cells already done are skipped unless `resume` is false,
/// `max_cells` bounds how many pending cells run now, and `dry_run` only plans. Returns the
/// run summary as a dict (cell counts, failures, and the `digest` once every cell is
/// done). The GIL is released while the campaign runs. Raises `ValueError` on an invalid
/// campaign and `RuntimeError` when the run cannot proceed.
#[pyfunction]
#[pyo3(signature = (campaign, out_dir, workers=0, resume=true, max_cells=None, dry_run=false))]
fn iq_campaign<'py>(
    py: Python<'py>,
    campaign: &str,
    out_dir: &str,
    workers: usize,
    resume: bool,
    max_cells: Option<usize>,
    dry_run: bool,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::campaign::{run, LoadedCampaign, RunOptions};
    let path = std::path::Path::new(campaign);
    let loaded = if !campaign.contains('\n') && path.is_file() {
        LoadedCampaign::load(path)
    } else {
        LoadedCampaign::load_text(campaign, std::path::Path::new("./campaign.toml"))
    }
    .map_err(PyValueError::new_err)?;
    let opts = RunOptions {
        workers,
        no_resume: !resume,
        max_cells,
        dry_run,
    };
    let out = std::path::PathBuf::from(out_dir);
    let summary = py
        .detach(|| run(&loaded, &out, &opts))
        .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
    let v = serde_json::to_value(&summary).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Rebuild a campaign's scorecards, HTML report and digest from the cells in `out_dir`.
/// Returns a dict with the cell and row counts, the rows failing a bar, and the `digest`
/// (None until every cell is done). Raises `RuntimeError` when `out_dir` holds no campaign.
#[pyfunction]
fn iq_campaign_report<'py>(py: Python<'py>, out_dir: &str) -> PyResult<Bound<'py, PyAny>> {
    let s = crate::iq::campaign::report::build(std::path::Path::new(out_dir))
        .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
    let v = serde_json::to_value(&s).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Apply a receiver front-end / interference-mitigation chain to complex samples
/// (`i`/`q` lists sampled at `fs_hz`) and return the filtered samples as `samples_i` /
/// `samples_q`. The stages mirror `kshana iq frontend`: an optional band-pass
/// (`bandpass_lo`/`bandpass_hi` Hz), adaptive `notch`, pulse `blank`ing, frequency-domain
/// `excise`, `agc`, and a `bits`-bit quantiser (preceded by an automatic AGC unless `no_agc`).
/// Raises `ValueError` on an invalid chain.
#[pyfunction]
#[pyo3(signature = (i, q, fs_hz, bandpass_lo=None, bandpass_hi=None, bandpass_transition=None, bandpass_atten=60.0, notch=false, notch_r=0.95, notch_mu=0.05, blank=None, blank_hold=0, excise=false, excise_fft=256, excise_pfa=1e-3, agc=false, agc_tau=1e-3, bits=None, quant_step=None, no_agc=false))]
#[allow(clippy::too_many_arguments)]
fn iq_frontend<'py>(
    py: Python<'py>,
    i: Vec<f64>,
    q: Vec<f64>,
    fs_hz: f64,
    bandpass_lo: Option<f64>,
    bandpass_hi: Option<f64>,
    bandpass_transition: Option<f64>,
    bandpass_atten: f64,
    notch: bool,
    notch_r: f64,
    notch_mu: f64,
    blank: Option<f64>,
    blank_hold: usize,
    excise: bool,
    excise_fft: usize,
    excise_pfa: f64,
    agc: bool,
    agc_tau: f64,
    bits: Option<u32>,
    quant_step: Option<f64>,
    no_agc: bool,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::iq::frontend::Stage;
    let mut samples = samples_from(&i, &q)?;
    let bandpass = match (bandpass_lo, bandpass_hi) {
        (Some(lo), Some(hi)) => Some((lo, hi)),
        (None, None) => None,
        _ => {
            return Err(PyValueError::new_err(
                "bandpass needs both bandpass_lo and bandpass_hi",
            ))
        }
    };
    let params = FrontendParams {
        bandpass,
        bandpass_transition_hz: bandpass_transition,
        bandpass_atten_db: bandpass_atten,
        notch,
        notch_r,
        notch_mu,
        blank,
        blank_hold,
        excise,
        excise_fft,
        excise_pfa,
        agc,
        agc_tau_s: agc_tau,
        bits,
        quant_step,
        no_agc,
    };
    let mut chain = build_chain(&params, fs_hz).map_err(PyValueError::new_err)?;
    chain.process(&mut samples);
    let oi: Vec<f64> = samples.iter().map(|s| s.re).collect();
    let oq: Vec<f64> = samples.iter().map(|s| s.im).collect();
    json_to_py(py, &serde_json::json!({ "samples_i": oi, "samples_q": oq }))
}

/// A text or bytes argument: a Python `str` is taken as UTF-8.
#[derive(FromPyObject)]
enum TextOrBytes {
    Text(String),
    Bytes(Vec<u8>),
}

impl TextOrBytes {
    fn bytes(&self) -> &[u8] {
        match self {
            Self::Text(t) => t.as_bytes(),
            Self::Bytes(b) => b,
        }
    }
}

/// Score a bounded excerpt of a vessel's NMEA 0183 stream the way `kshana receiver-trust
/// live` scores it. `session_toml` declares the vessel (`[platform] kind = "vessel"`);
/// `nmea` is the excerpt (`str` or `bytes`, at most 2 MiB and 20,000 epochs, and it must hold
/// the calibration window). Returns a dict with `schema` (`"1.3"`), `epochs` (one dict per
/// epoch: `state`, `score`, `deductions`, `alarms`, `position`, `advisory`, `geometry`,
/// ...; each deduction has a `class`: `geometry`, `environment` or `inconsistency`),
/// `last_pksht` and `summary` (counts by state, `lowest_score`, `final_score`,
/// `first_untrusted_t_s`, and `geometry`: epochs by geometry status and by class). `geometry` is
/// the satellite geometry (`good`, `degraded` with reasons, or `unknown` when the excerpt has no
/// GSV or GSA), with a `label` (`good`, `fair`, `poor`) always reported; the score deducts for
/// geometry only when it is `poor`. The
/// bounded form of the live command: no socket is opened and the gate is not applied (both
/// are command-line only). Advisory software, not type-approved navigation equipment (IEC 61108, IEC 61162): the operator remains responsible for the navigation of the vessel. Raises `ValueError` on an invalid session or
/// excerpt.
#[pyfunction]
fn receiver_trust_replay<'py>(
    py: Python<'py>,
    session_toml: &str,
    nmea: TextOrBytes,
) -> PyResult<Bound<'py, PyAny>> {
    let r = crate::surface::assess_vessel_excerpt(
        session_toml,
        nmea.bytes(),
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let v = serde_json::to_value(&r).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Assess a vessel's NMEA 0183 log as a batch run: `session_toml` (`[platform] kind =
/// "vessel"`, no `[log]` needed) and the log (`str` or `bytes`). Returns the result document
/// as a dict, the same as `kshana receiver-trust` writes to `result.json`: the score model,
/// the monitors that ran and every epoch's 0-100 score with its deductions. Advisory software, not type-approved navigation equipment (IEC 61108, IEC 61162): the operator remains responsible for the navigation of the vessel.
/// Raises `ValueError` on an invalid session or log.
#[pyfunction]
fn assess_vessel_log<'py>(
    py: Python<'py>,
    session_toml: &str,
    log: TextOrBytes,
) -> PyResult<Bound<'py, PyAny>> {
    let j = crate::surface::assess_vessel_log_json(
        session_toml,
        log.bytes(),
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let v: serde_json::Value =
        serde_json::from_str(&j).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// Build a signed evidence pack for a window of a vessel's NMEA log (`kshana receiver-trust
/// evidence`, in memory). `from_s`/`to_s` are seconds since the log's first epoch.
/// `seed_hex` is the Ed25519 signing-key seed (64 hex digits); omitted, one is generated
/// from the operating system's randomness and returned. `created_utc` is RFC 3339 UTC; `None`
/// is now, `"none"` leaves it out (a reproducible pack). Returns a dict: `files` (name to
/// `bytes`), `public_key`, `seed_hex` (the seed used: keep it private), `epochs_in_window`
/// and `slice` (`[start, end]` of the log bytes bundled, or `None` for the whole log). A
/// pack is a technical record, not a legal opinion. Raises `ValueError`.
#[pyfunction]
#[pyo3(signature = (session_toml, log, from_s, to_s, title=None, created_utc=None, seed_hex=None))]
#[allow(clippy::too_many_arguments)]
fn evidence_create<'py>(
    py: Python<'py>,
    session_toml: &str,
    log: TextOrBytes,
    from_s: f64,
    to_s: f64,
    title: Option<String>,
    created_utc: Option<String>,
    seed_hex: Option<String>,
) -> PyResult<Bound<'py, PyAny>> {
    use pyo3::types::{PyBytes, PyDict};
    let seed = seed_hex
        .as_deref()
        .map(|h| crate::surface::hex32("seed_hex", h))
        .transpose()
        .map_err(PyValueError::new_err)?;
    let created = match created_utc.as_deref() {
        None => Some(crate::surface::now_rfc3339_utc()),
        Some("none") => None,
        Some(t) => Some(t.to_string()),
    };
    let p = crate::surface::evidence_create(
        session_toml,
        log.bytes(),
        from_s,
        to_s,
        title.as_deref(),
        created.as_deref(),
        seed,
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let files = PyDict::new(py);
    for (k, v) in &p.files {
        files.set_item(k, PyBytes::new(py, v))?;
    }
    let out = PyDict::new(py);
    out.set_item("files", files)?;
    out.set_item("public_key", p.public_key)?;
    out.set_item("seed_hex", hex::encode(p.seed))?;
    out.set_item("epochs_in_window", p.epochs_in_window)?;
    out.set_item("slice", p.slice.map(|(a, b)| vec![a, b]))?;
    Ok(out.into_any())
}

/// Verify an evidence pack: `files` maps names to `bytes` (or `str`). Checks every hash, the
/// chain and the signature; with `public_key` (64 hex digits, obtained from the signer by
/// another route) also that the signer is the one expected, with `full_log` that the log
/// you hold is the one recorded, and with `require_timestamp` that a timestamp token is
/// present. Returns the report as a dict: `ok`, `failures`, `checks`, `signer_fingerprint`,
/// `signer_pinned`, `notes`, plus `verdict` (`"verified"`, `"intact-signer-not-pinned"` or
/// `"failed"`) and `message`. Without a public key a pass is only `"intact-signer-not-pinned"`:
/// the signature proves the pack is intact against the key it names itself, which anyone can
/// generate. Raises `ValueError` on a malformed key.
#[pyfunction]
#[pyo3(signature = (files, public_key=None, full_log=None, require_timestamp=false))]
fn evidence_verify<'py>(
    py: Python<'py>,
    files: std::collections::BTreeMap<String, TextOrBytes>,
    public_key: Option<String>,
    full_log: Option<TextOrBytes>,
    require_timestamp: bool,
) -> PyResult<Bound<'py, PyAny>> {
    let files: crate::evidence::Files = files
        .into_iter()
        .map(|(k, v)| (k, v.bytes().to_vec()))
        .collect();
    let r = crate::surface::evidence_verify(
        &files,
        public_key.as_deref(),
        full_log.as_ref().map(TextOrBytes::bytes),
        require_timestamp,
    )
    .map_err(PyValueError::new_err)?;
    json_to_py(py, &r)
}

/// Bind an RFC 3161 timestamp token to an evidence pack, in memory (`kshana evidence
/// attach-timestamp`). `files` is a pack as `evidence_create` returns it; `token` is the raw
/// bytes of the `.tsr` file, or its base64 text. The token is stored as `timestamp.tsr` beside
/// the signed manifest, not inside it, and the pack with the token must still verify or
/// `ValueError` is raised. An existing token is kept unless `replace=True`. This does NOT
/// verify the timestamp authority's signature or certificate chain: use `openssl ts -verify`.
/// Returns `{"files": {name: bytes}, "notes": [str]}`; verify later with
/// `require_timestamp=True` to insist the token is present.
#[pyfunction]
#[pyo3(signature = (files, token, replace=false))]
fn evidence_attach_timestamp<'py>(
    py: Python<'py>,
    files: std::collections::BTreeMap<String, TextOrBytes>,
    token: TextOrBytes,
    replace: bool,
) -> PyResult<Bound<'py, PyAny>> {
    use pyo3::types::{PyBytes, PyDict};
    let files: crate::evidence::Files = files
        .into_iter()
        .map(|(k, v)| (k, v.bytes().to_vec()))
        .collect();
    let token = match &token {
        TextOrBytes::Bytes(b) => b.clone(),
        TextOrBytes::Text(t) => crate::permalink::base64_decode(t.trim())
            .ok_or_else(|| PyValueError::new_err("token text is not base64"))?,
    };
    let t = crate::surface::evidence_attach_timestamp(
        &files,
        &token,
        replace,
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let out_files = PyDict::new(py);
    for (k, v) in &t.files {
        out_files.set_item(k, PyBytes::new(py, v))?;
    }
    let out = PyDict::new(py);
    out.set_item("files", out_files)?;
    out.set_item("notes", t.notes)?;
    Ok(out.into_any())
}

/// Generate an Ed25519 signing key for evidence packs, in memory (`kshana evidence keygen`).
/// Returns `{"seed_hex", "public_key", "fingerprint", "warning"}`. **The seed is the private
/// key**: keep it out of logs, chats and version control, and give verifiers only
/// `public_key`, by a route they trust. This is offered in Python only, because a private key
/// should not pass through a web page or an agent conversation. Pass `seed_hex` to
/// `evidence_create` to sign with it.
#[pyfunction]
fn evidence_keygen<'py>(py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
    let k = crate::surface::evidence_keygen();
    json_to_py(
        py,
        &serde_json::json!({
            "seed_hex": k.seed_hex,
            "public_key": k.public_key,
            "fingerprint": k.fingerprint,
            "warning": "seed_hex is the PRIVATE signing key. Keep it secret: never log it, paste it into a chat or commit it. Give verifiers only public_key, by a route they trust.",
        }),
    )
}

/// Build a GNSS interference map from CSV text (`source` is `"adsb"` or `"ais"`; the input
/// formats are in `docs/INTERFERENCE-MAP.md`). `dataset` is an approved preset
/// (`adsb-lol`, `noaa-marinecadastre`, `kystverket`) or `"custom"`, which also needs
/// `licence`, `licence_url` and `attribution` so the output carries them. `land_geojson`
/// (AIS only) is an optional land-polygon file. Returns one dict per UTC day with
/// `file_name`, `date`, `cells_published`, `cells_flagged` and the
/// `kshana-interference-map/v1` GeoJSON text in `geojson`. Aggregate only: identifiers are
/// never returned, and a degraded cell does not name interference as the cause. Raises
/// `ValueError` on bad input.
#[pyfunction]
#[pyo3(signature = (source, csv, dataset, cell_deg=None, licence=None, licence_url=None, attribution=None, land_geojson=None))]
#[allow(clippy::too_many_arguments)]
fn interference_map<'py>(
    py: Python<'py>,
    source: &str,
    csv: &str,
    dataset: &str,
    cell_deg: Option<f64>,
    licence: Option<String>,
    licence_url: Option<String>,
    attribution: Option<String>,
    land_geojson: Option<String>,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::surface::{interference_map as build, CustomDataset, MapSource, MAX_INPUT_BYTES};
    let custom = (licence.is_some() || licence_url.is_some() || attribution.is_some()).then(|| {
        CustomDataset {
            licence: licence.unwrap_or_default(),
            licence_url: licence_url.unwrap_or_default(),
            attribution: attribution.unwrap_or_default(),
        }
    });
    let days = build(
        MapSource::parse(source).map_err(PyValueError::new_err)?,
        csv,
        dataset,
        cell_deg,
        custom.as_ref(),
        land_geojson.as_deref(),
        MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let v = serde_json::Value::Array(
        days.into_iter()
            .map(|d| {
                serde_json::json!({
                    "file_name": d.file_name,
                    "date": d.date,
                    "cells_published": d.cells_published,
                    "cells_flagged": d.cells_flagged,
                    "geojson": d.geojson,
                })
            })
            .collect(),
    );
    json_to_py(py, &v)
}

/// Export a scenario's vehicle motion and events for a laboratory GNSS simulator
/// (`kshana bench-export`, `docs/TEST-BENCH.md`), in memory: nothing is written to disk.
/// Applies to the `gnss-ins`, `jamming` and `gnss-sim` kinds; others raise `ValueError` with
/// the reason. `epoch` is the UTC instant of motion time zero, `YYYY-MM-DDTHH:MM:SS` with an
/// optional `Z` (default 2024-01-01T00:00:00Z); the output is byte-identical for the same
/// scenario and epoch. Returns `{"files": {suffix: text}, "notes": [str], "notice": str}`:
/// `files` holds `.motion.csv`, `.motion.json`, `.nmea`, `.events.csv`, `.events.toml` and,
/// when the sample grid is millisecond-regular, `.waypoints.txt`; `notes` says why a file was
/// left out; keep `notice` with the files. No radio-frequency or baseband signal is written.
#[pyfunction]
#[pyo3(signature = (toml, epoch=None))]
fn bench_export<'py>(
    py: Python<'py>,
    toml: &str,
    epoch: Option<&str>,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::surface::{BENCH_NOTICE, MAX_INPUT_BYTES};
    let e = crate::surface::bench_export(toml, epoch, MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    let files: serde_json::Map<String, serde_json::Value> =
        e.files.into_iter().map(|(k, v)| (k, v.into())).collect();
    json_to_py(
        py,
        &serde_json::json!({"files": files, "notes": e.notes, "notice": BENCH_NOTICE}),
    )
}

/// Export an `orbit` scenario's propagated constellation as SP3-c precise-ephemeris text (the
/// artifact `--export-sp3` writes). Raises `ValueError` if the scenario is not an orbit kind.
#[pyfunction]
fn export_sp3(toml: &str) -> PyResult<String> {
    crate::api::export_sp3(toml).map_err(PyValueError::new_err)
}

/// Export a constellation's mean elements as a CCSDS OMM catalogue string (`--export-omm`).
/// Raises `ValueError` if the scenario cannot produce one.
#[pyfunction]
fn export_omm(toml: &str) -> PyResult<String> {
    crate::api::export_omm(toml).map_err(PyValueError::new_err)
}

/// Export the velocity-carrying state as a CCSDS OEM 2.0 ephemeris string (`--export-oem`).
/// Raises `ValueError` if the scenario cannot produce one.
#[pyfunction]
fn export_oem(toml: &str) -> PyResult<String> {
    crate::api::export_oem(toml).map_err(PyValueError::new_err)
}

/// Which interoperability formats apply to a scenario, without running it: a list of dicts
/// `{format, applies, reason, spec_url}` for `czml`, `kml`, `geojson`, `stk` and `sigmf`.
/// Raises `ValueError` on invalid TOML.
#[pyfunction]
fn export_formats<'py>(py: Python<'py>, toml: &str) -> PyResult<Bound<'py, PyAny>> {
    let v = crate::surface::export_formats(toml, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_to_py(py, &v)
}

/// Export a scenario in one interoperability format (`czml`, `kml`, `geojson`, `stk` or
/// `sigmf`), in memory: nothing is written. Returns `{format, spec_url, files}`; each file is
/// `{suffix, bytes, sha256, encoding, content}` where `encoding` is `"utf-8"` (the content is
/// the text) or `"base64"` (a binary file). Times are UTC and the same scenario gives
/// byte-identical files. Raises `ValueError` with the reason when the format does not apply.
#[pyfunction]
fn export_scenario<'py>(py: Python<'py>, toml: &str, format: &str) -> PyResult<Bound<'py, PyAny>> {
    let e = crate::surface::export_scenario(toml, format, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_to_py(py, &e.to_json())
}

/// Write a GeoJSON route (a `LineString`, or a `Feature` or `FeatureCollection` holding one)
/// into a scenario of a kind that flies a waypoint track (`terrain-nav`, `terrain-slam`,
/// `gravity-map`, `combined-altpnt`) and return the new TOML. Raises `ValueError` with the
/// reason otherwise.
#[pyfunction]
fn import_route(toml: &str, geojson: &str) -> PyResult<String> {
    crate::surface::import_route(toml, geojson, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)
}

/// Run a scenario and return its time series as an animation: `format` is `"svg"` (one
/// animated SVG, no script; default), `"html"` (one self-contained player page) or
/// `"frames"` (numbered static SVG frames and `manifest.json`, at most 120). Options: `fps`
/// (1 to 60, default 12), `duration_s` (0.5 to 600, default 8), `width` (480 to 3840 px, default
/// 960). Returns `{summary, files}`: what was drawn, and `{name: text}`. A kind with no
/// sampled time axis raises `ValueError`. Nothing is written.
#[pyfunction]
#[pyo3(signature = (toml, format="svg", fps=None, duration_s=None, width=None))]
fn animate_scenario<'py>(
    py: Python<'py>,
    toml: &str,
    format: &str,
    fps: Option<u32>,
    duration_s: Option<f64>,
    width: Option<u32>,
) -> PyResult<Bound<'py, PyAny>> {
    let a = crate::surface::animate_scenario(
        toml,
        Some(format),
        fps,
        duration_s,
        width,
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let files: serde_json::Map<String, serde_json::Value> =
        a.files.into_iter().map(|(k, v)| (k, v.into())).collect();
    json_to_py(
        py,
        &serde_json::json!({"summary": a.summary, "files": files}),
    )
}

/// The bundled reference scenarios: `{count, scenarios: [{name, kind, about}]}`; `kind` limits
/// the list to one scenario kind. Every one runs as it stands. Raises `ValueError` on an
/// unknown kind.
#[pyfunction]
#[pyo3(signature = (kind=None))]
fn list_examples<'py>(py: Python<'py>, kind: Option<&str>) -> PyResult<Bound<'py, PyAny>> {
    let v = crate::surface::list_examples(kind).map_err(PyValueError::new_err)?;
    json_to_py(py, &v)
}

/// The TOML text of one bundled reference scenario, byte for byte the repository's file.
/// Raises `ValueError` for an unknown name or one that exists but is not bundled.
#[pyfunction]
fn get_example(name: &str) -> PyResult<&'static str> {
    crate::surface::get_example(name).map_err(PyValueError::new_err)
}

/// Fill the public-framework mapping from result documents given as text: which rows of
/// five resilience frameworks and standards (`docs/compliance/`) the runs support evidence
/// for, which they do not, and the gap each row keeps. `runs` is a list of dicts with
/// `label` (the name to show), `result` (the result JSON text) and optionally `scenario`
/// (the scenario TOML text, which names the scenario kind a result does not). At most 64
/// runs; nothing is read from disk. Returns `{"report": dict, "markdown": str}`; the report
/// carries `statement` verbatim and lists every unusable input in `unrecognised`. A status
/// says the runs support evidence for a row's capabilities; it is not a finding that a
/// framework is met. Raises `ValueError` on a missing key or an over-size input.
#[pyfunction]
fn compliance_report<'py>(
    py: Python<'py>,
    runs: Vec<std::collections::BTreeMap<String, String>>,
) -> PyResult<Bound<'py, PyAny>> {
    use crate::surface::{ComplianceRunText, MAX_INPUT_BYTES};
    let mut texts = Vec::with_capacity(runs.len());
    for (i, r) in runs.iter().enumerate() {
        let get = |k: &str| r.get(k).cloned();
        texts.push(ComplianceRunText {
            label: get("label")
                .ok_or_else(|| PyValueError::new_err(format!("runs[{i}] needs `label`")))?,
            result_json: get("result")
                .ok_or_else(|| PyValueError::new_err(format!("runs[{i}] needs `result`")))?,
            scenario_toml: get("scenario"),
        });
    }
    let out = crate::surface::compliance_report(&texts, MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_to_py(
        py,
        &serde_json::json!({"report": out.report, "markdown": out.markdown}),
    )
}

/// Check Galileo OSNMA authentication of I/NAV pages held in memory, with the verifier
/// `kshana osnma verify` runs; returns the dict `--json` prints. Advisory: this is analysis and
/// monitoring output and not type-approved navigation equipment. The authentication status
/// reported here is not a navigation integrity service and must not be the sole basis for any
/// navigation or safety decision. `data` is the pages (text, or the published vector CSV) or a
/// u-blox UBX byte stream, as `bytes` or `str`; `format` is `"pages"`, `"vector-csv"`, `"ubx"` or
/// `""` (by content); `start_gst` is a vector CSV's start in GST seconds; `merkle_root_hex` is the
/// trusted Merkle root (64 hex digits, or `""`); `public_keys` are trusted keys as `ID:TYPE:HEX`;
/// `reference_time` is the present time in GST seconds from an independent source and
/// `max_time_error` the widest distance of a sub-frame from it (below 30; needs a reference
/// time). Without a reference time freshness is not enforced and the report says so. Raises
/// `ValueError` on bad input. Nothing is read from or written to disk.
#[pyfunction]
#[allow(clippy::too_many_arguments)]
#[pyo3(signature = (data, format="", start_gst=None, merkle_root_hex="", public_keys=None, reference_time=None, max_time_error=None))]
fn osnma_verify<'py>(
    py: Python<'py>,
    data: TextOrBytes,
    format: &str,
    start_gst: Option<i64>,
    merkle_root_hex: &str,
    public_keys: Option<Vec<String>>,
    reference_time: Option<i64>,
    max_time_error: Option<i64>,
) -> PyResult<Bound<'py, PyAny>> {
    let to_u32 = |name: &str, v: Option<i64>| -> PyResult<Option<u32>> {
        v.map(|n| {
            u32::try_from(n).map_err(|_| {
                PyValueError::new_err(format!("{name} must be between 0 and 2**32 - 1"))
            })
        })
        .transpose()
    };
    let (start_gst, reference_time, max_time_error) = (
        to_u32("start_gst", start_gst)?,
        to_u32("reference_time", reference_time)?,
        to_u32("max_time_error", max_time_error)?,
    );
    let out = crate::surface::osnma_verify(
        data.bytes(),
        format,
        start_gst,
        merkle_root_hex,
        &public_keys.unwrap_or_default(),
        reference_time,
        max_time_error,
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)?;
    let v: serde_json::Value =
        serde_json::from_str(&out).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// The static public-framework mapping as Markdown, led by the statement every report
/// carries: one table per framework (`sources=False`) or the source documents they cite,
/// with versions and URLs (`sources=True`).
#[pyfunction]
#[pyo3(signature = (sources=false))]
fn compliance_mapping(sources: bool) -> String {
    crate::surface::compliance_mapping(sources)
}

/// Render a trust stream or a batch result in one telemetry format with the encoders
/// `kshana trust-telemetry` uses (no process, socket or clock). `format` is `"prometheus"`,
/// `"cef"`, `"leef"` or `"otlp"`; `input_kind` is `"jsonl"` (the live per-epoch stream, the
/// default) or `"result"` (a `receiver-trust` result document); `host` names the sender in
/// syslog events (default `kshana`). `syslog_all` gives one event per epoch instead of one per
/// band change; `expose_position` adds the receiver-reported position to the metrics;
/// `unix_nano` is the export time of the `"otlp"` payload. Syslog lines carry no timestamp.
/// Returns the text. Raises `ValueError` on an unknown format or an input with no epoch.
#[pyfunction]
#[pyo3(signature = (input, format, host="", input_kind="jsonl", expose_position=false, syslog_all=false, unix_nano=0))]
fn trust_telemetry(
    input: &str,
    format: &str,
    host: &str,
    input_kind: &str,
    expose_position: bool,
    syslog_all: bool,
    unix_nano: i128,
) -> PyResult<String> {
    use crate::surface::{self, TelemetryFormat, TelemetryInput, TelemetryOptions};
    let bad = |e: String| PyValueError::new_err(e);
    let unix_nano = u64::try_from(unix_nano)
        .map_err(|_| PyValueError::new_err("unix_nano must be between 0 and 2**64 - 1"))?;
    surface::trust_telemetry(
        input,
        TelemetryFormat::parse(format).map_err(bad)?,
        host,
        TelemetryInput::parse(input_kind).map_err(bad)?,
        TelemetryOptions {
            expose_position,
            syslog_all,
            unix_nano,
        },
        surface::MAX_INPUT_BYTES,
    )
    .map_err(bad)
}

/// Share of a route (GeoJSON LineString or `lat,lon` CSV text) through degraded cells of
/// one or more interference maps (the `geojson` of [`interference_map`]), optionally limited
/// to `date_from`..`date_to` (`YYYY-MM-DD`). Returns the report as JSON text. Cells not
/// observed are not evidence of a clear route, and this is not a forecast. Raises
/// `ValueError` on bad input.
#[pyfunction]
#[pyo3(signature = (route, maps, date_from=None, date_to=None))]
fn route_exposure(
    route: &str,
    maps: Vec<String>,
    date_from: Option<String>,
    date_to: Option<String>,
) -> PyResult<String> {
    let refs: Vec<&str> = maps.iter().map(String::as_str).collect();
    crate::surface::route_exposure(
        route,
        &refs,
        date_from.as_deref(),
        date_to.as_deref(),
        crate::surface::MAX_INPUT_BYTES,
    )
    .map_err(PyValueError::new_err)
}

/// Generate synthetic bridge NMEA 0183 for crew training from a `nmea-scenario` TOML.
/// Returns a dict with `nmea` (CRLF text), `log_json` (the instructor log,
/// `kshana-nmea-training/1`) and `log_text`. `seed` replaces the scenario's seed. Text
/// only: it is for training and testing, never for a vessel's live navigation systems.
/// Raises `ValueError` on an invalid scenario.
#[pyfunction]
#[pyo3(signature = (toml, seed=None))]
fn nmea_training<'py>(
    py: Python<'py>,
    toml: &str,
    seed: Option<u64>,
) -> PyResult<Bound<'py, PyAny>> {
    let t = crate::surface::nmea_training(toml, seed, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_to_py(
        py,
        &serde_json::json!({"nmea": t.nmea, "log_json": t.log_json, "log_text": t.log_text}),
    )
}

fn request_text(py: Python<'_>, request: &Bound<'_, PyAny>) -> PyResult<String> {
    match request.extract::<String>() {
        Ok(s) => Ok(s),
        Err(_) => py
            .import("json")?
            .call_method1("dumps", (request,))?
            .extract(),
    }
}

fn json_text_to_py<'py>(py: Python<'py>, out: &str) -> PyResult<Bound<'py, PyAny>> {
    let v: serde_json::Value =
        serde_json::from_str(out).map_err(|e| PyValueError::new_err(e.to_string()))?;
    json_to_py(py, &v)
}

/// The satellites a site sees under a skyline and their geometry. `request` is JSON or TOML
/// text, or a dict: `site`, `time_s`, `systems` (preset names or `leo-pvt` system tables)
/// and/or `satellites` (`az_deg`/`el_deg` or `ecef_m`), `mask_deg`, `skyline`, `clock`
/// (`"per-system"` or `"common"`), `sigma_uere_m`, `confidence`. Returns a dict: the
/// satellites with `visible` and `masked`, the DOPs, the ENU cofactor and covariance and the
/// error ellipses, labelled MODELLED, with the advisory. Raises `ValueError`.
#[pyfunction]
fn sky_geometry<'py>(py: Python<'py>, request: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
    let text = request_text(py, request)?;
    let out = crate::surface::sky_geometry(&text, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_text_to_py(py, &out)
}

/// `n` (1 to 5000) seeded least-squares position errors under independent Gaussian range
/// errors on the visible set of `request` (as for `sky_geometry`), with their empirical
/// covariance, RMS and the fraction inside the 95 % ellipse. Raises `ValueError`.
#[pyfunction]
#[pyo3(signature = (request, seed=0, n=1000))]
fn sky_monte_carlo<'py>(
    py: Python<'py>,
    request: &Bound<'py, PyAny>,
    seed: u64,
    n: usize,
) -> PyResult<Bound<'py, PyAny>> {
    let text = request_text(py, request)?;
    let out = crate::surface::sky_monte_carlo(&text, seed, n, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_text_to_py(py, &out)
}

/// The named skyline presets with their REPRESENTATIVE label, provenance and the cut every
/// degree of azimuth.
#[pyfunction]
fn skyline_presets(py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
    json_text_to_py(py, &crate::surface::skyline_presets())
}

/// A skyline section (text or dict) reduced to the normalised 360-entry table the engine uses.
/// Raises `ValueError` on an invalid section.
#[pyfunction]
fn skyline_profile<'py>(
    py: Python<'py>,
    skyline: &Bound<'py, PyAny>,
) -> PyResult<Bound<'py, PyAny>> {
    let text = request_text(py, skyline)?;
    let out = crate::surface::skyline_profile(&text, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)?;
    json_text_to_py(py, &out)
}

/// The skyplot of `request` (as for `sky_geometry`) as static SVG text. Raises `ValueError`.
#[pyfunction]
fn skyplot_svg(py: Python<'_>, request: &Bound<'_, PyAny>) -> PyResult<String> {
    let text = request_text(py, request)?;
    crate::surface::skyplot_svg(&text, crate::surface::MAX_INPUT_BYTES)
        .map_err(PyValueError::new_err)
}

/// The GNSS IQ signal names [`iq_scene`], [`iq_acquire`] and [`iq_track`] accept.
#[pyfunction]
fn iq_signals() -> Vec<String> {
    crate::iq::cli::signal_names()
        .iter()
        .map(|s| s.to_string())
        .collect()
}

#[pymodule]
fn kshana(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<PyRunOutput>()?;
    m.add_function(wrap_pyfunction!(run, m)?)?;
    m.add_function(wrap_pyfunction!(run_full, m)?)?;
    m.add_function(wrap_pyfunction!(run_typed, m)?)?;
    m.add_function(wrap_pyfunction!(scenario_kinds, m)?)?;
    m.add_function(wrap_pyfunction!(validate_toml, m)?)?;
    m.add_function(wrap_pyfunction!(list_kinds, m)?)?;
    m.add_function(wrap_pyfunction!(error_kind, m)?)?;
    m.add_function(wrap_pyfunction!(version, m)?)?;
    m.add_function(wrap_pyfunction!(receiver_trust, m)?)?;
    m.add_function(wrap_pyfunction!(receiver_trust_replay, m)?)?;
    m.add_function(wrap_pyfunction!(assess_vessel_log, m)?)?;
    m.add_function(wrap_pyfunction!(evidence_create, m)?)?;
    m.add_function(wrap_pyfunction!(evidence_verify, m)?)?;
    m.add_function(wrap_pyfunction!(interference_map, m)?)?;
    m.add_function(wrap_pyfunction!(compliance_report, m)?)?;
    m.add_function(wrap_pyfunction!(bench_export, m)?)?;
    m.add_function(wrap_pyfunction!(iq_sweep, m)?)?;
    m.add_function(wrap_pyfunction!(iq_info, m)?)?;
    m.add_function(wrap_pyfunction!(iq_inventory, m)?)?;
    m.add_function(wrap_pyfunction!(evidence_attach_timestamp, m)?)?;
    m.add_function(wrap_pyfunction!(evidence_keygen, m)?)?;
    m.add_function(wrap_pyfunction!(export_sp3, m)?)?;
    m.add_function(wrap_pyfunction!(export_omm, m)?)?;
    m.add_function(wrap_pyfunction!(export_oem, m)?)?;
    m.add_function(wrap_pyfunction!(export_formats, m)?)?;
    m.add_function(wrap_pyfunction!(export_scenario, m)?)?;
    m.add_function(wrap_pyfunction!(import_route, m)?)?;
    m.add_function(wrap_pyfunction!(animate_scenario, m)?)?;
    m.add_function(wrap_pyfunction!(list_examples, m)?)?;
    m.add_function(wrap_pyfunction!(get_example, m)?)?;
    m.add_function(wrap_pyfunction!(compliance_mapping, m)?)?;
    m.add_function(wrap_pyfunction!(osnma_verify, m)?)?;
    m.add_function(wrap_pyfunction!(trust_telemetry, m)?)?;
    m.add_function(wrap_pyfunction!(route_exposure, m)?)?;
    m.add_function(wrap_pyfunction!(nmea_training, m)?)?;
    m.add_function(wrap_pyfunction!(iq_scene, m)?)?;
    m.add_function(wrap_pyfunction!(iq_scene_broadcast, m)?)?;
    m.add_function(wrap_pyfunction!(iq_acquire, m)?)?;
    m.add_function(wrap_pyfunction!(iq_acq_surface, m)?)?;
    m.add_function(wrap_pyfunction!(iq_track, m)?)?;
    m.add_function(wrap_pyfunction!(iq_loop_designs, m)?)?;
    m.add_function(wrap_pyfunction!(iq_read_epochs, m)?)?;
    m.add_function(wrap_pyfunction!(iq_labfit, m)?)?;
    m.add_function(wrap_pyfunction!(iq_frontend, m)?)?;
    m.add_function(wrap_pyfunction!(iq_monitor, m)?)?;
    m.add_function(wrap_pyfunction!(iq_signals, m)?)?;
    m.add_function(wrap_pyfunction!(iq_test_conditions, m)?)?;
    m.add_function(wrap_pyfunction!(iq_campaign, m)?)?;
    m.add_function(wrap_pyfunction!(iq_campaign_report, m)?)?;
    m.add_function(wrap_pyfunction!(sky_geometry, m)?)?;
    m.add_function(wrap_pyfunction!(sky_monte_carlo, m)?)?;
    m.add_function(wrap_pyfunction!(skyline_presets, m)?)?;
    m.add_function(wrap_pyfunction!(skyline_profile, m)?)?;
    m.add_function(wrap_pyfunction!(skyplot_svg, m)?)?;
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
    Ok(())
}