osint-dashboard/app/fire_aircraft.py

"""Flag firefighting ADS-B aircraft within 20 miles of an active fire."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from uuid import uuid4

from sqlalchemy import text

from database import async_session
from live_layers import _haversine_km


RADIUS_MILES = 20.0
RADIUS_KM = RADIUS_MILES * 1.609344
_recent: dict[tuple[str, str], datetime] = {}
_COOLDOWN = timedelta(minutes=5)

# ICAO type designators commonly used on wildfire air tankers, scoopers,
# helitack, and air-attack platforms. Uppercase; compared case-insensitively.
FIREFIGHTER_ICAO = frozenset({
    "AT802", "AT8T", "AT8P", "AT8B",
    "C130", "C30J", "C130J",
    "DC10", "MD10", "MD11", "MD87",
    "B737", "B38M",
    "CL2T", "CL215", "CL415", "CL5T",
    "S64", "SK64",
    "UH1", "UH1Y", "UH60", "H60", "S70",
    "B412", "B212", "B205",
    "AS50", "AS350", "A119", "A109",
    "B350", "BE20",
    "OV10",
    "RJ85", "RJ1H", "B461", "B462", "B463",
    "PC12",
    "TBM7", "TBM8", "TBM9",
    "C208",
    "DH8D", "Q400",
})


def _icao(marker: dict) -> str:
    extra = marker.get("extra") or {}
    return str(extra.get("type") or extra.get("t") or "").strip().upper()


def is_firefighter(marker: dict) -> bool:
    return _icao(marker) in FIREFIGHTER_ICAO


def _haversine_mi(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    return _haversine_km(lat1, lon1, lat2, lon2) / 1.609344


def correlate_aircraft_to_fires(
    fires: list[dict],
    aircraft: list[dict],
    radius_mi: float = RADIUS_MILES,
) -> list[dict]:
    hits: list[dict] = []
    for fire in fires:
        flat, flon = fire.get("lat"), fire.get("lon")
        if flat is None or flon is None:
            continue
        fid = str(fire.get("id") or fire.get("label") or "fire")
        for ac in aircraft:
            if not is_firefighter(ac):
                continue
            alat, alon = ac.get("lat"), ac.get("lon")
            if alat is None or alon is None:
                continue
            dist = _haversine_mi(float(flat), float(flon), float(alat), float(alon))
            if dist > radius_mi:
                continue
            hits.append({
                "fire_id": fid,
                "fire_lat": float(flat),
                "fire_lon": float(flon),
                "aircraft_hex": str(ac.get("id")),
                "aircraft_type": _icao(ac),
                "aircraft_lat": float(alat),
                "aircraft_lon": float(alon),
                "distance_mi": round(dist, 2),
                "label": ac.get("label") or ac.get("id"),
            })
    return hits


async def persist_hits(hits: list[dict], seen_at: datetime | None = None) -> int:
    if not hits:
        return 0
    ts = seen_at or datetime.now(timezone.utc)
    n = 0
    async with async_session() as session:
        for h in hits:
            try:
                await session.execute(
                    text(
                        """
                        INSERT INTO fire_aircraft_hits
                            (id, fire_id, fire_lat, fire_lon,
                             aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
                             distance_mi, seen_at)
                        VALUES (
                            CAST(:id AS uuid), :fire_id, :fire_lat, :fire_lon,
                            :aircraft_hex, :aircraft_type, :aircraft_lat, :aircraft_lon,
                            :distance_mi, :seen_at
                        )
                        ON CONFLICT (fire_id, aircraft_hex, seen_at) DO NOTHING
                        """
                    ),
                    {
                        "id": str(uuid4()),
                        "fire_id": h["fire_id"],
                        "fire_lat": h["fire_lat"],
                        "fire_lon": h["fire_lon"],
                        "aircraft_hex": h["aircraft_hex"],
                        "aircraft_type": h["aircraft_type"],
                        "aircraft_lat": h["aircraft_lat"],
                        "aircraft_lon": h["aircraft_lon"],
                        "distance_mi": h["distance_mi"],
                        "seen_at": ts.replace(microsecond=0),
                    },
                )
                n += 1
            except Exception:
                continue
        try:
            await session.commit()
        except Exception:
            return 0
    return n


async def recent_hits(limit: int = 200) -> list[dict]:
    try:
        async with async_session() as session:
            rows = (await session.execute(
                text(
                    """
                    SELECT fire_id, fire_lat, fire_lon,
                           aircraft_hex, aircraft_type, aircraft_lat, aircraft_lon,
                           distance_mi, seen_at
                    FROM fire_aircraft_hits
                    ORDER BY seen_at DESC
                    LIMIT :limit
                    """
                ),
                {"limit": limit},
            )).mappings().all()
        out = []
        for r in rows:
            item = dict(r)
            if item.get("seen_at") is not None:
                item["seen_at"] = item["seen_at"].isoformat()
            out.append(item)
        return out
    except Exception:
        return []


async def correlate_and_notify(fires: list[dict], aircraft: list[dict]) -> list[dict]:
    hits = correlate_aircraft_to_fires(fires, aircraft)
    if not hits:
        return []
    now = datetime.now(timezone.utc)
    fresh = []
    for h in hits:
        key = (h["fire_id"], h["aircraft_hex"])
        prev = _recent.get(key)
        if prev is not None and now - prev < _COOLDOWN:
            continue
        _recent[key] = now
        fresh.append(h)
    if not fresh:
        return []
    await persist_hits(fresh, seen_at=now)
    from ws_manager import manager
    for h in fresh:
        body = {**h, "seen_at": now.isoformat()}
        await manager.publish_point(
            "fire_aircraft", body, lat=h["aircraft_lat"], lon=h["aircraft_lon"],
        )
        await manager.publish_point(
            "fire_aircraft", body, lat=h["fire_lat"], lon=h["fire_lon"],
        )
    return fresh