Godstrument/workers/world_planes.py

#!/usr/bin/env python3
"""A swarm of real aircraft over a patch of sky, turned into a drone.
OpenSky live states give a count (thickness), mean altitude (pitch/register),
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
import urllib.request

from pythonosc.udp_client import SimpleUDPClient

NAME = "world_planes"
API = "https://opensky-network.org/api/states/all"
UA = "Godstrument/1.0 (live instrument; contact monsterrobotparty@gmail.com)"

# OpenSky anonymous budget is ~400 credits/day; a bbox query costs 2-4 credits,
# so <200 requests/day -> poll no faster than ~1 per 450s or the feed 429s out
# and freezes for the rest of the day. (Add OAuth2 creds for the 4000 tier.)
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.

    Raises urllib.error.HTTPError so the caller can detect 429.
    """
    s, w, n, e = bbox
    url = f"{API}?lamin={s}&lomin={w}&lamax={n}&lomax={e}"
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


def reduce_states(data):
    """Compute (count, mean_alt, mean_speed) from a states payload.

    index 7 = geo_altitude (m, may be null), index 9 = velocity (m/s, may be null).
    Nulls are ignored in the means. Returns floats; means are 0.0 when no data.
    """
    states = (data or {}).get("states") or []
    count = float(len(states))
    alts = []
    speeds = []
    for row in states:
        if not row:
            continue
        alt = row[7] if len(row) > 7 else None
        spd = row[9] if len(row) > 9 else None
        if alt is not None:
            try:
                alts.append(float(alt))
            except (TypeError, ValueError):
                pass
        if spd is not None:
            try:
                speeds.append(float(spd))
            except (TypeError, ValueError):
                pass
    avgalt = sum(alts) / len(alts) if alts else 0.0
    avgspeed = sum(speeds) / len(speeds) if speeds else 0.0
    return count, avgalt, avgspeed


def parse_bbox(text):
    parts = [p.strip() for p in text.split(",")]
    if len(parts) != 4:
        raise argparse.ArgumentTypeError('--bbox must be "s,w,n,e"')
    try:
        return tuple(float(p) for p in parts)
    except ValueError:
        raise argparse.ArgumentTypeError("--bbox values must be numbers")


def main():
    ap = argparse.ArgumentParser(description="OpenSky aircraft swarm -> OSC drone")
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=9000)
    ap.add_argument("--bbox", type=parse_bbox, default=(45.0, 0.0, 52.0, 10.0),
                    help='Bounding box "s,w,n,e" (default Western Europe)')
    args = ap.parse_args()

    client = SimpleUDPClient(args.host, args.port)
    poll = POLL_OK
    print(f"[{NAME}] emitting /gs/planes/count,/gs/planes/avgalt,/gs/planes/avgspeed every {POLL_OK:.0f}s")

    # Last known values, re-emitted every REEMIT seconds between polls.
    last = None  # (count, avgalt, avgspeed)
    next_poll = 0.0

    while True:
        now = time.monotonic()
        if now >= next_poll:
            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
                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(3600)         # 429 -> a stale cache beats nothing,
                    if _stale is not None:            #   but a day-old sky beats neither
                        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(3600)         # network down -> fall back, bounded 1h
                    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.
        if last is not None:
            count, avgalt, avgspeed = last
            try:
                client.send_message("/gs/planes/count", float(count))
                client.send_message("/gs/planes/avgalt", float(avgalt))
                client.send_message("/gs/planes/avgspeed", float(avgspeed))
            except OSError as ex:
                print(f"[{NAME}] OSC send error: {ex}")

        time.sleep(REEMIT)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)