356 lines
15 KiB
Python
356 lines
15 KiB
Python
"""The ODMS API client.
|
|
|
|
Methods return parsed JSON, or the body as text for the rendered formats --
|
|
deliberately not model classes. ODMS adds fields as the catalogue grows, and a
|
|
client that maps them onto fixed classes turns each addition into a release.
|
|
|
|
Every method is one request. No caching, no implicit pagination, no background
|
|
refresh: those are policy, they differ per service, and hiding them here is how
|
|
a shared client becomes a shared problem.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Iterable
|
|
|
|
from ._http import DEFAULT_MAX_RETRIES, DEFAULT_TIMEOUT_S, Transport
|
|
from .errors import NotFound
|
|
|
|
#: Rendering formats ``query()``/``download()`` accept. ``json`` is the parsed
|
|
#: default; the rest come back as text.
|
|
FORMATS = (
|
|
"json", "tle", "tle_alpha5", "omm_xml", "csv", "kvn",
|
|
# CCSDS SANA element sets, propagated then converted, served as CSV.
|
|
"cartpv", "keplerian", "keplerianmean", "equinoctial", "geodetic", "adbarv",
|
|
)
|
|
|
|
#: How a query selects elements in time.
|
|
MODES = ("latest", "at", "range")
|
|
|
|
#: Filters ``query()`` forwards. Anything else raises rather than being silently
|
|
#: dropped -- a typo'd filter that quietly returns the whole catalogue is a much
|
|
#: worse failure than an exception.
|
|
FILTERS = (
|
|
"q", "norad", "cospar", "internal_id", "name", "source", "limit",
|
|
"mode", "datetime", "start", "end",
|
|
)
|
|
|
|
|
|
class OdmsClient:
|
|
"""Talks to an ODMS deployment.
|
|
|
|
``base_url`` is the site root (``https://odms.tmtc.yksa.space``), not the API
|
|
prefix -- the paths are this client's business.
|
|
|
|
``token`` is optional. Without one you see the public catalogue and the
|
|
read-only endpoints, which is all most consumers need. With one you also see
|
|
satellites flagged non-public, and you can reach the gated endpoints: OEM
|
|
generation, message push, frame transforms, TLE fitting and bulk propagation.
|
|
|
|
Safe to share between threads when httpx is installed (its client is
|
|
thread-safe) and when it is not (urllib opens a connection per call). Not
|
|
safe to share across a fork -- build one per process.
|
|
|
|
::
|
|
|
|
with OdmsClient("https://odms.tmtc.yksa.space", token=TOKEN) as odms:
|
|
iss = odms.latest("iss")
|
|
oem = odms.oem("iss", start, stop, step_s=60)
|
|
"""
|
|
|
|
#: API prefix, and the element-query path under it. Both are attributes so a
|
|
#: deployment mounted elsewhere, or one still serving the older
|
|
#: ``tle/query/`` alias, can be reached without subclassing anything.
|
|
api_prefix = "api/v1"
|
|
query_path = "omm/query/"
|
|
|
|
def __init__(self, base_url: str, *, token: str | None = None,
|
|
timeout_s: float = DEFAULT_TIMEOUT_S,
|
|
max_retries: int = DEFAULT_MAX_RETRIES,
|
|
user_agent: str = "odms-client/1.0",
|
|
session: Any = None):
|
|
self._transport = Transport(
|
|
base_url, token=token, timeout_s=timeout_s, max_retries=max_retries,
|
|
user_agent=user_agent, session=session,
|
|
)
|
|
|
|
# -- lifecycle ----------------------------------------------------------
|
|
|
|
def close(self) -> None:
|
|
self._transport.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
self.close()
|
|
|
|
# -- elements -----------------------------------------------------------
|
|
|
|
def query(self, *, format: str = "json", **filters) -> Any:
|
|
"""Search the element catalogue. The one endpoint behind every lookup.
|
|
|
|
Filters select *which* elements; ``format`` selects how they come back.
|
|
``json`` (the default) returns ``{"count": n, "results": [...]}``;
|
|
anything else returns the rendered body as text.
|
|
|
|
``mode`` is ``latest`` (newest per satellite, by epoch not by download
|
|
time), ``at`` (with ``datetime=``: the newest element at or before it),
|
|
or ``range`` (with ``start=``/``end=``).
|
|
"""
|
|
params = self._filters(filters)
|
|
params["format"] = _one_of("format", format, FORMATS)
|
|
response = self._transport.request(
|
|
"GET", f"{self.api_prefix}/{self.query_path}", params=params,
|
|
)
|
|
return response.json() if format == "json" else response.text
|
|
|
|
def latest(self, key: str, **filters) -> dict | None:
|
|
"""The newest element for one satellite, or ``None`` if it has none.
|
|
|
|
``key`` is whatever identifies the object: internal id, NORAD number or
|
|
COSPAR designator. It is matched the same way ``satellite()`` matches.
|
|
"""
|
|
results = self.query(q=key, mode="latest", limit=1, **filters)["results"]
|
|
return results[0] if results else None
|
|
|
|
def at(self, key: str, when, **filters) -> dict | None:
|
|
"""The element in force at ``when``: newest with ``epoch <= when``."""
|
|
results = self.query(
|
|
q=key, mode="at", datetime=when, limit=1, **filters,
|
|
)["results"]
|
|
return results[0] if results else None
|
|
|
|
def history(self, key: str, start, end, *, limit: int = 100, **filters) -> list[dict]:
|
|
"""Every element for ``key`` between ``start`` and ``end``, oldest first.
|
|
|
|
``limit`` is capped at 500 server-side. This does not paginate for you:
|
|
a caller pulling a long history should walk it in windows, so that the
|
|
memory and the request count stay its own decision.
|
|
"""
|
|
return self.query(
|
|
q=key, mode="range", start=start, end=end, limit=limit, **filters,
|
|
)["results"]
|
|
|
|
def download(self, *, format: str, filename: str | None = None, **filters) -> str:
|
|
"""A rendered element document -- TLE text, OMM XML, CSV, KVN.
|
|
|
|
``filename`` asks ODMS to serve it as an attachment. It changes only the
|
|
Content-Disposition, so it matters when you are proxying the response to
|
|
a browser and not otherwise.
|
|
"""
|
|
params = self._filters(filters)
|
|
params["format"] = _one_of("format", format, FORMATS)
|
|
params["filename"] = filename
|
|
return self._transport.request(
|
|
"GET", f"{self.api_prefix}/{self.query_path}", params=params,
|
|
).text
|
|
|
|
# -- catalogue ----------------------------------------------------------
|
|
|
|
def satellites(self) -> list[dict]:
|
|
"""Tracked satellites. Anonymous sees public ones; a token sees all."""
|
|
return self._get("satellites/")["results"]
|
|
|
|
def satellite(self, key: str) -> dict:
|
|
"""One satellite, by internal id, NORAD number or COSPAR designator.
|
|
|
|
Raises :class:`~odms.errors.NotFound` for an unknown object *and* for a
|
|
non-public one seen anonymously -- indistinguishable on purpose.
|
|
"""
|
|
return self._get(f"satellites/{key}/")
|
|
|
|
def find(self, key: str) -> dict | None:
|
|
""":meth:`satellite`, but ``None`` instead of raising when absent."""
|
|
try:
|
|
return self.satellite(key)
|
|
except NotFound:
|
|
return None
|
|
|
|
def sources(self) -> list[dict]:
|
|
"""Configured public data sources, with last-run status and counters."""
|
|
return self._get("sources/")["results"]
|
|
|
|
# -- CCSDS messages -----------------------------------------------------
|
|
|
|
def opm(self, key: str, *, at=None, format: str = "kvn",
|
|
ref_frame: str | None = None) -> str:
|
|
"""Generate an OPM: single-epoch state plus osculating Keplerian.
|
|
|
|
``ref_frame`` defaults to TEME. A non-inertial frame (ITRF) yields a
|
|
Cartesian-only OPM, because osculating elements are inertial-only.
|
|
Frames beyond TEME/ITRF need ODMS on its Orekit backend.
|
|
"""
|
|
return self._get_text("orbital/opm/", {
|
|
"key": key, "at": at, "format": format, "ref_frame": ref_frame,
|
|
})
|
|
|
|
def oem(self, key: str, start, stop, *, step_s: int = 60,
|
|
format: str = "kvn", ref_frame: str | None = None) -> str:
|
|
"""Generate an OEM over a span. **Token required.**
|
|
|
|
Gated because it is the expensive one: the span is capped server-side
|
|
(7 days, 10 000 points by default) and exceeding either is a 400, not a
|
|
truncation.
|
|
"""
|
|
return self._get_text("orbital/oem/", {
|
|
"key": key, "start": start, "stop": stop, "step": step_s,
|
|
"format": format, "ref_frame": ref_frame,
|
|
})
|
|
|
|
def element_sets(self, key: str, *, at=None) -> dict:
|
|
"""Every CCSDS element set for one satellite at one instant."""
|
|
return self._get("orbital/elements/", {"key": key, "at": at})
|
|
|
|
def state(self, key: str, frame: str, *, at=None) -> dict:
|
|
"""Cartesian state in ``frame``. Non-TEME/ITRF needs the Orekit backend."""
|
|
return self._get("orbital/frames/", {"key": key, "frame": frame, "at": at})
|
|
|
|
def messages(self, key: str, *, limit: int = 50, offset: int = 0) -> dict:
|
|
"""Stored OPM/OEM for a satellite, newest first. Returns the page dict,
|
|
with its ``count``, so a caller can decide whether to walk further."""
|
|
return self._get("orbital/messages/", {
|
|
"key": key, "limit": limit, "offset": offset,
|
|
})
|
|
|
|
def message(self, message_id: int) -> dict:
|
|
"""One stored message: metadata, parsed states and the raw text."""
|
|
return self._get(f"orbital/messages/{message_id}/")
|
|
|
|
def push_message(self, text: str, *, key: str | None = None,
|
|
is_public: bool | None = None) -> dict:
|
|
"""Store an externally produced OPM/OEM. **Token required.**
|
|
|
|
With ``key`` omitted the satellite is resolved from the message's own
|
|
``OBJECT_ID``.
|
|
|
|
Not idempotent, and treated as such: a push that fails in transport is
|
|
never retried automatically, because it may well have been stored. Retry
|
|
it yourself only once you have checked :meth:`messages`.
|
|
"""
|
|
payload: dict = {"text": text}
|
|
if key is not None:
|
|
payload["key"] = key
|
|
if is_public is not None:
|
|
payload["is_public"] = is_public
|
|
return self._post("orbital/messages/push/", payload)
|
|
|
|
def transform(self, target_frame: str, *, text: str | None = None,
|
|
message_id: int | None = None, key: str | None = None,
|
|
format: str = "kvn") -> str:
|
|
"""Re-express an OPM/OEM in another frame. **Token required.**
|
|
|
|
The source is exactly one of: inline ``text``, a stored ``message_id``,
|
|
or ``key`` for that satellite's latest stored message.
|
|
"""
|
|
payload = _exactly_one(
|
|
{"text": text, "message_id": message_id, "key": key},
|
|
"transform() needs one of text=, message_id= or key=",
|
|
)
|
|
payload.update({"target_frame": target_frame, "format": format})
|
|
return self._transport.request(
|
|
"POST", f"{self.api_prefix}/orbital/transform/", json=payload,
|
|
).text
|
|
|
|
def tle_from_oem(self, *, message_id: int | None = None,
|
|
key: str | None = None, text: str | None = None) -> dict:
|
|
"""Least-squares fit a TLE to an OEM's states. **Token required.**
|
|
|
|
Returns the fitted lines with the fit's ``rms`` and ``sample_count`` --
|
|
check them. A fit that converged on a bad ephemeris still returns lines.
|
|
"""
|
|
payload = _exactly_one(
|
|
{"text": text, "message_id": message_id, "key": key},
|
|
"tle_from_oem() needs one of text=, message_id= or key=",
|
|
)
|
|
return self._post("orbital/tle-from-oem/", payload)
|
|
|
|
def propagate(self, key: str, timestamps: Iterable, *,
|
|
frame: str = "TEME") -> dict:
|
|
"""Propagate one satellite to many instants. **Token required.**
|
|
|
|
Each point uses the element whose epoch is *closest* to it, so this is
|
|
the right call for a long historical span and the wrong one if you need
|
|
every point to come from a single element set.
|
|
|
|
Capped server-side (5000 timestamps by default). A point that fails
|
|
carries an ``error`` instead of vectors rather than failing the batch.
|
|
"""
|
|
return self._post("orbital/propagate/", {
|
|
"key": key,
|
|
"timestamps": [_iso(t) for t in timestamps],
|
|
"frame": frame,
|
|
})
|
|
|
|
# -- decay forecasts ----------------------------------------------------
|
|
|
|
def decay_runs(self, key: str, *, kind: str | None = None) -> list[dict]:
|
|
"""Forecast summaries for a satellite, without their trajectories.
|
|
|
|
``kind`` is ``latest`` (the rolling forecast, one per satellite) or
|
|
``monthly`` (frozen snapshots anchored to the 1st, kept so past
|
|
forecasts can be read against what the elements actually did).
|
|
"""
|
|
return self._get("decay/runs/", {"key": key, "kind": kind})["results"]
|
|
|
|
def decay_run(self, run_id: int) -> dict:
|
|
"""One run with its full-resolution trajectory under ``series.runs[]``."""
|
|
return self._get(f"decay/runs/{run_id}/")
|
|
|
|
def decay_latest(self, key: str) -> dict | None:
|
|
"""The current rolling forecast, or ``None`` if there is not one yet.
|
|
|
|
No forecast is a normal state, not an error: an object with neither
|
|
recorded physical properties nor a fittable element history is skipped
|
|
deliberately rather than predicted from a coefficient nobody can justify.
|
|
"""
|
|
try:
|
|
return self._get("decay/latest/", {"key": key})
|
|
except NotFound:
|
|
return None
|
|
|
|
# -- internals ----------------------------------------------------------
|
|
|
|
def _get(self, path: str, params: dict | None = None) -> Any:
|
|
return self._transport.request(
|
|
"GET", f"{self.api_prefix}/{path}", params=params,
|
|
).json()
|
|
|
|
def _get_text(self, path: str, params: dict | None = None) -> str:
|
|
return self._transport.request(
|
|
"GET", f"{self.api_prefix}/{path}", params=params,
|
|
).text
|
|
|
|
def _post(self, path: str, payload: Any) -> Any:
|
|
return self._transport.request(
|
|
"POST", f"{self.api_prefix}/{path}", json=payload,
|
|
).json()
|
|
|
|
@staticmethod
|
|
def _filters(filters: dict) -> dict:
|
|
unknown = sorted(set(filters) - set(FILTERS))
|
|
if unknown:
|
|
raise TypeError(
|
|
f"unknown filter(s) {', '.join(unknown)}; "
|
|
f"expected any of {', '.join(FILTERS)}"
|
|
)
|
|
if "mode" in filters:
|
|
_one_of("mode", filters["mode"], MODES)
|
|
return dict(filters)
|
|
|
|
|
|
def _one_of(name: str, value: str, allowed) -> str:
|
|
if value not in allowed:
|
|
raise ValueError(f"{name}={value!r}; expected one of {', '.join(allowed)}")
|
|
return value
|
|
|
|
|
|
def _exactly_one(candidates: dict, message: str) -> dict:
|
|
given = {k: v for k, v in candidates.items() if v is not None}
|
|
if len(given) != 1:
|
|
raise TypeError(message)
|
|
return given
|
|
|
|
|
|
def _iso(value) -> str:
|
|
return value.isoformat() if hasattr(value, "isoformat") else str(value)
|