From f4a9cb7da69997a4506ffc8c19300f7bc5bdb779 Mon Sep 17 00:00:00 2001 From: James Smith Date: Thu, 11 Jun 2026 16:58:09 +0100 Subject: [PATCH] feat: add unified SQLite-backed TLE store Co-Authored-By: Claude Fable 5 --- tests/test_tle_store.py | 48 +++++++++++++++++ utils/tle_store.py | 111 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 159 insertions(+) create mode 100644 tests/test_tle_store.py create mode 100644 utils/tle_store.py diff --git a/tests/test_tle_store.py b/tests/test_tle_store.py new file mode 100644 index 0000000..f56a9ad --- /dev/null +++ b/tests/test_tle_store.py @@ -0,0 +1,48 @@ +"""Tests for the unified TLE store.""" + +import pytest + +from utils import tle_store + + +@pytest.fixture(autouse=True) +def _fresh_db(tmp_path, monkeypatch): + """Point the store at a throwaway database file.""" + monkeypatch.setattr(tle_store, "_DB_PATH", tmp_path / "tle.db") + tle_store._reset_for_tests() + + +SAMPLE = ( + "ISS (ZARYA)", + "1 25544U 98067A 23321.52083333 .00016717 00000-0 30171-3 0 9992", + "2 25544 51.6416 20.4567 0004561 45.3212 67.8912 15.49876543123456", +) + + +class TestTLEStore: + def test_seed_from_static_data(self): + """First access seeds from data/satellites.py TLE_SATELLITES.""" + tles = tle_store.all_tles() + assert "ISS" in tles + name, l1, l2 = tles["ISS"] + assert l1.startswith("1 ") + assert l2.startswith("2 ") + + def test_update_and_get(self): + tle_store.update({"TEST-SAT": SAMPLE}) + assert tle_store.get_tle("TEST-SAT") == SAMPLE + + def test_get_missing_returns_none(self): + assert tle_store.get_tle("NO-SUCH-SAT") is None + + def test_update_overwrites(self): + tle_store.update({"TEST-SAT": SAMPLE}) + newer = (SAMPLE[0], SAMPLE[1].replace("23321", "26100"), SAMPLE[2]) + tle_store.update({"TEST-SAT": newer}) + assert tle_store.get_tle("TEST-SAT") == newer + + def test_persists_across_reset(self): + """Data survives a cache reset (i.e., it actually hit the database).""" + tle_store.update({"TEST-SAT": SAMPLE}) + tle_store._reset_for_tests() + assert tle_store.get_tle("TEST-SAT") == SAMPLE diff --git a/utils/tle_store.py b/utils/tle_store.py new file mode 100644 index 0000000..7bd99dc --- /dev/null +++ b/utils/tle_store.py @@ -0,0 +1,111 @@ +"""Unified TLE store. + +Single source of truth for TLE data, shared by satellite tracking, +weather-satellite prediction, SSTV doppler, and the remote agent. +Backed by SQLite; seeded once from the static data/satellites.py. + +Replaces three previous stores: routes/satellite._tle_cache (which +persisted by rewriting data/satellites.py at runtime), +utils/weather_sat_predict._tle_cache, and the agent's own download. +""" + +import sqlite3 +import threading +from pathlib import Path + +from utils.logging import get_logger + +logger = get_logger("intercept.tle_store") + +_DB_PATH = Path(__file__).resolve().parent.parent / "instance" / "tle.db" +_lock = threading.Lock() +_cache: dict[str, tuple[str, str, str]] | None = None + + +def _connect() -> sqlite3.Connection: + _DB_PATH.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(_DB_PATH), check_same_thread=False) + conn.execute("PRAGMA journal_mode = WAL") + conn.execute( + """ + CREATE TABLE IF NOT EXISTS tle_entries ( + key TEXT PRIMARY KEY, + name TEXT NOT NULL, + line1 TEXT NOT NULL, + line2 TEXT NOT NULL, + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ) + """ + ) + return conn + + +def _seed_if_empty(conn: sqlite3.Connection) -> None: + count = conn.execute("SELECT COUNT(*) FROM tle_entries").fetchone()[0] + if count: + return + try: + from data.satellites import TLE_SATELLITES + except ImportError: + logger.warning("data/satellites.py unavailable; TLE store starts empty") + return + conn.executemany( + "INSERT OR REPLACE INTO tle_entries (key, name, line1, line2) VALUES (?, ?, ?, ?)", + [(key, name, l1, l2) for key, (name, l1, l2) in TLE_SATELLITES.items()], + ) + conn.commit() + logger.info(f"Seeded TLE store with {len(TLE_SATELLITES)} entries") + + +def _load() -> dict[str, tuple[str, str, str]]: + global _cache + if _cache is None: + with _lock: + if _cache is None: + conn = _connect() + try: + _seed_if_empty(conn) + rows = conn.execute("SELECT key, name, line1, line2 FROM tle_entries").fetchall() + _cache = {key: (name, l1, l2) for key, name, l1, l2 in rows} + finally: + conn.close() + return _cache + + +def all_tles() -> dict[str, tuple[str, str, str]]: + """Return all TLEs as {key: (name, line1, line2)}.""" + return dict(_load()) + + +def get_tle(key: str) -> tuple[str, str, str] | None: + """Return (name, line1, line2) for a satellite key, or None.""" + return _load().get(key) + + +def update(entries: dict[str, tuple[str, str, str]]) -> None: + """Insert or replace TLE entries and refresh the cache.""" + if not entries: + return + with _lock: + conn = _connect() + try: + conn.executemany( + "INSERT OR REPLACE INTO tle_entries (key, name, line1, line2, updated_at)" + " VALUES (?, ?, ?, ?, datetime('now'))", + [(key, name, l1, l2) for key, (name, l1, l2) in entries.items()], + ) + conn.commit() + finally: + conn.close() + global _cache + _cache = None # rebuilt on next read + + +def _reset_for_tests() -> None: + """Clear the in-memory cache so the next read hits the database. + + Seeding only runs when the table is empty, so a reset never overwrites + entries written by a test. + """ + global _cache + _cache = None