kshana 0.33.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>,
}

#[pymethods]
impl PyRunOutput {
    /// The result document parsed into a Python `dict` (figures of merit,
    /// time series, provenance, ...). NumPy users can wrap any numeric list with
    /// `numpy.asarray(...)`.
    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)
    }

    /// 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()`, and `.data()`). 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. 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, auto_coherent_periods, 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::{replay, CarrierLoop, ChannelInit, EpochOutput, LoopConfig};
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`. Raises `ValueError` on an invalid scene or channel.
#[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))]
#[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,
) -> PyResult<Bound<'py, PyAny>> {
    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,
    };
    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;
        if let Some(ch) =
            build_channel(&chan, carrier, params.seed, start_tow).map_err(PyValueError::new_err)?
        {
            scene.set_channel(ch);
        }
    }
    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)
}

/// 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 grid = acquire(&samples, &spec, code, &cfg).map_err(PyValueError::new_err)?;
        let r = grid.result;
        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))
}

/// Acquire then track each PRN over complex samples. Returns a dict with one entry per
/// channel (`code` and a list of per-epoch dicts: `doppler_hz`, `code_phase_chips`, `pli`,
/// `phase_lock`, `cn0_nwpr_dbhz`, the prompt `i_prompt`/`q_prompt`, ...). The loop design
/// starts from the GPS-L1-C/A-like default; any of `pll_bw`, `fll_bw`, `dll_bw`, `spacing`,
/// `coherent` overrides it. The initialising acquisition integrates `acq_coherent` code
/// periods coherently; the default `None` is 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.
#[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=1, doppler_max=5000.0, max_seconds=None))]
#[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: usize,
    doppler_max: f64,
    max_seconds: Option<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,
    };
    // Acquisition to initialise each channel.
    // Default auto (≈4 ms coherent); `acq_coherent=1` restores the 0.32 one-period search.
    let acq_coherent = acq_coherent
        .unwrap_or_else(|| auto_coherent_periods(codes[0].period_s()))
        .max(1);
    let acq = AcqConfig {
        coherent_periods: acq_coherent,
        noncoherent: acq_noncoherent.max(1),
        doppler_max_hz: doppler_max,
        doppler_step_hz: 2.0 / (3.0 * acq_coherent as f64 * codes[0].period_s()),
        pfa: 1e-3,
    };
    let mut inits = Vec::new();
    for code in &codes {
        let grid = acquire(&samples, &spec, code, &acq).map_err(PyValueError::new_err)?;
        if !grid.result.acquired {
            return Err(PyValueError::new_err(format!(
                "{}: not acquired (statistic {:.2} < threshold {:.2})",
                code.name(),
                grid.result.statistic,
                grid.result.threshold
            )));
        }
        let arc: std::sync::Arc<dyn SpreadingCode + Send + Sync> =
            std::sync::Arc::new(code.clone());
        inits.push(ChannelInit::from_acquisition(
            arc,
            &grid.result,
            &spec,
            0,
            periods_per_bit,
        ));
    }
    // One loop design from the overrides.
    let mut cfg = LoopConfig::default();
    if let Some(d) = spacing {
        cfg.spacing_chips = d;
    }
    if let Some(bn) = dll_bw {
        cfg.dll_bn_hz = bn;
    }
    if let Some(n) = coherent {
        cfg.coherent_periods = n.max(1);
    }
    if pll_bw.is_some() || fll_bw.is_some() {
        cfg.carrier = CarrierLoop::FllAssistedPll {
            pll_order: 2,
            pll_bn_hz: pll_bw.unwrap_or(15.0),
            fll_order: 1,
            fll_bn_hz: fll_bw.unwrap_or(10.0),
        };
    }
    let max_samples = max_seconds.map(|s| (s * fs_hz).round() as u64);
    let mut src = crate::iq::VecSource::new(spec, samples);
    let results = replay(&mut src, &inits, &[cfg], max_samples)
        .map_err(|e| PyValueError::new_err(e.to_string()))?;
    let chans: Vec<serde_json::Value> = codes
        .iter()
        .zip(&results[0].channels)
        .map(|(code, epochs)| {
            serde_json::json!({
                "code": code.name(),
                "epochs": epochs.iter().map(epoch_value).collect::<Vec<_>>(),
            })
        })
        .collect();
    json_to_py(
        py,
        &serde_json::json!({ "fs_hz": fs_hz, "channels": chans }),
    )
}

/// One tracking epoch as a JSON value for the Python surface.
fn epoch_value(e: &EpochOutput) -> serde_json::Value {
    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,
        "i_prompt": e.prompt.re,
        "q_prompt": e.prompt.im,
        "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,
    })
}

/// 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. Raises `ValueError` on an invalid scenario or unreadable log.
#[pyfunction]
fn iq_labfit<'py>(py: Python<'py>, toml: &str) -> PyResult<Bound<'py, PyAny>> {
    let out = crate::iq::labfit::run_toml(toml).map_err(PyValueError::new_err)?;
    let report: serde_json::Value =
        serde_json::from_str(&out.json).map_err(|e| PyValueError::new_err(e.to_string()))?;
    let v = serde_json::json!({
        "report": report,
        "residuals_csv": out.residuals_csv,
        "predictions_csv": out.predictions_csv,
        "markdown": out.markdown,
    });
    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 }))
}

/// 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!(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_track, m)?)?;
    m.add_function(wrap_pyfunction!(iq_labfit, m)?)?;
    m.add_function(wrap_pyfunction!(iq_frontend, m)?)?;
    m.add_function(wrap_pyfunction!(iq_signals, m)?)?;
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
    Ok(())
}