GODSIGH/ais_bridge.py

#!/usr/bin/env python3
"""AIS bridge (Wave 7) — real live global shipping for GODSIGH.

Holds a websocket to aisstream.io (free API key), keeps an in-memory table of
every ship seen recently, and atomically writes a snapshot JSON into the web
root every WRITE_SEC. The browser layer polls that static file — no server
code in the request path, works identically under dev serve.py and prod nginx
(the layer cache-busts with a rolling query param, so CDN caching is moot).

Usage:  python3 ais_bridge.py <credentials.json> <output.json>
  e.g. dev : ~/.venvs/godsigh/bin/python ais_bridge.py aiscredentials.json ais_snapshot.json
       prod: /home/humanjing/.venvs/godsigh/bin/python ais_bridge.py \
               /home/humanjing/godsigh-secret/aiscredentials.json \
               /home/humanjing/godsigh/ais_snapshot.json

The key file may be full JSON, a brace-less "key": "…" fragment, or a raw
token. Reconnects with backoff forever. Heartbeat: ~/.jobs/godsigh-ais.status
(fleet jobs convention).
"""
import json
import os
import re
import sys
import tempfile
import threading
import time

import websocket  # websocket-client

STALE_SEC = 900        # drop ships not heard from in 15 min
WRITE_SEC = 20         # snapshot cadence
MAX_SHIPS = 20000      # snapshot cap (newest first) — keeps the JSON sane
URL = "wss://stream.aisstream.io/v0/stream"

ships = {}             # mmsi -> dict
lock = threading.Lock()
stats = {"msgs": 0, "connected_at": None, "last_msg": None}


def read_key(path):
    raw = open(path).read().strip()
    for attempt in (raw, "{" + raw + "}"):
        try:
            d = json.loads(attempt)
            if isinstance(d, dict):
                for v in d.values():
                    if isinstance(v, str) and len(v) >= 16:
                        return v
        except Exception:
            pass
    m = re.findall(r"[A-Za-z0-9_-]{16,}", raw)
    if not m:
        raise SystemExit("no key found in " + path)
    return m[0]


def on_message(_ws, raw):
    try:
        m = json.loads(raw)
    except Exception:
        return
    meta = m.get("MetaData") or {}
    mmsi = meta.get("MMSI")
    if not mmsi:
        return
    now = time.time()
    stats["msgs"] += 1
    stats["last_msg"] = now
    t = m.get("MessageType")
    with lock:
        rec = ships.get(mmsi) or {"mmsi": mmsi}
        if t == "PositionReport":
            p = (m.get("Message") or {}).get("PositionReport") or {}
            rec.update(
                lat=p.get("Latitude"), lon=p.get("Longitude"),
                sog=p.get("Sog"), cog=p.get("Cog"),
                hdg=p.get("TrueHeading"), nav=p.get("NavigationalStatus"),
                ts=now,
            )
        elif t == "ShipStaticData":
            s = (m.get("Message") or {}).get("ShipStaticData") or {}
            rec["type"] = s.get("Type")
            dest = (s.get("Destination") or "").strip()
            if dest:
                rec["dest"] = dest[:24]
        name = (meta.get("ShipName") or "").strip()
        if name:
            rec["name"] = name[:32]
        # MetaData carries a position too — fallback for static-only sightings
        if rec.get("lat") is None and meta.get("latitude") is not None:
            rec["lat"], rec["lon"], rec["ts"] = meta.get("latitude"), meta.get("longitude"), now
        ships[mmsi] = rec


def writer(out_path):
    jobs = os.path.expanduser("~/.jobs")
    os.makedirs(jobs, exist_ok=True)
    while True:
        time.sleep(WRITE_SEC)
        now = time.time()
        with lock:
            for k in [k for k, v in ships.items() if now - (v.get("ts") or 0) > STALE_SEC]:
                del ships[k]
            rows = [v for v in ships.values() if v.get("lat") is not None]
        rows.sort(key=lambda r: r.get("ts", 0), reverse=True)
        rows = rows[:MAX_SHIPS]
        snap = {"ts": int(now), "count": len(rows), "ships": [
            {k: r.get(k) for k in ("mmsi", "name", "lat", "lon", "sog", "cog", "hdg", "nav", "type", "dest") if r.get(k) is not None}
            for r in rows
        ]}
        body = json.dumps(snap, separators=(",", ":"))
        fd, tmp = tempfile.mkstemp(dir=os.path.dirname(os.path.abspath(out_path)))
        with os.fdopen(fd, "w") as f:
            f.write(body)
        os.replace(tmp, out_path)  # atomic — readers never see a torn file
        os.chmod(out_path, 0o644)  # mkstemp births 0600 — nginx (other uid) must read it
        with open(os.path.join(jobs, "godsigh-ais.status"), "w") as f:
            f.write(json.dumps({"ts": int(now), "ships": len(rows), "msgs": stats["msgs"]}))
        print(f"[ais] {len(rows)} ships · {stats['msgs']} msgs", flush=True)


def main():
    if len(sys.argv) != 3:
        raise SystemExit(__doc__)
    key = read_key(sys.argv[1])
    threading.Thread(target=writer, args=(sys.argv[2],), daemon=True).start()
    backoff = 2
    while True:
        try:
            ws = websocket.WebSocketApp(
                URL,
                on_open=lambda w: (
                    stats.__setitem__("connected_at", time.time()),
                    w.send(json.dumps({
                        "APIKey": key,
                        "BoundingBoxes": [[[-90, -180], [90, 180]]],
                        "FilterMessageTypes": ["PositionReport", "ShipStaticData"],
                    })),
                    print("[ais] connected + subscribed (global)", flush=True),
                ),
                on_message=on_message,
                on_error=lambda w, e: print(f"[ais] error: {e}", flush=True),
            )
            ws.run_forever(ping_interval=30, ping_timeout=10)
        except Exception as e:
            print(f"[ais] crashed: {e}", flush=True)
        print(f"[ais] reconnecting in {backoff}s", flush=True)
        time.sleep(backoff)
        backoff = min(backoff * 2, 60)


if __name__ == "__main__":
    main()