yksa-orbital/yksa_orbital/wire.py
2026-08-18 22:01:53 +08:00

66 lines
2.5 KiB
Python

"""Render a stored OMM dict as the CCSDS message the Orekit sidecar consumes.
The sidecar takes CCSDS messages, not TLE lines, so nothing on this path
truncates an epoch to eight digits or drops a mass. This module is the single
place that turns ODMS's stored GP dict into one.
"""
from __future__ import annotations
from odm import OmmRecord, PropagationError, as_omm_kvn
#: CCSDS defines MEAN_MOTION_DOT as the *first derivative* of the mean motion,
#: while TLE line 1 carries half of it -- and Space-Track's OMM output copies
#: the TLE field across unchanged, so that is the convention in our stored data.
#: Orekit reads the spec. Doubling here keeps the element set Orekit reconstructs
#: identical to the one we hold, and is confined to this wire format: the OMM we
#: publish to users stays byte-compatible with Space-Track's.
_NDOT_TLE_TO_CCSDS = 2.0
#: Orekit's OMM parser rejects an SGP4 message missing any of these, and
#: upstream feeds do sometimes omit the bookkeeping ones. They do not affect
#: propagation; the values only have to be present and well-formed.
_SGP4_DEFAULTS = {
"EPHEMERIS_TYPE": 0,
"CLASSIFICATION_TYPE": "U",
"ELEMENT_SET_NO": 999,
"REV_AT_EPOCH": 1,
"BSTAR": 0.0,
"MEAN_MOTION_DOT": 0.0,
"MEAN_MOTION_DDOT": 0.0,
}
_REQUIRED = ("EPOCH", "MEAN_MOTION", "ECCENTRICITY", "INCLINATION")
def omm_message(omm: dict, *, object_name: str = "") -> str:
"""One stored OMM dict as CCSDS KVN, ready to POST to the sidecar."""
if not isinstance(omm, dict):
raise PropagationError(f"expected an OMM dict, got {type(omm).__name__}")
missing = [key for key in _REQUIRED if omm.get(key) in (None, "")]
if missing:
raise PropagationError(
f"OMM is missing {', '.join(missing)}; cannot build a CCSDS message"
)
fields = dict(omm)
theory = str(fields.get("MEAN_ELEMENT_THEORY") or "SGP4")
if theory.upper().startswith(("SGP", "SDP")):
for key, default in _SGP4_DEFAULTS.items():
if fields.get(key) in (None, ""):
fields[key] = default
fields["MEAN_MOTION_DOT"] = (
float(fields["MEAN_MOTION_DOT"]) * _NDOT_TLE_TO_CCSDS
)
return as_omm_kvn([OmmRecord(
omm=fields,
object_name=object_name or str(fields.get("OBJECT_NAME") or ""),
object_id=str(fields.get("OBJECT_ID") or ""),
originator="ODMS",
mean_element_theory=theory,
)])
def omm_messages(omms: list[dict]) -> list[str]:
return [omm_message(omm) for omm in omms]