API Python

Kshana livre une interface de programmation d'application (API) Python de premier plan, construite avec PyO3 et maturin. Ses wheels abi3 sont construits sur l'interface binaire d'application (ABI) stable de Python, de sorte qu'un seul wheel par plateforme fonctionne avec CPython ≥ 3,9 ; les wheels couvrent Linux, macOS et Windows, chacun en x86-64 et en ARM 64 bits (voir WHEEL_TAGS.md). Un stub de types fourni (kshana.pyi + py.typed) donne aux éditeurs et à mypy/pyright toute l'information de typage.

pip install kshana            # from the Python Package Index, PyPI (release wheels)
# or, from a checkout:
pip install maturin && maturin develop --features python

Démarrage rapide#

Exécutez ces extraits depuis la racine d'une copie du dépôt : ils lisent des fichiers de scénario sous scenarios/. La ligne NumPy nécessite pip install numpy.

import kshana

toml = open("scenarios/clock-holdover.toml").read()

# Typed result with a parsed-dict accessor:
out = kshana.run_typed(toml)
print(out.summary)
data = out.data()                         # a Python dict — no JSON re-parsing
fom = data["quantum"]["fom"]              # figures of merit, quantum side
classical_fom = data["classical"]["fom"]  # … and the classical baseline
print(out.json[:80], "...")               # raw JSON also available
open("chart.svg", "w").write(out.svg)     # the chart SVG

# NumPy arrays straight from the result (NumPy is imported on call; it is not a
# dependency of the package):
curve = out.columns("quantum.adev_curve")  # {"tau_s": ndarray, "adev": ndarray, ...}
adev = curve["adev"]
everything = out.series()                  # every record series, keyed by path

Surface#

Signatures de l'IQ GNSS et de la confiance du récepteur#

Les signatures complètes, telles que kshana.pyi les énonce (chaque mot-clé après les arguments obligatoires a une valeur par défaut ; les docstrings du stub donnent le sens de chacun). Encapsulez les listes samples_i / samples_q et les entrées i / q avec numpy.asarray(...) pour obtenir des tableaux.

def receiver_trust(toml: str) -> RunOutput: ...

def receiver_trust_replay(session_toml: str, nmea: str | bytes) -> dict: ...
def assess_vessel_log(session_toml: str, log: str | bytes) -> dict: ...
def jammertest_import(catalog_json: str, plans: list[str], log_start_utc: str, location: int | None = None,
                      local_offset_h: float = 2.0, year: int | None = None) -> dict: ...
def jammertest_score(session_toml: str, log: str | bytes, catalog_json: str, plans: list[str],
                     location: int | None = None, local_offset_h: float = 2.0,
                     log_start_utc: str | None = None, year: int | None = None,
                     recover_hold_s: float = 10.0, guard_s: float = 60.0) -> dict: ...
def evidence_create(session_toml: str, log: str | bytes, from_s: float, to_s: float,
                    title: str | None = None, created_utc: str | None = None,
                    seed_hex: str | None = None) -> dict: ...
def evidence_verify(files: dict[str, str | bytes], public_key: str | None = None,
                    full_log: str | bytes | None = None,
                    require_timestamp: bool = False) -> dict: ...
def interference_map(source: str, csv: str, dataset: str, cell_deg: float | None = None,
                     licence: str | None = None, licence_url: str | None = None,
                     attribution: str | None = None, land_geojson: str | None = None) -> list[dict]: ...
def route_exposure(route: str, maps: list[str], date_from: str | None = None,
                   date_to: str | None = None) -> str: ...
def export_sp3(toml: str) -> str: ...
def export_omm(toml: str) -> str: ...
def export_oem(toml: str) -> str: ...
def export_formats(toml: str) -> list[dict]: ...
def export_scenario(toml: str, format: str) -> dict: ...
def import_route(toml: str, geojson: str) -> str: ...
def animate_scenario(toml: str, format: str = "svg", fps: int | None = None,
                     duration_s: float | None = None, width: int | None = None) -> dict: ...
