"""Live map overlays: parsers, TTL cache, and upstream fetchers. Moving objects (aircraft, vessels, trains) and alerts are vectors. Radar / GIBS fire tiles are rasters served directly to the browser — this module only returns tile *templates* and GeoJSON/JSON for vectors. Third-party APIs that leak keys, lack CORS, or rate-limit by IP are fetched here (FastAPI), never from Leaflet. Viewport bbox only; never a global ADS-B or AIS poll. """ from __future__ import annotations import asyncio import math import time from datetime import datetime, timezone from typing import Any, Awaitable, Callable import httpx from config import OSINT_USER_AGENT MARKER_FIELDS = ("id", "lat", "lon", "heading", "speed", "label", "extra") ADSB_LOL_BASE = "https://api.adsb.lol" AMTRAKER_TRAINS = "https://api.amtraker.com/v3/trains" RAINVIEWER_MAPS = "https://api.rainviewer.com/public/weather-maps.json" NWS_ALERTS = "https://api.weather.gov/alerts/active" IEM_SBW = "https://mesonet.agron.iastate.edu/geojson/sbw.geojson" WFIGS_INCIDENTS = ( "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" "WFIGS_Incident_Locations_Current/FeatureServer/0/query" ) WFIGS_PERIMETERS = ( "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" "WFIGS_Interagency_Perimeters_Current/FeatureServer/0/query" ) NHC_STORMS = "https://www.nhc.noaa.gov/CurrentStorms.json" IEM_NEXRAD = "https://mesonet.agron.iastate.edu/cache/tile.py/1.0.0/nexrad-n0q/{z}/{x}/{y}.png" GIBS_THERMAL = ( "https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/" "VIIRS_SNPP_Thermal_Anomalies_375m_All/default/{time}/" "GoogleMapsCompatible_Level9/{z}/{y}/{x}.png" ) CONUS = (-125.0, 24.0, -66.0, 50.0) MAX_RADIUS_NM = 150 DEFAULT_LIMIT = 2000 _HTTP_TIMEOUT = 25.0 _COMPASS = { "N": 0, "NE": 45, "E": 90, "SE": 135, "S": 180, "SW": 225, "W": 270, "NW": 315, "NNE": 22, "ENE": 67, "ESE": 112, "SSE": 157, "SSW": 202, "WSW": 247, "WNW": 292, "NNW": 337, } _cache: dict[str, tuple[float, Any]] = {} _cache_lock = asyncio.Lock() # Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker. vessel_last_known: dict[str, dict] = {} vessel_lock = asyncio.Lock() def overlay_catalog() -> dict: """Tile templates + vector endpoint map for the layer panel. No secrets.""" today = datetime.now(timezone.utc).date().isoformat() return { "radar_iem": { "id": "radar_iem", "title": "IEM NEXRAD (CONUS)", "kind": "raster", "tileUrl": IEM_NEXRAD, "maxZoom": 18, "attribution": "Iowa Environmental Mesonet", }, "radar_rainviewer": { "id": "radar_rainviewer", "title": "RainViewer (global)", "kind": "raster", "tileUrl": None, # filled from /api/map/radar frames "maxZoom": 7, "attribution": 'Weather data by RainViewer', }, "gibs_thermal": { "id": "gibs_thermal", "title": "GIBS VIIRS thermal anomalies", "kind": "raster", "tileUrl": GIBS_THERMAL.replace("{time}", today), "timeTemplate": GIBS_THERMAL, "maxZoom": 9, "attribution": "NASA GIBS / EOSDIS", }, "nws_alerts": {"id": "nws_alerts", "kind": "geojson", "endpoint": "/api/weather-alerts"}, "wfigs_incidents": {"id": "wfigs_incidents", "kind": "points", "endpoint": "/api/fire-incidents"}, "wfigs_perimeters": {"id": "wfigs_perimeters", "kind": "geojson", "endpoint": "/api/fire-perimeters"}, "aircraft": {"id": "aircraft", "kind": "points", "endpoint": "/api/aircraft"}, "vessels": {"id": "vessels", "kind": "points", "endpoint": "/api/vessels"}, "trains": {"id": "trains", "kind": "points", "endpoint": "/api/trains"}, "storms": {"id": "storms", "kind": "points", "endpoint": "/api/storms"}, } def parse_bbox(bbox: str) -> tuple[float, float, float, float]: """Parse 'minlon,minlat,maxlon,maxlat' into four floats.""" parts = [p.strip() for p in (bbox or "").split(",")] if len(parts) != 4: raise ValueError("bbox must be 'minlon,minlat,maxlon,maxlat'") try: minlon, minlat, maxlon, maxlat = (float(p) for p in parts) except ValueError as exc: raise ValueError("bbox values must be floats") from exc return minlon, minlat, maxlon, maxlat def bbox_center_radius_nm( minlon: float, minlat: float, maxlon: float, maxlat: float, ) -> tuple[float, float, int]: """Viewport center + half-diagonal radius in nautical miles, clamped ≤ 150.""" lat = (minlat + maxlat) / 2.0 lon = (minlon + maxlon) / 2.0 # Half the diagonal of the box, in nm (1 deg lat ≈ 60 nm). dlat = abs(maxlat - minlat) / 2.0 dlon = abs(maxlon - minlon) / 2.0 km = _haversine_km(lat, lon, lat + dlat, lon + dlon) nm = km / 1.852 radius = max(1, min(MAX_RADIUS_NM, int(math.ceil(nm)))) return lat, lon, radius def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: r = 6371.0 p1, p2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlmb = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2 return 2 * r * math.asin(min(1.0, math.sqrt(a))) def to_marker( id_: str, lat: float | None, lon: float | None, heading: float | None = None, speed: float | None = None, label: str | None = None, extra: dict | None = None, ) -> dict: return { "id": str(id_), "lat": lat, "lon": lon, "heading": heading, "speed": speed, "label": label or str(id_), "extra": extra or {}, } def filter_points_bbox( points: list[dict], minlon: float, minlat: float, maxlon: float, maxlat: float, limit: int = DEFAULT_LIMIT, ) -> list[dict]: out = [] for p in points: lat, lon = p.get("lat"), p.get("lon") if lat is None or lon is None: continue if minlat <= lat <= maxlat and minlon <= lon <= maxlon: out.append(p) if len(out) >= limit: break return out def _f(value: object) -> float | None: if value is None or value == "": return None try: return float(value) except (TypeError, ValueError): return None def _heading(value: object) -> float | None: if value is None or value == "": return None if isinstance(value, str): key = value.strip().upper() if key in _COMPASS: return float(_COMPASS[key]) num = _f(value) if num is None: return None if num < 0 or num > 360: return None return num def transform_adsb_lol(payload: dict | list | None) -> list[dict]: """Map ADSB.lol v2 aircraft list to shared markers. Dedup on hex.""" if payload is None: return [] if isinstance(payload, list): aircraft = payload else: aircraft = payload.get("ac") or payload.get("aircraft") or [] seen: set[str] = set() out: list[dict] = [] for ac in aircraft: hex_id = str(ac.get("hex") or "").strip().lower() lat, lon = _f(ac.get("lat")), _f(ac.get("lon")) if not hex_id or lat is None or lon is None: continue if hex_id in seen: continue seen.add(hex_id) flight = str(ac.get("flight") or "").strip() or hex_id out.append(to_marker( hex_id, lat, lon, heading=_heading(ac.get("track")), speed=_f(ac.get("gs")), label=flight, extra={ "hex": hex_id, "reg": ac.get("r"), "type": ac.get("t"), "alt_baro": ac.get("alt_baro"), "squawk": ac.get("squawk"), "emergency": ac.get("emergency"), "category": ac.get("category"), "seen_pos": ac.get("seen_pos"), "src": "adsb.lol", }, )) return out def transform_amtraker(payload: dict | None) -> list[dict]: """Flatten Amtraker `{trainNum: [Train, ...]}` to one marker per trainID.""" if not payload or not isinstance(payload, dict): return [] out: list[dict] = [] for _num, trains in payload.items(): if not isinstance(trains, list): continue for tr in trains: if not isinstance(tr, dict): continue lat, lon = _f(tr.get("lat")), _f(tr.get("lon")) if lat is None or lon is None: continue tid = str(tr.get("trainID") or tr.get("trainId") or "").strip() tnum = str(tr.get("trainNum") or _num) route = str(tr.get("routeName") or "").strip() label = f"{route} #{tnum}".strip() if route else f"Train {tnum}" late = tr.get("late") if late is None: late = tr.get("lateMin") out.append(to_marker( tid or tnum, lat, lon, heading=_heading(tr.get("heading")), speed=_f(tr.get("velocity") or tr.get("speed")), label=label, extra={ "route": route, "trainNum": tnum, "late_min": late, "iconColor": tr.get("iconColor"), "stations": tr.get("stations") or [], "src": "amtraker", }, )) return out def transform_ais_frame(frame: dict | None) -> dict | None: """Decode one AISStream JSON envelope to a marker (or None if unusable).""" if not frame or not isinstance(frame, dict): return None meta = frame.get("MetaData") or {} mmsi = meta.get("MMSI") or meta.get("mmsi") if mmsi is None: return None name = str(meta.get("ShipName") or meta.get("shipName") or "").strip() lat = _f(meta.get("Latitude") if "Latitude" in meta else meta.get("latitude")) lon = _f(meta.get("Longitude") if "Longitude" in meta else meta.get("longitude")) msg = frame.get("Message") or {} pos = ( msg.get("PositionReport") or msg.get("StandardClassBPositionReport") or msg.get("ExtendedClassBPositionReport") or {} ) extra: dict[str, Any] = {"src": "aisstream", "mmsi": mmsi} if frame.get("MessageType") == "ShipStaticData": static = msg.get("ShipStaticData") or {} dest = str(static.get("Destination") or static.get("destination") or "").strip() extra["dest"] = dest extra["static"] = True if not name: name = str(static.get("Name") or static.get("name") or "").strip() if lat is None or lon is None: # Static-only update: caller merges onto last-known by MMSI. return to_marker(str(mmsi), None, None, label=name or str(mmsi), extra=extra) heading = pos.get("TrueHeading") if heading in (511, 511.0, None): heading = pos.get("Cog") sog = pos.get("Sog") navstat = pos.get("NavigationalStatus") extra["navstat"] = navstat extra["cog"] = pos.get("Cog") extra["dest"] = extra.get("dest") if lat is None or lon is None: return None return to_marker( str(mmsi), lat, lon, heading=_heading(heading), speed=_f(sog), label=name or str(mmsi), extra=extra, ) def transform_wfigs_incidents(fc: dict | None) -> list[dict]: if not fc: return [] out: list[dict] = [] for feat in fc.get("features") or []: geom = feat.get("geometry") or {} coords = geom.get("coordinates") or [] if len(coords) < 2: continue lon, lat = _f(coords[0]), _f(coords[1]) if lat is None or lon is None: continue props = feat.get("properties") or {} name = str(props.get("IncidentName") or "Incident").strip() out.append(to_marker( name, lat, lon, label=name, extra={ "acres": props.get("IncidentSize"), "contained": props.get("PercentContained"), "state": props.get("POOState"), "category": props.get("IncidentTypeCategory"), "cause": props.get("FireCause"), "discovered": props.get("FireDiscoveryDateTime"), "src": "wfigs", }, )) return out def transform_nhc_storms(payload: dict | None) -> list[dict]: if not payload: return [] storms = payload.get("activeStorms") or [] class_label = { "TD": "Tropical Depression", "TS": "Tropical Storm", "HU": "Hurricane", "STD": "Subtropical Depression", "STS": "Subtropical Storm", "PTC": "Potential Tropical Cyclone", "PC": "Post-Tropical Cyclone", } out: list[dict] = [] for s in storms: lat = _f(s.get("latitudeNumeric") or s.get("lat")) lon = _f(s.get("longitudeNumeric") or s.get("lon")) if lat is None or lon is None: continue sid = str(s.get("id") or s.get("binNumber") or "").strip() name = str(s.get("name") or "Storm").strip() klass = str(s.get("classification") or "").upper() kind = class_label.get(klass, klass or "Storm") out.append(to_marker( sid or name, lat, lon, heading=_heading(s.get("movementDir")), speed=_f(s.get("movementSpeed")), label=f"{kind} {name}".strip(), extra={ "classification": klass, "intensity_kt": s.get("intensity"), "src": "nhc", }, )) return out def rainviewer_tile_url(host: str, path: str, size: int = 256, color: int = 2, smooth: int = 1, snow: int = 1) -> str: host = host.rstrip("/") path = path if path.startswith("/") else f"/{path}" return f"{host}{path}/{size}/{{z}}/{{x}}/{{y}}/{color}/{smooth}_{snow}.png" def _headers() -> dict[str, str]: return {"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"} async def _ttl_get(key: str, ttl: float, factory: Callable[[], Awaitable[Any]]) -> Any: now = time.monotonic() hit = _cache.get(key) if hit and now - hit[0] < ttl: return hit[1] async with _cache_lock: hit = _cache.get(key) if hit and time.monotonic() - hit[0] < ttl: return hit[1] value = await factory() _cache[key] = (time.monotonic(), value) return value async def _get_json(url: str, params: dict | None = None) -> Any: async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, follow_redirects=True, headers=_headers()) as client: resp = await client.get(url, params=params) resp.raise_for_status() return resp.json() async def fetch_aircraft(bbox: str, limit: int = DEFAULT_LIMIT) -> list[dict]: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) lat, lon, radius = bbox_center_radius_nm(minlon, minlat, maxlon, maxlat) cache_key = f"adsb:{lat:.2f}:{lon:.2f}:{radius}" async def _load(): url = f"{ADSB_LOL_BASE}/v2/lat/{lat:.4f}/lon/{lon:.4f}/dist/{radius}" return transform_adsb_lol(await _get_json(url)) rows = await _ttl_get(cache_key, 8.0, _load) return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: async def _load(): return transform_amtraker(await _get_json(AMTRAKER_TRAINS)) rows = await _ttl_get("amtraker:trains", 20.0, _load) if bbox: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) return rows[:limit] async def fetch_vessels(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: async with vessel_lock: rows = [dict(v) for v in vessel_last_known.values() if v.get("lat") is not None and v.get("lon") is not None] if bbox: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit) return rows[:limit] async def upsert_vessel(marker: dict) -> None: """Merge an AIS marker into last-known by MMSI. Static-only updates names.""" vid = str(marker.get("id") or "") if not vid: return async with vessel_lock: prev = vessel_last_known.get(vid, {}) extra = {**(prev.get("extra") or {}), **(marker.get("extra") or {})} lat = marker.get("lat") if marker.get("lat") is not None else prev.get("lat") lon = marker.get("lon") if marker.get("lon") is not None else prev.get("lon") label = marker.get("label") if not label or label == vid: label = prev.get("label") or vid vessel_last_known[vid] = { **to_marker( vid, lat, lon, heading=marker.get("heading") if marker.get("heading") is not None else prev.get("heading"), speed=marker.get("speed") if marker.get("speed") is not None else prev.get("speed"), label=label, extra=extra, ), "seen_at": datetime.now(timezone.utc).isoformat(), } def _wfigs_params(bbox: str | None) -> dict: params = { "where": "1=1", "outSR": "4326", "f": "geojson", "resultRecordCount": 2000, } if bbox: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) params["geometry"] = f"{minlon},{minlat},{maxlon},{maxlat}" params["geometryType"] = "esriGeometryEnvelope" params["inSR"] = "4326" params["spatialRel"] = "esriSpatialRelIntersects" return params async def fetch_fire_incidents(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]: params = _wfigs_params(bbox) params["outFields"] = ( "IncidentName,IncidentSize,FireDiscoveryDateTime,POOState," "PercentContained,IncidentTypeCategory,FireCause" ) async def _load(): return transform_wfigs_incidents(await _get_json(WFIGS_INCIDENTS, params)) rows = await _ttl_get(f"wfigs:inc:{bbox or 'all'}", 600.0, _load) return rows[:limit] async def fetch_fire_perimeters(bbox: str | None) -> dict: params = _wfigs_params(bbox) params["outFields"] = ( "poly_IncidentName,poly_GISAcres,attr_IncidentSize," "attr_PercentContained,attr_FireDiscoveryDateTime" ) async def _load(): return await _get_json(WFIGS_PERIMETERS, params) fc = await _ttl_get(f"wfigs:per:{bbox or 'all'}", 600.0, _load) if not isinstance(fc, dict): return {"type": "FeatureCollection", "features": []} return fc async def fetch_weather_alerts(area: str | None, bbox: str | None) -> dict: """Cached NWS active alerts + IEM storm-based warning polygons.""" async def _load(): nws_params: dict[str, str] = {"status": "actual"} if area: nws_params["area"] = area.upper() elif bbox: minlon, minlat, maxlon, maxlat = parse_bbox(bbox) nws_params["bbox"] = f"{minlon},{minlat},{maxlon},{maxlat}" nws_fc: dict = {"features": []} sbw_fc: dict = {"features": []} try: nws_fc = await _get_json(NWS_ALERTS, nws_params) except Exception: nws_fc = {"features": []} try: sbw_fc = await _get_json(IEM_SBW) except Exception: sbw_fc = {"features": []} features = [] for feat in nws_fc.get("features") or []: props = feat.get("properties") or {} props["source"] = "nws" feat["properties"] = props features.append(feat) for feat in sbw_fc.get("features") or []: props = feat.get("properties") or {} props["source"] = "iem-sbw" # IEM uses `ps` (phenomenon) / `wfo`; map a display event. if "event" not in props: props["event"] = props.get("ps") or "Storm-based warning" feat["properties"] = props if bbox: # Cheap reject: skip if no geometry; keep otherwise (polygons). if not feat.get("geometry"): continue features.append(feat) return {"type": "FeatureCollection", "features": features} key = f"alerts:{area or ''}:{bbox or ''}" return await _ttl_get(key, 30.0, _load) async def fetch_radar_meta() -> dict: async def _load(): data = await _get_json(RAINVIEWER_MAPS) host = data.get("host") or "https://tilecache.rainviewer.com" radar = (data.get("radar") or {}) past = radar.get("past") or [] nowcast = radar.get("nowcast") or [] frames = [] for fr in past + nowcast: path = fr.get("path") if not path: continue frames.append({ "time": fr.get("time"), "path": path, "tileUrl": rainviewer_tile_url(host, path), }) latest = frames[-1]["tileUrl"] if frames else None return { "provider": "rainviewer", "host": host, "tileUrl": latest, "frames": frames, "attribution": "Weather data by RainViewer", "iemTileUrl": IEM_NEXRAD, } return await _ttl_get("radar:rv", 300.0, _load) async def fetch_storms() -> list[dict]: async def _load(): return transform_nhc_storms(await _get_json(NHC_STORMS)) return await _ttl_get("nhc:storms", 300.0, _load)