🤝 world_planes: read the shared OpenSky cache (quota-collision fix)
Reader side of the OpenSky shared-cache v1 contract (mirrored in GODSIGH SPEC2.md §4). Path ~/.cache/godverse/opensky-states.json (env OPENSKY_CACHE_FILE), content the raw global /states/all body, freshness = mtime within 120 s. Each poll cycle now checks the cache first: on a fresh hit it reads the global body, filters the states to our bbox client-side (indices 5=lon, 6=lat), and skips the HTTP call entirely (logs "cache hit"). On 429 or a network failure it falls back to a stale cache rather than nothing. This worker only ever polls a bbox, so it never WRITES the cache (never partial data into a global-contract file) — it is a pure reader, and degrades to its old ≥450 s bbox fetch when no writer is around (i.e. prod). Result: godstrument + GODSIGH running together spend ~one poller's quota instead of colliding and 429-ing each other out. Verified: unit tests for read_cache freshness/stale/missing + bbox filtering, and an integration run where a fresh cache is served (count filtered to bbox, zero HTTP). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
301005bdbc
commit
6d408a8bdb
@ -5,6 +5,7 @@ and mean groundspeed (intensity) — the sky's traffic becomes a slow chord.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
@ -23,6 +24,57 @@ POLL_OK = 450.0 # normal poll interval (s)
|
||||
POLL_429 = 600.0 # backoff poll interval after HTTP 429
|
||||
REEMIT = 10.0 # re-emit last values every 10s
|
||||
|
||||
# --- OpenSky shared cache v1 (reader side) --------------------------------
|
||||
# Path: ~/.cache/godverse/opensky-states.json, override via OPENSKY_CACHE_FILE.
|
||||
# Content: the raw, unmodified OpenSky /states/all body (global, no bbox), written
|
||||
# atomically by WHOEVER fetched it (GODSIGH's dev proxy is the writer). Freshness =
|
||||
# file mtime. We treat the cache as fresh within 120 s; on a fresh hit we skip our
|
||||
# own HTTP entirely and just filter the global states to our bbox, so a godstrument
|
||||
# + GODSIGH running together spend ~one poller's quota instead of two. This worker
|
||||
# only ever polls a bbox, so it never WRITES the cache (never partial data into a
|
||||
# global-contract file) — it is a pure reader, and degrades to its old bbox fetch
|
||||
# when no writer is around (i.e. prod).
|
||||
CACHE_FILE = os.environ.get("OPENSKY_CACHE_FILE") or \
|
||||
os.path.expanduser("~/.cache/godverse/opensky-states.json")
|
||||
CACHE_FRESH = 120.0 # a cache within this many seconds of its mtime is "fresh"
|
||||
|
||||
|
||||
def read_cache(max_age):
|
||||
"""Return (data, age_seconds) from the shared cache, or None. If max_age is not
|
||||
None, returns None when the cache is older than that (i.e. not fresh)."""
|
||||
try:
|
||||
mtime = os.path.getmtime(CACHE_FILE)
|
||||
except OSError:
|
||||
return None
|
||||
age = time.time() - mtime
|
||||
if max_age is not None and age > max_age:
|
||||
return None
|
||||
try:
|
||||
with open(CACHE_FILE, "r", encoding="utf-8") as f:
|
||||
return json.load(f), age
|
||||
except (OSError, json.JSONDecodeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def filter_states_bbox(data, bbox):
|
||||
"""Filter a global states payload to our bbox client-side (the cache is global).
|
||||
OpenSky state vectors: index 5 = longitude, index 6 = latitude."""
|
||||
s, w, n, e = bbox
|
||||
states = (data or {}).get("states") or []
|
||||
out = []
|
||||
for row in states:
|
||||
if not row or len(row) < 7:
|
||||
continue
|
||||
lon, lat = row[5], row[6]
|
||||
if lon is None or lat is None:
|
||||
continue
|
||||
try:
|
||||
if s <= float(lat) <= n and w <= float(lon) <= e:
|
||||
out.append(row)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
return {"states": out}
|
||||
|
||||
|
||||
def fetch(bbox, timeout=25):
|
||||
"""Fetch OpenSky states for the bbox. Returns parsed JSON dict.
|
||||
@ -95,25 +147,42 @@ def main():
|
||||
while True:
|
||||
now = time.monotonic()
|
||||
if now >= next_poll:
|
||||
try:
|
||||
data = fetch(args.bbox)
|
||||
last = reduce_states(data)
|
||||
cached = read_cache(CACHE_FRESH) # a fresh global cache -> skip HTTP entirely
|
||||
if cached is not None:
|
||||
data, age = cached
|
||||
last = reduce_states(filter_states_bbox(data, args.bbox))
|
||||
if poll != POLL_OK:
|
||||
print(f"[{NAME}] recovered; back to {POLL_OK:.0f}s poll")
|
||||
poll = POLL_OK
|
||||
except urllib.error.HTTPError as ex:
|
||||
if ex.code == 429:
|
||||
retry_after = ex.headers.get("X-Rate-Limit-Retry-After-Seconds")
|
||||
try:
|
||||
poll = max(POLL_429, float(retry_after))
|
||||
except (TypeError, ValueError):
|
||||
poll = POLL_429
|
||||
print(f"[{NAME}] HTTP 429 rate limited; backing off to {poll:.0f}s")
|
||||
else:
|
||||
print(f"[{NAME}] HTTP {ex.code}; retrying in {poll:.0f}s")
|
||||
except (urllib.error.URLError, TimeoutError, OSError,
|
||||
json.JSONDecodeError, ValueError) as ex:
|
||||
print(f"[{NAME}] fetch error: {ex}; retrying in {poll:.0f}s")
|
||||
print(f"[{NAME}] cache hit ({age:.0f}s old) -> {int(last[0])} aircraft in bbox")
|
||||
else:
|
||||
try:
|
||||
data = fetch(args.bbox)
|
||||
last = reduce_states(data)
|
||||
if poll != POLL_OK:
|
||||
print(f"[{NAME}] recovered; back to {POLL_OK:.0f}s poll")
|
||||
poll = POLL_OK
|
||||
except urllib.error.HTTPError as ex:
|
||||
if ex.code == 429:
|
||||
retry_after = ex.headers.get("X-Rate-Limit-Retry-After-Seconds")
|
||||
try:
|
||||
poll = max(POLL_429, float(retry_after))
|
||||
except (TypeError, ValueError):
|
||||
poll = POLL_429
|
||||
print(f"[{NAME}] HTTP 429 rate limited; backing off to {poll:.0f}s")
|
||||
else:
|
||||
print(f"[{NAME}] HTTP {ex.code}; retrying in {poll:.0f}s")
|
||||
_stale = read_cache(None) # 429 -> a stale cache beats nothing
|
||||
if _stale is not None:
|
||||
last = reduce_states(filter_states_bbox(_stale[0], args.bbox))
|
||||
print(f"[{NAME}] serving stale cache ({_stale[1]:.0f}s old)")
|
||||
except (urllib.error.URLError, TimeoutError, OSError,
|
||||
json.JSONDecodeError, ValueError) as ex:
|
||||
print(f"[{NAME}] fetch error: {ex}; retrying in {poll:.0f}s")
|
||||
_stale = read_cache(None) # network down -> fall back to a stale cache
|
||||
if _stale is not None:
|
||||
last = reduce_states(filter_states_bbox(_stale[0], args.bbox))
|
||||
print(f"[{NAME}] serving stale cache ({_stale[1]:.0f}s old)")
|
||||
next_poll = time.monotonic() + poll
|
||||
|
||||
# Emit (or re-emit) the last known values.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user