wave2 phase 4: historical record & replay (true time-travel)
Scrubbing off-live now REPLAYS actual recorded traffic instead of just hiding
aircraft (orbits were already computable in the past; live traffic was not).
- record.py (new, dev-only daemon): every 180s fetches through the local proxy
(so the shared cache dedupes quota — never hits OpenSky directly), reduces to
airborne essentials, zlib-compresses into SQLite data/history.db (WAL), keeps
72h, prunes + checkpoints hourly. Wired into John's ~/.jobs heartbeat.
- serve.py: GET history/aircraft?t=<epoch_s> returns the nearest snapshot within
±10 min (read-only per-request sqlite connection — WAL means never locks the
recorder), else a JSON 404.
- aircraft.js: onClockTick drives replay when not-live. Build loop factored into
renderPlanes() shared by live + replay. Replayed aircraft show amber ("replay
· <time> · N aircraft"); 404 → "no history for this time" + hide (this is the
prod-without-recorder path too). Requests coalesce so a scrub landing mid-fetch
still resolves to its final snapshot; debounced on requested-time distance.
- data/ gitignored (recorder db is local-only, never committed or deployed).
Verified: history endpoint hit/404/400; frontend replay across -1h/-2h/-3h maps
to distinct snapshots, 404 transitions, live-resume; recorder daemon writes
heartbeat + snapshots.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jing
parent
108680462e
commit
767884146d
2
.gitignore
vendored
2
.gitignore
vendored
@ -1,3 +1,5 @@
|
||||
.DS_Store
|
||||
*.pyc
|
||||
__pycache__/
|
||||
# Local recorder database (record.py) — dev-only, never committed or deployed.
|
||||
data/
|
||||
|
||||
@ -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');
|
||||
}
|
||||
|
||||
function onClockTick(currentTime, live) {
|
||||
isLive = live;
|
||||
if (!isLive) {
|
||||
bc.show = false; // pause: hide + skip fetches
|
||||
// ---- 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;
|
||||
if (enabled) kick();
|
||||
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) {
|
||||
lastReplayT = null; // leaving replay — reset the debounce
|
||||
pendingReplayT = null;
|
||||
bc.show = enabled;
|
||||
if (enabled) kick(); // resume the live poll loop exactly as before
|
||||
return;
|
||||
}
|
||||
// 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) {
|
||||
|
||||
130
record.py
Normal file
130
record.py
Normal file
@ -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/<job>.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")
|
||||
48
serve.py
48
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=<epoch_s> 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=<epoch_s> → 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=<epoch_s>"})
|
||||
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))
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user