Seed the world with live ADS-B traffic from adsb.retina.fm

retina_simulation/live_adsb.py

@@ -0,0 +1,147 @@
+"""Live ADS-B client — pulls real aircraft from adsb.retina.fm to seed the world.
+
+The endpoint speaks the adsb.lol ``/v2/point/{lat}/{lon}/{radius_nm}`` shape
+(tar1090 aircraft objects under ``ac``), so the parsing here is the same as
+the backend's adsb.lol client; it is a separate, dependency-free copy because
+this package must not import the backend, and because the two feeds serve
+different purposes: the backend polls adsb.lol for *external truth* to score
+nodes against, while this client feeds ``SimulationWorld.ingest_live_aircraft``
+so that the synthetic nodes fly real trajectories.
+
+Rows come back normalised to what the world needs, with ``captured_at`` as an
+absolute wall-clock time (``seen_pos`` resolved against the fetch), so a
+dead-reckoned position can be extrapolated to "now" on ingest.
+"""
+
+import gzip
+import json
+import logging
+import time
+import urllib.error
+import urllib.request
+
+log = logging.getLogger(__name__)
+
+DEFAULT_BASE_URL = "https://adsb.retina.fm"
+_TIMEOUT_S = 8
+_USER_AGENT = "retina-simulation/1.0 (+https://github.com/offworldlabs/retina-simulation)"
+# One last-good result per area may stand in for a failed fetch this long.
+# Longer than the world's own staleness window (LIVE_STALE_S) is pointless:
+# the world would drop the aircraft anyway before the cache stopped serving.
+_CACHE_MAX_AGE_S = 90.0
+
+
+def _num(v, default=0.0) -> float:
+    """Coerce a tar1090 numeric field; sentinels like ``"ground"`` → default."""
+    if isinstance(v, bool):
+        return default
+    if isinstance(v, (int, float)):
+        return float(v)
+    return default
+
+
+def parse_point_response(data: dict, fetched_at: float) -> list[dict]:
+    """Normalise one ``/v2/point`` payload into world-ready rows.
+
+    Rows without a position, and aircraft reported on the ground (tar1090's
+    ``alt_baro: "ground"``), are dropped: a parked airliner is not a radar
+    target the fleet should be echoing at zero altitude.
+    """
+    rows = []
+    for ac in data.get("ac", []) or []:
+        if not isinstance(ac, dict):
+            continue
+        hex_code = str(ac.get("hex") or "").strip().lower()
+        lat = ac.get("lat")
+        lon = ac.get("lon")
+        if not hex_code or not isinstance(lat, (int, float)) or not isinstance(lon, (int, float)):
+            continue
+        alt_baro = ac.get("alt_baro")
+        if alt_baro == "ground" or not isinstance(alt_baro, (int, float)):
+            continue
+        seen_pos = ac.get("seen_pos")
+        captured_at = fetched_at - seen_pos if isinstance(seen_pos, (int, float)) else fetched_at
+        rows.append(
+            {
+                "hex": hex_code,
+                "flight": (ac.get("flight") or "").strip(),
+                "lat": float(lat),
+                "lon": float(lon),
+                "alt_baro": float(alt_baro),  # ft
+                "gs": _num(ac.get("gs")),  # knots
+                "track": _num(ac.get("track")),  # deg
+                "baro_rate": _num(ac.get("baro_rate")),  # ft/min
+                "captured_at": captured_at,  # epoch s
+            }
+        )
+    return rows
+
+
+class LiveAdsbClient:
+    """Polls adsb.retina.fm (or any adsb.lol-shaped server) for one or more areas."""
+
+    def __init__(self, areas: list[dict], base_url: str = DEFAULT_BASE_URL):
+        """
+        Args:
+            areas: dicts with name, lat, lon and optional radius_nm (default 80).
+            base_url: server root; ``/v2/point/...`` is appended.
+        """
+        self.base_url = base_url.rstrip("/")
+        self.areas = [a for a in areas if isinstance(a, dict) and "lat" in a and "lon" in a]
+        self._cache: dict[str, list[dict]] = {}
+        self._cache_ts: dict[str, float] = {}
+        self.last_status: dict[str, bool] = {}
+        self.last_error: str | None = None
+
+    def _url(self, area: dict) -> str:
+        return f"{self.base_url}/v2/point/{area['lat']}/{area['lon']}/{area.get('radius_nm', 80)}"
+
+    def _get(self, url: str) -> dict:
+        req = urllib.request.Request(
+            url,
+            headers={
+                "Accept": "application/json",
+                "Accept-Encoding": "gzip",
+                "User-Agent": _USER_AGENT,
+            },
+        )
+        with urllib.request.urlopen(req, timeout=_TIMEOUT_S) as resp:  # noqa: S310 — https URL built from config
+            raw = resp.read()
+            if (resp.headers or {}).get("Content-Encoding") == "gzip":
+                raw = gzip.decompress(raw)
+        return json.loads(raw)
+
+    def _last_good(self, name: str) -> list[dict]:
+        ts = self._cache_ts.get(name)
+        if ts is None or time.monotonic() - ts > _CACHE_MAX_AGE_S:
+            self._cache.pop(name, None)
+            self._cache_ts.pop(name, None)
+            return []
+        return self._cache.get(name, [])
+
+    def fetch_area(self, area: dict) -> list[dict]:
+        name = str(area.get("name") or f"{area['lat']},{area['lon']}")
+        try:
+            data = self._get(self._url(area))
+            rows = parse_point_response(data, time.time())
+        except (urllib.error.URLError, OSError, ValueError) as e:
+            self.last_status[name] = False
+            self.last_error = str(e)
+            return self._last_good(name)
+        self._cache[name] = rows
+        self._cache_ts[name] = time.monotonic()
+        self.last_status[name] = True
+        self.last_error = None
+        return rows
+
+    def fetch_all(self) -> list[dict]:
+        """Every configured area, deduplicated by hex (first area wins)."""
+        seen: set[str] = set()
+        out: list[dict] = []
+        for area in self.areas:
+            for row in self.fetch_area(area):
+                if row["hex"] in seen:
+                    continue
+                seen.add(row["hex"])
+                out.append(row)
+        return out

