Python API

Kshana tarjoaa ensiluokkaisen Python-ohjelmointirajapinnan (API), joka on rakennettu työkaluilla PyO3 ja maturin. Sen wheel-paketit (abi3) rakennetaan vakaata Pythonin binäärirajapintaa (ABI) vasten, joten yksi wheel-paketti alustaa kohti toimii kaikilla CPython-versioilla ≥ 3,9; paketit kattavat Linuxin, macOS:n ja Windowsin, kukin x86-64- ja 64-bittiselle ARM-arkkitehtuurille (ks. WHEEL_TAGS.md). Mukana toimitettu tyyppiprofiili (kshana.pyi + py.typed) antaa editoreille sekä työkaluille mypy ja pyright täydet tyyppitiedot.

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

Pikaopas#

Aja nämä katkelmat repositorion tarkistetun kopion juuresta: ne lukevat skenaariotiedostoja hakemistosta scenarios/. NumPy-rivi tarvitsee komennon 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

Pinta#

GNSS-IQ:n ja vastaanottimen luottamuksen allekirjoitukset#

Täydet allekirjoitukset sellaisina kuin kshana.pyi ne esittää (jokaisella pakollisten argumenttien jälkeisellä avainsanalla on oletus; profiilin docstring-kommentit kertovat kunkin merkityksen). Kääri listat samples_i / samples_q ja syötteet i / q taulukoiksi kutsulla 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]: ...

Laboratorioon toistettava kampanja Pythonista ja sen raportti (kampanjatiedosto ja sen tallenteet ovat sinun; valmis solu ohitetaan seuraavalla kutsulla; iq_read_epochs lukee seurantaepookkitiedoston, jonka ajo kshana iq track --epochs kirjoitti):

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

Kutsussa iq_track asetus acq_coherent määrää, kuinka monta koodijaksoa luovutushankinta integroi koherentisti. Versiosta 0.33.0 alkaen oletus on None, mikä tarkoittaa automaattista: noin 4 ms (4 jaksoa koodille, jonka pituus on 1 ms, kuten GPS L1 C/A, ja 1 jakso koodille, jonka pituus on vähintään 4 ms), jolloin kanava aloittaa riittävän läheltä, jotta taajuussilmukka voi vetäytyä sisään; asetus acq_coherent=1 säilyttää version 0,32 yhden jakson haun.

Tyypit#

Tiedosto kshana.pyi tyypittää suljetut sanastot tarkistimille (mypy, pyright): ScenarioKind (jokainen tyypin nimi, Literal), ErrorKind (neljä tunnistetta, jotka error_kind palauttaa), ProvenanceClass sekä sanakirjatyypit (TypedDict) KindMeta (lista näistä on kutsun scenario_kinds() paluuarvo) ja FieldUnit (tuloksen lohkon units alkio, joka laajentaa pakollisten avainten kantatyyppiä _FieldUnitBase). Ne ovat olemassa vain tyyppitarkistimille eikä niitä voi tuoda ajonaikaisesti; tests/typed_api_sync.rs pitää ne yhtäläisinä moottorin listojen kanssa.

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#

Toistettavuustaulukko#

realtime-frame-eop, lunar-time-budget, lunar-jamming, telecom-timing ja leo-navmsg julkaisevat aina taulukon; moonlight-service-volume julkaisee sellaisen, kun vientipaikka (export_site_lat_deg + export_site_lon_deg) on määritetty. Jokainen muu tyyppi palauttaa csv = None. Teksti on samat tavut, jotka komentorivi (CLI) kirjoittaa tiedostoksi <scenario>.table.csv, joten julkaistun taulukon wheel-paketista toistava tarkastaja ei koskaan joudu turvautumaan komentoriviin:

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

Tuloksen rakenne#

Laatumittarit ja aikasarja sijaitsevat ajon puolen mukaan, eivät ylätasolla. Kelloajon asiakirja on:

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

joten laatumittarit ovat data["quantum"]["fom"], eivät koskaan data["figures_of_merit"]. Muut tyypit tuottavat omat pakettinsa; täydellinen kenttäkohtainen viite yksiköineen ja lähdeviitteineen on field-units-schema.json, ja lukijan opas on SCHEMA.md.

Huomautukset#

• validate_toml ja error_kind suorittavat skenaarion, joten ne tuovat esiin jäsennys-, konfiguraatio- ja ajonaikaiset virheet. Ne eivät koskaan nosta poikkeusta.
• run / run_typed nostavat virheen ValueError virheellisestä skenaariosta.
• Tulokset ovat toistettavia: skenaario kantaa mukanaan kentän seed, ja moottori tallentaa kentän scenario_hash, joten sama syöte tuottaa tavu tavulta saman tulosteen samalla alustalla (ks. REPRODUCIBILITY.md).
• .array, .columns ja .series palauttavat täsmälleen saman numpy.asarray(...) samoista arvoista kuin .data() (sama dtype, bitti bitiltä sama; tests/python/test_numpy_runoutput.py). Ne tuovat NumPyn kutsuttaessa ja nostavat virheen ImportError ilman sitä; NumPy pysyy paketin ajonaikaisten vaatimusten ulkopuolella. KeyError tarkoittaa, että polku puuttuu, ja ValueError, että siellä oleva arvo ei ole numeerinen taulukko (tai ei lista tietueita). Ajettavat muistikirjat ovat hakemistossa notebooks/.

Ei Pythonissa: pitkäkestoiset prosessit#

kshana receiver-trust live (virtaprosessi portilla ja valinnaisella TCP-palvelimella --listen) sekä telemetrian viejät (Prometheus, OTLP, syslog; docs/TRUST-TELEMETRY.md) ja suoratoisto (kshana nmea-scenario --tcp/--udp) ovat vain komentorivillä: sidos palaa kutsun palatessa eikä pidä socketia auki. Python kattaa kunkin eräversion: receiver_trust lokille, receiver_trust_replay virtakatkelmalle (pistemäärät ja tilat; porttia ei sovelleta) ja nmea_training generoidulle tekstille.