Initial commit

.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

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.

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"

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)

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}

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)

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

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)

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
+``<ndm>`` 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 <meanElements> 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 <tleParameters> 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 ``<ndm>`` 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 ``<omm>`` 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 <root>.
+        omm_iter = list(root.findall("omm"))
+        if not omm_iter:
+            raise ValueError(f"expected <ndm> or <omm> 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 ``<omm>`` 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()
+            )
+    #  <userDefinedParameters> 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 ``<ndm>`` / ``<omm>`` 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 ``<omm>...</omm>`` element back into a tree."""
+    return _parse_omm_root(chunk)
+
+def _serialise_omm_element(omm_el: ET.Element) -> str:
+    """Render a single ``<omm>`` 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")

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,
+    )

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,
+        }

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 ``<omm>...</omm>`` 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)

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)

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

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*"]

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

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

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
+``<ndm>`` 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 = """\
+<ndm xsi:noNamespaceSchemaLocation="https://sanaregistry.org/r/ndmxml_unqualified/ndmxml-3.0.0-master-3.0.xsd">
+<omm id="CCSDS_OMM_VERS" version="3.0">
+<header>
+<COMMENT>GENERATED VIA SPACE-TRACK.ORG API</COMMENT>
+<CREATION_DATE>2026-01-08T05:20:27</CREATION_DATE>
+<ORIGINATOR>18 SPCS</ORIGINATOR>
+</header>
+<body>
+<segment>
+<metadata>
+<OBJECT_NAME>TBA - TO BE ASSIGNED</OBJECT_NAME>
+<OBJECT_ID>2025-313AU</OBJECT_ID>
+<CENTER_NAME>EARTH</CENTER_NAME>
+<REF_FRAME>TEME</REF_FRAME>
+<TIME_SYSTEM>UTC</TIME_SYSTEM>
+<MEAN_ELEMENT_THEORY>SGP4</MEAN_ELEMENT_THEORY>
+</metadata>
+<data>
+<meanElements>
+<EPOCH>2026-01-06T22:50:56.049504</EPOCH>
+<MEAN_MOTION>15.21286088</MEAN_MOTION>
+<ECCENTRICITY>0.00095778</ECCENTRICITY>
+<INCLINATION>97.4119</INCLINATION>
+<RA_OF_ASC_NODE>84.2715</RA_OF_ASC_NODE>
+<ARG_OF_PERICENTER>195.9225</ARG_OF_PERICENTER>
+<MEAN_ANOMALY>164.1711</MEAN_ANOMALY>
+</meanElements>
+<tleParameters>
+<EPHEMERIS_TYPE>0</EPHEMERIS_TYPE>
+<CLASSIFICATION_TYPE>U</CLASSIFICATION_TYPE>
+<NORAD_CAT_ID>67290</NORAD_CAT_ID>
+<ELEMENT_SET_NO>999</ELEMENT_SET_NO>
+<REV_AT_EPOCH>143</REV_AT_EPOCH>
+<BSTAR>0.00053078883000</BSTAR>
+<MEAN_MOTION_DOT>0.00011823</MEAN_MOTION_DOT>
+<MEAN_MOTION_DDOT>0.0000000000000</MEAN_MOTION_DDOT>
+</tleParameters>
+<userDefinedParameters>
+<USER_DEFINED parameter="SEMIMAJOR_AXIS">6880.097</USER_DEFINED>
+<USER_DEFINED parameter="PERIOD">94.657</USER_DEFINED>
+<USER_DEFINED parameter="APOAPSIS">508.552</USER_DEFINED>
+<USER_DEFINED parameter="PERIAPSIS">495.373</USER_DEFINED>
+<USER_DEFINED parameter="OBJECT_TYPE">UNKNOWN</USER_DEFINED>
+<USER_DEFINED parameter="RCS_SIZE"/>
+<USER_DEFINED parameter="COUNTRY_CODE"/>
+<USER_DEFINED parameter="LAUNCH_DATE"/>
+<USER_DEFINED parameter="SITE"/>
+<USER_DEFINED parameter="DECAY_DATE"/>
+<USER_DEFINED parameter="FILE">4962752</USER_DEFINED>
+<USER_DEFINED parameter="GP_ID">308346717</USER_DEFINED>
+</userDefinedParameters>
+</data>
+</segment>
+</body>
+</omm>
+<omm id="CCSDS_OMM_VERS" version="3.0">
+<header>
+<COMMENT>GENERATED VIA SPACE-TRACK.ORG API</COMMENT>
+<CREATION_DATE>2026-01-08T15:46:24</CREATION_DATE>
+<ORIGINATOR>18 SPCS</ORIGINATOR>
+</header>
+<body>
+<segment>
+<metadata>
+<OBJECT_NAME>TBA - TO BE ASSIGNED</OBJECT_NAME>
+<OBJECT_ID>2025-313AU</OBJECT_ID>
+<CENTER_NAME>EARTH</CENTER_NAME>
+<REF_FRAME>TEME</REF_FRAME>
+<TIME_SYSTEM>UTC</TIME_SYSTEM>
+<MEAN_ELEMENT_THEORY>SGP4</MEAN_ELEMENT_THEORY>
+</metadata>
+<data>
+<meanElements>
+<EPOCH>2026-01-08T11:09:24.647040</EPOCH>
+<MEAN_MOTION>15.21321390</MEAN_MOTION>
+<ECCENTRICITY>0.00096980</ECCENTRICITY>
+<INCLINATION>97.4116</INCLINATION>
+<RA_OF_ASC_NODE>85.7604</RA_OF_ASC_NODE>
+<ARG_OF_PERICENTER>190.1005</ARG_OF_PERICENTER>
+<MEAN_ANOMALY>170.0038</MEAN_ANOMALY>
+</meanElements>
+<tleParameters>
+<EPHEMERIS_TYPE>0</EPHEMERIS_TYPE>
+<CLASSIFICATION_TYPE>U</CLASSIFICATION_TYPE>
+<NORAD_CAT_ID>67290</NORAD_CAT_ID>
+<ELEMENT_SET_NO>999</ELEMENT_SET_NO>
+<REV_AT_EPOCH>166</REV_AT_EPOCH>
+<BSTAR>0.00055242000000</BSTAR>
+<MEAN_MOTION_DOT>0.00012321</MEAN_MOTION_DOT>
+<MEAN_MOTION_DDOT>0.0000000000000</MEAN_MOTION_DDOT>
+</tleParameters>
+</data>
+</segment>
+</body>
+</omm>
+</ndm>
+"""
+
+
+@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("<?xml")
+    assert "<omm" in xml and 'id="CCSDS_OMM_VERS"' in xml and 'version="3.0"' in xml
+
+    assert "<CREATION_DATE>" in xml
+    assert "<ORIGINATOR>YKSA</ORIGINATOR>" 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 "<tleParameters>" 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 "<tleParameters>" not in xml
+    assert "<MEAN_ELEMENT_THEORY>DSST</MEAN_ELEMENT_THEORY>" in xml
+
+
+def test_omm_xml_parses_spacetrack_archive_format():
+    """The Space-Track archive ships <ndm>-wrapped multi-document OMM with an
+    undeclared xsi: prefix and a vendor-specific <userDefinedParameters> 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 <ndm>.
+    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 <omm> 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 "<omm" in a.omm_xml and "</omm>" in a.omm_xml
+    assert "userDefinedParameters" in a.omm_xml
+    assert "SEMIMAJOR_AXIS" in a.omm_xml

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

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