Add transponder outages (frac_adsb_outage)

retina_simulation/orchestrator.py

@@ -690,6 +690,10 @@ def build_ground_truth_payload(aircraft_summaries: list[dict]) -> list[dict]:
                 "object_type": ac.get("object_type", "aircraft"),
                 "is_anomalous": ac.get("is_anomalous", False),
                 "has_adsb": ac.get("has_adsb", False),
+                # Transponder present but currently silent.  has_adsb stays
+                # True — the truth is "this aircraft has a transponder and it
+                # is off right now", which is what verifies a known-track hold.
+                "adsb_silent": ac.get("adsb_silent", False),
                 "adsb_callsign": ac.get("adsb_callsign") or None,
                 "anomaly_event": ac.get("anomaly_event") or None,
             }
@@ -717,19 +721,26 @@ def build_real_adsb_body(payload: list[dict]) -> dict:
 def build_adsb_push_payload(aircraft_summaries: list[dict]) -> list[dict]:
     """Remap world aircraft summaries to the server ADS-B push schema.
 
-    Transponder-equipped aircraft only.  This push IS the simulated ADS-B
+    Transponder-equipped aircraft that are actually broadcasting only.  This push IS the simulated ADS-B
     broadcast, and a dark target by definition emits none — pushing it with
     its object id standing in for the hex minted a fake transponder per dark
     aircraft on the server, so every dark solve keyed mn-adsb-* and the dark
     lane stayed permanently empty.  Dark aircraft still reach the server
     through the ground-truth push, where the object id is the intended key
     (build_ground_truth_payload above).
+
+    An aircraft inside a transponder outage is skipped for the same reason:
+    this push IS the broadcast, so a silent aircraft must stop appearing in
+    it — otherwise the outage is invisible to the server and the known-track
+    hold it exists to exercise is never entered.
     """
     payload_aircraft = []
     for ac in aircraft_summaries:
         hex_code = ac.get("adsb_hex") or ""
         if not hex_code:
             continue
