93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
"""Live flight overlay, backed by the OpenSky Network REST API.
|
|
|
|
Positions are high-churn and ephemeral — unlike articles/markets/conflict
|
|
events we don't persist history to SQLite, just keep the latest global
|
|
snapshot in memory and let the scheduler refresh it.
|
|
|
|
Military classification is a best-effort heuristic (see
|
|
data/military_ranges.yaml for sourcing and caveats) — there is no public
|
|
"is this military" flag on live ADS-B data.
|
|
"""
|
|
|
|
import logging
|
|
from functools import lru_cache
|
|
|
|
import httpx
|
|
import yaml
|
|
|
|
from .config import settings
|
|
|
|
log = logging.getLogger("newsatlas.flights")
|
|
|
|
STATES_URL = "https://opensky-network.org/api/states/all"
|
|
|
|
_latest_flights: list[dict] = []
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def _military_rules() -> tuple[list[str], list[str]]:
|
|
with open(settings.military_ranges_file, encoding="utf-8") as f:
|
|
data = yaml.safe_load(f) or {}
|
|
icao_prefixes = [r["prefix"].lower() for r in data.get("icao24_prefixes", [])]
|
|
callsign_prefixes = [r["prefix"].upper() for r in data.get("callsign_prefixes", [])]
|
|
return icao_prefixes, callsign_prefixes
|
|
|
|
|
|
def _is_military(icao24: str, callsign: str) -> bool:
|
|
icao_prefixes, callsign_prefixes = _military_rules()
|
|
icao24 = (icao24 or "").lower()
|
|
callsign = (callsign or "").strip().upper()
|
|
if any(icao24.startswith(p) for p in icao_prefixes):
|
|
return True
|
|
if any(callsign.startswith(p) for p in callsign_prefixes):
|
|
return True
|
|
return False
|
|
|
|
|
|
def poll_flights() -> int:
|
|
global _latest_flights
|
|
|
|
auth = None
|
|
if settings.opensky_username and settings.opensky_password:
|
|
auth = (settings.opensky_username, settings.opensky_password)
|
|
|
|
try:
|
|
resp = httpx.get(STATES_URL, auth=auth, timeout=30)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
except Exception:
|
|
log.exception("Failed to fetch OpenSky states")
|
|
return 0
|
|
|
|
flights = []
|
|
for s in payload.get("states") or []:
|
|
icao24, callsign, origin_country = s[0], s[1], s[2]
|
|
lon, lat, baro_altitude, on_ground = s[5], s[6], s[7], s[8]
|
|
velocity, heading, vertical_rate = s[9], s[10], s[11]
|
|
if lat is None or lon is None or on_ground:
|
|
continue
|
|
|
|
callsign = (callsign or "").strip()
|
|
flights.append(
|
|
{
|
|
"icao24": icao24,
|
|
"callsign": callsign,
|
|
"origin_country": origin_country,
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"altitude_m": baro_altitude,
|
|
"velocity_ms": velocity,
|
|
"heading": heading,
|
|
"vertical_rate_ms": vertical_rate,
|
|
"is_military": _is_military(icao24, callsign),
|
|
}
|
|
)
|
|
|
|
_latest_flights = flights
|
|
log.info("Flight poll complete: %d airborne aircraft (%d flagged military)", len(flights), sum(f["is_military"] for f in flights))
|
|
return len(flights)
|
|
|
|
|
|
def get_latest_flights() -> list[dict]:
|
|
return _latest_flights
|