diff --git a/.gitignore b/.gitignore index f6d5072..3c1b716 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .DS_Store *.pyc __pycache__/ +# Local recorder database (record.py) — dev-only, never committed or deployed. +data/ diff --git a/js/layers/aircraft.js b/js/layers/aircraft.js index 4624c2e..a7a66f7 100644 --- a/js/layers/aircraft.js +++ b/js/layers/aircraft.js @@ -1,6 +1,12 @@ // Live aircraft (ADS-B via OpenSky) as a BillboardCollection primitive. // Thousands of billboards rebuilt wholesale each poll; entities would crawl. // SPEC §6.3. +// +// Wave 2: when the clock is scrubbed off wall-clock now, instead of just +// hiding, the layer REPLAYS actual recorded traffic from history/aircraft?t= +// (served by serve.py from record.py's SQLite db). Replayed aircraft show with +// an amber status. Where no recorder ran (prod), the endpoint 404s and the +// layer degrades to the original hide-when-scrubbed behavior. export default function create(ctx) { const { viewer, CONFIG, lib, ui, Cesium } = ctx; @@ -10,16 +16,64 @@ export default function create(ctx) { let enabled = true; // HUD checkbox let isLive = true; // clock at wall-clock now - let inFlight = false; // a fetch is currently running + let inFlight = false; // a live fetch is currently running let timer = null; // pending setTimeout handle let delay = CONFIG.aircraft.pollMs; // current cadence (grows on 429) + // Replay state (only used when scrubbed off-live). + let lastReplayT = null; // epoch_s of the last snapshot we requested (debounce) + let replayInFlight = false; + let pendingReplayT = null; // newest target requested while a fetch is in flight + ui.addLayer('aircraft', 'Aircraft (live ADS-B)', true, (on) => { enabled = on; bc.show = on && isLive; - if (on) kick(); + if (on && isLive) kick(); }); + // ---- normalized plane shape: {icao24, callsign, country, lon, lat, alt, vel, track} ---- + function stateToPlane(s) { + return { + icao24: s[0], callsign: (s[1] || '').trim(), country: s[2], + lon: s[5], lat: s[6], alt: s[7], vel: s[9], track: s[10], + }; + } + // Recorded snapshot plane: [icao24, callsign, lon, lat, baro_alt, track, vel]. + function snapPlaneToPlane(a) { + return { + icao24: a[0], callsign: a[1], country: '—', + lon: a[2], lat: a[3], alt: a[4], vel: a[6], track: a[5], + }; + } + + // Factored build loop — rebuilds the whole collection from a plane array. + function renderPlanes(planes) { + bc.removeAll(); + let count = 0; + for (const p of planes) { + if (p.lon == null || p.lat == null) continue; + bc.add({ + position: Cesium.Cartesian3.fromDegrees(p.lon, p.lat, p.alt || 0), + image: glyph, + color: lib.cz(lib.altitudeColor(CONFIG, p.alt)), + rotation: lib.headingToRotation(p.track), // lib negates — do NOT negate again + scale: 0.55, + id: { + layer: 'aircraft', + icao24: p.icao24, + callsign: p.callsign, + country: p.country, + alt: p.alt, + vel: p.vel, + track: p.track, + }, + }); + count++; + } + return count; + } + + // ---- live poll loop ------------------------------------------------------- // Poll only while enabled, live, and the tab is visible. function canPoll() { return enabled && isLive && !document.hidden; @@ -89,35 +143,13 @@ export default function create(ctx) { } const states = Array.isArray(data && data.states) ? data.states : []; - - bc.removeAll(); - let count = 0; + const planes = []; for (const s of states) { if (!s) continue; - const lon = s[5]; - const lat = s[6]; - const onGround = s[8]; - if (lon == null || lat == null || onGround) continue; - const alt = s[7]; - const track = s[10]; - bc.add({ - position: Cesium.Cartesian3.fromDegrees(lon, lat, alt || 0), - image: glyph, - color: lib.cz(lib.altitudeColor(CONFIG, alt)), - rotation: lib.headingToRotation(track), // lib negates — do NOT negate again - scale: 0.55, - id: { - layer: 'aircraft', - icao24: s[0], - callsign: (s[1] || '').trim(), - country: s[2], - alt, - vel: s[9], - track, - }, - }); - count++; + if (s[5] == null || s[6] == null || s[8]) continue; // no position or on-ground + planes.push(stateToPlane(s)); } + const count = renderPlanes(planes); delay = CONFIG.aircraft.pollMs; // healthy response — reset cadence const stamp = data.time ? new Date(data.time * 1000).toLocaleTimeString() : '—'; @@ -125,14 +157,72 @@ export default function create(ctx) { ui.setStatus('aircraft', `${count} aircraft · ${stamp}${quota}`, 'ok'); } + // ---- replay (scrubbed off-live) ------------------------------------------- + // Requests coalesce: while a history fetch is in flight, the newest requested + // time is stashed and processed on completion — so a scrub that lands (even + // paused, where onClockTick stops firing) still gets its final snapshot. + function requestReplay(tSec) { + pendingReplayT = tSec; + drainReplay(); + } + + async function drainReplay() { + if (replayInFlight || pendingReplayT === null) return; + const tSec = pendingReplayT; + pendingReplayT = null; + replayInFlight = true; + lastReplayT = tSec; + try { + const res = await fetch(`history/aircraft?t=${tSec}`); + if (isLive) return; // clock snapped back to live mid-fetch — discard + if (res.status === 404) { + bc.removeAll(); + bc.show = false; + ui.setStatus('aircraft', 'no history for this time', 'warn'); + return; + } + if (!res.ok) { + ui.setStatus('aircraft', `history HTTP ${res.status}`, 'warn'); + return; + } + let snap; + try { + snap = await res.json(); + } catch (err) { + ui.setStatus('aircraft', 'bad history response', 'warn'); + return; + } + if (isLive) return; + const count = renderPlanes((snap.planes || []).map(snapPlaneToPlane)); + bc.show = enabled; + const stamp = snap.t ? new Date(snap.t * 1000).toLocaleTimeString() : '—'; + // Amber = replayed traffic, not live (the LIVE/SCRUBBED chip already says SCRUBBED). + ui.setStatus('aircraft', `replay · ${stamp} · ${count} aircraft`, 'warn'); + } catch (err) { + ui.setStatus('aircraft', 'replay fetch error', 'warn'); + } finally { + replayInFlight = false; + // A newer target arrived mid-fetch — process it now (covers a paused clock). + if (pendingReplayT !== null && !isLive) drainReplay(); + } + } + function onClockTick(currentTime, live) { isLive = live; - if (!isLive) { - bc.show = false; // pause: hide + skip fetches + if (isLive) { + lastReplayT = null; // leaving replay — reset the debounce + pendingReplayT = null; + bc.show = enabled; + if (enabled) kick(); // resume the live poll loop exactly as before return; } - bc.show = enabled; - if (enabled) kick(); + // Not live → replay recorded traffic (or gracefully hide if none recorded). + if (!enabled) { bc.show = false; return; } + const tSec = Math.round(Cesium.JulianDate.toDate(currentTime).getTime() / 1000); + // Debounce on requested-time DISTANCE, not wall time: a scrub gesture fires + // many ticks — only refetch when the target moves >60 s from the last snapshot. + if (lastReplayT !== null && Math.abs(tSec - lastReplayT) <= 60) return; + requestReplay(tSec); } function handlePick(picked) { diff --git a/record.py b/record.py new file mode 100644 index 0000000..0cf41cd --- /dev/null +++ b/record.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""GODSIGH history recorder — dev-side daemon (NOT shipped to prod). + +Snapshots live aircraft every 180 s into a local SQLite db so the timeline can +replay ACTUAL past traffic instead of just hiding aircraft when scrubbed off +now. Fetches through the local dev proxy (http://127.0.0.1:8137/proxy/opensky) +so the shared OpenSky cache dedupes quota with the live app — it never calls +upstream directly. If the dev server is down it simply skips the cycle. + +Prod has no recorder; the app's history endpoint 404s gracefully there, so the +aircraft layer just falls back to hide-when-scrubbed. Wire this into John's +heartbeat convention: status at ~/.jobs/godsigh-recorder.status. + +Run it alongside the dev server: python3 record.py +""" +import json +import sqlite3 +import sys +import time +import urllib.request +import zlib +from datetime import datetime +from pathlib import Path + +PROXY_URL = "http://127.0.0.1:8137/proxy/opensky" +DB_PATH = Path(__file__).resolve().parent / "data" / "history.db" +SNAPSHOT_SEC = 180 # one snapshot every 3 min +RETENTION_SEC = 72 * 3600 # keep 72 h (~70 MB steady-state) +PRUNE_EVERY_SEC = 3600 # prune + checkpoint hourly +JOB = "godsigh-recorder" + + +def heartbeat(count, note=""): + # John's convention: overwrite ~/.jobs/.status every iteration so + # "is the recorder alive?" is one `cat` away. + try: + d = Path.home() / ".jobs" + d.mkdir(exist_ok=True) + (d / f"{JOB}.status").write_text( + f"{datetime.now():%Y-%m-%d %H:%M:%S} | {count} snapshots | {note}\n") + except OSError: + pass + + +def connect(): + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + con = sqlite3.connect(str(DB_PATH)) + con.execute("PRAGMA journal_mode=WAL") + con.execute("CREATE TABLE IF NOT EXISTS snapshots (ts INTEGER PRIMARY KEY, kind TEXT, body BLOB)") + con.commit() + return con + + +def reduce_states(data): + # Reduce OpenSky /states/all to airborne essentials: + # [icao24, callsign, lon, lat, baro_alt, track, vel] + planes = [] + for s in data.get("states") or []: + if not s: + continue + lon, lat, on_ground = s[5], s[6], s[8] + if lon is None or lat is None or on_ground: + continue + planes.append([s[0], (s[1] or "").strip(), lon, lat, s[7], s[10], s[9]]) + return planes + + +def record_once(con): + req = urllib.request.Request(PROXY_URL, headers={"User-Agent": "godsigh-recorder/0.1"}) + with urllib.request.urlopen(req, timeout=45) as r: + data = json.loads(r.read()) + t = int(data.get("time") or time.time()) + snap = {"t": t, "planes": reduce_states(data)} + body = zlib.compress(json.dumps(snap, separators=(",", ":")).encode()) + # INSERT OR REPLACE: if a stale cache hands us the same upstream time twice + # (quota exhausted), we keep one row per timestamp instead of erroring. + con.execute("INSERT OR REPLACE INTO snapshots (ts, kind, body) VALUES (?, ?, ?)", + (t, "aircraft", body)) + con.commit() + return t, len(snap["planes"]) + + +def prune(con): + cutoff = int(time.time()) - RETENTION_SEC + con.execute("DELETE FROM snapshots WHERE ts < ?", (cutoff,)) + con.commit() + con.execute("PRAGMA wal_checkpoint(TRUNCATE)") + + +def count_snaps(con): + return con.execute("SELECT COUNT(*) FROM snapshots").fetchone()[0] + + +def main(): + con = connect() + try: + prune(con) + except sqlite3.Error as e: + print(f"[recorder] initial prune failed: {e}", file=sys.stderr) + last_prune = time.time() + n = count_snaps(con) + heartbeat(n, "starting") + print(f"[recorder] db={DB_PATH} snapshots={n} — every {SNAPSHOT_SEC}s via {PROXY_URL}") + while True: + try: + t, planes = record_once(con) + n = count_snaps(con) + note = f"planes={planes} t={t}" + print(f"[recorder] snapshot ok — {note} (retained {n})") + heartbeat(n, note) + except Exception as e: + # Dev server down or upstream hiccup — skip this cycle, retry next + # tick. Never fetch OpenSky directly (that would defeat the cache). + print(f"[recorder] snapshot skipped: {e}", file=sys.stderr) + heartbeat(n, f"skip: {e}") + now = time.time() + if now - last_prune > PRUNE_EVERY_SEC: + try: + prune(con) + last_prune = now + except sqlite3.Error as e: + print(f"[recorder] prune failed: {e}", file=sys.stderr) + time.sleep(SNAPSHOT_SEC) + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\n[recorder] stopped") diff --git a/serve.py b/serve.py index 212400f..5672076 100644 --- a/serve.py +++ b/serve.py @@ -6,12 +6,16 @@ lets GODSIGH and its sibling project godstrument poll OpenSky from the same IP without racing each other into the anonymous daily quota. The contract is mirrored verbatim in godstrument's brief — keep the two in sync. """ +import json import os +import sqlite3 import sys import tempfile import time import urllib.error +import urllib.parse import urllib.request +import zlib from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -30,6 +34,12 @@ UPSTREAMS = { # may serve a stale cache rather than nothing. OPENSKY_CACHE_FRESH_SEC = 120 +# ---- Historical replay (record.py writes this db; dev-only) ----------------- +# GET history/aircraft?t= returns the nearest recorded snapshot within +# ±10 min, or 404. Prod has no recorder → the file is absent → graceful 404. +HISTORY_DB = Path(__file__).resolve().parent / "data" / "history.db" +HISTORY_TOLERANCE_SEC = 600 + def opensky_cache_file(): override = os.environ.get("OPENSKY_CACHE_FILE") @@ -93,6 +103,8 @@ class Handler(SimpleHTTPRequestHandler): super().end_headers() def do_GET(self): + if self.path.startswith("/history/"): + return self._serve_history() if not self.path.startswith("/proxy/"): return super().do_GET() name, _, query = self.path[len("/proxy/"):].partition("?") @@ -161,6 +173,42 @@ class Handler(SimpleHTTPRequestHandler): self.end_headers() self.wfile.write(body) + def _send_json(self, status, obj, extra=None): + self._send(status, "application/json", json.dumps(obj).encode(), extra) + + def _serve_history(self): + # /history/aircraft?t= → nearest snapshot within ±10 min, or 404. + kind, _, query = self.path[len("/history/"):].partition("?") + params = urllib.parse.parse_qs(query) + t_raw = (params.get("t") or [None])[0] + if kind != "aircraft" or t_raw is None: + return self._send_json(400, {"error": "usage: history/aircraft?t="}) + try: + t = int(float(t_raw)) + except ValueError: + return self._send_json(400, {"error": "t must be epoch seconds"}) + if not HISTORY_DB.exists(): + return self._send_json(404, {"error": "no history for this time"}) + try: + # Read-only, short-lived connection per request — WAL lets us read + # without ever locking out the recorder's writes. + con = sqlite3.connect(f"file:{HISTORY_DB}?mode=ro", uri=True, timeout=2) + try: + row = con.execute( + "SELECT ts, body FROM snapshots WHERE kind='aircraft' " + "ORDER BY ABS(ts - ?) ASC LIMIT 1", (t,)).fetchone() + finally: + con.close() + except sqlite3.Error as e: + return self._send_json(500, {"error": f"history db error: {e}"}) + if not row or abs(row[0] - t) > HISTORY_TOLERANCE_SEC: + return self._send_json(404, {"error": "no history for this time"}) + try: + snap = zlib.decompress(row[1]) # already-JSON bytes: {"t":..,"planes":[...]} + except zlib.error: + return self._send_json(500, {"error": "corrupt snapshot"}) + return self._send(200, "application/json", snap, {"X-Godsigh-History": "hit"}) + def log_message(self, fmt, *args): sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))