Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-06-18 18:39:47 -07:00
Code Issues Packages Projects Releases Wiki Activity

refactor: satellite routes read/write TLEs via unified store

Browse Source 
data/satellites.py is no longer rewritten at runtime; it remains as
the read-only seed for the store.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
James Smith
2026-06-11 17:12:23 +01:00
parent 74d5663f73
commit 1c72e15c7c
2 changed files with 51 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
routes/satellite.py
+44 -65
View File
@@ -12,7 +12,7 @@ import requests
from flask import Blueprint, Response, jsonify, make_response, render_template, request
from config import SHARED_OBSERVER_LOCATION_ENABLED
from data.satellites import TLE_SATELLITES
from utils import tle_store
from utils.database import (
add_tracked_satellite,
bulk_add_tracked_satellites,
@@ -47,8 +47,11 @@ MAX_RESPONSE_SIZE = 1024 * 1024
# Allowed hosts for TLE fetching
ALLOWED_TLE_HOSTS = ["celestrak.org", "celestrak.com", "www.celestrak.org", "www.celestrak.com"]
# Local TLE cache (can be updated via API)
_tle_cache = dict(TLE_SATELLITES)
def _get_tle_cache() -> dict:
"""All TLEs from the unified store."""
return tle_store.all_tles()
# Ground track cache: key=(sat_name, tle_line1[:20]) -> (track_data, computed_at_timestamp)
# TTL is 1800 seconds (30 minutes)
@@ -72,27 +75,26 @@ _BUILTIN_NORAD_TO_KEY = {
def _load_db_satellites_into_cache():
"""Load user-tracked satellites from DB into the TLE cache."""
global _tle_cache
"""Load user-tracked satellites from DB into the TLE store."""
try:
db_sats = get_tracked_satellites()
loaded = 0
new_entries: dict = {}
current = _get_tle_cache()
for sat in db_sats:
if sat["tle_line1"] and sat["tle_line2"]:
# Use a cache key derived from name (sanitised)
cache_key = sat["name"].replace(" ", "-").upper()
if cache_key not in _tle_cache:
_tle_cache[cache_key] = (sat["name"], sat["tle_line1"], sat["tle_line2"])
loaded += 1
if loaded:
logger.info(f"Loaded {loaded} user-tracked satellites into TLE cache")
if cache_key not in current:
new_entries[cache_key] = (sat["name"], sat["tle_line1"], sat["tle_line2"])
if new_entries:
tle_store.update(new_entries)
logger.info(f"Loaded {len(new_entries)} user-tracked satellites into TLE store")
except Exception as e:
logger.warning(f"Failed to load DB satellites into TLE cache: {e}")
logger.warning(f"Failed to load DB satellites into TLE store: {e}")
def get_cached_tle(name: str) -> tuple[str, str, str] | None:
"""Return (name, line1, line2) from the live TLE cache, or None if not found."""
return _tle_cache.get(name)
"""Return (name, line1, line2) from the live TLE store, or None if not found."""
return _get_tle_cache().get(name)
def _normalize_satellite_name(value: object) -> str:
@@ -164,27 +166,28 @@ def _resolve_satellite_request(
]
)
tles = _get_tle_cache()
seen: set[str] = set()
for key in candidate_keys:
norm = _normalize_satellite_name(key)
if norm in seen:
continue
seen.add(norm)
if key in _tle_cache:
tle_data = _tle_cache[key]
if key in tles:
tle_data = tles[key]
break
if norm in _tle_cache:
tle_data = _tle_cache[norm]
if norm in tles:
tle_data = tles[norm]
break
if tle_data is None and tracked and tracked.get("tle_line1") and tracked.get("tle_line2"):
display_name = tracked.get("name") or sat_key or str(norad_id or "UNKNOWN")
tle_data = (display_name, tracked["tle_line1"], tracked["tle_line2"])
_tle_cache[_normalize_satellite_name(display_name)] = tle_data
tle_store.update({_normalize_satellite_name(display_name): tle_data})
if tle_data is None and sat_key:
normalized = _normalize_satellite_name(sat_key)
for key, value in _tle_cache.items():
for key, value in tles.items():
if key == normalized or _normalize_satellite_name(value[0]) == normalized:
tle_data = value
break
@@ -251,21 +254,20 @@ def _start_satellite_tracker():
tle1 = sat_rec.get("tle_line1")
tle2 = sat_rec.get("tle_line2")
if not tle1 or not tle2:
# Fall back to TLE cache. Try the builtin NORAD-ID key first
# Fall back to TLE store. Try the builtin NORAD-ID key first
# (e.g. 'ISS'), then the name-derived key as a last resort.
try:
norad_int = int(norad_id)
except (TypeError, ValueError):
norad_int = 0
tles = _get_tle_cache()
builtin_key = _BUILTIN_NORAD_TO_KEY.get(norad_int)
cache_key = (
builtin_key
if (builtin_key and builtin_key in _tle_cache)
else sat_name.replace(" ", "-").upper()
builtin_key if (builtin_key and builtin_key in tles) else sat_name.replace(" ", "-").upper()
)
if cache_key not in _tle_cache:
if cache_key not in tles:
continue
tle_entry = _tle_cache[cache_key]
tle_entry = tles[cache_key]
tle1 = tle_entry[1]
tle2 = tle_entry[2]
@@ -660,13 +662,14 @@ def get_satellite_position():
iss_data = _fetch_iss_realtime(lat, lon)
if iss_data:
# Add orbit track if requested (using TLE for track prediction)
if include_track and "ISS" in _tle_cache:
iss_tle = _get_tle_cache().get("ISS")
if include_track and iss_tle:
try:
if ts is None:
ts = _get_timescale()
now = ts.now()
now_dt = now.utc_datetime()
tle_data = _tle_cache["ISS"]
tle_data = iss_tle
satellite = EarthSatellite(tle_data[1], tle_data[2], tle_data[0], ts)
orbit_track = []
for minutes_offset in range(-45, 46, 1):
@@ -798,8 +801,6 @@ def refresh_tle_data() -> list:
This can be called at startup or periodically to keep TLE data fresh.
Returns list of satellite names that were updated.
"""
global _tle_cache
name_mappings = {
"ISS (ZARYA)": "ISS",
"NOAA 15": "NOAA-15",
@@ -813,6 +814,8 @@ def refresh_tle_data() -> list:
}
updated = []
new_entries: dict = {}
current = _get_tle_cache()
for group in ["stations", "weather", "noaa"]:
url = f"https://celestrak.org/NORAD/elements/gp.php?GROUP={group}&FORMAT=tle"
@@ -833,8 +836,8 @@ def refresh_tle_data() -> list:
internal_name = name_mappings.get(name, name)
if internal_name in _tle_cache:
_tle_cache[internal_name] = (name, line1, line2)
if internal_name in current:
new_entries[internal_name] = (name, line1, line2)
if internal_name not in updated:
updated.append(internal_name)
@@ -843,39 +846,12 @@ def refresh_tle_data() -> list:
logger.warning(f"Error fetching TLE group {group}: {e}")
continue
if updated:
_persist_tle_cache()
if new_entries:
tle_store.update(new_entries)
return updated
def _persist_tle_cache() -> None:
"""Write updated TLE data back to data/satellites.py so restarts don't reload stale values."""
import os
satellites_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "satellites.py")
try:
lines = [
"# TLE data for satellite tracking (updated periodically)\n",
'# To update: click "Update TLE" in satellite dashboard or SSTV mode\n',
"# Data source: CelesTrak (celestrak.org)\n",
"TLE_SATELLITES = {\n",
]
for key, val in _tle_cache.items():
name, line1, line2 = val
escaped_name = name.replace("'", "\\'")
escaped_key = key.replace("'", "\\'")
lines.append(f" '{escaped_key}': ('{escaped_name}',\n")
lines.append(f" '{line1}',\n")
lines.append(f" '{line2}'),\n")
lines.append("}\n")
with open(satellites_path, "w") as f:
f.writelines(lines)
logger.info(f"Persisted {len(_tle_cache)} TLE entries to data/satellites.py")
except Exception as e:
logger.warning(f"Failed to persist TLE cache to disk: {e}")
@satellite_bp.route("/update-tle", methods=["POST"])
def update_tle():
"""Update TLE data from CelesTrak (API endpoint)."""
@@ -966,7 +942,6 @@ def list_tracked_satellites():
@satellite_bp.route("/tracked", methods=["POST"])
def add_tracked_satellites_endpoint():
"""Add one or more tracked satellites."""
global _tle_cache
data = request.get_json(silent=True)
if not data:
return api_error("No data provided", 400)
@@ -975,6 +950,7 @@ def add_tracked_satellites_endpoint():
sat_list = data if isinstance(data, list) else [data]
normalized: list[dict] = []
new_tle_entries: dict = {}
for sat in sat_list:
norad_id = str(sat.get("norad_id", sat.get("norad", "")))
name = sat.get("name", "")
@@ -995,10 +971,13 @@ def add_tracked_satellites_endpoint():
}
)
# Also inject into TLE cache if we have TLE data
# Also inject into TLE store if we have TLE data
if tle1 and tle2:
cache_key = name.replace(" ", "-").upper()
_tle_cache[cache_key] = (name, tle1, tle2)
new_tle_entries[cache_key] = (name, tle1, tle2)
if new_tle_entries:
tle_store.update(new_tle_entries)
# Single inserts preserve previous behavior; list inserts use DB-level bulk path.
if len(normalized) == 1:
tests/test_satellite.py
+7 -13
View File
@@ -9,20 +9,14 @@ from routes.satellite import satellite_bp
@pytest.fixture(autouse=True)
def _isolate_tle_state(monkeypatch):
"""Keep TLE updates off the real data/satellites.py and reset the cache.
def _isolate_tle_state(tmp_path, monkeypatch):
"""Point the TLE store at a throwaway DB so tests never touch real data."""
from utils import tle_store
Without this, the update-tle test rewrites the tracked data file on
every run and leaks 'ISS' into the module-global cache, breaking
later tests that depend on cache contents.
"""
import routes.satellite as sat
monkeypatch.setattr(sat, "_persist_tle_cache", lambda: None)
saved = dict(sat._tle_cache)
monkeypatch.setattr(tle_store, "_DB_PATH", tmp_path / "tle.db")
tle_store._reset_for_tests()
yield
sat._tle_cache.clear()
sat._tle_cache.update(saved)
tle_store._reset_for_tests()
@pytest.fixture
@@ -119,7 +113,7 @@ def test_tracker_position_has_no_observer_fields():
tle_key = (ISS_TLE[0], ISS_TLE[1][:20])
stub_track = [{"lat": 0.0, "lon": float(i), "past": i < 45} for i in range(91)]
with (
patch("routes.satellite._tle_cache", {"ISS": ISS_TLE}),
patch("routes.satellite._get_tle_cache", return_value={"ISS": ISS_TLE}),
patch("routes.satellite.get_tracked_satellites") as mock_tracked,
patch("routes.satellite._track_cache", {tle_key: (stub_track, 1e18)}),
patch("routes.satellite._get_timescale", return_value=real_ts),