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