def list_examples(kind: str | None = None) -> dict: ...
def get_example(name: str) -> str: ...
def evidence_attach_timestamp(files: dict[str, str | bytes], token: str | bytes,
                              replace: bool = False) -> dict: ...
def evidence_keygen() -> dict: ...
def iq_sweep(i: list[float], q: list[float], fs_hz: float, signal: str, prns: list[int],
             **options) -> dict: ...
def iq_info(path: str, hash: bool = False) -> dict: ...
def iq_inventory(directory: str, recursive: bool = False, hash: bool = False) -> list[dict]: ...
def bench_export(toml: str, epoch: str | None = None) -> dict: ...
def compliance_report(runs: list[dict[str, str]]) -> dict: ...
def compliance_mapping(sources: bool = False) -> str: ...
def standards_pack_list() -> dict[str, Any]: ...
def standards_pack_run(pack_id: str) -> dict[str, Any]: ...
def lunanet_afs_check_spec() -> dict[str, Any]: ...
def lunanet_afs_check_symbols(symbols: str, max_sp_errors: int = 0) -> dict[str, Any]: ...
def lunanet_afs_generate(frames: int = 1, toi0: int = 0) -> str: ...
def evidence_round_trip(scoring_toml: str, scenario_toml: str, conditions: str, log: str | bytes, from_s: float, to_s: float, title: Optional[str] = None, created_utc: Optional[str] = None, seed_hex: Optional[str] = None) -> dict[str, Any]: ...
def bench_score(truth_json: str, predictions_json: str, threshold: float | None = None) -> str: ...
def bench_leaderboard(results: list[str]) -> str: ...
def osnma_verify(data: bytes | str, format: str = "", start_gst: int | None = None,
                 merkle_root_hex: str = "", public_keys: list[str] | None = None,
                 reference_time: int | None = None, max_time_error: int | None = None) -> dict: ...
def sky_geometry(request: str | dict[str, Any]) -> dict[str, Any]: ...
def sky_monte_carlo(request: str | dict[str, Any], seed: int = 0, n: int = 1000) -> dict[str, Any]: ...
def skyline_presets() -> list[dict[str, Any]]: ...
def skyline_profile(skyline: str | dict[str, Any]) -> dict[str, Any]: ...
def skyplot_svg(request: str | dict[str, Any]) -> str: ...
def trust_telemetry(input: str, format: str, host: str = "", input_kind: str = "jsonl",
                    expose_position: bool = False, syslog_all: bool = False,
                    unix_nano: int = 0) -> str: ...
def nmea_training(toml: str, seed: int | None = None) -> dict[str, str]: ...

def iq_signals() -> list[str]: ...

def iq_scene(
    fs_hz: float,
    duration_s: float,
    signal: str,
    prns: list[int],
    dopplers: Optional[list[float]] = ...,
    cn0_dbhz: Optional[float] = ...,
    center_hz: Optional[float] = ...,
    if_hz: float = ...,
    noise: bool = ...,
    noise_figure_db: float = ...,
    seed: int = ...,
    data: bool = ...,
    threads: int = ...,
    iono_stec: Optional[float] = ...,
    iono_vtec: Optional[float] = ...,
    iono_klobuchar: bool = ...,
    tropo: bool = ...,
    tropo_doy: float = ...,
    s4: Optional[float] = ...,
    scint_tau0: float = ...,
    sigma_phi: float = ...,
    multipath_height: Optional[float] = ...,
    multipath_ground: str = ...,
    land_mobile: bool = ...,
    nlos: bool = ...,
) -> dict[str, Any]: ...

def iq_scene_broadcast(
    fs_hz: float,
    window_s: float,
    nav_text: str,
    rx_lat: float,
    rx_lon: float,
    rx_alt: float,
    prns: Optional[list[int]] = ...,
    start_tow: float = ...,
    cn0_dbhz: Optional[float] = ...,
    center_hz: Optional[float] = ...,
    if_hz: float = ...,
    noise: bool = ...,
    noise_figure_db: float = ...,
    seed: int = ...,
    mask_deg: float = ...,
    threads: int = ...,
) -> dict[str, Any]: ...