retina_simulation/orchestrator.py

@@ -39,6 +39,7 @@
     fleet_summary,
     generate_fleet,
 )
+from retina_simulation.live_adsb import LiveAdsbClient
 from retina_simulation.tower_resolver import apply_tower_assignments, resolve_towers
 from retina_simulation.world import (
     MetroCell,
@@ -266,6 +267,10 @@ def __init__(
         self.metro_traffic_frac = min(1.0, max(0.0, metro_traffic_frac))
         self.connections: dict[str, NodeConnection] = {}
         self.world: SimulationWorld | None = None
+        # Runtime switch for the live ADS-B seed task (_seed_live_adsb):
+        # the task only runs when main_async wires it, but the backend's
+        # simulation config can pause it without a restart.
+        self.live_adsb_enabled = True
         self._running = False
         self._stats = {
             "total_frames": 0,
@@ -651,6 +656,7 @@ def get_stats(self) -> dict:
             "frames_per_sec": round(self._stats["total_frames"] / max(elapsed, 1), 1),
             "detections_per_sec": round(self._stats["total_detections"] / max(elapsed, 1), 1),
             "aircraft_count": len(self.world.aircraft) if self.world else 0,
+            "live_aircraft_count": len(self.world.live_aircraft) if self.world else 0,
             "ground_truth_snapshots": len(self.ground_truth),
         }
 
@@ -696,6 +702,10 @@ def build_ground_truth_payload(aircraft_summaries: list[dict]) -> list[dict]:
                 "adsb_silent": ac.get("adsb_silent", False),
                 "adsb_callsign": ac.get("adsb_callsign") or None,
                 "anomaly_event": ac.get("anomaly_event") or None,
+                # "live" for an aircraft mirrored from the ADS-B feed, "sim"
+                # for the world's own spawns — so the server can count and
+                # colour the two populations apart.
+                "source": ac.get("source") or "sim",
             }
         )
     return payload_aircraft
