commit 5303986377ce8feab31e347b798d4e4e27cf5cfe Author: ThePetrovich Date: Tue Aug 18 22:03:52 2026 +0800 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5e70718 --- /dev/null +++ b/.gitignore @@ -0,0 +1,155 @@ + +# Created by https://www.toptal.com/developers/gitignore/api/python +# Edit at https://www.toptal.com/developers/gitignore?templates=python + +### Python ### +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +pytestdebug.log + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ +doc/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ +pythonenv* + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# profiling data +.prof + +# Db and static files +*.sqlite3 +/media +/static +/postgres +/EXAMPLE_* + +# Docker +docker-compose.override.yml +docker-compose.override + +# End of https://www.toptal.com/developers/gitignore/api/python diff --git a/README.md b/README.md new file mode 100644 index 0000000..d7893b2 --- /dev/null +++ b/README.md @@ -0,0 +1,74 @@ +# odm + +Orbit Data Messages: read, write and convert orbital element sets. + +Everything ODMS knows about orbital *formats*, with nothing it knows about +Django, HTTP or a database. Dependencies are `sgp4` and the standard library. + +## Why it is a library + +The formats used to live in the service that happened to serve them first — +TLE rendering in a Django app's `formats.py`, the element-set catalogue in +another app, the OPM/OEM builders in a third, and the canonical record itself +implicit in a model row. That worked while one service read the data. It stops +working the moment a second one does: a format two services implement separately +is a format they will eventually disagree about, and the disagreement surfaces as +a satellite whose OMM and TLE describe slightly different orbits. + +So: the OMM is the record. TLE text, OMM XML, OMM KVN and Celestrak GP CSV are +*renderings* of it, and the CCSDS SANA element sets are conversions of the state +it propagates to. One definition each, here. + +## Use + +```python +from odm import OmmRecord, parse_tle, registry + +parsed = parse_tle(name, line1, line2) +record = OmmRecord(omm=parsed.omm, object_name=parsed.name, line1=line1, line2=line2) + +xml = registry.get("omm_xml").write([record]) +csv = registry.get("csv").write([record]) +kep = registry.get("keplerian").write([record], backend=backend, at=when) +``` + +Every writer takes the same arguments — `write(records, *, backend=None, +at=None)` — so a caller never has to know which kind it got. `backend` and `at` +matter only to the element-set formats, which propagate before converting. + +## Modules + +| Module | What | +|---|---| +| `records` | `OmmRecord`, `ParsedTLE`, `StateVector` — what readers produce and writers consume. | +| `tle` | TLE text in and out. Lossy by construction; see the module docstring. | +| `omm` | CCSDS 502.0-B-3 OMM, XML and KVN. The authoritative rendering. | +| `gp` | Celestrak GP CSV, in their exact column order. | +| `norad` | Catalog-ID normalisation, ALPHA-5, and the temp-ID fallback. | +| `orbits` | Mean-element derivations: period, apogee, perigee, altitudes. | +| `conversions` | State vector → one SANA element set. | +| `element_sets` | The SANA catalogue, and computing a set for an OMM. | +| `messages` | CCSDS OPM/OEM: build and parse. | +| `registry` | Every output format in one table. | +| `propagator` | The seam a propagation backend implements. | + +## The propagator seam + +`odm` does no orbital mechanics beyond mean-element algebra. Anything needing a +propagated state takes a `PropagatorBackend` — `element_sets.compute_element_set`, +`messages.build_opm`, `messages.build_oem`, the element-set writers. + +The backend is always **passed in**, never looked up. A library that reached for +a service's configured default would only work inside that service, which is the +thing this package exists not to be. Implementations live in the `yksa_orbital` +Django app (pure-Python SGP4, and an HTTP client to the Orekit sidecar). + +## What is deliberately not here + +- **Storage.** ODMS holds elements in Django, `track` holds them elsewhere, the + sidecar holds none. The format code works on records; the service adapts. +- **Serving.** Content types and filenames are in `registry` because they are + properties of the format, but nothing here builds a response. +- **The simulation model.** Force models, atmospheres and propagator settings + belong to the Orekit sidecar (`services/orekit/config.py`), which is their one + definition for the same reason this package is the formats' one definition. diff --git a/odm/__init__.py b/odm/__init__.py new file mode 100644 index 0000000..a10bafc --- /dev/null +++ b/odm/__init__.py @@ -0,0 +1,128 @@ +"""Orbit Data Messages: read, write and convert element sets. + +Everything ODMS knows about orbital *formats*, with nothing it knows about +Django, HTTP or a database. The OMM is the record; TLE text, OMM XML, OMM KVN +and Celestrak GP CSV are renderings of it, and the CCSDS SANA element sets are +conversions of the state it propagates to. + +:: + + from odm import OmmRecord, parse_tle, as_omm_xml, registry + + record = OmmRecord(omm=parse_tle(name, l1, l2).omm) + xml = registry.get("omm_xml").write([record]) + +The package is standard library plus ``sgp4``. It is separate from the services +because more than one of them now reads ODMS data -- and a format that two +services implement separately is a format they will eventually disagree about. + +Layout +------ + +``records`` the in-memory types every reader and writer works on +``tle`` TLE text in and out (lossy by construction -- see the module) +``omm`` CCSDS OMM, XML and KVN +``gp`` Celestrak's GP CSV +``norad`` catalog-ID normalisation, ALPHA-5 and the temp-ID scheme +``orbits`` mean-element derivations (period, apogee, perigee, ...) +``conversions`` state vector -> one SANA element set +``element_sets`` the SANA catalogue, and computing a set for an OMM +``messages`` CCSDS OPM/OEM: build and parse +``registry`` every output format in one table +``propagator`` the seam a propagation backend implements +""" + +from .element_sets import ( + API_ELEMENT_SET_KEYS, + DISPLAY_SET_KEYS, + ELEMENT_SETS, + ElementSet, + compute_element_set, + display_payload, + get_set, +) +from .gp import GP_FIELDS, as_celestrak_csv +from .messages import ( + MessageMeta, + ParsedMessage, + build_oem, + build_opm, + parse_message, + parsed_states_to_statevectors, + parsed_to_json, +) +from .norad import ( + ALPHA5_LETTERS, + ALPHA5_MAX, + ALPHA5_MIN, + TEMP_NORAD_MAX, + TEMP_NORAD_MIN, + coerce_norad_to_int, + from_alpha5, + normalise_norad, + temp_tle_norad, + tle_catalog_field, + to_alpha5, +) +from .omm import ( + CCSDS_OMM_VERSION, + OMM_DEFAULT_METADATA, + as_omm_kvn, + as_omm_xml, + format_ccsds_datetime, + omm_xml_to_records, + parse_omm_epoch, +) +from .propagator import BackendBusy, PropagationError, PropagatorBackend +from .records import OmmRecord, ParsedTLE, StateVector +from .tle import as_plaintext, as_txt_txt, indexed_name, parse_tle, tle_from_gp + +__all__ = [ + "ALPHA5_LETTERS", + "ALPHA5_MAX", + "ALPHA5_MIN", + "API_ELEMENT_SET_KEYS", + "BackendBusy", + "CCSDS_OMM_VERSION", + "DISPLAY_SET_KEYS", + "ELEMENT_SETS", + "ElementSet", + "GP_FIELDS", + "MessageMeta", + "OMM_DEFAULT_METADATA", + "OmmRecord", + "ParsedMessage", + "ParsedTLE", + "PropagationError", + "PropagatorBackend", + "StateVector", + "TEMP_NORAD_MAX", + "TEMP_NORAD_MIN", + "as_celestrak_csv", + "as_omm_kvn", + "as_omm_xml", + "as_plaintext", + "as_txt_txt", + "build_oem", + "build_opm", + "coerce_norad_to_int", + "compute_element_set", + "display_payload", + "format_ccsds_datetime", + "from_alpha5", + "get_set", + "indexed_name", + "normalise_norad", + "omm_xml_to_records", + "parse_message", + "parse_omm_epoch", + "parse_tle", + "parsed_states_to_statevectors", + "parsed_to_json", + "temp_tle_norad", + "tle_catalog_field", + "tle_from_gp", + "to_alpha5", +] + +__version__ = "1.0.0" diff --git a/odm/conversions.py b/odm/conversions.py new file mode 100644 index 0000000..76d8659 --- /dev/null +++ b/odm/conversions.py @@ -0,0 +1,261 @@ +"""Convert a propagated :class:`StateVector` (+ the source OMM) into the CCSDS +SANA orbital element sets this service exposes. + +Definitions follow the SANA registry (https://sanaregistry.org/r/orbital_elements): +CARTPV, KEPLERIAN (osculating), KEPLERIANMEAN, KEPLERIANMEANSGP-4, EQUINOCTIAL, +GEODETIC, ADBARV. Every function returns a flat ``{COMPONENT_KEY: float}`` dict +whose keys line up with :data:`odm.element_sets.ELEMENT_SETS`. + +Pure stdlib ``math`` (no numpy). The inertial frame is whatever the backend +emits (TEME for SGP4); a future Orekit backend can supply a rigorously-defined +frame without touching this module. +""" + +from __future__ import annotations + +import math + +from .orbits import MU_EARTH_KM3_S2 as MU +from .orbits import derive as derive_mean + +from .records import StateVector + +TWO_PI = 2.0 * math.pi + + +# --- small vector helpers --------------------------------------------------- + +def _dot(a, b) -> float: + return a[0] * b[0] + a[1] * b[1] + a[2] * b[2] + + +def _cross(a, b) -> tuple[float, float, float]: + return ( + a[1] * b[2] - a[2] * b[1], + a[2] * b[0] - a[0] * b[2], + a[0] * b[1] - a[1] * b[0], + ) + + +def _norm(a) -> float: + return math.sqrt(_dot(a, a)) + + +def _wrap360(deg: float) -> float: + return deg % 360.0 + + +# --- element-set conversions ------------------------------------------------ + +def to_cartpv(sv: StateVector, omm: dict | None = None) -> dict: + """Cartesian position + velocity in the inertial (TEME) frame, km / km·s⁻¹.""" + x, y, z = sv.r_km + xd, yd, zd = sv.v_kms + return {"X": x, "Y": y, "Z": z, "X_DOT": xd, "Y_DOT": yd, "Z_DOT": zd} + + +def to_keplerian(sv: StateVector, omm: dict | None = None) -> dict: + """Osculating Keplerian elements from the inertial state (classic RV→COE).""" + r = sv.r_km + v = sv.v_kms + r_mag = _norm(r) + v_mag = _norm(v) + + h = _cross(r, v) + h_mag = _norm(h) + n = _cross((0.0, 0.0, 1.0), h) + n_mag = _norm(n) + + # eccentricity vector + rv = _dot(r, v) + e_vec = tuple( + ((v_mag * v_mag - MU / r_mag) * r[i] - rv * v[i]) / MU for i in range(3) + ) + ecc = _norm(e_vec) + + energy = v_mag * v_mag / 2.0 - MU / r_mag + sma = -MU / (2.0 * energy) if abs(energy) > 1e-12 else float("inf") + + inc = math.acos(_clamp(h[2] / h_mag, -1.0, 1.0)) if h_mag else 0.0 + + if n_mag > 1e-12: + raan = math.acos(_clamp(n[0] / n_mag, -1.0, 1.0)) + if n[1] < 0.0: + raan = TWO_PI - raan + else: # equatorial: RAAN undefined, take 0 + raan = 0.0 + + if n_mag > 1e-12 and ecc > 1e-12: + argp = math.acos(_clamp(_dot(n, e_vec) / (n_mag * ecc), -1.0, 1.0)) + if e_vec[2] < 0.0: + argp = TWO_PI - argp + else: + argp = 0.0 + + if ecc > 1e-12: + ta = math.acos(_clamp(_dot(e_vec, r) / (ecc * r_mag), -1.0, 1.0)) + if rv < 0.0: + ta = TWO_PI - ta + else: # circular: measure from ascending node (argument of latitude) + if n_mag > 1e-12: + ta = math.acos(_clamp(_dot(n, r) / (n_mag * r_mag), -1.0, 1.0)) + if r[2] < 0.0: + ta = TWO_PI - ta + else: + ta = math.atan2(r[1], r[0]) % TWO_PI + + return { + "SEMI_MAJOR_AXIS": sma, + "ECCENTRICITY": ecc, + "INCLINATION": math.degrees(inc), + "RA_OF_ASC_NODE": _wrap360(math.degrees(raan)), + "ARG_OF_PERICENTER": _wrap360(math.degrees(argp)), + "TRUE_ANOMALY": _wrap360(math.degrees(ta)), + } + + +def to_equinoctial(sv: StateVector, omm: dict | None = None) -> dict: + """Equinoctial elements (a, a_f, a_g, L, χ, ψ, f_r) from osculating Keplerian. + + ``f_r`` is the retrograde factor: +1 for direct orbits, −1 for retrograde + (i > 90°), chosen to keep the χ/ψ pair well-defined near the poles. ``L`` is + the mean longitude. + """ + kep = to_keplerian(sv) + a = kep["SEMI_MAJOR_AXIS"] + ecc = kep["ECCENTRICITY"] + inc = math.radians(kep["INCLINATION"]) + raan = math.radians(kep["RA_OF_ASC_NODE"]) + argp = math.radians(kep["ARG_OF_PERICENTER"]) + ta = math.radians(kep["TRUE_ANOMALY"]) + + f_r = 1.0 if kep["INCLINATION"] <= 90.0 else -1.0 + + a_f = ecc * math.cos(argp + f_r * raan) + a_g = ecc * math.sin(argp + f_r * raan) + + tan_half = math.tan(inc / 2.0) + factor = tan_half if f_r > 0 else (1.0 / tan_half if tan_half else 0.0) + chi = factor * math.sin(raan) + psi = factor * math.cos(raan) + + # true anomaly -> mean anomaly for the mean longitude L + ecc_anom = 2.0 * math.atan2( + math.sqrt(max(1.0 - ecc, 0.0)) * math.sin(ta / 2.0), + math.sqrt(1.0 + ecc) * math.cos(ta / 2.0), + ) + mean_anom = ecc_anom - ecc * math.sin(ecc_anom) + lon = mean_anom + argp + f_r * raan + + return { + "SEMI_MAJOR_AXIS": a, + "A_F": a_f, + "A_G": a_g, + "L": _wrap360(math.degrees(lon)), + "CHI": chi, + "PSI": psi, + "F_R": f_r, + } + + +def to_geodetic(sv: StateVector, omm: dict | None = None) -> dict: + """Earth-relative geodetic set (λ, Φ_GD, β, A, h, v_rel) from the ECEF view.""" + if sv.geodetic is None or sv.ecef_km is None or sv.ecef_v_kms is None: + raise ValueError("backend did not supply an Earth-fixed state for GEODETIC") + lat, lon, alt = sv.geodetic + v_rel = _norm(sv.ecef_v_kms) + east, north, up = _enu(lat, lon, sv.ecef_v_kms) + azimuth = _wrap360(math.degrees(math.atan2(east, north))) + flight_path = math.degrees(math.asin(_clamp(up / v_rel, -1.0, 1.0))) if v_rel else 0.0 + return { + "LON": lon, + "LAT": lat, + "FLIGHT_PATH_ANGLE": flight_path, + "AZIMUTH": azimuth, + "ALTITUDE": alt, + "V_REL": v_rel, + } + + +def to_adbarv(sv: StateVector, omm: dict | None = None) -> dict: + """Inertial spherical set (α, δ, β, A, r, v) from the inertial state.""" + r = sv.r_km + v = sv.v_kms + r_mag = _norm(r) + v_mag = _norm(v) + ra = _wrap360(math.degrees(math.atan2(r[1], r[0]))) + dec = math.degrees(math.asin(_clamp(r[2] / r_mag, -1.0, 1.0))) if r_mag else 0.0 + + # Local frame about the radial direction (geocentric up = r̂). + up = tuple(c / r_mag for c in r) if r_mag else (0.0, 0.0, 1.0) + east = _unit(_cross((0.0, 0.0, 1.0), up)) or (1.0, 0.0, 0.0) + north = _cross(up, east) + v_e, v_n, v_u = _dot(v, east), _dot(v, north), _dot(v, up) + azimuth = _wrap360(math.degrees(math.atan2(v_e, v_n))) + flight_path = math.degrees(math.asin(_clamp(v_u / v_mag, -1.0, 1.0))) if v_mag else 0.0 + + return { + "RA": ra, + "DEC": dec, + "FLIGHT_PATH_ANGLE": flight_path, + "AZIMUTH": azimuth, + "RADIUS": r_mag, + "SPEED": v_mag, + } + + +def to_keplerian_mean(sv: StateVector | None, omm: dict) -> dict: + """Mean Keplerian (a, e, i, Ω, ω, M) from the OMM mean elements. + + ``a`` is derived from the mean motion via Kepler's third law (reusing + :func:`odm.orbits.derive`); the angles come straight from the + stored mean elements. No propagation needed. + """ + info = derive_mean(omm) + if info is None: + raise ValueError("OMM is missing mean elements required for KEPLERIANMEAN") + return { + "SEMI_MAJOR_AXIS": info.semi_major_axis_km, + "ECCENTRICITY": info.eccentricity, + "INCLINATION": info.inclination_deg, + "RA_OF_ASC_NODE": info.raan_deg, + "ARG_OF_PERICENTER": info.arg_perigee_deg, + "MEAN_ANOMALY": info.mean_anomaly_deg, + } + + +def to_keplerian_mean_sgp4(sv: StateVector | None, omm: dict) -> dict: + """KEPLERIANMEANSGP-4: the mean Keplerian set plus the SGP4 B* drag term. + + This mirrors what the stored TLE/OMM already carries; B* is reported as-is + (1/earth-radii) rather than converted to the registry's AGOM/BTERM (m²/kg), + which needs an atmosphere-density assumption we deliberately do not bake in. + """ + out = to_keplerian_mean(sv, omm) + out["BSTAR"] = float(omm.get("BSTAR") or 0.0) + return out + + +# --- internals -------------------------------------------------------------- + +def _clamp(value: float, low: float, high: float) -> float: + return max(low, min(high, value)) + + +def _unit(a): + mag = _norm(a) + if mag < 1e-15: + return None + return (a[0] / mag, a[1] / mag, a[2] / mag) + + +def _enu(lat_deg: float, lon_deg: float, vec) -> tuple[float, float, float]: + """Project an ECEF vector onto the local East/North/Up frame at (lat, lon).""" + lat = math.radians(lat_deg) + lon = math.radians(lon_deg) + sin_lat, cos_lat = math.sin(lat), math.cos(lat) + sin_lon, cos_lon = math.sin(lon), math.cos(lon) + east = (-sin_lon, cos_lon, 0.0) + north = (-sin_lat * cos_lon, -sin_lat * sin_lon, cos_lat) + up = (cos_lat * cos_lon, cos_lat * sin_lon, sin_lat) + return _dot(vec, east), _dot(vec, north), _dot(vec, up) diff --git a/odm/element_sets.py b/odm/element_sets.py new file mode 100644 index 0000000..cb7be74 --- /dev/null +++ b/odm/element_sets.py @@ -0,0 +1,190 @@ +"""Data-driven catalog of the CCSDS SANA element sets this service exposes. + +One :class:`ElementSet` per registry entry is the single source of truth for +both the API serialization (CSV columns) and the on-page conversion tables: +component ``key`` names the machine/CSV column, ``symbol`` + ``unit`` label the +display. :func:`compute_element_set` propagates when a set needs a state vector +and dispatches to :mod:`odm.conversions`. The propagator is always passed in -- +the catalogue has no opinion about which one, and a library that reached for a +service's configured default would be unusable from anywhere else. + +Registry reference: https://sanaregistry.org/r/orbital_elements +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Callable + +from . import conversions +from .propagator import PropagatorBackend +from .records import StateVector + + +@dataclass(frozen=True) +class Component: + key: str # machine / CSV column name + symbol: str # display symbol + unit: str # display unit ("" for dimensionless) + + +@dataclass(frozen=True) +class ElementSet: + key: str # URL / format key, e.g. "keplerian" + name: str # SANA identifier, e.g. "KEPLERIAN" + status: str # "Assigned" / "Provisional" + frame: str # reference frame label + needs_state: bool # requires propagation to a state vector + components: tuple[Component, ...] + compute: Callable[[StateVector | None, dict], dict] + description: str = "" + + +def _c(*items: tuple[str, str, str]) -> tuple[Component, ...]: + return tuple(Component(*i) for i in items) + + +ELEMENT_SETS: dict[str, ElementSet] = { + "cartpv": ElementSet( + key="cartpv", name="CARTPV", status="Assigned", frame="TEME", + needs_state=True, compute=conversions.to_cartpv, + components=_c( + ("X", "X", "km"), ("Y", "Y", "km"), ("Z", "Z", "km"), + ("X_DOT", "Ẋ", "km/s"), ("Y_DOT", "Ẏ", "km/s"), ("Z_DOT", "Ż", "km/s"), + ), + description="Cartesian inertial position and velocity.", + ), + "keplerian": ElementSet( + key="keplerian", name="KEPLERIAN", status="Assigned", frame="TEME", + needs_state=True, compute=conversions.to_keplerian, + components=_c( + ("SEMI_MAJOR_AXIS", "a", "km"), ("ECCENTRICITY", "e", ""), + ("INCLINATION", "i", "deg"), ("RA_OF_ASC_NODE", "Ω", "deg"), + ("ARG_OF_PERICENTER", "ω", "deg"), ("TRUE_ANOMALY", "ν", "deg"), + ), + description="Osculating Keplerian elements from the inertial state.", + ), + "keplerianmean": ElementSet( + key="keplerianmean", name="KEPLERIANMEAN", status="Assigned", frame="TEME", + needs_state=False, compute=conversions.to_keplerian_mean, + components=_c( + ("SEMI_MAJOR_AXIS", "a", "km"), ("ECCENTRICITY", "e", ""), + ("INCLINATION", "i", "deg"), ("RA_OF_ASC_NODE", "Ω", "deg"), + ("ARG_OF_PERICENTER", "ω", "deg"), ("MEAN_ANOMALY", "M", "deg"), + ), + description="Mean Keplerian elements derived from the OMM mean motion.", + ), + "keplerianmeansgp4": ElementSet( + key="keplerianmeansgp4", name="KEPLERIANMEANSGP-4", status="Provisional", + frame="TEME", needs_state=False, compute=conversions.to_keplerian_mean_sgp4, + components=_c( + ("SEMI_MAJOR_AXIS", "a", "km"), ("ECCENTRICITY", "e", ""), + ("INCLINATION", "i", "deg"), ("RA_OF_ASC_NODE", "Ω", "deg"), + ("ARG_OF_PERICENTER", "ω", "deg"), ("MEAN_ANOMALY", "M", "deg"), + ("BSTAR", "B*", "1/ER"), + ), + description="SGP4 mean elements as stored (B* reported natively).", + ), + "equinoctial": ElementSet( + key="equinoctial", name="EQUINOCTIAL", status="Assigned", frame="TEME", + needs_state=True, compute=conversions.to_equinoctial, + components=_c( + ("SEMI_MAJOR_AXIS", "a", "km"), ("A_F", "a_f", ""), ("A_G", "a_g", ""), + ("L", "L", "deg"), ("CHI", "χ", ""), ("PSI", "ψ", ""), ("F_R", "f_r", "±1"), + ), + description="Equinoctial elements (retrograde factor f_r).", + ), + "geodetic": ElementSet( + key="geodetic", name="GEODETIC", status="Assigned", frame="WGS84 ECEF", + needs_state=True, compute=conversions.to_geodetic, + components=_c( + ("LON", "λ", "deg"), ("LAT", "Φ_GD", "deg"), + ("FLIGHT_PATH_ANGLE", "β", "deg"), ("AZIMUTH", "A", "deg"), + ("ALTITUDE", "h", "km"), ("V_REL", "v_rel", "km/s"), + ), + description="Earth-relative geodetic sub-satellite state.", + ), + "adbarv": ElementSet( + key="adbarv", name="ADBARV", status="Assigned", frame="TEME", + needs_state=True, compute=conversions.to_adbarv, + components=_c( + ("RA", "α", "deg"), ("DEC", "δ", "deg"), + ("FLIGHT_PATH_ANGLE", "β", "deg"), ("AZIMUTH", "A", "deg"), + ("RADIUS", "r", "km"), ("SPEED", "v", "km/s"), + ), + description="Inertial spherical position and velocity.", + ), +} + +#: Element sets offered as downloadable API formats (KEPLERIANMEANSGP-4 is +#: already covered by the native tle/omm formats, so it stays display-only). +API_ELEMENT_SET_KEYS: tuple[str, ...] = ( + "cartpv", "keplerian", "keplerianmean", "equinoctial", "geodetic", "adbarv", +) + + +def get_set(key: str) -> ElementSet: + try: + return ELEMENT_SETS[key] + except KeyError as exc: + raise ValueError(f"unknown element set {key!r}") from exc + + +#: Element sets shown on the satellite detail "Conversions" tab (all of them). +DISPLAY_SET_KEYS: tuple[str, ...] = tuple(ELEMENT_SETS) + + +def display_payload(omm: dict, backend: PropagatorBackend, *, + at: datetime | None = None) -> dict: + """Compute every display element set for one OMM in one pass. + + Returns ``{"epoch": iso|None, "sets": [{key, name, status, frame, + description, components: [{key, symbol, unit, value}]}]}`` -- the single + shape used by both the detail-page initial render and the JSON endpoint the + "propagate to" control fetches. Sets that fail to compute are skipped. + """ + epoch = None + sets = [] + for key in DISPLAY_SET_KEYS: + eset = ELEMENT_SETS[key] + try: + result = compute_element_set(key, omm, backend, at=at) + except Exception: # noqa: BLE001 -- a bad set shouldn't blank the page + continue + if epoch is None and result["epoch"] is not None: + epoch = result["epoch"] + values = result["values"] + sets.append({ + "key": eset.key, + "name": eset.name, + "status": eset.status, + "frame": eset.frame, + "description": eset.description, + "components": [ + {"key": c.key, "symbol": c.symbol, "unit": c.unit, + "value": values.get(c.key)} + for c in eset.components + ], + }) + return {"epoch": epoch.isoformat() if epoch else None, "sets": sets} + + +def compute_element_set( + key: str, omm: dict, backend: PropagatorBackend, *, + at: datetime | None = None, +) -> dict: + """Compute one element set's values from an OMM dict. + + Returns ``{"epoch": datetime|None, "frame": str, "values": {COMPONENT: float}}``. + Sets flagged ``needs_state`` are propagated (at ``at`` or the OMM epoch); + mean sets read the OMM directly. + """ + eset = get_set(key) + state = None + epoch = None + if eset.needs_state: + state = backend.state_at(omm, at) + epoch = state.epoch + values = eset.compute(state, omm) + return {"epoch": epoch, "frame": eset.frame, "values": values} diff --git a/odm/gp.py b/odm/gp.py new file mode 100644 index 0000000..6f0183a --- /dev/null +++ b/odm/gp.py @@ -0,0 +1,76 @@ +"""Celestrak's GP CSV rendering. + +Column set and order are Celestrak's, exactly, so the output drops into tooling +written against their ``FORMAT=csv`` queries without a translation step. Values +are formatted through :mod:`odm.omm` rather than restringified here, so a number +reads the same whichever rendering you asked for. +""" + +from __future__ import annotations + +import csv +import io +from datetime import datetime +from typing import Iterable + +from .omm import format_omm_value +from .records import OmmRecord + +GP_FIELDS = ( + "OBJECT_NAME", + "OBJECT_ID", + "EPOCH", + "MEAN_MOTION", + "ECCENTRICITY", + "INCLINATION", + "RA_OF_ASC_NODE", + "ARG_OF_PERICENTER", + "MEAN_ANOMALY", + "EPHEMERIS_TYPE", + "CLASSIFICATION_TYPE", + "NORAD_CAT_ID", + "ELEMENT_SET_NO", + "REV_AT_EPOCH", + "BSTAR", + "MEAN_MOTION_DOT", + "MEAN_MOTION_DDOT", +) + +def as_celestrak_csv(records: Iterable["OmmRecord"]) -> str: + """Render GP CSV in Celestrak's special-request column order. + + Columns are :data:`GP_FIELDS` -- the exact field set and order Celestrak + emits for ``FORMAT=csv`` GP queries -- so the output drops straight into + tooling written against Celestrak's CSV. One row per element, read from the + record's canonical ``omm`` dict (falling back to the record's metadata for + OBJECT_NAME / OBJECT_ID). + """ + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(GP_FIELDS) + for rec in records: + omm = rec.omm or {} + row = [] + for key in GP_FIELDS: + if key == "OBJECT_NAME": + value = omm.get(key) or rec.object_name + elif key == "OBJECT_ID": + value = omm.get(key) or rec.object_id + else: + value = omm.get(key) + row.append(_csv_cell(key, value)) + writer.writerow(row) + return buf.getvalue() + +def _csv_cell(key: str, value) -> str: + """Stringify one GP field for CSV output. + + EPOCH is passed through as-is (already an ISO string); other values reuse + :func:`format_omm_value` so numeric/enum formatting matches the XML/KVN + paths. Missing values render as an empty cell. + """ + if value is None: + return "" + if key == "EPOCH": + return format_omm_value(key, value) if isinstance(value, datetime) else str(value) + return format_omm_value(key, value) diff --git a/odm/messages.py b/odm/messages.py new file mode 100644 index 0000000..bbf87a3 --- /dev/null +++ b/odm/messages.py @@ -0,0 +1,461 @@ +"""CCSDS 502.0 OPM / OEM message build + parse (KVN and XML). + +* **OPM** (Orbit Parameter Message) -- a single-epoch state: Cartesian state + vector + osculating Keplerian elements, generated by propagating an OMM. +* **OEM** (Orbit Ephemeris Message) -- a time-ordered run of Cartesian state + vectors over a span, generated from a backend ephemeris. + +The same functions parse externally-provided messages (for the ingest path) into +a normalized dict. Datetime formatting reuses +:func:`odm.omm.format_ccsds_datetime` so generated messages match +the OMM path. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, replace +from datetime import datetime, timezone +from xml.etree import ElementTree as ET + +from .omm import format_ccsds_datetime, parse_omm_epoch + +from . import conversions +from .propagator import PropagatorBackend +from .records import StateVector + +OPM_VERSION = "3.0" +OEM_VERSION = "3.0" +GM_EARTH = "398600.4418" + +_STATE_KEYS = ("X", "Y", "Z", "X_DOT", "Y_DOT", "Z_DOT") +_KEPLERIAN_KEYS = ( + "SEMI_MAJOR_AXIS", "ECCENTRICITY", "INCLINATION", + "RA_OF_ASC_NODE", "ARG_OF_PERICENTER", "TRUE_ANOMALY", +) + +#: Frames in which osculating Keplerian elements are meaningful (pseudo-inertial). +#: For an Earth-fixed frame (ITRF) we emit the Cartesian state only. +INERTIAL_FRAMES = frozenset({"TEME", "GCRF", "EME2000", "J2000", "TOD", "MOD"}) + + +@dataclass +class MessageMeta: + object_name: str = "" + object_id: str = "" + originator: str = "YKSA" + center_name: str = "EARTH" + ref_frame: str = "TEME" + time_system: str = "UTC" + + +@dataclass +class ParsedMessage: + message_type: str # "OPM" | "OEM" + raw_format: str # "KVN" | "XML" + meta: MessageMeta + epoch: datetime | None = None # OPM state epoch + start_time: datetime | None = None # OEM span + stop_time: datetime | None = None + states: list[dict] = field(default_factory=list) # [{epoch, X, Y, Z, X_DOT...}] + creation_date: datetime | None = None + + +# --- number/datetime helpers ----------------------------------------------- + +def _num(value) -> str: + return repr(float(value)) + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +# --- OPM build -------------------------------------------------------------- + +def build_opm( + omm: dict, + meta: MessageMeta, + at: datetime | None = None, + *, + backend: PropagatorBackend, + fmt: str = "kvn", + ref_frame: str | None = None, +) -> str: + """Propagate ``omm`` to a single-epoch OPM in ``ref_frame`` (default TEME). + + A non-TEME ``ref_frame`` transforms the propagated state via the backend; + Keplerian elements are only emitted for pseudo-inertial frames. + """ + target = (ref_frame or meta.ref_frame or "TEME").upper() + sv = _express(backend.state_at(omm, at), target, backend) + meta = replace(meta, ref_frame=target) + state = conversions.to_cartpv(sv) + kep = conversions.to_keplerian(sv) if target in INERTIAL_FRAMES else None + if fmt == "xml": + return _opm_xml(meta, sv.epoch, state, kep) + return _opm_kvn(meta, sv.epoch, state, kep) + + +def _express(sv: StateVector, target: str, backend: PropagatorBackend) -> StateVector: + """Return ``sv`` re-expressed in ``target`` (a no-op when already there).""" + if not target or target == sv.frame: + return sv + return backend.transform([sv], sv.frame, target)[0] + + +def _opm_kvn(meta, epoch, state, kep) -> str: + lines = [ + f"CCSDS_OPM_VERS = {OPM_VERSION}", + f"CREATION_DATE = {format_ccsds_datetime(_now())}", + f"ORIGINATOR = {meta.originator}", + f"OBJECT_NAME = {meta.object_name}", + f"OBJECT_ID = {meta.object_id}", + f"CENTER_NAME = {meta.center_name}", + f"REF_FRAME = {meta.ref_frame}", + f"TIME_SYSTEM = {meta.time_system}", + f"EPOCH = {format_ccsds_datetime(epoch)}", + ] + lines += [f"{k} = {_num(state[k])}" for k in _STATE_KEYS] + if kep is not None: + lines += [f"{k} = {_num(kep[k])}" for k in _KEPLERIAN_KEYS] + lines.append(f"GM = {GM_EARTH}") + return "\n".join(lines) + "\n" + + +def _opm_xml(meta, epoch, state, kep) -> str: + opm = ET.Element("opm", attrib={"id": "CCSDS_OPM_VERS", "version": OPM_VERSION}) + header = ET.SubElement(opm, "header") + ET.SubElement(header, "CREATION_DATE").text = format_ccsds_datetime(_now()) + ET.SubElement(header, "ORIGINATOR").text = meta.originator + body = ET.SubElement(opm, "body") + segment = ET.SubElement(body, "segment") + _append_metadata(segment, meta) + data = ET.SubElement(segment, "data") + sv_el = ET.SubElement(data, "stateVector") + ET.SubElement(sv_el, "EPOCH").text = format_ccsds_datetime(epoch) + for k in _STATE_KEYS: + ET.SubElement(sv_el, k).text = _num(state[k]) + if kep is not None: + kep_el = ET.SubElement(data, "keplerianElements") + for k in _KEPLERIAN_KEYS: + ET.SubElement(kep_el, k).text = _num(kep[k]) + ET.SubElement(kep_el, "GM").text = GM_EARTH + ET.indent(opm, space=" ") + return ET.tostring(opm, encoding="utf-8", xml_declaration=True).decode("utf-8") + + +# --- OEM build -------------------------------------------------------------- + +def build_oem( + omm: dict, + meta: MessageMeta, + start: datetime, + stop: datetime, + step_s: float, + *, + backend: PropagatorBackend, + fmt: str = "kvn", + ref_frame: str | None = None, +) -> str: + """Propagate ``omm`` to an OEM ephemeris in ``ref_frame`` (default TEME). + + A non-TEME ``ref_frame`` batch-transforms the whole run through the backend + (one round-trip to the Orekit sidecar) before serialising. + """ + target = (ref_frame or meta.ref_frame or "TEME").upper() + states = backend.ephemeris(omm, start, stop, step_s) + if target != "TEME": + states = backend.transform(states, "TEME", target) + meta = replace(meta, ref_frame=target) + rows = [ + {"epoch": sv.epoch, **conversions.to_cartpv(sv)} for sv in states + ] + if fmt == "xml": + return _oem_xml(meta, start, stop, rows) + return _oem_kvn(meta, start, stop, rows) + + +def _oem_kvn(meta, start, stop, rows) -> str: + lines = [ + f"CCSDS_OEM_VERS = {OEM_VERSION}", + f"CREATION_DATE = {format_ccsds_datetime(_now())}", + f"ORIGINATOR = {meta.originator}", + "", + "META_START", + f"OBJECT_NAME = {meta.object_name}", + f"OBJECT_ID = {meta.object_id}", + f"CENTER_NAME = {meta.center_name}", + f"REF_FRAME = {meta.ref_frame}", + f"TIME_SYSTEM = {meta.time_system}", + f"START_TIME = {format_ccsds_datetime(start)}", + f"STOP_TIME = {format_ccsds_datetime(stop)}", + "META_STOP", + "", + ] + for row in rows: + cells = " ".join(_num(row[k]) for k in _STATE_KEYS) + lines.append(f"{format_ccsds_datetime(row['epoch'])} {cells}") + return "\n".join(lines) + "\n" + + +def _oem_xml(meta, start, stop, rows) -> str: + oem = ET.Element("oem", attrib={"id": "CCSDS_OEM_VERS", "version": OEM_VERSION}) + header = ET.SubElement(oem, "header") + ET.SubElement(header, "CREATION_DATE").text = format_ccsds_datetime(_now()) + ET.SubElement(header, "ORIGINATOR").text = meta.originator + body = ET.SubElement(oem, "body") + segment = ET.SubElement(body, "segment") + metadata = _append_metadata(segment, meta) + ET.SubElement(metadata, "START_TIME").text = format_ccsds_datetime(start) + ET.SubElement(metadata, "STOP_TIME").text = format_ccsds_datetime(stop) + data = ET.SubElement(segment, "data") + for row in rows: + sv_el = ET.SubElement(data, "stateVector") + ET.SubElement(sv_el, "EPOCH").text = format_ccsds_datetime(row["epoch"]) + for k in _STATE_KEYS: + ET.SubElement(sv_el, k).text = _num(row[k]) + ET.indent(oem, space=" ") + return ET.tostring(oem, encoding="utf-8", xml_declaration=True).decode("utf-8") + + +def _append_metadata(segment: ET.Element, meta: MessageMeta) -> ET.Element: + metadata = ET.SubElement(segment, "metadata") + ET.SubElement(metadata, "OBJECT_NAME").text = meta.object_name + ET.SubElement(metadata, "OBJECT_ID").text = meta.object_id + ET.SubElement(metadata, "CENTER_NAME").text = meta.center_name + ET.SubElement(metadata, "REF_FRAME").text = meta.ref_frame + ET.SubElement(metadata, "TIME_SYSTEM").text = meta.time_system + return metadata + + +# --- parse (ingest) --------------------------------------------------------- + +def parse_message(text: str) -> ParsedMessage: + """Parse an OPM or OEM message (KVN or XML) into a :class:`ParsedMessage`.""" + stripped = text.lstrip() + if stripped.startswith("<"): + return _parse_xml(stripped) + return _parse_kvn(text) + + +def _parse_kvn(text: str) -> ParsedMessage: + lines = text.splitlines() + kind = "OEM" if any("CCSDS_OEM_VERS" in ln for ln in lines) else "OPM" + meta = MessageMeta() + result = ParsedMessage(message_type=kind, raw_format="KVN", meta=meta) + + in_meta = False + for raw in lines: + line = raw.strip() + if not line or line.startswith("COMMENT"): + continue + if line in ("META_START", "META_STOP"): + in_meta = line == "META_START" + continue + if " = " in line: + key, value = (p.strip() for p in line.split("=", 1)) + _assign_kv(result, key, value) + continue + # A non key/value line inside/after an OEM is an ephemeris row. + if kind == "OEM": + row = _parse_ephemeris_row(line) + if row: + result.states.append(row) + return result + + +def _assign_kv(result: ParsedMessage, key: str, value: str) -> None: + meta = result.meta + if key == "ORIGINATOR": + meta.originator = value + elif key == "OBJECT_NAME": + meta.object_name = value + elif key == "OBJECT_ID": + meta.object_id = value + elif key == "CENTER_NAME": + meta.center_name = value + elif key == "REF_FRAME": + meta.ref_frame = value + elif key == "TIME_SYSTEM": + meta.time_system = value + elif key == "CREATION_DATE": + result.creation_date = parse_omm_epoch(value) + elif key == "EPOCH": + result.epoch = parse_omm_epoch(value) + result.states.append({"epoch": result.epoch}) + elif key == "START_TIME": + result.start_time = parse_omm_epoch(value) + elif key == "STOP_TIME": + result.stop_time = parse_omm_epoch(value) + elif key in _STATE_KEYS and result.states: + result.states[-1][key] = _to_float(value) + + +def _parse_ephemeris_row(line: str) -> dict | None: + parts = line.split() + if len(parts) < 7: + return None + epoch = parse_omm_epoch(parts[0]) + if epoch is None: + return None + row = {"epoch": epoch} + for key, token in zip(_STATE_KEYS, parts[1:7]): + row[key] = _to_float(token) + return row + + +def _parse_xml(text: str) -> ParsedMessage: + root = ET.fromstring(text) + kind = "OEM" if root.tag.lower().endswith("oem") else "OPM" + meta = MessageMeta() + result = ParsedMessage(message_type=kind, raw_format="XML", meta=meta) + + header = root.find("header") + if header is not None: + meta.originator = _text(header, "ORIGINATOR", meta.originator) + result.creation_date = parse_omm_epoch(_text(header, "CREATION_DATE", "")) + + segment = root.find("./body/segment") + if segment is not None: + md = segment.find("metadata") + if md is not None: + meta.object_name = _text(md, "OBJECT_NAME", "") + meta.object_id = _text(md, "OBJECT_ID", "") + meta.center_name = _text(md, "CENTER_NAME", meta.center_name) + meta.ref_frame = _text(md, "REF_FRAME", meta.ref_frame) + meta.time_system = _text(md, "TIME_SYSTEM", meta.time_system) + result.start_time = parse_omm_epoch(_text(md, "START_TIME", "")) + result.stop_time = parse_omm_epoch(_text(md, "STOP_TIME", "")) + data = segment.find("data") + if data is not None: + for sv_el in data.findall("stateVector"): + row = {"epoch": parse_omm_epoch(_text(sv_el, "EPOCH", ""))} + for k in _STATE_KEYS: + row[k] = _to_float(_text(sv_el, k, "")) + result.states.append(row) + if kind == "OPM" and result.states: + result.epoch = result.states[0]["epoch"] + return result + + +def parsed_to_json(parsed: ParsedMessage) -> dict: + """Flatten a :class:`ParsedMessage` into a JSON-serializable dict for storage.""" + def _iso(dt): + return dt.isoformat() if dt else None + + return { + "message_type": parsed.message_type, + "raw_format": parsed.raw_format, + "originator": parsed.meta.originator, + "object_name": parsed.meta.object_name, + "object_id": parsed.meta.object_id, + "center_name": parsed.meta.center_name, + "ref_frame": parsed.meta.ref_frame, + "time_system": parsed.meta.time_system, + "creation_date": _iso(parsed.creation_date), + "epoch": _iso(parsed.epoch), + "start_time": _iso(parsed.start_time), + "stop_time": _iso(parsed.stop_time), + "states": [ + {**{k: v for k, v in state.items() if k != "epoch"}, + "epoch": _iso(state.get("epoch"))} + for state in parsed.states + ], + } + + +# --- re-express / derive from an ingested message --------------------------- + +def parsed_states_to_statevectors(parsed: ParsedMessage) -> list[StateVector]: + """Turn a parsed message's Cartesian rows into frame-tagged StateVectors. + + Rows missing an epoch or any of the six state components are skipped (an OPM + header without its state vector, say). The frame comes from the message's + ``REF_FRAME`` so downstream transforms know where the data starts. + """ + frame = (parsed.meta.ref_frame or "TEME").upper() + out: list[StateVector] = [] + for s in parsed.states: + epoch = s.get("epoch") + if epoch is None or any(s.get(k) is None for k in _STATE_KEYS): + continue + out.append(StateVector( + epoch=epoch, + frame=frame, + r_km=(s["X"], s["Y"], s["Z"]), + v_kms=(s["X_DOT"], s["Y_DOT"], s["Z_DOT"]), + )) + return out + + +def transform_message( + parsed: ParsedMessage, + target_frame: str, + *, + backend: PropagatorBackend, + fmt: str = "kvn", +) -> str: + """Re-express an ingested OPM/OEM in ``target_frame`` and re-serialise it. + + The source frame is the message's own ``REF_FRAME``; the states are pushed + through the backend's frame transform. OPM Keplerian elements are re-derived + only when the target is pseudo-inertial. This is the "source OPM/OEM in any + coordinate system" path -- the tracking service can publish in its native + frame and ODMS serves it in whatever the consumer wants. + """ + target = target_frame.upper() + source = (parsed.meta.ref_frame or "TEME").upper() + svs = parsed_states_to_statevectors(parsed) + if not svs: + raise ValueError("message has no complete Cartesian states to transform") + if target != source: + svs = backend.transform(svs, source, target) + meta = replace(parsed.meta, ref_frame=target) + + if parsed.message_type == "OEM": + rows = [{"epoch": sv.epoch, **conversions.to_cartpv(sv)} for sv in svs] + start = parsed.start_time or svs[0].epoch + stop = parsed.stop_time or svs[-1].epoch + if fmt == "xml": + return _oem_xml(meta, start, stop, rows) + return _oem_kvn(meta, start, stop, rows) + + sv = svs[0] + state = conversions.to_cartpv(sv) + kep = conversions.to_keplerian(sv) if target in INERTIAL_FRAMES else None + if fmt == "xml": + return _opm_xml(meta, sv.epoch, state, kep) + return _opm_kvn(meta, sv.epoch, state, kep) + + +def fit_tle_from_message( + parsed: ParsedMessage, + template_omm: dict, + *, + backend: PropagatorBackend, +) -> dict: + """Fit a TLE to an ingested OEM's states via the backend orbit-fit. + + ``template_omm`` seeds the object identity (NORAD id / designator). Returns + the backend's ``{"line1", "line2", ...}``. Needs a backend with a real fit + (Orekit); the SGP4 backend refuses. + """ + svs = parsed_states_to_statevectors(parsed) + if len(svs) < 2: + raise ValueError("need at least two states to fit a TLE") + frame = (parsed.meta.ref_frame or "TEME").upper() + return backend.fit_tle(svs, frame, template_omm) + + +def _text(el: ET.Element, tag: str, default: str) -> str: + child = el.find(tag) + if child is None or child.text is None: + return default + return child.text.strip() + + +def _to_float(value: str): + try: + return float(value) + except (TypeError, ValueError): + return None diff --git a/odm/norad.py b/odm/norad.py new file mode 100644 index 0000000..57493d2 --- /dev/null +++ b/odm/norad.py @@ -0,0 +1,106 @@ +"""NORAD catalog ID normalisation and TLE catalog-field rendering. + +The canonical ID is stored as a string: legacy numeric IDs are zero-padded to +5 digits so ``"25544"`` stays comparable across sources, while Space-Track's +future alphanumeric IDs (e.g. ``"A1234"``) are preserved verbatim. +Two conventions are supported (see :func:`tle_catalog_field`): + +* **temp** -- the fallback; any ID that does not fit five digits is + mapped into the **90000-99999** pseudo-range via :func:`temp_tle_norad`. +* **ALPHA-5** -- Space-Track's scheme for catalog numbers 100000-339999, where + the leading two digits are replaced by a letter. It is numeric-only and + *not* backward compatible with five-digit parsers, so it is an opt-in + rendering, never the default. +""" + +from __future__ import annotations + +import hashlib + +# Reserved pseudo-range for synthetic numeric NORAD substitutes. +TEMP_NORAD_MIN = 90000 +TEMP_NORAD_MAX = 99999 +_TEMP_NORAD_SPAN = TEMP_NORAD_MAX - TEMP_NORAD_MIN + 1 + +# ALPHA-5: letters encode the leading "10".."33"; I and O are skipped to avoid +# confusion with 1 and 0. +ALPHA5_LETTERS = "ABCDEFGHJKLMNPQRSTUVWXYZ" +ALPHA5_MIN = 100000 +ALPHA5_MAX = 339999 + + +def normalise_norad(value: object) -> str | None: + """Return the canonical string form of a NORAD catalog ID.""" + if value is None: + return None + if isinstance(value, bytes): + value = value.decode("ascii", errors="replace") + s = str(value).strip() + if not s: + return None + if s.isdigit(): + return s.zfill(5) + return s + + +def temp_tle_norad(value: object) -> int: + """Return a deterministic 5-digit pseudo-NORAD in [90000, 99999].""" + raw = "" if value is None else str(value).strip() + digest = hashlib.blake2b(raw.encode("utf-8"), digest_size=4).digest() + n = int.from_bytes(digest, "big") + return TEMP_NORAD_MIN + (n % _TEMP_NORAD_SPAN) + + +def _as_int(value: object) -> int | None: + """Parse a NORAD value to a plain int, or None if it is not numeric.""" + if value is None: + return None + if isinstance(value, bool): + return None + if isinstance(value, int): + return value + s = str(value).strip() + return int(s) if s.isdigit() else None + + +def to_alpha5(value: int) -> str: + """Encode a numeric catalog ID in [100000, 339999] as a 5-char ALPHA-5 field.""" + if not ALPHA5_MIN <= value <= ALPHA5_MAX: + raise ValueError(f"{value} outside ALPHA-5 range [{ALPHA5_MIN}, {ALPHA5_MAX}]") + return f"{ALPHA5_LETTERS[value // 10000 - 10]}{value % 10000:04d}" + + +def from_alpha5(field: str) -> int | None: + """Decode a 5-char ALPHA-5 field back to its integer catalog number. + + Returns None when ``field`` is not a valid ALPHA-5 token. + """ + field = field.strip() + if len(field) != 5 or not field[1:].isdigit(): + return None + idx = ALPHA5_LETTERS.find(field[0].upper()) + if idx < 0: + return None + return (idx + 10) * 10000 + int(field[1:]) + + +def tle_catalog_field(value: object, *, alpha5: bool = False) -> str: + """Return the 5-character catalog field for a TLE line. + + Numeric IDs <= 99999 render as zero-padded digits. With ``alpha5=True``, + numeric IDs in [100000, 339999] use the ALPHA-5 letter prefix. Anything + that cannot fit five characters (alphanumeric future IDs, out-of-range + numbers) falls back to a deterministic 9xxxx pseudo-ID. + """ + n = _as_int(value) + if n is not None and 0 <= n <= 99999: + return f"{n:05d}" + if alpha5 and n is not None and ALPHA5_MIN <= n <= ALPHA5_MAX: + return to_alpha5(n) + return f"{temp_tle_norad(value):05d}" + + +def coerce_norad_to_int(value: object) -> int: + """Best-effort int form for TLE rendering (legacy temp-ID scheme).""" + n = _as_int(value) + return n if n is not None else temp_tle_norad(value) diff --git a/odm/omm.py b/odm/omm.py new file mode 100644 index 0000000..92c74c6 --- /dev/null +++ b/odm/omm.py @@ -0,0 +1,411 @@ +"""CCSDS 502.0-B-3 OMM, in both XML and Keyword-Value Notation. + +The authoritative rendering of an element set: full precision, real catalog IDs, +and the metadata (originator, theory, frame, time system) that says what the +numbers mean. The XML path handles both the single-document form and the +```` multi-document wrapper Space-Track ships archives in. + +XML and KVN deliberately share their key groups and value formatting, so the two +cannot drift into disagreeing about the same record. + +Reference: CCSDS 502.0-B-3 -- Orbit Data Messages (Blue Book). +""" + +from __future__ import annotations + +import re +from datetime import datetime, timezone +from typing import Iterable, Iterator +from xml.etree import ElementTree as ET + +from .records import OmmRecord + +CCSDS_OMM_VERSION = "3.0" + +CCSDS_NDM_SCHEMA = ( + "https://sanaregistry.org/r/ndmxml_unqualified/" "ndmxml-3.0.0-master-3.0.xsd" +) + +XSI_NS = "http://www.w3.org/2001/XMLSchema-instance" + +OMM_DEFAULT_METADATA = { + "CENTER_NAME": "EARTH", + "REF_FRAME": "TEME", + "TIME_SYSTEM": "UTC", + "MEAN_ELEMENT_THEORY": "SGP4", +} + +# CCSDS-mandatory fields that go into for every OMM, regardless +# of theory. SEMI_MAJOR_AXIS is the spec's preferred form; we emit MEAN_MOTION +# because SGP4-derived data carries it natively and Celestrak's GP JSON uses +# the same. +_OMM_MEAN_ELEMENT_KEYS = ( + "EPOCH", + "MEAN_MOTION", + "ECCENTRICITY", + "INCLINATION", + "RA_OF_ASC_NODE", + "ARG_OF_PERICENTER", + "MEAN_ANOMALY", +) + +# Emitted as when MEAN_ELEMENT_THEORY is an SGP* variant. +_OMM_TLE_PARAM_KEYS = ( + "EPHEMERIS_TYPE", + "CLASSIFICATION_TYPE", + "NORAD_CAT_ID", + "ELEMENT_SET_NO", + "REV_AT_EPOCH", + "BSTAR", + "MEAN_MOTION_DOT", + "MEAN_MOTION_DDOT", +) + +def as_omm_xml( + records: Iterable["OmmRecord"], + *, + originator: str = "YKSA", + creation_date: datetime | None = None, +) -> str: + """Render CCSDS 502.0-B-3 OMM XML wrapped in an ```` container.""" + creation = creation_date or datetime.now(timezone.utc) + ndm = ET.Element( + "ndm", + attrib={ + f"{{{XSI_NS}}}noNamespaceSchemaLocation": CCSDS_NDM_SCHEMA, + }, + ) + # ElementTree only registers the xmlns:xsi prefix if we pre-register it. + ET.register_namespace("xsi", XSI_NS) + + materialised = list(records) + for rec in materialised: + if rec.omm_xml: + try: + ndm.append(_parse_omm_chunk(rec.omm_xml)) + continue + except ET.ParseError: + pass # fall through to synthesis + ndm.append(_synthesise_omm_document(rec, originator, creation)) + + if len(materialised) == 1 and not materialised[0].omm_xml: + only = ndm[0] + ET.indent(only, space=" ", level=0) + return ET.tostring(only, encoding="utf-8", xml_declaration=True).decode("utf-8") + + ET.indent(ndm, space=" ", level=0) + return ET.tostring(ndm, encoding="utf-8", xml_declaration=True).decode("utf-8") + +def omm_xml_to_records(payload: str | bytes) -> Iterator["OmmRecord"]: + """Parse CCSDS OMM XML and yield one record per ```` element. + + Stray ``xsi:`` references without an ``xmlns:xsi`` declaration (as in + Space-Track samples) are auto-repaired before parsing. + """ + text = _xml_payload_to_str(payload) + root = _parse_omm_root(text) + + if root.tag == "ndm": + omm_iter = list(root.findall("omm")) + elif root.tag == "omm": + omm_iter = [root] + else: + # Unwrap one level of unknown container -- some vendors emit . + omm_iter = list(root.findall("omm")) + if not omm_iter: + raise ValueError(f"expected or root, got <{root.tag}>") + + for omm_el in omm_iter: + # Header is per-OMM in the multi-document case; share it across + # whatever segments the OMM contains (usually exactly one). + header_ctx = _read_omm_header(omm_el) + body = omm_el.find("body") + if body is None: + continue + segments = list(body.findall("segment")) + if not segments: + continue + for segment in segments: + rec = _segment_to_record(segment, header_ctx) + if len(segments) == 1: + rec.omm_xml = _serialise_omm_element(omm_el) + yield rec + +def as_omm_kvn( + records: Iterable["OmmRecord"], + *, + originator: str = "YKSA", + creation_date: datetime | None = None, +) -> str: + """Render CCSDS 502.0-B-3 OMM in Keyword-Value Notation (KVN). + + Flat ``KEYWORD = value`` layout (as Space-Track's ``/format/kvn`` emits), + reusing the same key groups and value formatting as the XML path so the two + representations stay in lock-step. Multiple records are separated by a blank + line; each begins with its own ``CCSDS_OMM_VERS`` line. + """ + creation = creation_date or datetime.now(timezone.utc) + blocks: list[str] = [] + for rec in records: + lines = [f"CCSDS_OMM_VERS = {CCSDS_OMM_VERSION}"] + hdr_date = rec.creation_date or creation + lines.append(f"CREATION_DATE = {format_ccsds_datetime(hdr_date)}") + lines.append(f"ORIGINATOR = {rec.originator or originator}") + + lines.append(f"OBJECT_NAME = {rec.object_name or rec.omm.get('OBJECT_NAME', '')}") + lines.append(f"OBJECT_ID = {rec.object_id or rec.omm.get('OBJECT_ID', '')}") + lines.append(f"CENTER_NAME = {rec.center_name}") + lines.append(f"REF_FRAME = {rec.ref_frame}") + lines.append(f"TIME_SYSTEM = {rec.time_system}") + lines.append(f"MEAN_ELEMENT_THEORY = {rec.mean_element_theory}") + + for key in _OMM_MEAN_ELEMENT_KEYS: + value = rec.omm.get(key) + if value is None: + continue + lines.append(f"{key} = {format_omm_value(key, value)}") + + if rec.is_sgp_variant: + for key in _OMM_TLE_PARAM_KEYS: + value = rec.omm.get(key) + if value is None: + continue + lines.append(f"{key} = {format_omm_value(key, value)}") + + blocks.append("\n".join(lines)) + return "\n\n".join(blocks) + "\n" + +def _synthesise_omm_document( + rec: OmmRecord, + originator: str, + creation_date: datetime, +) -> ET.Element: + """Build one full ```` element from an :class:`OmmRecord`. + + Used when the record has no upstream XML to splice in (i.e. it came from + a TLE / JSON / internal source). + """ + omm = ET.Element( + "omm", attrib={"id": "CCSDS_OMM_VERS", "version": CCSDS_OMM_VERSION} + ) + + header = ET.SubElement(omm, "header") + hdr_date = rec.creation_date or creation_date + ET.SubElement(header, "CREATION_DATE").text = format_ccsds_datetime(hdr_date) + ET.SubElement(header, "ORIGINATOR").text = rec.originator or originator + + body = ET.SubElement(omm, "body") + body.append(_omm_segment(rec)) + return omm + +def _omm_segment(rec: OmmRecord) -> ET.Element: + segment = ET.Element("segment") + metadata = ET.SubElement(segment, "metadata") + ET.SubElement(metadata, "OBJECT_NAME").text = rec.object_name or rec.omm.get( + "OBJECT_NAME", "" + ) + ET.SubElement(metadata, "OBJECT_ID").text = rec.object_id or rec.omm.get( + "OBJECT_ID", "" + ) + ET.SubElement(metadata, "CENTER_NAME").text = rec.center_name + ET.SubElement(metadata, "REF_FRAME").text = rec.ref_frame + ET.SubElement(metadata, "TIME_SYSTEM").text = rec.time_system + ET.SubElement(metadata, "MEAN_ELEMENT_THEORY").text = rec.mean_element_theory + + data = ET.SubElement(segment, "data") + mean = ET.SubElement(data, "meanElements") + for key in _OMM_MEAN_ELEMENT_KEYS: + value = rec.omm.get(key) + if value is None: + continue + ET.SubElement(mean, key).text = format_omm_value(key, value) + + if rec.is_sgp_variant and any( + rec.omm.get(k) is not None for k in _OMM_TLE_PARAM_KEYS + ): + tle_params = ET.SubElement(data, "tleParameters") + for key in _OMM_TLE_PARAM_KEYS: + value = rec.omm.get(key) + if value is None: + continue + ET.SubElement(tle_params, key).text = format_omm_value(key, value) + + return segment + +def _segment_to_record( + segment: ET.Element, + header_ctx: dict, +) -> OmmRecord: + md = segment.find("metadata") + md_text = ( + {child.tag: (child.text or "").strip() for child in md} + if md is not None + else {} + ) + + omm_dict: dict = {} + data = segment.find("data") + for block in () if data is None else ("meanElements", "tleParameters"): + node = data.find(block) + if node is None: + continue + for child in node: + omm_dict[child.tag] = coerce_omm_value( + child.tag, (child.text or "").strip() + ) + # are ignored + + omm_dict.setdefault("OBJECT_NAME", md_text.get("OBJECT_NAME", "")) + omm_dict.setdefault("OBJECT_ID", md_text.get("OBJECT_ID", "")) + + return OmmRecord( + omm=omm_dict, + object_name=md_text.get("OBJECT_NAME", ""), + object_id=md_text.get("OBJECT_ID", ""), + originator=header_ctx.get("ORIGINATOR", "UNKNOWN"), + center_name=md_text.get("CENTER_NAME", OMM_DEFAULT_METADATA["CENTER_NAME"]), + ref_frame=md_text.get("REF_FRAME", OMM_DEFAULT_METADATA["REF_FRAME"]), + time_system=md_text.get("TIME_SYSTEM", OMM_DEFAULT_METADATA["TIME_SYSTEM"]), + mean_element_theory=md_text.get( + "MEAN_ELEMENT_THEORY", + OMM_DEFAULT_METADATA["MEAN_ELEMENT_THEORY"], + ), + creation_date=header_ctx.get("CREATION_DATE"), + ) + +def _read_omm_header(omm_el: ET.Element) -> dict: + """Extract ORIGINATOR / CREATION_DATE / COMMENT from an OMM header.""" + header = omm_el.find("header") + if header is None: + return {} + out: dict = {} + for child in header: + text = (child.text or "").strip() + if not text: + continue + if child.tag == "CREATION_DATE": + out["CREATION_DATE"] = _parse_ccsds_datetime(text) + else: + out[child.tag] = text + return out + +def _parse_ccsds_datetime(text: str) -> datetime: + """Lenient parser -- Space-Track ships microsecond precision, no offset.""" + t = text.strip() + if t.endswith("Z"): + t = t[:-1] + try: + dt = datetime.fromisoformat(t) + except ValueError: + return datetime.now(timezone.utc) + return ( + dt.replace(tzinfo=timezone.utc) + if dt.tzinfo is None + else dt.astimezone(timezone.utc) + ) + +def parse_omm_epoch(value: object) -> datetime | None: + """Parse an ``OMM["EPOCH"]`` value into an aware UTC datetime. + + Accepts ``datetime`` instances, ISO-8601 strings (with or without Z), and + naive timestamps (assumed UTC). Returns ``None`` on anything that can't be parsed. + """ + if value is None: + return None + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + text = str(value).strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + if "+" not in text and text.count("-") <= 2: + text = text + "+00:00" + try: + dt = datetime.fromisoformat(text) + except ValueError: + return None + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + +# Real-world Space-Track archive XML uses xsi: prefix without declaring its +# namespace. We auto-repair before parsing rather than fail-hard. +_NS_DECL_INJECTION = re.compile( + r"(<(?:ndm|omm)\b)([^>]*xsi:[^>]*>)", + flags=re.IGNORECASE, +) + +def _xml_payload_to_str(payload: str | bytes) -> str: + if isinstance(payload, bytes): + return payload.decode("utf-8", errors="replace") + return str(payload) + +def _parse_omm_root(text: str) -> ET.Element: + """Parse a Space-Track-style payload, repairing missing xsi declarations.""" + candidates = [text] + if "xmlns:xsi" not in text and "xsi:" in text: + candidates.append(_inject_xsi_namespace(text)) + last_exc: ET.ParseError | None = None + for variant in candidates: + try: + return ET.fromstring(variant) + except ET.ParseError as exc: + last_exc = exc + assert last_exc is not None # pragma: no cover + raise last_exc + +def _inject_xsi_namespace(text: str) -> str: + """Add ``xmlns:xsi=...`` to the first ```` / ```` tag that uses xsi:.""" + + def _patch(match: re.Match) -> str: + return f'{match.group(1)} xmlns:xsi="{XSI_NS}"{match.group(2)}' + + return _NS_DECL_INJECTION.sub(_patch, text, count=1) + +def _parse_omm_chunk(chunk: str) -> ET.Element: + """Parse a single stored ``...`` element back into a tree.""" + return _parse_omm_root(chunk) + +def _serialise_omm_element(omm_el: ET.Element) -> str: + """Render a single ```` element back to its byte form for storage.""" + return str(ET.tostring(omm_el, encoding="unicode")) + +def format_omm_value(key: str, value) -> str: + if key == "EPOCH": + if isinstance(value, datetime): + return format_ccsds_datetime(value) + return str(value) + if key == "NORAD_CAT_ID": + return str(value) + if key in ("EPHEMERIS_TYPE", "ELEMENT_SET_NO", "REV_AT_EPOCH"): + return str(int(value)) + if key == "CLASSIFICATION_TYPE": + return str(value)[:1] or "U" + if isinstance(value, float): + # Default float repr; OMM is forgiving about precision. + return repr(value) + return str(value) + +def coerce_omm_value(key: str, text: str): + if not text: + return None + if key == "EPOCH": + return text + if key == "NORAD_CAT_ID": + return text + if key in ("EPHEMERIS_TYPE", "ELEMENT_SET_NO", "REV_AT_EPOCH"): + try: + return int(text) + except ValueError: + return None + if key == "CLASSIFICATION_TYPE": + return text[:1] + try: + return float(text) + except ValueError: + return text + +def format_ccsds_datetime(dt: datetime) -> str: + """ISO 8601 without offset suffix, milliseconds, UTC (per CCSDS convention).""" + if dt.tzinfo is not None: + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) + return dt.isoformat(timespec="milliseconds") diff --git a/odm/orbits.py b/odm/orbits.py new file mode 100644 index 0000000..55e62bd --- /dev/null +++ b/odm/orbits.py @@ -0,0 +1,81 @@ +"""Derive human-readable orbital quantities from an OMM dict. + +Implements the standard SGP4-era conventions: + +* mu_earth (GM) = 398 600.4418 km^3/s^2 +* Earth's mean radius = 6 378.135 km +* Mean motion in rev/day -> orbital period in minutes +* Semi-major axis from Kepler's third law on the mean motion +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass + + +MU_EARTH_KM3_S2 = 398_600.4418 +R_EARTH_KM = 6_378.135 + +@dataclass(frozen=True) +class OrbitalInfo: + """Derived human-readable orbital quantities.""" + + mean_motion_rev_per_day: float + period_minutes: float + semi_major_axis_km: float + eccentricity: float + inclination_deg: float + raan_deg: float + arg_perigee_deg: float + mean_anomaly_deg: float + perigee_alt_km: float + apogee_alt_km: float + + @property + def period_hms(self) -> str: + """Return the orbital period as ``HH:MM:SS``.""" + total_seconds = int(round(self.period_minutes * 60)) + h, rem = divmod(total_seconds, 3600) + m, s = divmod(rem, 60) + return f"{h:02d}:{m:02d}:{s:02d}" + + +def derive(omm: dict) -> OrbitalInfo | None: + """Build :class:`OrbitalInfo` from an OMM mean-element dict. + + Returns ``None`` when the dict is missing one of the fields required to + derive both SMA and perigee/apogee. + """ + try: + mean_motion = float(omm["MEAN_MOTION"]) + ecc = float(omm["ECCENTRICITY"]) + inc = float(omm["INCLINATION"]) + raan = float(omm["RA_OF_ASC_NODE"]) + argp = float(omm["ARG_OF_PERICENTER"]) + ma = float(omm["MEAN_ANOMALY"]) + except (KeyError, TypeError, ValueError): + return None + + if mean_motion <= 0: + return None + + n_rad_per_sec = 2.0 * math.pi * mean_motion / 86_400.0 + semi_major_axis_km = (MU_EARTH_KM3_S2 / (n_rad_per_sec ** 2)) ** (1.0 / 3.0) + + period_minutes = 1440.0 / mean_motion + perigee_alt_km = semi_major_axis_km * (1.0 - ecc) - R_EARTH_KM + apogee_alt_km = semi_major_axis_km * (1.0 + ecc) - R_EARTH_KM + + return OrbitalInfo( + mean_motion_rev_per_day=mean_motion, + period_minutes=period_minutes, + semi_major_axis_km=semi_major_axis_km, + eccentricity=ecc, + inclination_deg=inc, + raan_deg=raan, + arg_perigee_deg=argp, + mean_anomaly_deg=ma, + perigee_alt_km=perigee_alt_km, + apogee_alt_km=apogee_alt_km, + ) diff --git a/odm/propagator.py b/odm/propagator.py new file mode 100644 index 0000000..f04e13c --- /dev/null +++ b/odm/propagator.py @@ -0,0 +1,158 @@ +"""The propagator seam: what a propagation backend must be able to do. + +An interface, not an implementation. Everything in this package that needs a +state vector -- the element-set conversions, the OPM/OEM builders -- takes one of +these rather than reaching for a particular propagator, which is what lets the +same code run against the pure-Python SGP4 path and against a rigorous +flight-dynamics engine without knowing which it got. + +Implementations live in the service that owns the deployment (see the +``yksa_orbital`` Django app). Nothing here does any orbital mechanics. +""" + +from __future__ import annotations + +import abc +from datetime import datetime + +from .records import StateVector + + +class PropagationError(Exception): + """Raised when a backend cannot produce a state for the requested time.""" + + +class BackendBusy(PropagationError): + """The backend refused the work because it is at capacity, not because the + request was wrong. + + A subclass so existing ``except PropagationError`` handlers keep working, + but callers that can wait -- the Celery decay tasks -- should catch this + first and retry rather than recording a failed forecast. Nothing about the + request needs to change for it to succeed later. + """ + +class PropagatorBackend(abc.ABC): + """Turns a canonical OMM mean-element dict into state vectors.""" + + #: Short backend identifier stored on generated artifacts (e.g. ``"sgp4"``). + name: str = "abstract" + + #: Reference frames this backend can express a Cartesian state in, offered + #: by the detail page's coordinate-system dropdown. The pure-Python backend + #: only knows TEME + a GMST Earth-fixed frame; Orekit adds rigorous frames. + display_frames: tuple[str, ...] = ("TEME",) + + @abc.abstractmethod + def state_at(self, omm: dict, at: datetime | None = None) -> StateVector: + """Return the state at ``at`` (aware UTC), or at the OMM epoch if ``None``.""" + + @abc.abstractmethod + def ephemeris( + self, omm: dict, start: datetime, stop: datetime, step_s: float, + ) -> list[StateVector]: + """Return states from ``start`` to ``stop`` inclusive, every ``step_s`` seconds.""" + + def transform( + self, states: list[StateVector], frame_in: str, frame_out: str, + ) -> list[StateVector]: + """Re-express Cartesian ``states`` from ``frame_in`` to ``frame_out``. + + Returns new :class:`StateVector` instances whose ``r_km`` / ``v_kms`` + are in ``frame_out`` (and whose ``frame`` label is ``frame_out``). The + default handles the identity case only; backends override to add real + frame math (SGP4: TEME↔ITRF via GMST; Orekit: the full rigorous set). + """ + if frame_in == frame_out: + return list(states) + raise PropagationError( + f"{self.name} backend cannot transform {frame_in!r}->{frame_out!r}" + ) + + def fit_tle( + self, states: list[StateVector], frame: str, template_omm: dict, + ) -> dict: + """Fit a TLE to a run of Cartesian states (an OEM ephemeris). + + ``template_omm`` seeds the fit with the object's identity (NORAD id, + designator). Returns ``{"line1", "line2", ...}``. Only backends with a + real orbit-fit (Orekit) implement this; the default refuses. + """ + raise PropagationError(f"{self.name} backend cannot fit TLEs from ephemerides") + + def decay(self, omm: dict, spacecraft: dict, **options) -> dict: + """Propagate to re-entry under drag and return the decay trajectory. + + ``spacecraft`` carries the drag/SRP properties resolved by + :mod:`yksa_tle.predictions.ballistic` (a ballistic coefficient, or mass + + areas + coefficients). ``options`` are forwarded to the engine + (``decay_altitude_km``, ``strengths``, ``max_years``, ...). + + Returns ``{"runs": [{"strength", "decayed", "decay_epoch", "decay_days", + "points": [...]}, ...], ...}``. Only backends with a numerical + propagator and an atmosphere model (Orekit) can answer this; SGP4's + analytical theory has no drag integration to run, so the default + refuses rather than returning a number nobody should trust. + """ + raise PropagationError( + f"{self.name} backend cannot predict decay; use the orekit backend" + ) + + def fit_drag(self, elements: list[dict], **options) -> dict: + """Fit a ballistic coefficient from an observed element history. + + ``elements`` is a chronological list of stored OMM dicts. Returns + ``{"ballistic_coefficient", "adot_m_per_s", "r_squared", ...}``. + + Needs an atmosphere model to convert the observed decay rate into a + coefficient, so like :meth:`decay` this is Orekit-only. + """ + raise PropagationError( + f"{self.name} backend cannot fit drag; use the orekit backend" + ) + + def space_weather( + self, start: datetime, stop: datetime, **options, + ) -> dict: + """Observed daily F10.7 / Ap between two dates. + + Returns ``{"times": [...], "f107": [...], "ap": [...]}``. Reads a space- + weather provider, which lives in the sidecar, so like :meth:`fit_drag` + this is Orekit-only. + """ + raise PropagationError( + f"{self.name} backend cannot read space weather; use the orekit backend" + ) + + def state_in_frames( + self, omm: dict, at: datetime | None = None, frames=None, + ) -> dict: + """Return the state expressed in one or more reference frames. + + Shape:: + + {"epoch", "element_epoch", "frames": [...], + "states": {frame: {"r_km": [...], "v_kms": [...]}}, "geodetic": [...]} + + The default implementation derives everything from :meth:`state_at` + (TEME, plus an Earth-fixed ``ITRF`` view when the backend fills ECEF). + Backends with real frame transforms (Orekit) override this. + """ + sv = self.state_at(omm, at) + available: dict[str, dict] = { + "TEME": {"r_km": list(sv.r_km), "v_kms": list(sv.v_kms)}, + } + if sv.ecef_km is not None: + available["ITRF"] = { + "r_km": list(sv.ecef_km), + "v_kms": list(sv.ecef_v_kms) if sv.ecef_v_kms is not None else None, + } + requested = list(frames) if frames else list(self.display_frames) + states = {f: available[f] for f in requested if f in available} + return { + "epoch": sv.epoch.isoformat() if sv.epoch else None, + "element_epoch": sv.element_epoch.isoformat() if sv.element_epoch else None, + "frames": list(states), + "states": states, + "geodetic": list(sv.geodetic) if sv.geodetic else None, + } diff --git a/odm/records.py b/odm/records.py new file mode 100644 index 0000000..f04e0f7 --- /dev/null +++ b/odm/records.py @@ -0,0 +1,92 @@ +"""The in-memory representations every format reads into and writes out of. + +Nothing here knows about a file, a wire protocol or a database row. A reader +produces one of these; a writer consumes one. That is the whole point of the +package: the OMM is the record, and TLE text, OMM XML, KVN and GP CSV are four +renderings of it -- so a new rendering is one writer, not another parallel idea +of what an element set is. + +Deliberately not the service's storage model. ODMS holds these rows in Django, +`track` holds them somewhere else, and the sidecar holds none at all; the format +code has to work for all three, so it works on records and the service adapts. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime + + +@dataclass +class OmmRecord: + """One element set, with the metadata that says what its numbers mean. + + ``omm`` is the canonical mean-element dict under CCSDS field names + (``EPOCH``, ``MEAN_MOTION``, ``ECCENTRICITY``, ... plus the SGP-only + ``tleParameters`` fields). Everything else is provenance: who produced it, + under which theory, in which frame and time system -- which is why an OMM + can be read back exactly and a TLE cannot. + """ + + omm: dict + object_name: str = "" + object_id: str = "" + originator: str = "UNKNOWN" + center_name: str = "EARTH" + ref_frame: str = "TEME" + time_system: str = "UTC" + mean_element_theory: str = "SGP4" + creation_date: datetime | None = None + + omm_xml: str = "" + """Original ``...`` chunk from upstream, if available. + + Kept so a record ingested as XML is republished byte-identically rather than + re-synthesised: an upstream document is evidence, and re-rendering it loses + whatever the producer said that we do not model. + """ + + line1: str = "" + line2: str = "" + """The TLE lines as issued, when the record came with them. + + A rendering, not data -- but the *issued* one, so it is preserved rather + than regenerated. Regeneration is only correct when it produces exactly + these, and for alphanumeric catalog IDs it cannot (see :mod:`odm.norad`). + """ + + @property + def is_sgp_variant(self) -> bool: + return self.mean_element_theory.upper().startswith(("SGP", "SDP")) + + +@dataclass(frozen=True) +class ParsedTLE: + """In-memory representation produced by :func:`odm.tle.parse_tle`.""" + + name: str + line1: str + line2: str + omm: dict + + +@dataclass(frozen=True) +class StateVector: + """A single propagated state, frame-agnostic at the seam. + + ``r_km`` / ``v_kms`` are the inertial (TEME for SGP4) position and velocity + and are always populated. ``ecef_km`` / ``ecef_v_kms`` and ``geodetic`` + (lat_deg, lon_deg, alt_km) are Earth-fixed convenience views a backend may + fill; conversions that need them should tolerate ``None`` from backends that + do not. + """ + + epoch: datetime + frame: str + r_km: tuple[float, float, float] + v_kms: tuple[float, float, float] + ecef_km: tuple[float, float, float] | None = None + ecef_v_kms: tuple[float, float, float] | None = None + geodetic: tuple[float, float, float] | None = None + element_epoch: datetime | None = None + warnings: list[str] = field(default_factory=list) diff --git a/odm/registry.py b/odm/registry.py new file mode 100644 index 0000000..67793a5 --- /dev/null +++ b/odm/registry.py @@ -0,0 +1,224 @@ +"""Every output format in one table, keyed by the name the API uses. + +This is the answer to "what formats does ODMS emit?" -- a question that used to +need reading a Django renderer module, a schema Literal, an element-set catalogue +and a controller. A format is a key, the HTTP facts about it (content type, +filename, encoding), and a writer over :class:`~odms.records.OmmRecord`. Adding +one is a `Format` entry; nothing else has to learn about it. + +Writers all take the same arguments so a caller never has to know which kind it +got:: + + write(records, *, backend=None, at=None) -> str + +``backend`` and ``at`` matter only to the element-set formats, which propagate +before converting. The rest ignore them. + +``json`` is registered but has no writer: it is the API's own serialisation of +its storage rows, which is a service concern rather than a format. It appears +here so the format list is complete and validation has one source. +""" + +from __future__ import annotations + +import csv +import io +from dataclasses import dataclass +from datetime import datetime +from typing import Callable, Iterable + +from . import element_sets +from .gp import as_celestrak_csv +from .omm import as_omm_kvn, as_omm_xml, parse_omm_epoch +from .records import OmmRecord +from .tle import as_plaintext, indexed_name, tle_from_gp + +DEFAULT_ORIGINATOR = "YKSA" + + +@dataclass(frozen=True) +class Format: + """One addressable output format.""" + + key: str + content_type: str + filename: str + encoding: str = "utf-8" + #: ``None`` for formats the service serialises itself (``json``). + write: Callable[..., str] | None = None + #: True when the writer propagates, and so needs a ``backend``. + needs_backend: bool = False + + +# --- TLE text --------------------------------------------------------------- + + +def labelled_rows(records: Iterable[OmmRecord], *, + alpha5: bool = False) -> list[tuple[str, str, str]]: + """``(name, line1, line2)`` per record, with duplicates numbered. + + A range query returns many elements for one object, and three-line text has + nowhere to put an epoch -- so identical name lines would make the records + indistinguishable in a file. Repeats get a ``[NN]`` suffix; a single record + per object is left alone. + """ + records = list(records) + totals: dict[str, int] = {} + for record in records: + key = _identity(record) + totals[key] = totals.get(key, 0) + 1 + + seen: dict[str, int] = {} + rows: list[tuple[str, str, str]] = [] + for record in records: + key = _identity(record) + seen[key] = seen.get(key, 0) + 1 + total = totals[key] + name = indexed_name( + record.object_name, index=seen[key] if total > 1 else None, total=total, + ) + line1, line2 = tle_lines(record, alpha5=alpha5) + rows.append((name, line1, line2)) + return rows + + +def tle_lines(record: OmmRecord, *, alpha5: bool = False) -> tuple[str, str]: + """The record's TLE lines: as issued, or re-rendered from the OMM. + + ALPHA-5 always re-renders. The issued lines use the temp-ID scheme, so a + numeric catalog number at or above 100000 only carries its letter prefix if + we build the field ourselves. Falls back to the issued lines when the OMM + cannot be rendered -- which is the normal case for an object whose ID has no + five-character form at all. + """ + if alpha5 and (record.omm or {}).get("NORAD_CAT_ID") is not None: + try: + _, line1, line2 = tle_from_gp(record.omm, alpha5=True) + return line1, line2 + except (KeyError, ValueError, TypeError): + pass + if record.line1 and record.line2: + return record.line1, record.line2 + _, line1, line2 = tle_from_gp(record.omm) + return line1, line2 + + +def _identity(record: OmmRecord) -> str: + return record.object_id or record.object_name or "" + + +def _write_tle(records, **_kw) -> str: + return "".join(as_plaintext(*row) for row in labelled_rows(records)) + + +def _write_tle_alpha5(records, **_kw) -> str: + return "".join(as_plaintext(*row) for row in labelled_rows(records, alpha5=True)) + + +# --- OMM and GP ------------------------------------------------------------- + + +def _write_omm_xml(records, **_kw) -> str: + return as_omm_xml(records, originator=DEFAULT_ORIGINATOR) + + +def _write_omm_kvn(records, **_kw) -> str: + return as_omm_kvn(records, originator=DEFAULT_ORIGINATOR) + + +def _write_gp_csv(records, **_kw) -> str: + return as_celestrak_csv(records) + + +# --- SANA element sets ------------------------------------------------------ + + +def _element_set_writer(set_key: str) -> Callable[..., str]: + """A CSV writer for one SANA element set. + + Columns are ``OBJECT_NAME, NORAD_CAT_ID, EPOCH, FRAME`` plus the set's own + components. Each record is propagated to ``at`` (or its own epoch) and + converted. A record that cannot be propagated is skipped rather than failing + the batch: one manoeuvring object must not cost a caller the other 199. + """ + + def write(records, *, backend=None, at: datetime | None = None, **_kw) -> str: + eset = element_sets.get_set(set_key) + component_keys = [c.key for c in eset.components] + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow( + ["OBJECT_NAME", "NORAD_CAT_ID", "EPOCH", "FRAME", *component_keys] + ) + for record in records: + try: + result = element_sets.compute_element_set( + set_key, record.omm, backend, at=at, + ) + except Exception: # noqa: BLE001 -- one bad record must not fail the batch + continue + omm = record.omm or {} + epoch = result["epoch"] or parse_omm_epoch(omm.get("EPOCH")) + values = result["values"] + writer.writerow([ + record.object_name or omm.get("OBJECT_NAME") or "", + omm.get("NORAD_CAT_ID") or "", + epoch.isoformat() if epoch else "", + result["frame"], + *[_number(values.get(k)) for k in component_keys], + ]) + return buf.getvalue() + + return write + + +def _number(value) -> str: + return "" if value is None else repr(float(value)) + + +# --- the table -------------------------------------------------------------- + +FORMATS: dict[str, Format] = { + "json": Format( + key="json", content_type="application/json", filename="yksa-tle.json", + ), + "tle": Format( + key="tle", content_type="text/plain", filename="yksa-tle.tle", + encoding="ascii", write=_write_tle, + ), + "tle_alpha5": Format( + key="tle_alpha5", content_type="text/plain", filename="yksa-tle-alpha5.tle", + encoding="ascii", write=_write_tle_alpha5, + ), + "omm_xml": Format( + key="omm_xml", content_type="application/xml", filename="yksa-omm.xml", + write=_write_omm_xml, + ), + "kvn": Format( + key="kvn", content_type="text/plain", filename="yksa-omm.kvn", + write=_write_omm_kvn, + ), + "csv": Format( + key="csv", content_type="text/csv", filename="yksa-gp.csv", + write=_write_gp_csv, + ), +} + +# Registered from the element-set catalogue itself, so the two cannot drift into +# offering different sets. +for _key in element_sets.API_ELEMENT_SET_KEYS: + FORMATS[_key] = Format( + key=_key, content_type="text/csv", filename=f"yksa-{_key}.csv", + write=_element_set_writer(_key), needs_backend=True, + ) + + +def get(key: str) -> Format: + try: + return FORMATS[key] + except KeyError as exc: + raise ValueError(f"unknown format {key!r}") from exc + + +def keys() -> tuple[str, ...]: + return tuple(FORMATS) diff --git a/odm/tle.py b/odm/tle.py new file mode 100644 index 0000000..a6dfb07 --- /dev/null +++ b/odm/tle.py @@ -0,0 +1,232 @@ +"""TLE text: reading three lines in, rendering three lines out. + +The TLE is a *derived* rendering, not the record. Line 1 has narrow fixed-width +fields that lose precision on MEAN_MOTION_DOT, MEAN_MOTION_DDOT and BSTAR +against the OMM they came from, and its five-character catalog field cannot hold +a modern alphanumeric ID at all (see :mod:`odm.norad`). Round-tripping through +here is lossy by construction; round-tripping through :mod:`odm.omm` is not. + +Reference for the line layout: https://celestrak.org/columns/v04n03/ +""" + +from __future__ import annotations + +import math +from datetime import datetime, timedelta, timezone +from typing import Iterable + +from sgp4.api import Satrec + +from .norad import tle_catalog_field +from .records import ParsedTLE + +_SGP4_NDOT_TO_CELESTRAK = (1440.0**2) / (2.0 * math.pi) + +_SGP4_NDDOT_TO_CELESTRAK = (1440.0**3) / (2.0 * math.pi) + +def parse_tle(line0: str, line1: str, line2: str) -> ParsedTLE: + """Parse a 3-line TLE block and return a :class:`ParsedTLE` with OMM dict.""" + line0 = (line0 or "").strip().lstrip("0 ").strip() + line1 = line1.rstrip() + line2 = line2.rstrip() + if not (line1.startswith("1 ") and line2.startswith("2 ")): + raise ValueError("TLE lines must start with '1 ' and '2 '") + if len(line1) < 69 or len(line2) < 69: + raise ValueError("TLE lines must be 69 characters long") + + sat = Satrec.twoline2rv(line1, line2) + epoch = _epoch_from_satrec(sat) + omm_dict: dict = { + "OBJECT_NAME": line0, + "OBJECT_ID": _intl_designator(line1), + "EPOCH": epoch.isoformat().replace("+00:00", ""), + "MEAN_MOTION": _no_kozai_to_rev_per_day(sat.no_kozai), + "ECCENTRICITY": sat.ecco, + "INCLINATION": _rad2deg(sat.inclo), + "RA_OF_ASC_NODE": _rad2deg(sat.nodeo), + "ARG_OF_PERICENTER": _rad2deg(sat.argpo), + "MEAN_ANOMALY": _rad2deg(sat.mo), + "EPHEMERIS_TYPE": int(line1[62]) if line1[62].isdigit() else 0, + "CLASSIFICATION_TYPE": line1[7] or "U", + "NORAD_CAT_ID": sat.satnum, + "ELEMENT_SET_NO": int(line1[64:68].strip() or 0), + "REV_AT_EPOCH": int(line2[63:68].strip() or 0), + "BSTAR": sat.bstar, + "MEAN_MOTION_DOT": sat.ndot * _SGP4_NDOT_TO_CELESTRAK, + "MEAN_MOTION_DDOT": sat.nddot * _SGP4_NDDOT_TO_CELESTRAK, + } + return ParsedTLE(name=line0, line1=line1, line2=line2, omm=omm_dict) + +def tle_from_gp(gp: dict, *, alpha5: bool = False) -> tuple[str, str, str]: + """Render a 3-line TLE block from a GP JSON dict. + + The 5-char catalog field follows :func:`tle_catalog_field`: numeric IDs + render as digits, ``alpha5=True`` enables the ALPHA-5 letter prefix for + 100000-339999, and anything else falls back to a deterministic 9xxxx + pseudo-ID (backwards compatible with Orbitron and similar tools). + """ + cat = tle_catalog_field(gp["NORAD_CAT_ID"], alpha5=alpha5) + classification = (gp.get("CLASSIFICATION_TYPE") or "U")[:1] + intl = (gp.get("OBJECT_ID") or "").strip() + intl_l1 = _format_intl_for_tle(intl) + epoch_yy, epoch_day = _epoch_to_yyddd(gp["EPOCH"]) + n_dot = float(gp.get("MEAN_MOTION_DOT") or 0.0) + n_ddot = float(gp.get("MEAN_MOTION_DDOT") or 0.0) + bstar = float(gp.get("BSTAR") or 0.0) + ephem_type = int(gp.get("EPHEMERIS_TYPE") or 0) + elset = int(gp.get("ELEMENT_SET_NO") or 0) + + line1_body = ( + f"1 {cat}{classification} {intl_l1:<8s} " + f"{epoch_yy:02d}{epoch_day:012.8f} " + f"{_format_signed_decimal(n_dot)} " + f"{_format_assumed_exponent(n_ddot)} " + f"{_format_assumed_exponent(bstar)} " + f"{ephem_type:1d} {elset:4d}" + ) + line1 = _with_checksum(line1_body) + + incl = float(gp["INCLINATION"]) + node = float(gp["RA_OF_ASC_NODE"]) + ecc = float(gp["ECCENTRICITY"]) + argp = float(gp["ARG_OF_PERICENTER"]) + ma = float(gp["MEAN_ANOMALY"]) + mm = float(gp["MEAN_MOTION"]) + rev = int(gp.get("REV_AT_EPOCH") or 0) + + line2_body = ( + f"2 {cat} " + f"{incl:8.4f} {node:8.4f} " + f"{int(round(ecc * 1e7)):07d} " + f"{argp:8.4f} {ma:8.4f} " + f"{mm:11.8f}{rev:5d}" + ) + line2 = _with_checksum(line2_body) + + name = (gp.get("OBJECT_NAME") or "").strip() + return name, line1, line2 + +def as_plaintext(name: str, line1: str, line2: str) -> str: + """Three-line block with a trailing newline.""" + return f"{name}\n{line1}\n{line2}\n" + +def as_txt_txt(records: Iterable[tuple[str, str, str]]) -> str: + """txt-style multi-satellite text file (CRLF line endings, name pad 24).""" + parts: list[str] = [] + for name, line1, line2 in records: + padded = (name or "")[:24].ljust(24) + parts.append(padded) + parts.append(line1) + parts.append(line2) + return "\r\n".join(parts) + "\r\n" + +def indexed_name(base: str, *, index: int | None, total: int) -> str: + """Compose a TLE-name-line label for batch exports. + + When *total* > 1 (the same satellite appears more than once in the + output set), a ``[NN]`` suffix is appended so consumers can tell the + records apart in plaintext / .txt downloads. + """ + base = (base or "").strip() + if not base: + base = "UNKNOWN" + if total <= 1 or index is None: + return base[:24] + suffix = f" [{index:02d}]" + # Reserve room for the suffix; clip the base if needed so the total + # never exceeds 24 chars. + budget = 24 - len(suffix) + return f"{base[:budget].rstrip()}{suffix}" + +def _epoch_from_satrec(sat: Satrec) -> datetime: + """Reconstruct a UTC datetime from a Satrec's epochyr/epochdays.""" + year = sat.epochyr + if year < 57: + year += 2000 + elif year < 100: + year += 1900 + day_of_year = sat.epochdays # 1-based, fractional + return datetime(year, 1, 1, tzinfo=timezone.utc) + timedelta(days=day_of_year - 1) + +def _intl_designator(line1: str) -> str: + raw = line1[9:17].rstrip() + if not raw: + return "" + yy = raw[:2] + rest = raw[2:].strip() + if not yy.isdigit(): + return raw + year = int(yy) + century = 2000 if year < 57 else 1900 + return f"{century + year:04d}-{rest}" + +def _format_intl_for_tle(intl: str) -> str: + """Convert "2024-001A" back to TLE line-1 format "24001A".""" + if not intl: + return "" + if "-" in intl: + year, rest = intl.split("-", 1) + try: + return f"{int(year) % 100:02d}{rest}" + except ValueError: + return intl + return intl + +def _epoch_to_yyddd(epoch_iso: str) -> tuple[int, float]: + iso = epoch_iso + if iso.endswith("Z"): + iso = iso[:-1] + "+00:00" + if "+" not in iso and iso.count("-") <= 2: + iso = iso + "+00:00" + dt = datetime.fromisoformat(iso) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + dt_utc = dt.astimezone(timezone.utc) + start = datetime(dt_utc.year, 1, 1, tzinfo=timezone.utc) + day_of_year = (dt_utc - start).total_seconds() / 86400.0 + 1.0 + return dt_utc.year % 100, day_of_year + +def _rad2deg(value: float) -> float: + return math.degrees(value) + +def _no_kozai_to_rev_per_day(no_kozai: float) -> float: + """Convert sgp4's stored mean motion (rad/min) to revolutions/day.""" + return no_kozai * 1440.0 / (2.0 * math.pi) + +def _format_signed_decimal(value: float) -> str: + """Format like '-.00002182' / ' .00002182' (10 chars, no leading zero).""" + sign = "-" if value < 0 else " " + body = f"{abs(value):.8f}" # "0.00002182" + # Drop the leading "0" to get ".00002182". + if body.startswith("0."): + body = body[1:] + return f"{sign}{body}" + +def _format_assumed_exponent(value: float) -> str: + """Format like ' 00000-0' / '-12345-3' (8 chars, assumed leading decimal). + + Encodes value = 0.MANTISSA * 10**EXP. Mantissa is 5 digits. + """ + if value == 0.0: + return " 00000-0" + sign = "-" if value < 0 else " " + v = abs(value) + exp = int(math.floor(math.log10(v))) + 1 # mantissa in [0.1, 1.0) + mantissa = v / (10.0**exp) + mant_digits = int(round(mantissa * 1e5)) + if mant_digits >= 100000: + mant_digits //= 10 + exp += 1 + exp_sign = "-" if exp < 0 else "+" + return f"{sign}{mant_digits:05d}{exp_sign}{abs(exp):d}" + +def _with_checksum(body: str) -> str: + """Pad to 68 chars and append the standard mod-10 checksum digit.""" + body = body[:68].ljust(68) + total = 0 + for ch in body: + if ch.isdigit(): + total += int(ch) + elif ch == "-": + total += 1 + return f"{body}{total % 10}" diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2881197 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,17 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "odm" +version = "1.0.0" +description = "Orbit Data Messages: read, write and convert orbital element sets" +readme = "README.md" +requires-python = ">=3.10" +# sgp4 only, and only for reading TLE text (Satrec does the unpacking) and for +# the mean-element derivations. No Django, no HTTP, no framework: every service +# that touches ODMS data has to be able to install this. +dependencies = ["sgp4>=2.20"] + +[tool.setuptools.packages.find] +include = ["odm*"] diff --git a/tests/test_alpha5.py b/tests/test_alpha5.py new file mode 100644 index 0000000..d70ecdc --- /dev/null +++ b/tests/test_alpha5.py @@ -0,0 +1,98 @@ +"""Tests for the ALPHA-5 NORAD catalog-number encoding.""" + +from __future__ import annotations + +import pytest +from sgp4.api import Satrec + +from odm import ( + ALPHA5_MAX, + ALPHA5_MIN, + from_alpha5, + tle_catalog_field, + to_alpha5, +) +from odm import parse_tle, tle_from_gp + + +@pytest.mark.parametrize("value,expected", [ + (100000, "A0000"), + (108493, "A8493"), + (180000, "J0000"), # skips I + (234567, "P4567"), + (270354, "T0354"), + (339999, "Z9999"), +]) +def test_to_alpha5_known_values(value, expected): + assert to_alpha5(value) == expected + + +def test_to_alpha5_skips_i_and_o(): + encoded = "".join(to_alpha5(v * 10000)[0] for v in range(10, 34)) + assert "I" not in encoded and "O" not in encoded + assert len(set(encoded)) == 24 # all distinct + + +@pytest.mark.parametrize("value", [ALPHA5_MIN, 123456, 234567, ALPHA5_MAX]) +def test_alpha5_round_trips(value): + assert from_alpha5(to_alpha5(value)) == value + + +@pytest.mark.parametrize("value", [99999, ALPHA5_MAX + 1, 0, -1]) +def test_to_alpha5_rejects_out_of_range(value): + with pytest.raises(ValueError): + to_alpha5(value) + + +@pytest.mark.parametrize("field", ["", "ABCDE", "A123", "I0000", "O0000", "12345"]) +def test_from_alpha5_rejects_invalid(field): + assert from_alpha5(field) is None + + +def test_tle_catalog_field_plain_numeric_unchanged(): + assert tle_catalog_field("25544", alpha5=True) == "25544" + assert tle_catalog_field("25544", alpha5=False) == "25544" + + +def test_tle_catalog_field_alpha5_vs_temp(): + assert tle_catalog_field("234567", alpha5=True) == "P4567" + temp = tle_catalog_field("234567", alpha5=False) + assert temp.isdigit() and 90000 <= int(temp) <= 99999 + + +def test_tle_catalog_field_alphanumeric_always_pseudo(): + for flag in (True, False): + field = tle_catalog_field("A1234", alpha5=flag) + assert field.isdigit() and 90000 <= int(field) <= 99999 + + +def test_tle_from_gp_alpha5_round_trips_through_sgp4(): + iss = parse_tle( + "ISS (ZARYA)", + "1 25544U 98067A 24070.50000000 .00012345 00000-0 22000-3 0 9991", + "2 25544 51.6400 200.0000 0001234 90.0000 270.0000 15.50000000400000", + ) + gp = dict(iss.omm) + gp["NORAD_CAT_ID"] = "234567" + _, line1, line2 = tle_from_gp(gp, alpha5=True) + + assert len(line1) == 69 and len(line2) == 69 + assert line1[2:7] == "P4567" + assert line1[2:7] == line2[2:7] + # sgp4 must decode the ALPHA-5 field back to the real catalog number. + sat = Satrec.twoline2rv(line1, line2) + assert sat.satnum == 234567 + + +def test_tle_from_gp_defaults_to_temp_id(): + """Without alpha5, a 6-digit numeric ID must not overflow the 5-char field.""" + iss = parse_tle( + "ISS (ZARYA)", + "1 25544U 98067A 24070.50000000 .00012345 00000-0 22000-3 0 9991", + "2 25544 51.6400 200.0000 0001234 90.0000 270.0000 15.50000000400000", + ) + gp = dict(iss.omm) + gp["NORAD_CAT_ID"] = "234567" + _, line1, line2 = tle_from_gp(gp) + assert len(line1) == 69 and line1[2:7].isdigit() + assert 90000 <= int(line1[2:7]) <= 99999 diff --git a/tests/test_export_formats.py b/tests/test_export_formats.py new file mode 100644 index 0000000..961224a --- /dev/null +++ b/tests/test_export_formats.py @@ -0,0 +1,66 @@ +"""Tests for the CSV and KVN export renderers added for the ODMS service.""" + +from __future__ import annotations + +import csv +import io + +from odm import GP_FIELDS, OmmRecord, as_celestrak_csv, as_omm_kvn + + +def _record() -> OmmRecord: + omm = { + "OBJECT_NAME": "ISS (ZARYA)", + "OBJECT_ID": "1998-067A", + "EPOCH": "2026-05-13T22:14:00", + "MEAN_MOTION": 15.498, + "ECCENTRICITY": 0.000284, + "INCLINATION": 51.64, + "RA_OF_ASC_NODE": 20.23, + "ARG_OF_PERICENTER": 74.12, + "MEAN_ANOMALY": 10.2, + "EPHEMERIS_TYPE": 0, + "CLASSIFICATION_TYPE": "U", + "NORAD_CAT_ID": "25544", + "ELEMENT_SET_NO": 999, + "REV_AT_EPOCH": 100, + "BSTAR": 0.00012345, + "MEAN_MOTION_DOT": 7.2e-05, + "MEAN_MOTION_DDOT": 0.0, + } + return OmmRecord(omm=omm, object_name="ISS (ZARYA)", object_id="1998-067A", originator="YKSA") + + +def test_celestrak_csv_header_matches_gp_field_order(): + text = as_celestrak_csv([_record()]) + rows = list(csv.reader(io.StringIO(text))) + assert tuple(rows[0]) == GP_FIELDS + # One data row, NORAD in the right column. + assert rows[1][GP_FIELDS.index("NORAD_CAT_ID")] == "25544" + assert rows[1][GP_FIELDS.index("OBJECT_NAME")] == "ISS (ZARYA)" + + +def test_celestrak_csv_multiple_records(): + text = as_celestrak_csv([_record(), _record()]) + rows = list(csv.reader(io.StringIO(text))) + assert len(rows) == 3 # header + 2 + + +def test_omm_kvn_has_version_and_key_value_lines(): + text = as_omm_kvn([_record()], originator="YKSA") + assert text.startswith("CCSDS_OMM_VERS = 3.0") + kv = dict( + line.split(" = ", 1) + for line in text.splitlines() + if " = " in line + ) + assert kv["ORIGINATOR"] == "YKSA" + assert kv["NORAD_CAT_ID"] == "25544" + assert kv["MEAN_ELEMENT_THEORY"] == "SGP4" + assert kv["INCLINATION"] == "51.64" + + +def test_omm_kvn_separates_records_with_blank_line(): + text = as_omm_kvn([_record(), _record()]) + assert text.count("CCSDS_OMM_VERS = 3.0") == 2 + assert "\n\n" in text diff --git a/tests/test_formats.py b/tests/test_formats.py new file mode 100644 index 0000000..f9f17c5 --- /dev/null +++ b/tests/test_formats.py @@ -0,0 +1,288 @@ +"""TLE / OMM format conversion tests. + +Round-trip checks: ``parse_tle -> tle_from_gp`` should reproduce the underlying +orbit (compared via sgp4) to within float tolerance. OMM XML rendering must +emit every CCSDS-mandatory field, and the parser must handle the real-world +```` wrapper Space-Track ships in its archive exports. +""" + +from __future__ import annotations + +import math + +import pytest +from sgp4.api import Satrec + +from odm import ( + OmmRecord, + as_omm_xml, + as_txt_txt, + as_plaintext, + omm_xml_to_records, + parse_tle, + tle_from_gp, +) + + +ISS = ( + "ISS (ZARYA)", + "1 25544U 98067A 24070.50000000 .00012345 00000-0 22000-3 0 9991", + "2 25544 51.6400 200.0000 0001234 90.0000 270.0000 15.50000000400000", +) + +NOAA15 = ( + "NOAA 15", + "1 25338U 98030A 24070.20000000 .00000050 00000-0 31000-4 0 9990", + "2 25338 98.7000 100.0000 0010000 10.0000 350.0000 14.26000000900000", +) + + +# Real-world Space-Track archive sample (verbatim from user-supplied issue), +# trimmed to two records to keep the test fast. +SPACETRACK_ARCHIVE_XML = """\ + + +
+GENERATED VIA SPACE-TRACK.ORG API +2026-01-08T05:20:27 +18 SPCS +
+ + + +TBA - TO BE ASSIGNED +2025-313AU +EARTH +TEME +UTC +SGP4 + + + +2026-01-06T22:50:56.049504 +15.21286088 +0.00095778 +97.4119 +84.2715 +195.9225 +164.1711 + + +0 +U +67290 +999 +143 +0.00053078883000 +0.00011823 +0.0000000000000 + + +6880.097 +94.657 +508.552 +495.373 +UNKNOWN + + + + + +4962752 +308346717 + + + + + + +
+GENERATED VIA SPACE-TRACK.ORG API +2026-01-08T15:46:24 +18 SPCS +
+ + + +TBA - TO BE ASSIGNED +2025-313AU +EARTH +TEME +UTC +SGP4 + + + +2026-01-08T11:09:24.647040 +15.21321390 +0.00096980 +97.4116 +85.7604 +190.1005 +170.0038 + + +0 +U +67290 +999 +166 +0.00055242000000 +0.00012321 +0.0000000000000 + + + + + + +""" + + +@pytest.mark.parametrize("name,l1,l2", [ISS, NOAA15]) +def test_parse_tle_yields_expected_omm(name, l1, l2): + parsed = parse_tle(name, l1, l2) + assert parsed.omm["NORAD_CAT_ID"] == int(l1[2:7]) + assert math.isclose(parsed.omm["INCLINATION"], float(l2[8:16]), abs_tol=1e-4) + + +@pytest.mark.parametrize("name,l1,l2", [ISS, NOAA15]) +def test_round_trip_preserves_orbit(name, l1, l2): + parsed = parse_tle(name, l1, l2) + name_out, l1_out, l2_out = tle_from_gp(parsed.omm) + assert l1_out.startswith("1 ") + assert l2_out.startswith("2 ") + assert len(l1_out) == 69 + assert len(l2_out) == 69 + sat_a = Satrec.twoline2rv(l1, l2) + sat_b = Satrec.twoline2rv(l1_out, l2_out) + assert sat_a.satnum == sat_b.satnum + assert math.isclose(sat_a.inclo, sat_b.inclo, abs_tol=1e-5) + assert math.isclose(sat_a.nodeo, sat_b.nodeo, abs_tol=1e-5) + assert math.isclose(sat_a.ecco, sat_b.ecco, abs_tol=1e-7) + assert math.isclose(sat_a.argpo, sat_b.argpo, abs_tol=1e-5) + assert math.isclose(sat_a.mo, sat_b.mo, abs_tol=1e-5) + assert math.isclose(sat_a.no_kozai, sat_b.no_kozai, abs_tol=1e-7) + + +def test_txt_txt_format(): + name, l1, l2 = ISS + parsed = parse_tle(name, l1, l2) + out = as_txt_txt([(parsed.name, parsed.line1, parsed.line2)]) + lines = out.split("\r\n") + assert lines[0] == parsed.name[:24].ljust(24) + assert lines[1] == parsed.line1 + assert lines[2] == parsed.line2 + assert lines[3] == "" + + +def test_plaintext_format_has_trailing_newline(): + out = as_plaintext(*ISS) + assert out.endswith("\n") + assert out.count("\n") == 3 + + +def test_omm_xml_contains_all_mandatory_fields(): + parsed = parse_tle(*ISS) + rec = OmmRecord( + omm=parsed.omm, + object_name=parsed.name, + object_id=parsed.omm["OBJECT_ID"], + originator="YKSA", + mean_element_theory="SGP4", + ) + xml = as_omm_xml([rec], originator="YKSA") + + assert xml.startswith("" in xml + assert "YKSA" in xml + + for field in ( + "OBJECT_NAME", "OBJECT_ID", "CENTER_NAME", "REF_FRAME", + "TIME_SYSTEM", "MEAN_ELEMENT_THEORY", + ): + assert f"<{field}>" in xml, f"missing mandatory metadata field {field}" + + for field in ( + "EPOCH", "MEAN_MOTION", "ECCENTRICITY", "INCLINATION", + "RA_OF_ASC_NODE", "ARG_OF_PERICENTER", "MEAN_ANOMALY", + ): + assert f"<{field}>" in xml, f"missing mandatory meanElements field {field}" + + assert "" in xml + for field in ("NORAD_CAT_ID", "BSTAR", "MEAN_MOTION_DOT"): + assert f"<{field}>" in xml + + +def test_omm_xml_round_trip_preserves_orbit(): + parsed = parse_tle(*ISS) + rec = OmmRecord( + omm=parsed.omm, + object_name=parsed.name, + object_id=parsed.omm["OBJECT_ID"], + mean_element_theory="SGP4", + ) + xml = as_omm_xml([rec]) + parsed_back = list(omm_xml_to_records(xml)) + assert len(parsed_back) == 1 + rt = parsed_back[0] + assert rt.object_name == parsed.name + assert rt.mean_element_theory == "SGP4" + assert rt.ref_frame == "TEME" + assert math.isclose(rt.omm["INCLINATION"], parsed.omm["INCLINATION"], abs_tol=1e-9) + assert math.isclose(rt.omm["MEAN_MOTION"], parsed.omm["MEAN_MOTION"], abs_tol=1e-12) + # NORAD_CAT_ID round-trips through OMM XML as a string (to accommodate the + # future alphanumeric Space-Track catalog IDs). + assert str(rt.omm["NORAD_CAT_ID"]) == str(parsed.omm["NORAD_CAT_ID"]) + + +def test_omm_xml_omits_tle_params_for_non_sgp_theories(): + parsed = parse_tle(*ISS) + rec = OmmRecord( + omm=parsed.omm, + object_name=parsed.name, + object_id=parsed.omm["OBJECT_ID"], + mean_element_theory="DSST", + ) + xml = as_omm_xml([rec]) + assert "" not in xml + assert "DSST" in xml + + +def test_omm_xml_parses_spacetrack_archive_format(): + """The Space-Track archive ships -wrapped multi-document OMM with an + undeclared xsi: prefix and a vendor-specific block. + All three must be tolerated without losing structured data. + """ + records = list(omm_xml_to_records(SPACETRACK_ARCHIVE_XML)) + assert len(records) == 2 + + a, b = records + # First record's precise numeric fields must survive verbatim -- this is + # the exact reason we treat OMM as canonical instead of round-tripping + # through TLE line 1. + assert a.omm["NORAD_CAT_ID"] == "67290" + assert math.isclose(a.omm["MEAN_MOTION_DOT"], 0.00011823, abs_tol=1e-12) + assert math.isclose(a.omm["BSTAR"], 0.00053078883, abs_tol=1e-14) + assert math.isclose(a.omm["MEAN_MOTION"], 15.21286088, abs_tol=1e-12) + + # Second record carries its own creation date -- header context must be + # per-OMM, not shared across the whole . + assert b.omm["NORAD_CAT_ID"] == "67290" + assert math.isclose(b.omm["BSTAR"], 0.00055242, abs_tol=1e-14) + + # Metadata from the segment must be threaded through correctly. + assert a.mean_element_theory == "SGP4" + assert a.ref_frame == "TEME" + assert a.originator == "18 SPCS" + assert a.object_id == "2025-313AU" + + # The original chunk must be preserved verbatim so we can hand it + # back to API consumers unchanged. The userDefinedParameters block must + # still be present in the stored XML even though we discard it from the + # structured dict. + assert "" in a.omm_xml + assert "userDefinedParameters" in a.omm_xml + assert "SEMIMAJOR_AXIS" in a.omm_xml diff --git a/tests/test_norad.py b/tests/test_norad.py new file mode 100644 index 0000000..ed54133 --- /dev/null +++ b/tests/test_norad.py @@ -0,0 +1,24 @@ +"""Tests for the NORAD ID normaliser.""" + +from __future__ import annotations + +import pytest + +from odm import normalise_norad + + +@pytest.mark.parametrize("raw,expected", [ + (None, None), + ("", None), + (" ", None), + (25544, "25544"), + ("25544", "25544"), + (" 25544 ", "25544"), + ("123", "00123"), # zero-padded to 5 + (b"42", "00042"), # bytes accepted, zero-padded + ("A1234", "A1234"), # alphanumeric preserved verbatim + ("1998-067A", "1998-067A"), # hyphenated preserved verbatim + ("12345678", "12345678"), # >5 digits left intact +]) +def test_normalise_norad(raw, expected): + assert normalise_norad(raw) == expected diff --git a/tests/test_temp_norad.py b/tests/test_temp_norad.py new file mode 100644 index 0000000..8ce2f09 --- /dev/null +++ b/tests/test_temp_norad.py @@ -0,0 +1,53 @@ +"""Tests for the alphanumeric -> numeric TLE fallback.""" + +from __future__ import annotations + +from odm import ( + TEMP_NORAD_MAX, + TEMP_NORAD_MIN, + coerce_norad_to_int, + temp_tle_norad, +) + + +def test_temp_norad_stays_in_reserved_range(): + for ident in ("A1234", "ZZ-9999", "1998-067A", "X", ""): + n = temp_tle_norad(ident) + assert TEMP_NORAD_MIN <= n <= TEMP_NORAD_MAX + + +def test_temp_norad_is_deterministic(): + assert temp_tle_norad("A1234") == temp_tle_norad("A1234") + assert temp_tle_norad("A1234") != temp_tle_norad("A1235") + + +def test_coerce_norad_to_int_passes_through_numerics(): + assert coerce_norad_to_int(25544) == 25544 + assert coerce_norad_to_int("25544") == 25544 + assert coerce_norad_to_int(" 25544 ") == 25544 + + +def test_coerce_norad_to_int_substitutes_for_alphanumeric(): + n = coerce_norad_to_int("A1234") + assert TEMP_NORAD_MIN <= n <= TEMP_NORAD_MAX + # Stable: second call returns the same number. + assert coerce_norad_to_int("A1234") == n + + +def test_tle_from_gp_renders_for_alphanumeric_norad(): + """The TLE rendering must succeed even when NORAD_CAT_ID is non-numeric.""" + from odm import parse_tle, tle_from_gp + + iss = parse_tle( + "ISS (ZARYA)", + "1 25544U 98067A 24070.50000000 .00012345 00000-0 22000-3 0 9991", + "2 25544 51.6400 200.0000 0001234 90.0000 270.0000 15.50000000400000", + ) + gp = dict(iss.omm) + gp["NORAD_CAT_ID"] = "A1234" # alphanumeric future ID + _, line1, line2 = tle_from_gp(gp) + assert line1.startswith("1 ") and line2.startswith("2 ") + # The 5-digit slot at chars 2-7 must contain the synthetic 9xxxx number. + assert line1[2:7].isdigit() + assert 90000 <= int(line1[2:7]) <= 99999 + assert line1[2:7] == line2[2:7] # consistent across lines