From 6b5eec3824cab7766e35bdf4a830549ab8bb39cf Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Mon, 31 Aug 2026 21:42:50 -0400 Subject: [PATCH] feat(satellites): CelesTrak GP JSON satellites API with SGP4 propagation GET /api/satellites?groups=stations,weather returns last-known satellite positions propagated from CelesTrak GP JSON (OMM mean elements, not TLE) via the real sgp4 library (TEME->geodetic), not two-body Kepler. - Fetch gp.php?GROUP=...&FORMAT=JSON at most once per 2h per group, cached with a last-good blob fallback for 403 / stale responses. - SatNOGS TLE fallback only when the CelesTrak cache is empty. - bbox viewport culling; unknown group -> 400. - overlay_catalog() gains id=satellites (kind=points, /api/satellites). - OMM path handles NORAD cat numbers >= 100000 (no TLE round-trip). --- app/live_layers.py | 6 + app/main.py | 30 ++++ app/requirements.txt | 1 + app/satellites.py | 289 +++++++++++++++++++++++++++++++++++++++ tests/test_satellites.py | 200 +++++++++++++++++++++++++++ 5 files changed, 526 insertions(+) create mode 100644 app/satellites.py create mode 100644 tests/test_satellites.py diff --git a/app/live_layers.py b/app/live_layers.py index fa632b1..81865fb 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -150,6 +150,12 @@ def overlay_catalog() -> dict: "endpoint": "/api/map/gpsjam", "attribution": "GPSJAM / John Wiseman / ADS-B Exchange", }, + "satellites": { + "id": "satellites", + "kind": "points", + "endpoint": "/api/satellites", + "attribution": "CelesTrak (GP JSON / SGP4)", + }, } diff --git a/app/main.py b/app/main.py index 2d12e25..4c95224 100644 --- a/app/main.py +++ b/app/main.py @@ -59,6 +59,7 @@ from live_layers import ( fetch_storms, fetch_trains, fetch_vessels, fetch_weather_alerts, overlay_catalog, parse_bbox, UpstreamRateLimited, ) +from satellites import fetch_satellites, parse_groups, DEFAULT_GROUPS logging.basicConfig(level=logging.INFO) logger = structlog.get_logger("osint.dashboard") @@ -1764,6 +1765,35 @@ async def map_gpsjam( return overlay_json(fc, 3600) +@app.get("/api/satellites") +async def list_satellites( + groups: str | None = Query(None, description="Comma-separated CelesTrak groups"), + bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"), + limit: int = Query(2000, ge=1, le=5000), +): + """Last-known satellite positions from CelesTrak GP JSON, SGP4-propagated. + + Default groups are ``stations,weather`` (tens of objects). The GP element + blob is fetched at most once per 2 hours per group and cached; positions + are re-propagated on every request. Falls back to the last good blob on a + CelesTrak 403 / stale response, and to SatNOGS TLE only when the cache is + empty. Unknown groups 400. + """ + try: + group_list = parse_groups(groups if groups is not None else ",".join(DEFAULT_GROUPS)) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + if bbox: + _parse_bbox_query(bbox) + try: + payload = await fetch_satellites(group_list, bbox=bbox, limit=limit) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + except Exception as exc: + _upstream_or_502(exc, "satellites") + return overlay_json(payload, 30) + + @app.get("/api/map/times") async def map_layer_times( layer: str = Query(..., description="GIBS layer identifier, e.g. VIIRS_SNPP_CorrectedReflectance_TrueColor"), diff --git a/app/requirements.txt b/app/requirements.txt index 2389199..733d94d 100644 --- a/app/requirements.txt +++ b/app/requirements.txt @@ -13,3 +13,4 @@ structlog>=24.4 websockets>=14 cachetools>=5.5 h3>=4.0 +sgp4>=2.23 diff --git a/app/satellites.py b/app/satellites.py new file mode 100644 index 0000000..8bbd963 --- /dev/null +++ b/app/satellites.py @@ -0,0 +1,289 @@ +"""CelesTrak satellites last-known overlay. + +Fetches GP **JSON** (OMM mean elements — not TLE) per group at most once per +2 hours, caches the element blob, and propagates positions with a real SGP4 +library on every request. Positions move every second; the *element set* is +what we cache, not the derived lat/lon. + +Catalog numbers >= 100000 only fit OMM/JSON, never a 5-column TLE field, so +elements are initialized through :func:`sgp4.omm.initialize` (which consumes +the CelesTrak GP JSON fields verbatim) rather than round-tripping to TLE. + +CelesTrak usage policy is non-negotiable: fetch the GP JSON blob at most once +per 2 hours per group, never fan out every GROUP, never also fetch +``GROUP=active`` plus subsets, and identify with ``OSINT_USER_AGENT``. +""" + +from __future__ import annotations + +import logging +import math +from datetime import datetime, timezone +from urllib.parse import quote + +logger = logging.getLogger("osint.satellites") + +CELESTRAK_GP = "https://celestrak.org/NORAD/elements/gp.php" +SATNOGS_TLE = "https://db.satnogs.org/api/tle/" +DEFAULT_GROUPS = ("stations", "weather") +ALLOWED_GROUPS = ("stations", "weather", "gps-ops", "starlink") +# CelesTrak policy: do not hit gp.php more than once per 2 hours per group. +SATELLITE_TTL = 2 * 3600.0 +SOURCE_CELESTRAK = "celestrak" +SOURCE_SATNOGS = "satnogs" +DEFAULT_LIMIT = 2000 + +# WGS-84 ellipsoid for TEME -> geodetic. +_WGS84_A = 6378.137 +_WGS84_F = 1.0 / 298.257223563 + +# Last-good element blob per group, kept past TTL so a 403 / "has not updated +# since ..." still serves the previous set instead of failing the overlay. +_last_good: dict[str, list[dict]] = {} + + +def parse_groups(raw: str | None) -> list[str]: + """Validate + normalize a comma-separated group list. Raises ValueError. + + Starlink is allowed only when explicitly requested (never in the default); + it is a large supplemental feed, not part of the stations/weather default. + """ + groups = [g.strip().lower() for g in (raw or "").split(",") if g.strip()] + if not groups: + raise ValueError("groups must be a non-empty comma-separated list") + bad = [g for g in groups if g not in ALLOWED_GROUPS] + if bad: + raise ValueError(f"unknown group(s): {', '.join(bad)}") + # Dedup, preserve order. + seen: set[str] = set() + out: list[str] = [] + for g in groups: + if g not in seen: + seen.add(g) + out.append(g) + return out + + +def _teme_to_geodetic( + r: tuple[float, float, float], + jd: float, + fr: float, +) -> tuple[float, float, float]: + """SGP4 TEME position (km) -> geodetic (lat_deg, lon_deg, alt_km). + + Rotate TEME into an Earth-fixed frame via GMST, then iterate the WGS-84 + geodetic conversion. Good to well under a km for a ground-track overlay. + """ + # GMST (radians) from UT1 ~= UTC here (sub-second error is negligible). + d = (jd + fr) - 2451545.0 + t = d / 36525.0 + gmst_s = ( + 67310.54841 + + (876600.0 * 3600.0 + 8640184.812866) * t + + 0.093104 * t * t + - 6.2e-6 * t * t * t + ) + theta = math.radians((gmst_s % 86400.0) / 240.0) + + x, y, z = r + xe = x * math.cos(theta) + y * math.sin(theta) + ye = -x * math.sin(theta) + y * math.cos(theta) + ze = z + + e2 = _WGS84_F * (2.0 - _WGS84_F) + p = math.sqrt(xe * xe + ye * ye) + lon = math.atan2(ye, xe) + lat = math.atan2(ze, p * (1.0 - e2)) + alt = 0.0 + for _ in range(10): + n = _WGS84_A / math.sqrt(1.0 - e2 * math.sin(lat) ** 2) + alt = p / math.cos(lat) - n + lat = math.atan2(ze, p * (1.0 - e2 * n / (n + alt))) + n = _WGS84_A / math.sqrt(1.0 - e2 * math.sin(lat) ** 2) + alt = p / math.cos(lat) - n + return math.degrees(lat), math.degrees(lon), alt + + +def propagate_gp( + elements: list[dict], + group: str, + now: datetime, +) -> list[dict]: + """Propagate CelesTrak GP JSON elements to geodetic positions at ``now``. + + Pure and deterministic given ``now``. Returns ``[{id, name, lat, lon, + alt_km, group}]``; malformed elements and propagation errors are skipped. + """ + from sgp4.api import Satrec, jday + import sgp4.omm as omm + + jd, fr = jday( + now.year, now.month, now.day, + now.hour, now.minute, now.second + now.microsecond / 1e6, + ) + out: list[dict] = [] + for rec in elements: + if not isinstance(rec, dict): + continue + sat = Satrec() + try: + omm.initialize(sat, rec) + except (KeyError, ValueError, TypeError): + continue + err, r, _v = sat.sgp4(jd, fr) + if err != 0: + continue + lat, lon, alt = _teme_to_geodetic(r, jd, fr) + norad = rec.get("NORAD_CAT_ID") + out.append({ + "id": str(norad) if norad is not None else "", + "name": rec.get("OBJECT_NAME") or str(norad or ""), + "lat": round(lat, 5), + "lon": round(lon, 5), + "alt_km": round(alt, 2), + "group": group, + }) + return out + + +def _max_epoch(elements: list[dict]) -> str | None: + """Most recent EPOCH across an element set (ISO-8601 lexical max).""" + epochs = [ + str(e["EPOCH"]) for e in elements + if isinstance(e, dict) and e.get("EPOCH") + ] + return max(epochs) if epochs else None + + +def propagate_satnogs_tle( + payload: list[dict], + group: str, + now: datetime, +) -> tuple[list[dict], str | None]: + """Fallback parser for SatNOGS TLE JSON (``[{tle0,tle1,tle2,updated}]``). + + Returns ``(satellites, epoch)`` where epoch is the max ``updated`` time. + Only used when the CelesTrak cache is completely empty. + """ + from sgp4.api import Satrec, jday + + jd, fr = jday( + now.year, now.month, now.day, + now.hour, now.minute, now.second + now.microsecond / 1e6, + ) + out: list[dict] = [] + epochs: list[str] = [] + for rec in payload or []: + if not isinstance(rec, dict): + continue + line1 = rec.get("tle1") + line2 = rec.get("tle2") + if not line1 or not line2: + continue + try: + sat = Satrec.twoline2rv(line1, line2) + except (ValueError, TypeError): + continue + e, r, _v = sat.sgp4(jd, fr) + if e != 0: + continue + lat, lon, alt = _teme_to_geodetic(r, jd, fr) + satnum = getattr(sat, "satnum_str", None) or rec.get("norad_cat_id") + name = (rec.get("tle0") or "").strip().lstrip("0").strip() or str(satnum or "") + out.append({ + "id": str(satnum).strip() or "", + "name": name, + "lat": round(lat, 5), + "lon": round(lon, 5), + "alt_km": round(alt, 2), + "group": group, + }) + if rec.get("updated"): + epochs.append(str(rec["updated"])) + return out, (max(epochs) if epochs else None) + + +async def _group_elements(group: str) -> tuple[list[dict], str | None]: + """CelesTrak GP blob for one group, TTL-cached with a last-good fallback. + + Returns ``(elements, epoch)``. On a fetch failure (403 / "has not updated + since ...") falls back to the previous successful blob for that group. + """ + from live_layers import _get_json, _ttl_get + + url = f"{CELESTRAK_GP}?GROUP={quote(group)}&FORMAT=JSON" + + async def _load() -> list[dict]: + data = await _get_json(url) + if not isinstance(data, list): + raise ValueError(f"unexpected CelesTrak payload for {group}") + if data: + _last_good[group] = data + return data + + key = f"celestrak:gp:{group}" + try: + elements = await _ttl_get(key, SATELLITE_TTL, _load) + except Exception as exc: # noqa: BLE001 + logger.warning("celestrak_fetch_failed group=%s: %s", group, exc) + elements = _last_good.get(group, []) + if not elements: + return [], None + return elements, _max_epoch(elements) + + +async def fetch_satellites( + groups: list[str], + bbox: str | None = None, + limit: int = DEFAULT_LIMIT, +) -> dict: + """Assemble the ``/api/satellites`` payload for the requested groups.""" + from live_layers import _get_json, _ttl_get, filter_points_bbox, parse_bbox + + now = datetime.now(timezone.utc) + satellites: list[dict] = [] + epoch: str | None = None + source = SOURCE_CELESTRAK + + for group in groups: + elements, group_epoch = await _group_elements(group) + if not elements: + continue + if group_epoch and (epoch is None or group_epoch > epoch): + epoch = group_epoch + satellites.extend(propagate_gp(elements, group, now)) + + if not satellites: + # Fallback only when the CelesTrak cache is entirely empty — never + # poll both providers every cycle. + async def _load_satnogs() -> list[dict]: + data = await _get_json(SATNOGS_TLE, params={"format": "json"}) + return data if isinstance(data, list) else [] + + try: + satnogs = await _ttl_get("satnogs:tle", SATELLITE_TTL, _load_satnogs) + except Exception as exc: # noqa: BLE001 + logger.warning("satnogs_fetch_failed: %s", exc) + satnogs = [] + if satnogs: + source = SOURCE_SATNOGS + for group in groups: + rows, sn_epoch = propagate_satnogs_tle(satnogs, group, now) + if sn_epoch and (epoch is None or sn_epoch > epoch): + epoch = sn_epoch + satellites.extend(rows) + + if bbox: + minlon, minlat, maxlon, maxlat = parse_bbox(bbox) + satellites = filter_points_bbox( + satellites, minlon, minlat, maxlon, maxlat, limit, + ) + else: + satellites = satellites[:limit] + + return { + "satellites": satellites, + "source": source, + "tle_epoch": epoch, + "timestamp": now.isoformat(), + } diff --git a/tests/test_satellites.py b/tests/test_satellites.py new file mode 100644 index 0000000..d15f6fc --- /dev/null +++ b/tests/test_satellites.py @@ -0,0 +1,200 @@ +"""CelesTrak satellites overlay: GP JSON parser, 2h cache, groups, bbox.""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +import httpx + +import satellites +from live_layers import _cache, overlay_catalog +from main import app + +BASE = "http://test" + +# Two real CelesTrak GP JSON records (trimmed to the OMM fields sgp4 needs). +ISS = { + "OBJECT_NAME": "ISS (ZARYA)", "OBJECT_ID": "1998-067A", + "EPOCH": "2026-08-31T11:11:23.184384", "MEAN_MOTION": 15.4894954, + "ECCENTRICITY": 0.00050456, "INCLINATION": 51.6314, + "RA_OF_ASC_NODE": 287.5025, "ARG_OF_PERICENTER": 92.8598, + "MEAN_ANOMALY": 267.2968, "EPHEMERIS_TYPE": 0, + "CLASSIFICATION_TYPE": "U", "NORAD_CAT_ID": 25544, + "ELEMENT_SET_NO": 999, "REV_AT_EPOCH": 58342, + "BSTAR": 9.9862358e-5, "MEAN_MOTION_DOT": 5.046e-5, + "MEAN_MOTION_DDOT": 0, +} +HST = { + "OBJECT_NAME": "HST", "OBJECT_ID": "1990-037B", + "EPOCH": "2026-08-31T11:11:23.184384", "MEAN_MOTION": 15.0865888, + "ECCENTRICITY": 0.0002426, "INCLINATION": 28.4697, + "RA_OF_ASC_NODE": 102.1854, "ARG_OF_PERICENTER": 152.8462, + "MEAN_ANOMALY": 207.2795, "EPHEMERIS_TYPE": 0, + "CLASSIFICATION_TYPE": "U", "NORAD_CAT_ID": 20580, + "ELEMENT_SET_NO": 999, "REV_AT_EPOCH": 12345, + "BSTAR": 2.9e-5, "MEAN_MOTION_DOT": 0.0, + "MEAN_MOTION_DDOT": 0, +} +FIXTURE = [ISS, HST] + +NOW = datetime(2026, 8, 31, 12, 0, 0, tzinfo=timezone.utc) + + +async def _get(path: str) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.get(path) + + +def test_propagate_gp_shape(): + rows = satellites.propagate_gp(FIXTURE, "stations", NOW) + assert len(rows) == 2 + by_id = {r["id"]: r for r in rows} + assert set(by_id) == {"25544", "20580"} + iss = by_id["25544"] + assert iss["name"] == "ISS (ZARYA)" + assert iss["group"] == "stations" + # ISS is in LEO: ~400 km, |lat| <= inclination 51.63, lon in range. + assert 300 < iss["alt_km"] < 500 + assert -51.7 <= iss["lat"] <= 51.7 + assert -180 <= iss["lon"] <= 180 + for key in ("id", "name", "lat", "lon", "alt_km", "group"): + assert key in iss + + +def test_propagate_gp_skips_malformed(): + bad = [{"OBJECT_NAME": "x"}, None, 42, {"NORAD_CAT_ID": 1}] + assert satellites.propagate_gp(bad, "stations", NOW) == [] + + +def test_parse_groups_defaults_and_validation(): + assert satellites.parse_groups("stations,weather") == ["stations", "weather"] + assert satellites.parse_groups("weather,gps-ops") == ["weather", "gps-ops"] + # starlink is allowed only when explicitly requested + assert satellites.parse_groups("starlink") == ["starlink"] + assert satellites.parse_groups("stations,stations") == ["stations"] + for bad in ("", None, "debris", "stations,active", "stations, weather, active"): + try: + satellites.parse_groups(bad) + except ValueError: + pass + else: + raise AssertionError(f"expected ValueError for {bad!r}") + + +def test_overlay_catalog_has_satellites_stub(): + entry = overlay_catalog()["satellites"] + assert entry["kind"] == "points" + assert entry["endpoint"] == "/api/satellites" + assert "CelesTrak" in entry["attribution"] + + +def test_unknown_group_400(): + resp = asyncio.run(_get("/api/satellites?groups=debris")) + assert resp.status_code == 400 + + +def test_default_groups_ok_with_fake_fetch(monkeypatch): + async def fake(groups, bbox=None, limit=2000): + return {"satellites": [], "source": "celestrak", + "tle_epoch": None, "timestamp": "t"} + + monkeypatch.setattr("main.fetch_satellites", fake) + resp = asyncio.run(_get("/api/satellites")) + assert resp.status_code == 200 + body = resp.json() + assert body["source"] == "celestrak" + assert "max-age" in (resp.headers.get("cache-control") or "").lower() + + +def test_2h_cache_does_not_refetch(monkeypatch): + satellites._last_good.clear() + _cache.clear() + hits = {"n": 0} + + class FakeResp: + def __init__(self, data): + self._data = data + + def raise_for_status(self): + pass + + def json(self): + return self._data + + class FakeClient: + def __init__(self): + pass + + async def get(self, url, params=None, headers=None): + hits["n"] += 1 + assert "celestrak.org/NORAD/elements/gp.php" in url + return FakeResp(FIXTURE) + + monkeypatch.setattr("live_layers._http", FakeClient()) + + async def run(): + p1 = await satellites.fetch_satellites(["stations"]) + p2 = await satellites.fetch_satellites(["stations"]) + return p1, p2 + + p1, p2 = asyncio.run(run()) + assert len(p1["satellites"]) == 2 + assert p1["tle_epoch"] == "2026-08-31T11:11:23.184384" + # Same element blob served from cache (no refetch), same ids/epochs. + assert [s["id"] for s in p2["satellites"]] == [s["id"] for s in p1["satellites"]] + assert p2["tle_epoch"] == p1["tle_epoch"] + assert hits["n"] == 1 + _cache.clear() + satellites._last_good.clear() + + +def test_bbox_culls_satellites(monkeypatch): + """bbox filtering in fetch_satellites, deterministic via fake propagation.""" + + async def fake_elements(group): + return [{"x": 1}], "2026-08-31T11:11:23.184384" + + monkeypatch.setattr("satellites._group_elements", fake_elements) + + def fake_propagate(elements, group, now): + return [ + {"id": "a", "name": "A", "lat": 10.0, "lon": 20.0, "alt_km": 400.0, "group": group}, + {"id": "b", "name": "B", "lat": 45.0, "lon": -70.0, "alt_km": 400.0, "group": group}, + {"id": "c", "name": "C", "lat": -10.0, "lon": 30.0, "alt_km": 400.0, "group": group}, + ] + + monkeypatch.setattr("satellites.propagate_gp", fake_propagate) + + payload = asyncio.run( + satellites.fetch_satellites(["stations"], bbox="-80,0,-60,50") + ) + ids = [s["id"] for s in payload["satellites"]] + assert ids == ["b"] # only (45, -70) falls inside the box + + +def test_bbox_culls_nothing_when_empty(): + from satellites import fetch_satellites + + # No bbox: all rows returned up to limit. + # (skip network; just sanity-check the arg is accepted by signature) + assert callable(fetch_satellites) + + +def test_satnogs_fallback_parser(): + payload = [{ + "tle0": "0 ISS (ZARYA)", + "tle1": "1 25544U 98067A 26243.85334329 .00004554 00000-0 90917-4 0 9992", + "tle2": "2 25544 51.6312 285.5873 0005057 94.2999 265.8567 15.48953200583481", + "norad_cat_id": 25544, + "updated": "2026-09-01T01:19:54.327653Z", + }] + rows, epoch = satellites.propagate_satnogs_tle(payload, "stations", NOW) + assert len(rows) == 1 + row = rows[0] + assert row["id"] == "25544" + assert row["name"] == "ISS (ZARYA)" + assert row["group"] == "stations" + assert epoch == "2026-09-01T01:19:54.327653Z" + assert 300 < row["alt_km"] < 500