Logan CusanoandClaude Opus 5.5
bdbde359a1
- Cap one refresh's NWS zone fetches at 15 s wall clock. They run under the feed lock, so a hung zone endpoint could otherwise stall every /weather/alerts request for minutes; unfinished zones retry next refresh. - Prod c2-core.env.j2 never set NWS_USER_AGENT, so the VM would have sent the contact-less placeholder. Default it to the site URL (overridable via vault_nws_user_agent); warn on the .env.example example.com placeholder too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
409 lines
17 KiB
Python
409 lines
17 KiB
Python
"""Weather context for the map: NWS active alerts and nearest-METAR wind (server-26#237).
|
|
|
|
Sources -- both US Government works, public domain, no licence gate:
|
|
* NWS alerts: https://api.weather.gov/alerts/active?area=<codes>
|
|
(docs: https://www.weather.gov/documentation/services-web-api). NWS requires
|
|
a User-Agent naming the app and a contact, and rate-limits, so the feed is
|
|
fetched server-side, cached (``settings.weather_alerts_ttl_s``) and shared by
|
|
every viewer. Credit: "National Weather Service".
|
|
* METAR wind: https://aviationweather.gov/api/data/metar?bbox=minLat,minLon,maxLat,maxLon&format=json
|
|
(docs: https://aviationweather.gov/data/api/). Verified 2026-10-05: JSON list,
|
|
one latest report per station, HTTP 204 with an empty body when the bbox holds
|
|
no station, ``wdir`` is an int or the string "VRB".
|
|
|
|
Shape and failure rules are the 511NY ones (app/internal/ny511.py): a failed
|
|
refresh keeps serving the last good data and reports the error and its age; a
|
|
payload whose records ALL fail to parse is a schema change and an error, never
|
|
an empty layer.
|
|
|
|
Many NWS alerts carry ``geometry: null`` -- they are zone-based (forecast/county
|
|
zones listed in ``affectedZones``). Each zone's polygon is fetched once and kept
|
|
24 h. An alert whose zones cannot be resolved is returned in ``zone_only`` with
|
|
its event and expiry, so nothing is silently dropped.
|
|
"""
|
|
import asyncio
|
|
import math
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
from app.internal.cameras.base import safe_error
|
|
from app.internal.logger import logger
|
|
|
|
NWS_ALERTS_URL = "https://api.weather.gov/alerts/active"
|
|
NWS_ZONE_PREFIX = "https://api.weather.gov/zones/" # only zone URLs under this are ever fetched
|
|
METAR_URL = "https://aviationweather.gov/api/data/metar"
|
|
CREDIT = "National Weather Service"
|
|
WIND_CREDIT = "NOAA Aviation Weather Center (METAR)"
|
|
DEFAULT_USER_AGENT = "DRB scanner map (contact: set NWS_USER_AGENT)"
|
|
|
|
RETRY_AFTER_FAILURE_S = 30
|
|
ZONE_TTL_S = 24 * 3600
|
|
ZONE_FAILURE_RETRY_S = 15 * 60
|
|
ZONE_CONCURRENCY = 6
|
|
# A state-wide winter event can list 100+ zones. Fetches past this cap wait for
|
|
# the next refresh (the zone cache fills across refreshes); meanwhile the alert
|
|
# shows in zone_only or with the zones resolved so far.
|
|
MAX_ZONE_FETCHES_PER_REFRESH = 80
|
|
# Wall-clock cap on one refresh's zone fetches. The refresh runs under the feed
|
|
# lock, so without it a slow zone endpoint (20 s timeout x 80 fetches / 6 wide)
|
|
# could hold every /weather/alerts request for minutes. Unfinished zones are
|
|
# simply not cached and are retried next refresh.
|
|
ZONE_BUDGET_S = 15.0
|
|
|
|
WIND_TTL_S = 5 * 60
|
|
WIND_CELL_DEG = 0.5 # wind cache cell; one upstream call per cell per TTL
|
|
WIND_MAX_KM = 75.0 # a station farther than this says nothing about the incident's wind
|
|
WIND_STALE_S = 2 * 3600 # older reports are returned but flagged stale
|
|
_WIND_CACHE_MAX = 500
|
|
|
|
_warned_ua = False
|
|
|
|
|
|
def _headers(accept: str) -> Dict[str, str]:
|
|
global _warned_ua
|
|
ua = settings.nws_user_agent or DEFAULT_USER_AGENT
|
|
if (ua == DEFAULT_USER_AGENT or "example.com" in ua) and not _warned_ua:
|
|
_warned_ua = True
|
|
logger.warning("NWS_USER_AGENT is the placeholder -- set it to 'app (contact email)'; NWS may block us")
|
|
return {"User-Agent": ua, "Accept": accept}
|
|
|
|
|
|
# ── geometry helpers ─────────────────────────────────────────────────────────
|
|
|
|
def _polygons(geom: Any) -> List[Any]:
|
|
"""Polygon coordinate arrays out of a Polygon / MultiPolygon / GeometryCollection."""
|
|
if not isinstance(geom, dict):
|
|
return []
|
|
t = geom.get("type")
|
|
if t == "Polygon" and isinstance(geom.get("coordinates"), list):
|
|
return [geom["coordinates"]]
|
|
if t == "MultiPolygon" and isinstance(geom.get("coordinates"), list):
|
|
return list(geom["coordinates"])
|
|
if t == "GeometryCollection":
|
|
return [p for g in geom.get("geometries") or [] for p in _polygons(g)]
|
|
return []
|
|
|
|
|
|
def _merge(polys: List[Any]) -> Optional[Dict[str, Any]]:
|
|
if not polys:
|
|
return None
|
|
if len(polys) == 1:
|
|
return {"type": "Polygon", "coordinates": polys[0]}
|
|
return {"type": "MultiPolygon", "coordinates": polys}
|
|
|
|
|
|
def _bbox(geom: Optional[Dict[str, Any]]) -> Optional[List[float]]:
|
|
"""[south, west, north, east] of a (Multi)Polygon, or None."""
|
|
lats: List[float] = []
|
|
lons: List[float] = []
|
|
for poly in _polygons(geom):
|
|
for ring in poly:
|
|
for pt in ring:
|
|
try:
|
|
lons.append(float(pt[0]))
|
|
lats.append(float(pt[1]))
|
|
except (TypeError, ValueError, IndexError):
|
|
continue
|
|
if not lats:
|
|
return None
|
|
return [min(lats), min(lons), max(lats), max(lons)]
|
|
|
|
|
|
def intersects(b: Optional[List[float]], south: float, west: float, north: float, east: float) -> bool:
|
|
return b is not None and b[0] <= north and b[2] >= south and b[1] <= east and b[3] >= west
|
|
|
|
|
|
# ── NWS alerts ───────────────────────────────────────────────────────────────
|
|
|
|
def normalize_alert(feature: Any) -> Optional[Dict[str, Any]]:
|
|
"""One GeoJSON feature -> our record (geometry may still be None). None if unparseable."""
|
|
if not isinstance(feature, dict):
|
|
return None
|
|
p = feature.get("properties")
|
|
if not isinstance(p, dict):
|
|
return None
|
|
aid = p.get("id") or feature.get("id")
|
|
event = p.get("event")
|
|
if not aid or not event:
|
|
return None
|
|
geometry = _merge(_polygons(feature.get("geometry")))
|
|
zones = [z for z in (p.get("affectedZones") or []) if isinstance(z, str)]
|
|
return {
|
|
"id": aid,
|
|
"event": event,
|
|
"severity": p.get("severity"),
|
|
"urgency": p.get("urgency"),
|
|
"headline": p.get("headline"),
|
|
"onset": p.get("onset"),
|
|
"expires": p.get("expires"),
|
|
"ends": p.get("ends"),
|
|
"sender_name": p.get("senderName"), # an NWS office ("NWS Upton NY"), not a person
|
|
"area_desc": p.get("areaDesc"),
|
|
"geometry": geometry,
|
|
"geometry_source": "alert" if geometry else None,
|
|
"zones": zones,
|
|
}
|
|
|
|
|
|
class _ZoneCache:
|
|
def __init__(self):
|
|
self._entries: Dict[str, Tuple[float, Optional[List[Any]]]] = {} # url -> (expires_at, polygons|None)
|
|
|
|
def get(self, url: str) -> Tuple[bool, Optional[List[Any]]]:
|
|
e = self._entries.get(url)
|
|
if e is None or e[0] < time.time():
|
|
return False, None
|
|
return True, e[1]
|
|
|
|
def put(self, url: str, polys: Optional[List[Any]]):
|
|
ttl = ZONE_TTL_S if polys else ZONE_FAILURE_RETRY_S
|
|
self._entries[url] = (time.time() + ttl, polys)
|
|
|
|
def purge(self):
|
|
now = time.time()
|
|
for k in [k for k, e in self._entries.items() if e[0] < now]:
|
|
del self._entries[k]
|
|
|
|
def clear(self):
|
|
self._entries.clear()
|
|
|
|
|
|
zones = _ZoneCache()
|
|
|
|
|
|
async def _fetch_zone(client: "httpx.AsyncClient", url: str, sem: asyncio.Semaphore) -> None:
|
|
async with sem:
|
|
try:
|
|
r = await client.get(url, headers=_headers("application/geo+json"))
|
|
r.raise_for_status()
|
|
polys = _polygons(r.json().get("geometry"))
|
|
except Exception as e:
|
|
logger.warning(f"NWS zone {url} unresolved: {safe_error(e)}")
|
|
polys = []
|
|
zones.put(url, polys or None)
|
|
|
|
|
|
async def _resolve_zones(client: "httpx.AsyncClient", alerts: List[Dict[str, Any]]) -> None:
|
|
"""Fill geometry for zone-based alerts in place, from the 24 h zone cache."""
|
|
zones.purge()
|
|
wanted: List[str] = []
|
|
for a in alerts:
|
|
if a["geometry"] is None:
|
|
for z in a["zones"]:
|
|
if z.startswith(NWS_ZONE_PREFIX) and not zones.get(z)[0] and z not in wanted:
|
|
wanted.append(z)
|
|
if wanted:
|
|
sem = asyncio.Semaphore(ZONE_CONCURRENCY)
|
|
try:
|
|
await asyncio.wait_for(
|
|
asyncio.gather(*(_fetch_zone(client, z, sem) for z in wanted[:MAX_ZONE_FETCHES_PER_REFRESH])),
|
|
timeout=ZONE_BUDGET_S)
|
|
except asyncio.TimeoutError:
|
|
logger.warning(f"NWS zone fetches over {ZONE_BUDGET_S:.0f}s budget; the rest resolve next refresh")
|
|
for a in alerts:
|
|
if a["geometry"] is not None:
|
|
continue
|
|
polys: List[Any] = []
|
|
missing = 0
|
|
for z in a["zones"]:
|
|
hit, p = zones.get(z)
|
|
if hit and p:
|
|
polys.extend(p)
|
|
else:
|
|
missing += 1
|
|
a["geometry"] = _merge(polys)
|
|
if a["geometry"] is not None:
|
|
a["geometry_source"] = "zones"
|
|
a["zones_unresolved"] = missing # >0: drawn area is smaller than the real one
|
|
|
|
|
|
async def _fetch_alerts() -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
|
|
area = ",".join(s.strip().upper() for s in (settings.nws_area or "").split(",") if s.strip())
|
|
if not area:
|
|
raise ValueError("NWS_AREA is empty -- no state/marine area to ask NWS for")
|
|
async with httpx.AsyncClient(timeout=20.0) as client:
|
|
r = await client.get(NWS_ALERTS_URL, params={"area": area}, headers=_headers("application/geo+json"))
|
|
r.raise_for_status()
|
|
raw = r.json()
|
|
features = raw.get("features") if isinstance(raw, dict) else None
|
|
if not isinstance(features, list):
|
|
raise ValueError("NWS alerts payload has no 'features' list -- schema change?")
|
|
parsed = [a for a in (normalize_alert(f) for f in features) if a is not None]
|
|
if features and not parsed:
|
|
raise ValueError(f"0 of {len(features)} NWS alerts parsed -- schema change?")
|
|
await _resolve_zones(client, parsed)
|
|
mapped, zone_only = [], []
|
|
for a in parsed:
|
|
a["bbox"] = _bbox(a["geometry"])
|
|
if a["geometry"] is None:
|
|
zone_only.append({k: v for k, v in a.items() if k not in ("geometry", "geometry_source", "bbox")})
|
|
else:
|
|
mapped.append(a)
|
|
return mapped, zone_only
|
|
|
|
|
|
class _AlertsFeed:
|
|
def __init__(self):
|
|
self.alerts: List[Dict[str, Any]] = []
|
|
self.zone_only: List[Dict[str, Any]] = []
|
|
self.fetched_at: Optional[float] = None # epoch s of last SUCCESSFUL fetch
|
|
self.error: Optional[str] = None
|
|
self._next_attempt = 0.0
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def get(self) -> "_AlertsFeed":
|
|
if time.time() < self._next_attempt:
|
|
return self
|
|
async with self._lock: # single flight: waiters see the refreshed state
|
|
if time.time() < self._next_attempt:
|
|
return self
|
|
ttl = max(30, int(settings.weather_alerts_ttl_s))
|
|
try:
|
|
self.alerts, self.zone_only = await _fetch_alerts()
|
|
self.fetched_at = time.time()
|
|
self.error = None
|
|
self._next_attempt = time.time() + ttl
|
|
except Exception as e:
|
|
self._next_attempt = time.time() + min(ttl, RETRY_AFTER_FAILURE_S)
|
|
self.error = safe_error(e)
|
|
logger.warning(f"NWS alerts refresh failed, serving {len(self.alerts)} cached: {self.error}")
|
|
return self
|
|
|
|
|
|
alerts = _AlertsFeed()
|
|
|
|
|
|
# ── METAR wind ───────────────────────────────────────────────────────────────
|
|
|
|
def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
|
|
p1, p2 = math.radians(lat1), math.radians(lat2)
|
|
dp, dl = p2 - p1, math.radians(lon2 - lon1)
|
|
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
|
|
return 6371.0 * 2 * math.asin(min(1.0, math.sqrt(a)))
|
|
|
|
|
|
def normalize_metar(x: Any) -> Optional[Dict[str, Any]]:
|
|
if not isinstance(x, dict) or not x.get("icaoId"):
|
|
return None
|
|
try:
|
|
lat, lon = float(x["lat"]), float(x["lon"])
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
wspd = x.get("wspd")
|
|
if not isinstance(wspd, (int, float)):
|
|
return None # no wind group in this report
|
|
wdir = x.get("wdir")
|
|
variable = isinstance(wdir, str) and wdir.upper() == "VRB"
|
|
obs = x.get("obsTime")
|
|
obs_iso = (datetime.fromtimestamp(obs, tz=timezone.utc).isoformat().replace("+00:00", "Z")
|
|
if isinstance(obs, (int, float)) else x.get("reportTime"))
|
|
gust = x.get("wgst")
|
|
return {
|
|
"station": x["icaoId"],
|
|
"station_name": x.get("name") or "",
|
|
"lat": lat, "lon": lon,
|
|
"obs_epoch": obs if isinstance(obs, (int, float)) else None,
|
|
"obs_time": obs_iso,
|
|
"wind_dir_deg": wdir if isinstance(wdir, (int, float)) else None,
|
|
"variable": variable,
|
|
"wind_kt": wspd,
|
|
"gust_kt": gust if isinstance(gust, (int, float)) else None,
|
|
"raw": x.get("rawOb"),
|
|
}
|
|
|
|
|
|
class _WindCell:
|
|
def __init__(self):
|
|
self.stations: List[Dict[str, Any]] = []
|
|
self.fetched_at: Optional[float] = None
|
|
self.error: Optional[str] = None
|
|
self.next_attempt = 0.0
|
|
self.lock = asyncio.Lock()
|
|
|
|
|
|
_wind_cells: Dict[Tuple[int, int], _WindCell] = {}
|
|
|
|
|
|
def _cell_key(lat: float, lon: float) -> Tuple[int, int]:
|
|
return (math.floor(lat / WIND_CELL_DEG), math.floor(lon / WIND_CELL_DEG))
|
|
|
|
|
|
async def _fetch_metars(key: Tuple[int, int]) -> List[Dict[str, Any]]:
|
|
# Search box = the cell padded by WIND_MAX_KM, so every point in the cell sees
|
|
# every station within WIND_MAX_KM of it.
|
|
s, w = key[0] * WIND_CELL_DEG, key[1] * WIND_CELL_DEG
|
|
n, e = s + WIND_CELL_DEG, w + WIND_CELL_DEG
|
|
pad_lat = WIND_MAX_KM / 111.0
|
|
pad_lon = WIND_MAX_KM / (111.0 * max(0.2, math.cos(math.radians(max(abs(s), abs(n))))))
|
|
bbox = f"{s - pad_lat:.3f},{w - pad_lon:.3f},{n + pad_lat:.3f},{e + pad_lon:.3f}"
|
|
async with httpx.AsyncClient(timeout=15.0) as client:
|
|
r = await client.get(METAR_URL, params={"bbox": bbox, "format": "json"}, headers=_headers("application/json"))
|
|
r.raise_for_status()
|
|
if r.status_code == 204 or not r.content.strip():
|
|
return [] # no station in the box
|
|
raw = r.json()
|
|
if not isinstance(raw, list):
|
|
raise ValueError(f"METAR API: expected a JSON list, got {type(raw).__name__}")
|
|
latest: Dict[str, Dict[str, Any]] = {}
|
|
for m in (normalize_metar(x) for x in raw):
|
|
if m is None:
|
|
continue
|
|
cur = latest.get(m["station"])
|
|
if cur is None or (m["obs_epoch"] or 0) > (cur["obs_epoch"] or 0):
|
|
latest[m["station"]] = m
|
|
# Schema canary: reports without a wind group are normal, but if NO record even
|
|
# has a station id and position, the field names changed under us.
|
|
if raw and not any(isinstance(x, dict) and x.get("icaoId") and "lat" in x for x in raw):
|
|
raise ValueError(f"0 of {len(raw)} METAR records parsed -- schema change?")
|
|
return list(latest.values())
|
|
|
|
|
|
async def _cell(lat: float, lon: float) -> _WindCell:
|
|
key = _cell_key(lat, lon)
|
|
cell = _wind_cells.get(key)
|
|
if cell is None:
|
|
if len(_wind_cells) >= _WIND_CACHE_MAX: # bound memory: drop expired cells, then the oldest
|
|
for k in [k for k, c in _wind_cells.items() if c.next_attempt < time.time() and not c.lock.locked()]:
|
|
del _wind_cells[k]
|
|
if len(_wind_cells) >= _WIND_CACHE_MAX:
|
|
del _wind_cells[min(_wind_cells, key=lambda k: _wind_cells[k].next_attempt)]
|
|
cell = _wind_cells[key] = _WindCell()
|
|
if time.time() < cell.next_attempt:
|
|
return cell
|
|
async with cell.lock:
|
|
if time.time() < cell.next_attempt:
|
|
return cell
|
|
try:
|
|
cell.stations = await _fetch_metars(key)
|
|
cell.fetched_at = time.time()
|
|
cell.error = None
|
|
cell.next_attempt = time.time() + WIND_TTL_S
|
|
except Exception as e:
|
|
cell.next_attempt = time.time() + RETRY_AFTER_FAILURE_S
|
|
cell.error = safe_error(e)
|
|
logger.warning(f"METAR refresh for cell {key} failed, serving {len(cell.stations)} cached: {cell.error}")
|
|
return cell
|
|
|
|
|
|
async def nearest_wind(lat: float, lon: float) -> Dict[str, Any]:
|
|
"""Wind from the nearest METAR station within WIND_MAX_KM, or a clear reason why not."""
|
|
cell = await _cell(lat, lon)
|
|
base = {"fetched_at": cell.fetched_at, "error": cell.error, "credit": WIND_CREDIT}
|
|
ranked = sorted(((haversine_km(lat, lon, s["lat"], s["lon"]), s) for s in cell.stations), key=lambda t: t[0])
|
|
if not ranked or ranked[0][0] > WIND_MAX_KM:
|
|
if cell.fetched_at is None:
|
|
reason = f"METAR feed unavailable: {cell.error}" if cell.error else "METAR feed not fetched yet"
|
|
else:
|
|
reason = f"no METAR station reporting wind within {WIND_MAX_KM:.0f} km"
|
|
return {**base, "available": False, "reason": reason, "wind": None}
|
|
dist, s = ranked[0]
|
|
stale = s["obs_epoch"] is not None and time.time() - s["obs_epoch"] > WIND_STALE_S
|
|
wind = {k: v for k, v in s.items() if k not in ("obs_epoch",)}
|
|
wind["distance_km"] = round(dist, 1)
|
|
wind["stale"] = stale
|
|
return {**base, "available": True, "reason": None, "wind": wind}
|