158 lines
6.6 KiB
Python
158 lines
6.6 KiB
Python
"""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,
|
|
}
|