Python-API

Kshana liefert eine erstklassige Python-Programmierschnittstelle (API, application programming interface), gebaut mit PyO3 und maturin. Ihre abi3-Wheels werden gegen die stabile Python-Binärschnittstelle (ABI, application binary interface) gebaut, sodass ein Wheel je Plattform über CPython ≥ 3,9 hinweg funktioniert; die Wheels decken Linux, macOS und Windows ab, jeweils auf x86-64 und 64-Bit-ARM (siehe WHEEL_TAGS.md). Ein mitgeliefertes Typ-Stub (kshana.pyi + py.typed) gibt Editoren und mypy/pyright vollständige Typinformationen.

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

Schnellstart#

Führen Sie diese Snippets aus dem Wurzelverzeichnis eines Repository-Checkouts aus: Sie lesen Szenariodateien unter scenarios/. Die NumPy-Zeile braucht 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

Oberfläche#

Signaturen für GNSS-IQ und Empfängervertrauen#

Die vollständigen Signaturen, wie kshana.pyi sie angibt (jedes Schlüsselwort nach den erforderlichen Argumenten hat einen Standardwert; die Docstrings im Stub geben jeweils die Bedeutung an). Umhüllen Sie die Listen samples_i / samples_q und die Eingaben i / q für Arrays mit numpy.asarray(...).

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]: ...

Eine Lab-Replay-Kampagne aus Python, dann ihr Bericht (die Kampagnendatei und ihre Aufzeichnungen sind Ihre; eine abgeschlossene Zelle wird beim nächsten Aufruf übersprungen; iq_read_epochs liest eine Nachführ-Epochendatei, die ein Lauf von kshana iq track --epochs geschrieben hat):

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

acq_coherent von iq_track legt fest, über wie viele Codeperioden die Übergabe-Akquisition kohärent integriert. Ab 0.33.0 ist der Standardwert None, also automatisch: etwa 4 ms (4 Perioden bei einem Code von 1 ms wie GPS L1 C/A, 1 bei einem Code von 4 ms oder länger), sodass der Kanal nahe genug startet, damit die Frequenzschleife einrasten kann; acq_coherent=1 behält die Ein-Perioden-Suche von 0,32 bei.

Typen#

kshana.pyi typisiert die geschlossenen Vokabulare für Prüfprogramme (mypy, pyright): ScenarioKind (jeder Artname, ein Literal), ErrorKind (die vier Tags, die error_kind zurückgibt), ProvenanceClass und die TypedDicts KindMeta (wovon scenario_kinds() eine Liste zurückgibt) und FieldUnit (ein Eintrag im Block units eines Ergebnisses, der die Basis _FieldUnitBase mit den Pflichtschlüsseln erweitert). Sie existieren nur für Typprüfer und sind zur Laufzeit nicht importierbar; tests/typed_api_sync.rs hält sie gleich mit den Listen der Engine.

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#

Die Reproduzierbarkeitstabelle#

realtime-frame-eop, lunar-time-budget, lunar-jamming, telecom-timing und leo-navmsg veröffentlichen immer eine Tabelle; moonlight-service-volume veröffentlicht eine, wenn ein Exportstandort (export_site_lat_deg + export_site_lon_deg) konfiguriert ist. Jede andere Art gibt csv = None zurück. Der Text hat dieselben Bytes, die die CLI (command-line interface) als <scenario>.table.csv schreibt, sodass ein Prüfer, der eine veröffentlichte Tabelle aus dem Wheel reproduziert, nie auf die Kommandozeile wechseln muss:

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")

Form des Ergebnisses#

Die Gütemaße und die Zeitreihen liegen je Lauf-Seite, nicht auf oberster Ebene. Für einen Uhrenlauf lautet das Dokument:

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

die Gütemaße sind also data["quantum"]["fom"], nie data["figures_of_merit"]. Andere Arten geben ihre eigenen Packs aus; die vollständige Referenz je Feld, mit Einheiten und einem Quellverweis, ist field-units-schema.json, und der Leitfaden für Leser ist SCHEMA.md.

Hinweise#

• validate_toml und error_kind führen das Szenario aus, sodass sie Parse-, Konfigurations- und Laufzeitfehler sichtbar machen. Sie lösen nie eine Ausnahme aus.
• run / run_typed lösen bei einem ungültigen Szenario ValueError aus.
• Ergebnisse sind reproduzierbar: Ein Szenario trägt seinen seed, und die Engine erfasst einen scenario_hash, sodass dieselbe Eingabe auf derselben Plattform byte-identische Ausgabe liefert (siehe REPRODUCIBILITY.md).
• .array, .columns und .series geben genau numpy.asarray(...) derselben Werte zurück, die .data() liefert (gleicher dtype, bit-identisch; tests/python/test_numpy_runoutput.py). Sie importieren NumPy beim Aufruf und lösen ohne es ImportError aus; NumPy bleibt aus den Laufzeitanforderungen des Pakets heraus. KeyError bedeutet, dass der Pfad fehlt, ValueError, dass der Wert dort kein numerisches Array ist (oder keine Liste von Datensätzen). Lauffähige Notebooks stehen in notebooks/.

Nicht in Python: langlaufende Prozesse#

kshana receiver-trust live (ein Stream-Prozess mit einem Gate und einem optionalen TCP-Server --listen) und die Telemetrie-Exporter (Prometheus, OTLP, Syslog; docs/TRUST-TELEMETRY.md) sowie das Streaming von kshana nmea-scenario --tcp/--udp gibt es nur auf der Kommandozeile: Ein Binding kehrt zurück, wenn der Aufruf zurückkehrt, und hält keinen Socket offen. Python deckt die Batch-Form von jedem ab: receiver_trust für ein Log, receiver_trust_replay für einen Stream-Auszug (Werte und Zustände; das Gate wird nicht angewendet), nmea_training für den erzeugten Text.