mirror of
https://github.com/BigBodyCobain/Shadowbroker.git
synced 2026-05-28 01:52:28 +02:00
fix(gps-jamming): count nac_p=0 + lower thresholds so the layer actually fires
Three stacked filters meant the gps_jamming layer almost never lit up: 1. nac_p == 0 aircraft were dropped on the theory that "0 = old transponder." That's only half right — modern Mode-S Enhanced Surveillance transponders also fall back to nac_p=0 when they lose GPS lock entirely, which IS the jamming signature we want to catch. Discarding them was discarding the strongest signal. None (no field at all — typical for OpenSky-sourced records) is still skipped because absence-of-data isn't evidence. 2. GPS_JAMMING_MIN_AIRCRAFT was 5 per 1°x1° cell. Jamming hotspots (eastern Med, Russia/Ukraine border, Iran/Iraq) tend to have sparser traffic because pilots avoid them. Lowered to 3. 3. GPS_JAMMING_MIN_RATIO was 0.30. Combined with the (preserved) -1 noise cushion that made the effective bar high. Lowered to 0.20. The 1-aircraft noise cushion is intact so a single quirky transponder still can't flag a zone alone. Also extracted the detector loop into a pure ``detect_gps_jamming_zones()`` function at module scope so it's testable in isolation (was previously inlined inside ``_classify_and_publish``). The public signature accepts threshold overrides for ad-hoc re-tuning without code edits. 16 new tests cover nac_p=0 inclusion, None-skip preservation, MIN_AIRCRAFT lowering, MIN_RATIO lowering, noise cushion preservation, constant pinning, override behavior, lon/lng key compatibility, and robustness to empty/None inputs. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
BigBodyCobain
@@ -11,8 +11,13 @@ DEFAULT_TRAIL_TTL_S = 300 # 5 min - trail TTL for non-tracked flights
|
||||
HOLD_PATTERN_DEGREES = 300 # Total heading change to flag holding pattern
|
||||
GPS_JAMMING_NACP_THRESHOLD = 8 # NACp below this = degraded GPS signal
|
||||
GPS_JAMMING_GRID_SIZE = 1.0 # 1 degree grid for aggregation
|
||||
GPS_JAMMING_MIN_RATIO = 0.30 # 30% degraded aircraft to flag zone
|
||||
GPS_JAMMING_MIN_AIRCRAFT = 5 # Min aircraft in grid cell for statistical significance
|
||||
# Tuned 2026-05: previously 0.30 / 5 aircraft which — combined with the
|
||||
# -1 noise cushion in the detector AND the pre-fix nac_p==0 filter that
|
||||
# discarded jamming victims — meant the layer almost never lit up.
|
||||
# Lowering the bar so genuine jamming zones with sparser ADS-B coverage
|
||||
# clear (eastern Med, Russia/Ukraine border, Iran/Iraq).
|
||||
GPS_JAMMING_MIN_RATIO = 0.20 # 20% degraded aircraft to flag zone
|
||||
GPS_JAMMING_MIN_AIRCRAFT = 3 # Min aircraft in grid cell for statistical significance
|
||||
|
||||
# ─── Network & Circuit Breaker ──────────────────────────────────────────────
|
||||
CIRCUIT_BREAKER_TTL_S = 120 # Skip domain for 2 min after total failure
|
||||
|
||||
@@ -29,6 +29,88 @@ _RE_AIRLINE_CODE_1 = re.compile(r"^([A-Z]{3})\d")
|
||||
_RE_AIRLINE_CODE_2 = re.compile(r"^([A-Z]{3})[A-Z\d]")
|
||||
|
||||
|
||||
def detect_gps_jamming_zones(
|
||||
raw_flights: list[dict],
|
||||
*,
|
||||
min_aircraft: int | None = None,
|
||||
min_ratio: float | None = None,
|
||||
nacp_threshold: int | None = None,
|
||||
) -> list[dict]:
|
||||
"""Detect GPS interference zones from a snapshot of raw ADS-B aircraft.
|
||||
|
||||
Methodology mirrors GPSJam.org / Flightradar24: bin aircraft into 1°x1°
|
||||
grid cells, flag cells where the fraction of aircraft reporting degraded
|
||||
NACp clears a threshold.
|
||||
|
||||
Inputs
|
||||
------
|
||||
raw_flights:
|
||||
Iterable of dicts. Each item is expected to carry ``lat``, ``lng``
|
||||
(or ``lon``), and ``nac_p``. Records missing position OR missing
|
||||
``nac_p`` entirely (typical for OpenSky-sourced flights) are
|
||||
skipped — absence-of-data isn't evidence of anything.
|
||||
|
||||
nac_p == 0 IS counted as degraded. Pre-fix code skipped it on the theory
|
||||
that "0 = old transponder, never computed accuracy." That's only half
|
||||
right: modern Mode-S Enhanced Surveillance transponders also fall back
|
||||
to nac_p=0 when they lose GPS lock entirely — which is exactly the
|
||||
jamming signature we're trying to detect. Filtering 0 out was discarding
|
||||
the strongest evidence.
|
||||
|
||||
Denoising:
|
||||
1. Require ``min_aircraft`` per grid cell for statistical validity.
|
||||
2. Subtract 1 from degraded count per cell (GPSJam's technique) so
|
||||
a single quirky transponder can't flag an entire zone.
|
||||
3. Require ratio ``adjusted_degraded / total > min_ratio``.
|
||||
|
||||
All thresholds default to the module-level constants but can be
|
||||
overridden for testing.
|
||||
"""
|
||||
min_aircraft = GPS_JAMMING_MIN_AIRCRAFT if min_aircraft is None else int(min_aircraft)
|
||||
min_ratio = GPS_JAMMING_MIN_RATIO if min_ratio is None else float(min_ratio)
|
||||
nacp_threshold = (
|
||||
GPS_JAMMING_NACP_THRESHOLD if nacp_threshold is None else int(nacp_threshold)
|
||||
)
|
||||
|
||||
jamming_grid: dict[str, dict[str, int]] = {}
|
||||
for rf in raw_flights or []:
|
||||
rlat = rf.get("lat")
|
||||
rlng = rf.get("lng") if rf.get("lng") is not None else rf.get("lon")
|
||||
if rlat is None or rlng is None:
|
||||
continue
|
||||
nacp = rf.get("nac_p")
|
||||
if nacp is None:
|
||||
continue
|
||||
grid_key = f"{int(rlat)},{int(rlng)}"
|
||||
cell = jamming_grid.setdefault(grid_key, {"degraded": 0, "total": 0})
|
||||
cell["total"] += 1
|
||||
if nacp < nacp_threshold:
|
||||
cell["degraded"] += 1
|
||||
|
||||
jamming_zones: list[dict] = []
|
||||
for gk, counts in jamming_grid.items():
|
||||
if counts["total"] < min_aircraft:
|
||||
continue
|
||||
adjusted_degraded = max(counts["degraded"] - 1, 0)
|
||||
if adjusted_degraded == 0:
|
||||
continue
|
||||
ratio = adjusted_degraded / counts["total"]
|
||||
if ratio > min_ratio:
|
||||
lat_i, lng_i = gk.split(",")
|
||||
severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
|
||||
jamming_zones.append(
|
||||
{
|
||||
"lat": int(lat_i) + 0.5,
|
||||
"lng": int(lng_i) + 0.5,
|
||||
"severity": severity,
|
||||
"ratio": round(ratio, 2),
|
||||
"degraded": counts["degraded"],
|
||||
"total": counts["total"],
|
||||
}
|
||||
)
|
||||
return jamming_zones
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# OpenSky Network API Client (OAuth2)
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -724,56 +806,8 @@ def _classify_and_publish(all_adsb_flights):
|
||||
latest_data["military_flights"] = military_snapshot
|
||||
|
||||
# --- GPS Jamming Detection ---
|
||||
# Uses NACp (Navigation Accuracy Category – Position) from ADS-B to infer
|
||||
# GPS interference zones, similar to GPSJam.org / Flightradar24.
|
||||
# NACp < 8 = position accuracy worse than the FAA-mandated 0.05 NM.
|
||||
#
|
||||
# Denoising (to suppress false positives from old GA transponders):
|
||||
# 1. Skip nac_p == 0 ("unknown accuracy") — old transponders that never
|
||||
# computed accuracy, NOT evidence of jamming. Real jamming shows 1-7.
|
||||
# 2. Require minimum aircraft per grid cell for statistical validity.
|
||||
# 3. Subtract 1 from degraded count per cell (GPSJam's technique) so a
|
||||
# single quirky transponder can't flag an entire zone.
|
||||
# 4. Require the adjusted ratio to exceed the threshold.
|
||||
try:
|
||||
jamming_grid = {}
|
||||
raw_flights = raw_flights_snapshot
|
||||
for rf in raw_flights:
|
||||
rlat = rf.get("lat")
|
||||
rlng = rf.get("lng") or rf.get("lon")
|
||||
if rlat is None or rlng is None:
|
||||
continue
|
||||
nacp = rf.get("nac_p")
|
||||
if nacp is None or nacp == 0:
|
||||
continue
|
||||
grid_key = f"{int(rlat)},{int(rlng)}"
|
||||
if grid_key not in jamming_grid:
|
||||
jamming_grid[grid_key] = {"degraded": 0, "total": 0}
|
||||
jamming_grid[grid_key]["total"] += 1
|
||||
if nacp < GPS_JAMMING_NACP_THRESHOLD:
|
||||
jamming_grid[grid_key]["degraded"] += 1
|
||||
|
||||
jamming_zones = []
|
||||
for gk, counts in jamming_grid.items():
|
||||
if counts["total"] < GPS_JAMMING_MIN_AIRCRAFT:
|
||||
continue
|
||||
adjusted_degraded = max(counts["degraded"] - 1, 0)
|
||||
if adjusted_degraded == 0:
|
||||
continue
|
||||
ratio = adjusted_degraded / counts["total"]
|
||||
if ratio > GPS_JAMMING_MIN_RATIO:
|
||||
lat_i, lng_i = gk.split(",")
|
||||
severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
|
||||
jamming_zones.append(
|
||||
{
|
||||
"lat": int(lat_i) + 0.5,
|
||||
"lng": int(lng_i) + 0.5,
|
||||
"severity": severity,
|
||||
"ratio": round(ratio, 2),
|
||||
"degraded": counts["degraded"],
|
||||
"total": counts["total"],
|
||||
}
|
||||
)
|
||||
jamming_zones = detect_gps_jamming_zones(raw_flights_snapshot)
|
||||
with _data_lock:
|
||||
latest_data["gps_jamming"] = jamming_zones
|
||||
if jamming_zones:
|
||||
|
||||
@@ -0,0 +1,333 @@
|
||||
"""GPS jamming detection — nac_p=0 counted, lowered thresholds.
|
||||
|
||||
Background
|
||||
----------
|
||||
Pre-fix, the detector had three stacked filters that together meant the
|
||||
``gps_jamming`` layer almost never lit up:
|
||||
|
||||
1. ``nac_p == 0`` aircraft were dropped on the theory that "0 = old
|
||||
transponder." But modern Mode-S Enhanced Surveillance transponders
|
||||
also fall back to ``nac_p == 0`` when they lose GPS lock entirely —
|
||||
which is *exactly* the jamming signature we want to catch.
|
||||
2. ``GPS_JAMMING_MIN_AIRCRAFT = 5`` per 1°x1° cell.
|
||||
3. ``GPS_JAMMING_MIN_RATIO = 0.30`` adjusted ratio.
|
||||
|
||||
Combined with the existing ``-1`` noise cushion (``adjusted = degraded - 1``)
|
||||
the bar to clear required dense, busy airspace — but jamming hotspots
|
||||
(eastern Med, eastern Ukraine, Iran/Iraq) tend to have sparser traffic
|
||||
precisely because pilots avoid them.
|
||||
|
||||
These tests pin the new behavior:
|
||||
|
||||
* ``nac_p == 0`` is now counted as degraded.
|
||||
* ``nac_p == None`` (no field — typical for OpenSky records) is still
|
||||
skipped — absence isn't evidence.
|
||||
* Thresholds lowered to 3 aircraft / 0.20 ratio.
|
||||
* Public function signature accepts overrides so callers / future
|
||||
operators can re-tune without code edits.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# nac_p == 0 inclusion (the headline fix)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNacpZeroCounted:
|
||||
def test_cell_dominated_by_nacp_zero_now_fires(self):
|
||||
"""Three aircraft all reporting nac_p=0 in one cell, plus two
|
||||
with valid GPS. Pre-fix the three nac_p=0 records were skipped
|
||||
entirely (cell would have total=2, degraded=0, no zone). Post-fix
|
||||
they count as degraded — this IS the jamming signature."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
# All in 1°x1° cell at int(lat)=40, int(lng)=-100
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
|
||||
{"hex": "a2", "lat": 40.5, "lng": -100.5, "nac_p": 0},
|
||||
{"hex": "a3", "lat": 40.9, "lng": -100.9, "nac_p": 0},
|
||||
{"hex": "b1", "lat": 40.2, "lng": -100.3, "nac_p": 9},
|
||||
{"hex": "b2", "lat": 40.7, "lng": -100.7, "nac_p": 11},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# total=5, degraded=3, adjusted=2, ratio=0.40 > 0.20 → zone fires.
|
||||
assert len(zones) == 1
|
||||
assert zones[0]["degraded"] == 3
|
||||
assert zones[0]["total"] == 5
|
||||
assert zones[0]["ratio"] == 0.40
|
||||
# Grid-cell center coords.
|
||||
assert zones[0]["lat"] == 40.5
|
||||
assert zones[0]["lng"] == -99.5
|
||||
|
||||
def test_nacp_zero_alone_clears_min_aircraft(self):
|
||||
"""A cell with exactly 3 aircraft all reporting nac_p=0 must
|
||||
fire under the new MIN_AIRCRAFT=3 + MIN_RATIO=0.20 regime."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 50.1, "lng": 30.1, "nac_p": 0},
|
||||
{"hex": "a2", "lat": 50.5, "lng": 30.5, "nac_p": 0},
|
||||
{"hex": "a3", "lat": 50.9, "lng": 30.9, "nac_p": 0},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# total=3, degraded=3, adjusted=2, ratio=0.667 > 0.20 → fires.
|
||||
# severity is "medium" because 0.5 ≤ ratio < 0.75.
|
||||
assert len(zones) == 1
|
||||
assert zones[0]["severity"] == "medium"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# nac_p == None is still skipped (preserve OpenSky behavior)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNoneStillSkipped:
|
||||
def test_none_records_dont_add_to_grid(self):
|
||||
"""OpenSky's /states/all doesn't include nac_p, so its records
|
||||
arrive with the field absent (``rf.get("nac_p") is None``). These
|
||||
records must NOT count toward total — absence-of-data isn't
|
||||
evidence of either jamming OR working GPS."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
# 3 jammed + 4 OpenSky-style (no nac_p). Pre-fix and post-fix
|
||||
# behavior should be identical here: None always skipped.
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
|
||||
{"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 0},
|
||||
{"hex": "a3", "lat": 40.3, "lng": -100.3, "nac_p": 0},
|
||||
# OpenSky-style: no nac_p at all
|
||||
{"hex": "o1", "lat": 40.4, "lng": -100.4},
|
||||
{"hex": "o2", "lat": 40.5, "lng": -100.5},
|
||||
{"hex": "o3", "lat": 40.6, "lng": -100.6},
|
||||
{"hex": "o4", "lat": 40.7, "lng": -100.7},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# Only the 3 nac_p=0 records hit the grid. total=3, not 7.
|
||||
assert len(zones) == 1
|
||||
assert zones[0]["total"] == 3
|
||||
assert zones[0]["degraded"] == 3
|
||||
|
||||
def test_explicit_none_skipped(self):
|
||||
"""Same behavior when ``nac_p`` is present but set to None
|
||||
(defensive — adsb.lol shouldn't do this, but downstream
|
||||
normalizers might)."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 0.1, "lng": 0.1, "nac_p": None},
|
||||
{"hex": "a2", "lat": 0.2, "lng": 0.2, "nac_p": None},
|
||||
{"hex": "a3", "lat": 0.3, "lng": 0.3, "nac_p": None},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# No records counted → no zones.
|
||||
assert zones == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lowered MIN_AIRCRAFT (5 → 3)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMinAircraftLowered:
|
||||
def test_three_aircraft_cell_now_qualifies(self):
|
||||
"""Pre-fix MIN_AIRCRAFT=5 blocked sparse cells entirely. Post-fix
|
||||
the bar is 3 aircraft per cell, which is realistic for the actual
|
||||
jamming hotspots where traffic is thinner."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
|
||||
{"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 5},
|
||||
{"hex": "a3", "lat": 33.3, "lng": 44.3, "nac_p": 7},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# total=3, degraded=3, adjusted=2, ratio=0.667 — fires under new
|
||||
# rules, would have been blocked by MIN_AIRCRAFT=5 pre-fix.
|
||||
assert len(zones) == 1
|
||||
|
||||
def test_two_aircraft_cell_still_blocked(self):
|
||||
"""We didn't lower the bar to 2 — that would create too much
|
||||
single-transponder noise. Two aircraft per cell still doesn't
|
||||
qualify."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
|
||||
{"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 3},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
assert zones == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lowered MIN_RATIO (0.30 → 0.20)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMinRatioLowered:
|
||||
def test_ratio_between_old_and_new_threshold_fires(self):
|
||||
"""Construct a cell whose ratio sits in the (0.20, 0.30) window:
|
||||
fires under the new bar, would have been blocked pre-fix."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
# 10 aircraft, 4 degraded → adjusted=3, ratio=3/10=0.30.
|
||||
# Pre-fix threshold was > 0.30 strict — would NOT fire.
|
||||
# Post-fix threshold is > 0.20 — fires.
|
||||
feed = (
|
||||
[{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
|
||||
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(6)]
|
||||
)
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
assert len(zones) == 1
|
||||
assert zones[0]["degraded"] == 4
|
||||
assert zones[0]["total"] == 10
|
||||
assert zones[0]["ratio"] == 0.30
|
||||
|
||||
def test_ratio_at_or_below_new_threshold_does_not_fire(self):
|
||||
"""Ratio of exactly 0.20 must NOT fire (strict ``>`` comparison)."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
# 15 aircraft, 4 degraded → adjusted=3, ratio=3/15=0.20. Strictly
|
||||
# not greater than 0.20, so doesn't qualify.
|
||||
feed = (
|
||||
[{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
|
||||
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(11)]
|
||||
)
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
assert zones == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pre-existing noise cushion (-1) preserved
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNoiseCushionPreserved:
|
||||
def test_single_quirky_transponder_doesnt_fire(self):
|
||||
"""One degraded aircraft in a healthy cell shouldn't fire even
|
||||
under the relaxed thresholds. The ``-1`` adjustment in the
|
||||
detector exists for this reason."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = (
|
||||
[{"hex": "d1", "lat": 40.1, "lng": -100.1, "nac_p": 3}]
|
||||
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(10)]
|
||||
)
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
# total=11, degraded=1, adjusted=0 → cell short-circuits.
|
||||
assert zones == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants pinned (catches accidental rollback)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestConstantsPinned:
|
||||
def test_min_aircraft_is_three(self):
|
||||
from services.constants import GPS_JAMMING_MIN_AIRCRAFT
|
||||
assert GPS_JAMMING_MIN_AIRCRAFT == 3, (
|
||||
"MIN_AIRCRAFT must be 3; raising it back to 5 brings back the "
|
||||
"'jamming never shows' bug."
|
||||
)
|
||||
|
||||
def test_min_ratio_is_0_20(self):
|
||||
from services.constants import GPS_JAMMING_MIN_RATIO
|
||||
assert GPS_JAMMING_MIN_RATIO == 0.20, (
|
||||
"MIN_RATIO must be 0.20; raising it back to 0.30 brings back "
|
||||
"the 'jamming never shows' bug."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Overrides honored
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestOverridesHonored:
|
||||
def test_overrides_supersede_constants(self):
|
||||
"""The public signature accepts overrides so an operator can
|
||||
re-tune at the call site (e.g. for a more aggressive setup in
|
||||
an active conflict zone) without editing the module constants."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 3},
|
||||
{"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 3},
|
||||
]
|
||||
|
||||
# With defaults (min_aircraft=3) this is blocked. With override=2 it fires.
|
||||
assert detect_gps_jamming_zones(feed) == []
|
||||
zones = detect_gps_jamming_zones(feed, min_aircraft=2)
|
||||
assert len(zones) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# lon vs lng compatibility
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestLonLngCompat:
|
||||
def test_lon_key_accepted(self):
|
||||
"""adsb.lol records arrive with ``lon`` (no g). The OpenSky merge
|
||||
normalizes to ``lng`` but raw records flowing into the detector
|
||||
may use either. Make sure both work."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
|
||||
feed = [
|
||||
{"hex": "a1", "lat": 40.1, "lon": -100.1, "nac_p": 0},
|
||||
{"hex": "a2", "lat": 40.2, "lon": -100.2, "nac_p": 0},
|
||||
{"hex": "a3", "lat": 40.3, "lon": -100.3, "nac_p": 0},
|
||||
]
|
||||
|
||||
zones = detect_gps_jamming_zones(feed)
|
||||
|
||||
assert len(zones) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Empty / malformed inputs don't crash
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRobustness:
|
||||
def test_empty_feed(self):
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
assert detect_gps_jamming_zones([]) == []
|
||||
|
||||
def test_none_feed(self):
|
||||
"""The wrapper at the call site passes ``raw_flights_snapshot``
|
||||
which could in principle be None on a startup race. Handle it."""
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
assert detect_gps_jamming_zones(None) == []
|
||||
|
||||
def test_records_missing_position_skipped(self):
|
||||
from services.fetchers.flights import detect_gps_jamming_zones
|
||||
feed = [
|
||||
{"hex": "noloc", "nac_p": 0},
|
||||
{"hex": "nolat", "lng": -100.0, "nac_p": 0},
|
||||
{"hex": "nolng", "lat": 40.0, "nac_p": 0},
|
||||
]
|
||||
assert detect_gps_jamming_zones(feed) == []
|
||||
Reference in New Issue
Block a user