Merge pull request 'feat(satellites): CelesTrak GP JSON satellites API with SGP4 propagation' (#43) from osint-dashboard/t_fe710941-osint-celestrak-get-api-satellites into master

Reviewed-on: #43

app/live_layers.py

@@ -155,6 +155,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)",
+        },
         "infra_nuclear": {
             "id": "infra_nuclear",
             "kind": "points",

app/main.py

@@ -61,6 +61,7 @@ from live_layers import (
     fetch_storms, fetch_trains, fetch_vessels, fetch_weather_alerts,
     fetch_infrastructure, overlay_catalog, parse_bbox, UpstreamRateLimited,
 )
+from satellites import fetch_satellites, parse_groups, DEFAULT_GROUPS
 from place import reverse_geocode
 
 logging.basicConfig(level=logging.INFO)
@@ -1899,6 +1900,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)
+
+
 _INFRA_TYPES = frozenset({"nuclear"})
 
 

app/requirements.txt

@@ -13,3 +13,4 @@ structlog>=24.4
 websockets>=14
 cachetools>=5.5
 h3>=4.0
+sgp4>=2.23

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(),
+    }

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