def iq_acquire(
    i: list[float],
    q: list[float],
    fs_hz: float,
    signal: str,
    prns: list[int],
    if_hz: float = ...,
    center_hz: Optional[float] = ...,
    coherent: int = ...,
    noncoherent: int = ...,
    doppler_max: float = ...,
    doppler_step: Optional[float] = ...,
    pfa: float = ...,
) -> list[dict[str, Any]]: ...

def iq_acq_surface(
    i: list[float],
    q: list[float],
    fs_hz: float,
    signal: str,
    prn: int,
    if_hz: float = ...,
    center_hz: Optional[float] = ...,
    coherent: int = ...,
    noncoherent: int = ...,
    doppler_max: float = ...,
    doppler_step: Optional[float] = ...,
    pfa: float = ...,
) -> dict[str, Any]: ...

def iq_track(
    i: list[float],
    q: list[float],
    fs_hz: float,
    signal: str,
    prns: list[int],
    if_hz: float = ...,
    center_hz: Optional[float] = ...,
    pll_bw: Optional[float] = ...,
    fll_bw: Optional[float] = ...,
    dll_bw: Optional[float] = ...,
    spacing: Optional[float] = ...,
    coherent: Optional[int] = ...,
    periods_per_bit: Optional[int] = ...,
    acq_coherent: Optional[int] = ...,
    acq_noncoherent: int = ...,
    doppler_max: float = ...,
    max_seconds: Optional[float] = ...,
) -> dict[str, Any]: ...

def iq_frontend(
    i: list[float],
    q: list[float],
    fs_hz: float,
    bandpass_lo: Optional[float] = ...,
    bandpass_hi: Optional[float] = ...,
    bandpass_transition: Optional[float] = ...,
    bandpass_atten: float = ...,
    notch: bool = ...,
    notch_r: float = ...,
    notch_mu: float = ...,
    blank: Optional[float] = ...,
    blank_hold: int = ...,
    excise: bool = ...,
    excise_fft: int = ...,
    excise_pfa: float = ...,
    agc: bool = ...,
    agc_tau: float = ...,
    bits: Optional[int] = ...,
    quant_step: Optional[float] = ...,
    no_agc: bool = ...,
) -> dict[str, Any]: ...

def iq_labfit(toml: str) -> dict[str, Any]: ...
def iq_loop_designs(toml: str) -> list[dict[str, Any]]: ...
def iq_read_epochs(path: str) -> dict[str, Any]: ...
def iq_monitor(path: str, signal: Optional[str] = ..., prns: Optional[list[int]] = ...,
               power: bool = ..., spectral: bool = ..., settings: Optional[str] = ...,
               **options) -> dict[str, Any]: ...
def iq_test_conditions(conditions: str) -> dict[str, Any]: ...
def iq_campaign(campaign: str, out_dir: str, workers: int = 0, resume: bool = True,
                max_cells: Optional[int] = None, dry_run: bool = False) -> dict[str, Any]: ...
def iq_campaign_report(out_dir: str) -> dict[str, Any]: ...

Une campagne de rejeu de laboratoire depuis Python, puis son rapport (le fichier de campagne et ses enregistrements sont les vôtres ; une cellule terminée est ignorée à l'appel suivant ; iq_read_epochs lit un fichier d'époques de poursuite écrit par une exécution de kshana iq track --epochs) :

import kshana

summary = kshana.iq_campaign("lab.campaign.toml", "campaign-out", workers=4)
report = kshana.iq_campaign_report("campaign-out")
print(report["digest"])          # None until every cell is done

Le acq_coherent de iq_track fixe le nombre de périodes de code que l'acquisition de transfert intègre de façon cohérente. À partir de 0.33.0, il vaut par défaut None, ce qui signifie automatique : environ 4 ms (4 périodes pour un code de 1 ms comme le GPS L1 C/A, 1 pour un code de 4 ms ou plus), de sorte que le canal démarre assez près pour que la boucle de fréquence s'accroche ; acq_coherent=1 conserve la recherche à une période de 0,32.