+        if ac.get("adsb_silent"):
+            continue
         speed_ms = ac.get("speed_ms", 0)
         payload_aircraft.append(
             {
@@ -928,16 +939,29 @@ def _fetch():
                 orchestrator.world.frac_anomalous = float(cfg.get("frac_anomalous", 0.0))
                 orchestrator.world.frac_drone = float(cfg.get("frac_drone", 0.0))
                 orchestrator.world.frac_dark = float(cfg.get("frac_dark", 0.15))
+                # Default 0.0 (matching SimulationWorld) so a payload missing
+                # the key cannot switch outages on.  Raising it also re-rolls
+                # aircraft already in the air — see schedule_adsb_outages.
+                orchestrator.world.frac_adsb_outage = float(cfg.get("frac_adsb_outage", 0.0))
+                if orchestrator.world.frac_adsb_outage > 0.0:
+                    n_sched = orchestrator.world.schedule_adsb_outages()
+                    if n_sched:
+                        log.info(
+                            "ADS-B outages scheduled for %d in-flight aircraft (frac %.2f)",
+                            n_sched,
+                            orchestrator.world.frac_adsb_outage,
+                        )
                 if "min_aircraft" in cfg:
                     orchestrator.world.min_aircraft = int(cfg["min_aircraft"])
                 if "max_aircraft" in cfg:
                     orchestrator.world.max_aircraft = int(cfg["max_aircraft"])
                 last_updated_at = updated_at
                 log.info(
-                    "Simulation config updated: anomalous=%.2f drone=%.2f dark=%.2f aircraft=%d–%d",
+                    "Simulation config updated: anomalous=%.2f drone=%.2f dark=%.2f adsb_outage=%.2f aircraft=%d–%d",
                     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,
                 )

retina_simulation/world.py

@@ -108,6 +108,18 @@ def waypoints_for_metro(metro: str | None) -> list[tuple[float, float]]:
     return _REGIONAL_WAYPOINTS.get(metro.strip().lower(), _US_WAYPOINTS)
 
 
+# Transponder-outage timing.  An outage has to outlast the backend's 45 s
+# fix-age cap and the 60 s map expiry to be observable as "ADS-B went away"
+# rather than as a dropped update, and has to be short enough that a 20-minute
+# capture sees several start and end — hence a minute to four minutes, opening
+# 20 s to 2 min after spawn (long enough that the track is established first).
+_ADSB_OUTAGE_START_S = (20.0, 120.0)
+_ADSB_OUTAGE_DURATION_S = (60.0, 240.0)
+# Re-roll of already-flying aircraft when the knob is raised at runtime opens
+# sooner: verification should not have to wait for a whole fleet turnover.
+_ADSB_OUTAGE_RUNTIME_START_S = (5.0, 60.0)
+
+
 @dataclass
 class SimulatedAircraft:
     """A simulated aircraft in the world with lat/lon/alt position."""
@@ -151,6 +163,22 @@ class SimulatedAircraft:
     anomaly_fired: bool = False  # True once the event has been applied
     _pre_spoof_lat: float = 0.0  # real position before GPS spoof
     _pre_spoof_lon: float = 0.0
+    # Transponder outage (scheduled window in world-clock seconds).  The
+    # aircraft still HAS a transponder — has_adsb and adsb_hex are untouched —
+    # it just stops broadcasting for the window, which is what a real failed
+    # or switched-off transponder looks like to every downstream consumer.
+    adsb_outage_start_s: float | None = None
+    adsb_outage_end_s: float | None = None
+    # World clock as of the last step(), so adsb_silent can be a plain
+    # property: the aircraft has no back-reference to its world.
+    sim_now_s: float = 0.0
+
+    @property
+    def adsb_silent(self) -> bool:
+        """True while the world clock sits inside this aircraft's outage."""
+        if self.adsb_outage_start_s is None or self.adsb_outage_end_s is None:
+            return False
+        return self.adsb_outage_start_s <= self.sim_now_s < self.adsb_outage_end_s
 
 
 @dataclass
@@ -367,6 +395,11 @@ def __init__(
         # stay matched to this value.
         self.frac_drone: float = 0.0
         self.frac_dark: float = 0.15
+        # Fraction of ADS-B-equipped aircraft that suffer a transponder
+        # outage mid-flight.  Orthogonal to the frac_* type roll above (it is
+        # a fraction OF the ADS-B population, not of all spawns), and OFF by
+        # default so nothing changes for anyone who does not set it.
+        self.frac_adsb_outage: float = 0.0
         # remaining fraction = commercial aircraft with ADS-B
         # Hub-radial flight planning: when metro_cells is non-empty, this
         # fraction of spawns is routed through a metro cell; the rest are
@@ -586,6 +619,8 @@ def _spawn_aircraft(self, mode: str = "detection") -> SimulatedAircraft:
             letters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
             adsb_callsign = f"{''.join(random.choices(letters, k=3))}{random.randint(100, 9999)}"
 
+        outage_start, outage_end = self._roll_adsb_outage(has_adsb, _ADSB_OUTAGE_START_S)
+
         # Initial heading toward next waypoint
         next_wp = route[1] if len(route) > 1 else route[0]
         heading = _bearing_deg(lat, lon, next_wp[0], next_wp[1])
@@ -611,13 +646,51 @@ def _spawn_aircraft(self, mode: str = "detection") -> SimulatedAircraft:
             object_type=object_type,
             adsb_hex=adsb_hex,
             adsb_callsign=adsb_callsign,
+            adsb_outage_start_s=outage_start,
+            adsb_outage_end_s=outage_end,
+            sim_now_s=self._time,
             created_at=self._time,
             lifetime_s=random.uniform(180, 900) if object_type != "drone" else random.uniform(60, 300),
             waypoints=route,
             waypoint_idx=1,
             **self._maybe_schedule_anomaly(is_anomalous, object_type),
         )
 
+    def _roll_adsb_outage(self, has_adsb: bool, start_range: tuple[float, float]) -> tuple[float | None, float | None]:
+        """Roll a transponder-outage window, or (None, None) for no outage.
+
+        Dark aircraft are never candidates: they have no transponder to lose.
+        """
+        if not has_adsb or self.frac_adsb_outage <= 0.0:
+            return None, None
+        if random.random() >= self.frac_adsb_outage:
+            return None, None
+        start = self._time + random.uniform(*start_range)
+        return start, start + random.uniform(*_ADSB_OUTAGE_DURATION_S)
+
+    def schedule_adsb_outages(self) -> int:
+        """Roll outages for live ADS-B aircraft that have none scheduled yet.
+
+        Called when frac_adsb_outage is raised at runtime: without it the knob
+        only reaches aircraft spawned after the change, so verification would
+        have to wait for a full fleet turnover.  Aircraft that already carry a
+        window (including one that has already ended) are left alone, so
+        repeated config polls cannot re-roll the same aircraft every 5 s.
+        Returns the number newly scheduled.
+        """
+        scheduled = 0
+        for ac in self.aircraft:
+            if ac.adsb_outage_start_s is not None:
+                continue
+            start, end = self._roll_adsb_outage(ac.has_adsb, _ADSB_OUTAGE_RUNTIME_START_S)
+            if start is None:
+                continue
+            ac.adsb_outage_start_s = start
+            ac.adsb_outage_end_s = end
+            ac.sim_now_s = self._time
+            scheduled += 1
+        return scheduled
+
     # Expired aircraft are retired only once they are at least this far from
     # the world center — a target vanishing overhead reads as a tracking bug
     # on the map, so retirement happens off at the edges, beyond the ~60 km
@@ -691,6 +764,7 @@ def step(self, dt: float, mode: str = "detection"):
 
         # Update each aircraft
         for ac in self.aircraft:
+            ac.sim_now_s = self._time
             self._update_aircraft(ac, dt)
 
         self._enforce_separation(dt)
@@ -1000,8 +1074,10 @@ def generate_detections_for_node(self, node_id: str, timestamp_ms: int) -> dict:
             dopplers.append(round(doppler, 2))
             snrs.append(round(snr, 2))
 
-            # ADS-B entry
-            if ac.has_adsb:
+            # ADS-B entry.  A silent transponder gets exactly what a dark
+            # aircraft gets — a None slot — so the outage is indistinguishable
+            # downstream from "this echo carries no ADS-B tag".
+            if ac.has_adsb and not ac.adsb_silent:
                 has_any_adsb = True
                 speed_ms = ac.speed_km_s * 1000
                 # GPS spoof: ADS-B reports frozen pre-spoof position while
@@ -1069,6 +1145,7 @@ def get_aircraft_summary(self) -> list[dict]:
                 "object_type": ac.object_type,
                 "adsb_hex": ac.adsb_hex,
                 "adsb_callsign": ac.adsb_callsign,
+                "adsb_silent": ac.adsb_silent,
                 "anomaly_event": ac.anomaly_event,
             }
             for ac in self.aircraft