@@ -745,7 +755,7 @@ def build_adsb_push_payload(aircraft_summaries: list[dict]) -> list[dict]:
         payload_aircraft.append(
             {
                 "hex": hex_code,
-                "flight": "",
+                "flight": ac.get("adsb_callsign") or "",
                 "lat": round(ac["lat"], 5),
                 "lon": round(ac["lon"], 5),
                 "alt_baro": round(ac["alt_km"] * 1000 / 0.3048),
@@ -890,6 +900,70 @@ async def _push_real_adsb(
             log.debug("Real ADSB push failed: %s", e)
 
 
+async def _seed_live_adsb(
+    orchestrator: FleetOrchestrator,
+    feed_url: str,
+    areas: list[dict],
+    interval_s: float = 5.0,
+):
+    """Pull real aircraft from the live feed into the world every interval_s.
+
+    This is what makes the synthetic nodes fly real traffic: every aircraft
+    the feed reports over the metro becomes a world aircraft the nodes echo
+    (delay/Doppler from its real position and velocity), cast as ADS-B or
+    dark by frac_live_dark.  Unlike the opt-in adsb.lol relay
+    (_push_real_adsb), which only decorates the map with a second world's
+    positions, these aircraft ARE in the simulated world — so their ADS-B
+    tags reach the server through the ordinary node frames and 1 Hz push,
+    untagged, as simulated-world traffic.
+
+    The backend's live_adsb_enabled flag (via _poll_simulation_config)
+    pauses the pull and clears the live aircraft; re-enabling resumes on the
+    next poll.  A failed fetch leaves the world alone: aircraft coast on
+    their last velocity and expire only after the world's staleness window.
+    """
+    client = LiveAdsbClient(areas, base_url=feed_url)
+    loop = asyncio.get_event_loop()
+    log.info(
+        "Live ADS-B seeding started (feed=%s, %d area(s), interval=%.0fs)",
+        feed_url,
+        len(client.areas),
+        interval_s,
+    )
+    last_live = -1
+    was_enabled = True
+    while orchestrator._running:
+        await asyncio.sleep(interval_s)
+        if orchestrator.world is None:
+            continue
+        if not orchestrator.live_adsb_enabled:
+            if was_enabled:
+                n = orchestrator.world.clear_live_aircraft()
+                log.info("Live ADS-B seeding paused by simulation config (%d live aircraft removed)", n)
+                was_enabled = False
+            continue
+        was_enabled = True
+        try:
+            rows = await loop.run_in_executor(None, client.fetch_all)
+        except Exception as e:  # defensive: the client swallows its own errors
+            log.debug("Live ADS-B fetch failed: %s", e)
+            continue
+        if not any(client.last_status.values()):
+            log.warning("Live ADS-B feed unreachable (%s) — live aircraft coasting", client.last_error)
+            continue
+        stats = orchestrator.world.ingest_live_aircraft(rows)
+        if stats["live"] != last_live:
+            n_dark = sum(1 for ac in orchestrator.world.live_aircraft.values() if not ac.has_adsb)
+            log.info(
+                "Live ADS-B: %d aircraft in world (%d dark, frac_live_dark=%.2f; +%d new this poll)",
+                stats["live"],
+                n_dark,
+                orchestrator.world.frac_live_dark,
+                stats["created"],
+            )
+            last_live = stats["live"]
+
+
 async def _poll_simulation_config(
     orchestrator: FleetOrchestrator,
     base_url: str,
@@ -955,15 +1029,28 @@ def _fetch():
                     orchestrator.world.min_aircraft = int(cfg["min_aircraft"])
                 if "max_aircraft" in cfg:
                     orchestrator.world.max_aircraft = int(cfg["max_aircraft"])
+                # Live feed knobs.  Fallbacks match the world/orchestrator
+                # defaults (0.0 dark, seeding on) so an older backend that
+                # ships neither key changes nothing.  set_frac_live_dark
+                # re-casts the aircraft already in the air, so the slider
+                # takes effect at the next poll rather than the next fleet
+                # turnover.
+                n_live_dark = orchestrator.world.set_frac_live_dark(float(cfg.get("frac_live_dark", 0.0)))
+                orchestrator.live_adsb_enabled = bool(cfg.get("live_adsb_enabled", True))
                 last_updated_at = updated_at
                 log.info(
-                    "Simulation config updated: anomalous=%.2f drone=%.2f dark=%.2f adsb_outage=%.2f aircraft=%d–%d",
+                    "Simulation config updated: anomalous=%.2f drone=%.2f dark=%.2f adsb_outage=%.2f "
+                    "aircraft=%d–%d live_adsb=%s live_dark=%.2f (%d/%d live aircraft dark)",
                     orchestrator.world.frac_anomalous,
                     orchestrator.world.frac_drone,
                     orchestrator.world.frac_dark,
                     orchestrator.world.frac_adsb_outage,
                     orchestrator.world.min_aircraft,
                     orchestrator.world.max_aircraft,
+                    "on" if orchestrator.live_adsb_enabled else "off",
+                    orchestrator.world.frac_live_dark,
+                    n_live_dark,
+                    len(orchestrator.world.live_aircraft),
                 )
 
                 # Scene-change detection. Absent scene stamp (stale volume,
@@ -1296,18 +1383,43 @@ def _near_any_metro(node):
     # time metro config alone switched this on — the "realistic mix" arrived
     # with ghost planes attached.
     adsb_metros = getattr(args, "metros", "") or getattr(args, "metro", "") or ""
-    if getattr(args, "real_adsb", False) and args.validation_url and adsb_metros:
-        metro_areas = _parse_metro_areas(adsb_metros)
-        if metro_areas:
+    # Live ADS-B seeding: real aircraft over the metro join the simulated
+    # world and are echoed by the synthetic nodes (see _seed_live_adsb).
+    # Needs an area to query, so it is scoped like the relay below.
+    live_url = (getattr(args, "live_adsb_url", "") or "").strip()
+    live_seeding = False
+    if live_url and adsb_metros:
+        live_areas = _parse_metro_areas(adsb_metros)
+        if live_areas:
+            live_seeding = True
             tasks.append(
-                _push_real_adsb(
+                _seed_live_adsb(
                     orchestrator,
-                    args.validation_url,
-                    areas=metro_areas,
-                    interval_s=10.0,
+                    live_url,
+                    areas=live_areas,
+                    interval_s=max(1.0, float(getattr(args, "live_adsb_interval", 5.0) or 5.0)),
                 )
             )
-    elif adsb_metros and args.validation_url:
+    elif live_url:
+        log.warning("Live ADS-B seeding needs --metro/--metros to know where to look — disabled")
+    if getattr(args, "real_adsb", False) and args.validation_url and adsb_metros:
+        if live_seeding:
+            # Both would push the same real hexes, one tagged source=real and
+            # one as simulated-world traffic, and the server keys its cache
+            # by hex — the two writers would take turns owning each entry.
+            log.warning("--real-adsb ignored: live ADS-B seeding already carries real traffic in the simulated world")
+        else:
+            metro_areas = _parse_metro_areas(adsb_metros)
+            if metro_areas:
+                tasks.append(
+                    _push_real_adsb(
+                        orchestrator,
+                        args.validation_url,
+                        areas=metro_areas,
+                        interval_s=10.0,
+                    )
+                )
+    elif adsb_metros and args.validation_url and not live_seeding:
         log.info("Real ADS-B relay disabled (pass --real-adsb to inject adsb.lol traffic)")
 
     if args.validate and args.validation_url:
@@ -1424,6 +1536,21 @@ def main():
         "config alone, and the decoy transponders it added produced ghost "
         "planes on the map.",
     )
+    parser.add_argument(
+        "--live-adsb-url",
+        type=str,
+        default="",
+        help="Base URL of an adsb.lol-shaped live feed (e.g. https://adsb.retina.fm). "
+        "Real aircraft over the --metro/--metros areas then join the simulated "
+        "world and are echoed by the synthetic nodes; the backend's "
+        "frac_live_dark casts a share of them as dark. Empty = off.",
+    )
+    parser.add_argument(
+        "--live-adsb-interval",
+        type=float,
+        default=5.0,
+        help="Seconds between live feed polls (min 1)",
+    )
     parser.add_argument(
         "--no-hub-radial",
         action="store_true",