Types#

kshana.pyi type les vocabulaires fermés pour les vérificateurs (mypy, pyright) : ScenarioKind (chaque nom de type de scénario, un Literal), ErrorKind (les quatre étiquettes que renvoie error_kind), ProvenanceClass, et les TypedDict KindMeta (dont scenario_kinds() renvoie une liste) et FieldUnit (une entrée du bloc units d'un résultat, qui étend la base de clés obligatoires _FieldUnitBase). Ils n'existent que pour les vérificateurs de types et ne sont pas importables à l'exécution ; tests/typed_api_sync.rs les maintient égaux aux listes du moteur.

from typing import TYPE_CHECKING
import kshana
if TYPE_CHECKING:
    from kshana import ScenarioKind

names: "list[ScenarioKind]" = [k["name"] for k in kshana.scenario_kinds()]

RunOutput#

La table de reproductibilité#

realtime-frame-eop, lunar-time-budget, lunar-jamming, telecom-timing et leo-navmsg publient toujours une table ; moonlight-service-volume en publie une lorsqu'un site d'export (export_site_lat_deg + export_site_lon_deg) est configuré. Tout autre type de scénario renvoie csv = None. Le texte est constitué des mêmes octets que ceux que la CLI (interface en ligne de commande) écrit sous <scenario>.table.csv, de sorte qu'un relecteur qui reproduit une table publiée à partir du wheel n'a jamais à passer à la ligne de commande :

import kshana

out = kshana.run_typed(open("scenarios/realtime-frame-eop.toml").read())
if out.csv is not None:
    print(out.write_csv("table.csv"), "bytes written")

Forme du résultat#

Les indicateurs de performance et les séries temporelles se trouvent par côté d'exécution, et non au niveau supérieur. Pour une exécution d'horloge, le document est :

schema_version  engine_version  scenario_hash  seed  threshold_ns  units  figure_tiers
quantum   → spec · series · fom · adev_curve · filter_health
classical → spec · series · fom · adev_curve · filter_health

de sorte que les indicateurs de performance sont data["quantum"]["fom"], jamais data["figures_of_merit"]. D'autres types de scénario émettent leurs propres paquets ; la référence complète champ par champ, avec unités et pointeur vers la source, est field-units-schema.json et le guide du lecteur est SCHEMA.md.

Notes#

• validate_toml et error_kind exécutent le scénario, de sorte qu'ils font remonter les erreurs d'analyse, de configuration et d'exécution. Ils ne lèvent jamais d'exception.
• run / run_typed lèvent ValueError sur un scénario invalide.
• Les résultats sont reproductibles : un scénario porte sa seed et le moteur enregistre un scenario_hash, de sorte que la même entrée donne une sortie identique octet pour octet sur la même plateforme (voir REPRODUCIBILITY.md).
• .array, .columns et .series renvoient exactement numpy.asarray(...) des mêmes valeurs que donne .data() (même dtype, identiques au bit près ; tests/python/test_numpy_runoutput.py). Ils importent NumPy lorsqu'on les appelle et lèvent ImportError sans lui ; NumPy reste en dehors des dépendances d'exécution du paquet. KeyError signifie que le chemin est absent, ValueError que la valeur qui s'y trouve n'est pas un tableau numérique (ou pas une liste d'enregistrements). Des notebooks exécutables se trouvent dans notebooks/.

Pas en Python : les processus de longue durée#

kshana receiver-trust live (un processus de flux avec une passerelle et un serveur TCP --listen facultatif) et les exportateurs de télémétrie (Prometheus, OTLP, syslog ; docs/TRUST-TELEMETRY.md) et le flux kshana nmea-scenario --tcp/--udp relèvent uniquement de la ligne de commande : une liaison rend la main quand l'appel se termine et ne garde pas de socket ouvert. Python couvre la forme par lot de chacun : receiver_trust pour un journal, receiver_trust_replay pour un extrait de flux (scores et états ; la passerelle n'est pas appliquée), nmea_training pour le texte généré.