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).
This commit is contained in:
Sirius DevOps
parent
47c726d68d
commit
6b5eec3824
5 changed files with 526 additions and 0 deletions
|
|
@ -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)",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
30
app/main.py
30
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"),
|
||||
|
|
|
|||
|
|
@ -13,3 +13,4 @@ structlog>=24.4
|
|||
websockets>=14
|
||||
cachetools>=5.5
|
||||
h3>=4.0
|
||||
sgp4>=2.23
|
||||
|
|
|
|||
289
app/satellites.py
Normal file
289
app/satellites.py
Normal file
|
|
@ -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(),
|
||||
}
|
||||
200
tests/test_satellites.py
Normal file
200
tests/test_satellites.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue