From fb1b6956c204aaf11a90d09cbd1ec9a635516061 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 5 Oct 2026 02:54:33 -0400 Subject: [PATCH 1/2] Weather backend: NWS active alerts + nearest-METAR wind, cached (Refs #237) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET /weather/alerts — api.weather.gov/alerts/active for settings.nws_area, sent with the NWS-required User-Agent, cached weather_alerts_ttl_s (90 s) with single-flight refresh. Zone-based alerts (geometry null) get their polygons from affectedZones (24 h zone cache, api.weather.gov/zones/ only); any still unresolved are returned in zone_only, never dropped. Failed refresh keeps last-good data + error + fetched_at; a payload where no feature parses is a schema-change error. Optional bbox filter. GET /weather/wind — nearest METAR (aviationweather.gov /api/data/metar, bbox query) within 75 km, cached 5 min per 0.5-degree cell; VRB wind and stale reports flagged; available:false with a reason otherwise. Both behind require_service_or_firebase_token like /traffic. Co-Authored-By: Claude Opus 5.5 --- drb-c2-core/.env.example | 4 + drb-c2-core/app/config.py | 8 + drb-c2-core/app/internal/weather.py | 398 ++++++++++++++++++++++++++++ drb-c2-core/app/main.py | 3 +- drb-c2-core/app/routers/weather.py | 51 ++++ drb-c2-core/tests/test_weather.py | 218 +++++++++++++++ 6 files changed, 681 insertions(+), 1 deletion(-) create mode 100644 drb-c2-core/app/internal/weather.py create mode 100644 drb-c2-core/app/routers/weather.py create mode 100644 drb-c2-core/tests/test_weather.py diff --git a/drb-c2-core/.env.example b/drb-c2-core/.env.example index 33eb3e0..80f3996 100644 --- a/drb-c2-core/.env.example +++ b/drb-c2-core/.env.example @@ -31,6 +31,10 @@ NY511_API_KEY= NY511_API_BASE= # Ontario 511 developer key (camera provider "ontario511"; providers default OFF) ONTARIO511_API_KEY= +# NWS weather alerts / METAR wind (server-26#237). NWS requires app name + contact in the User-Agent +NWS_USER_AGENT=DRB (you@example.com) +# Comma list of NWS state/marine area codes served +NWS_AREA=NY # OpenAI — for transcription (Whisper), intelligence extraction, embeddings, and summaries OPENAI_API_KEY= diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 31d2eae..922aeed 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -59,6 +59,14 @@ class Settings(BaseSettings): # Camera providers (app/internal/cameras). Keys only; every provider is still # OFF until an admin enables it, and licence-gated ones need an acknowledgement. ontario511_api_key: Optional[str] = None + # Weather context (server-26#237). api.weather.gov requires a User-Agent that + # names the app AND a contact (e.g. "DRB (ops@example.com)"); the placeholder + # default must be overridden in every deploy -- NWS may block anonymous agents. + nws_user_agent: str = "DRB scanner map (contact: set NWS_USER_AGENT)" + # Comma list of NWS state/marine area codes for /alerts/active (the areas served). + nws_area: str = "NY" + # NWS alerts cache TTL; the issue asks 60-120 s (floored at 30 s in code). + weather_alerts_ttl_s: int = 90 # Gemini (intelligence extraction, embeddings, incident summaries) gemini_api_key: Optional[str] = None diff --git a/drb-c2-core/app/internal/weather.py b/drb-c2-core/app/internal/weather.py new file mode 100644 index 0000000..aa7aae5 --- /dev/null +++ b/drb-c2-core/app/internal/weather.py @@ -0,0 +1,398 @@ +"""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= + (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 + +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 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) + await asyncio.gather(*(_fetch_zone(client, z, sem) for z in wanted[:MAX_ZONE_FETCHES_PER_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} diff --git a/drb-c2-core/app/main.py b/drb-c2-core/app/main.py index c63038d..a2b4e5e 100644 --- a/drb-c2-core/app/main.py +++ b/drb-c2-core/app/main.py @@ -19,7 +19,7 @@ from app.internal.auth import ( ) from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic, cameras -from app.routers import outages +from app.routers import outages, weather from app.routers import discord as discord_router from app.routers import discord_link as discord_link_router from app.internal import dynsec @@ -150,6 +150,7 @@ app.include_router(cameras.router, dependencies=[Depends(require_service_or_fi app.include_router(cameras.admin_router) # auth: admin only, per endpoint (needs the principal for audit) app.include_router(outages.router, dependencies=[Depends(require_service_or_firebase_token)]) app.include_router(outages.admin_router) # auth: admin only, per endpoint (needs the principal for audit) +app.include_router(weather.router, dependencies=[Depends(require_service_or_firebase_token)]) # auth: any signed-in user/service, like traffic app.include_router(upload.router) # auth is per-node, handled inline app.include_router(admin.router) # auth is per-endpoint (read: firebase, write: admin) app.include_router(replay.router) # auth: admin only (every route spends or reads a replay run) diff --git a/drb-c2-core/app/routers/weather.py b/drb-c2-core/app/routers/weather.py new file mode 100644 index 0000000..b07a6d1 --- /dev/null +++ b/drb-c2-core/app/routers/weather.py @@ -0,0 +1,51 @@ +"""Weather context for the map (server-26#237): NWS alerts + nearest-METAR wind. + +Public-domain US Government data, cached server-side in app/internal/weather.py; +no licence gate (unlike 511NY / outage providers). +""" +from typing import Optional + +from fastapi import APIRouter, HTTPException, Query + +from app.internal import weather + +router = APIRouter(prefix="/weather", tags=["weather"]) + + +@router.get("/alerts") +async def get_alerts( + south: Optional[float] = Query(None, ge=-90, le=90), + west: Optional[float] = Query(None, ge=-180, le=180), + north: Optional[float] = Query(None, ge=-90, le=90), + east: Optional[float] = Query(None, ge=-180, le=180), +): + """Active NWS alerts with polygons, optionally filtered to those intersecting a bbox. + + ``zone_only`` holds alerts whose zones could not be resolved to a polygon; it + is never bbox-filtered (there is no shape to test) and never dropped. + """ + box = (south, west, north, east) + if any(v is not None for v in box) and any(v is None for v in box): + raise HTTPException(400, "give all of south, west, north, east or none") + if south is not None and (south > north or west > east): + raise HTTPException(400, "bbox must satisfy south<=north and west<=east") + feed = await weather.alerts.get() + items = feed.alerts + if south is not None: + items = [a for a in items if weather.intersects(a.get("bbox"), south, west, north, east)] + return { + "alerts": items, + "zone_only": feed.zone_only, + "fetched_at": feed.fetched_at, + "error": feed.error, + "credit": weather.CREDIT, + } + + +@router.get("/wind") +async def get_wind( + lat: float = Query(..., ge=-90, le=90), + lon: float = Query(..., ge=-180, le=180), +): + """Wind from the nearest METAR station, or ``available: false`` with a reason.""" + return await weather.nearest_wind(lat, lon) diff --git a/drb-c2-core/tests/test_weather.py b/drb-c2-core/tests/test_weather.py new file mode 100644 index 0000000..478f634 --- /dev/null +++ b/drb-c2-core/tests/test_weather.py @@ -0,0 +1,218 @@ +""" +server-26#237 — weather backend: NWS active alerts + nearest-METAR wind. + +Pins the failure shapes that would otherwise look like "no weather right now": +an upstream error keeps last-good data AND reports the error, a schema change is +an error (not an empty layer), and zone-based alerts (geometry null) are never +dropped. Also pins the NWS contract: User-Agent sent, one upstream call per TTL. +""" +import asyncio +from unittest.mock import patch + +from fastapi.testclient import TestClient + +from app.main import app +from app.internal import weather +from app.internal.auth import require_service_or_firebase_token + +client = TestClient(app) + +SQUARE = {"type": "Polygon", "coordinates": [[[-74.0, 41.0], [-73.5, 41.0], [-73.5, 41.5], [-74.0, 41.5], [-74.0, 41.0]]]} +ZONE_POLY = {"type": "Polygon", "coordinates": [[[-75.0, 42.0], [-74.5, 42.0], [-74.5, 42.5], [-75.0, 42.0]]]} +ZONE_A = "https://api.weather.gov/zones/forecast/NYZ067" +ZONE_B = "https://api.weather.gov/zones/forecast/NYZ999" + + +def _alert(aid, event, geometry=None, zones=()): + return {"id": f"https://api.weather.gov/alerts/{aid}", "type": "Feature", "geometry": geometry, + "properties": {"id": aid, "event": event, "severity": "Severe", "urgency": "Immediate", + "headline": f"{event} issued by NWS Upton NY", "onset": "2026-10-05T10:00:00-04:00", + "expires": "2026-10-05T12:00:00-04:00", "ends": None, "senderName": "NWS Upton NY", + "areaDesc": "Westchester", "affectedZones": list(zones)}} + + +class Resp: + def __init__(self, body=None, status=200): + self._body, self.status_code = body, status + self.content = b"" if body is None else b"{}" + + def raise_for_status(self): + if self.status_code >= 400: + raise RuntimeError(f"HTTP {self.status_code}") + + def json(self): + return self._body + + +class FakeHttp: + """Routes GETs by URL; records every call (url, params, headers).""" + def __init__(self, routes): + self.routes, self.calls = routes, [] + + def __call__(self, **kw): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *a): + pass + + async def get(self, url, params=None, headers=None): + self.calls.append((url, params, headers)) + r = self.routes.get(url) + if r is None: + return Resp(status=404) + return r() if callable(r) else r + + +def setup_function(): + app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True} + weather.alerts = weather._AlertsFeed() + weather.zones.clear() + weather._wind_cells.clear() + + +def teardown_function(): + app.dependency_overrides.pop(require_service_or_firebase_token, None) + + +def _nws(features, extra=None): + routes = {weather.NWS_ALERTS_URL: Resp({"type": "FeatureCollection", "features": features})} + routes.update(extra or {}) + return FakeHttp(routes) + + +def test_happy_path_normalises_and_sends_user_agent(): + http = _nws([_alert("a1", "Severe Thunderstorm Warning", SQUARE)]) + with patch.object(weather.httpx, "AsyncClient", http), \ + patch.object(weather.settings, "nws_user_agent", "DRB (ops@example.com)"), \ + patch.object(weather.settings, "nws_area", "ny, ct"): + feed = asyncio.run(weather.alerts.get()) + url, params, headers = http.calls[0] + assert params == {"area": "NY,CT"} + assert headers["User-Agent"] == "DRB (ops@example.com)" and headers["Accept"] == "application/geo+json" + a = feed.alerts[0] + assert {k: a[k] for k in ("id", "event", "severity", "urgency", "sender_name", "expires")} == { + "id": "a1", "event": "Severe Thunderstorm Warning", "severity": "Severe", "urgency": "Immediate", + "sender_name": "NWS Upton NY", "expires": "2026-10-05T12:00:00-04:00"} + assert a["geometry"] == SQUARE and a["geometry_source"] == "alert" + assert a["bbox"] == [41.0, -74.0, 41.5, -73.5] + assert feed.error is None and feed.fetched_at is not None and feed.zone_only == [] + + +def test_null_geometry_resolves_zones_or_lands_in_zone_only(): + http = _nws( + [_alert("z1", "Flood Watch", None, [ZONE_A]), # zone resolves -> polygon + _alert("z2", "Heat Advisory", None, [ZONE_B]), # zone 404s -> zone_only, not dropped + _alert("z3", "Wind Advisory", None, ["https://evil.example/zones/x"])], # never fetched + {ZONE_A: Resp({"geometry": ZONE_POLY, "properties": {}})}) + with patch.object(weather.httpx, "AsyncClient", http): + feed = asyncio.run(weather.alerts.get()) + assert [a["id"] for a in feed.alerts] == ["z1"] + assert feed.alerts[0]["geometry"] == ZONE_POLY and feed.alerts[0]["geometry_source"] == "zones" + assert feed.alerts[0]["zones_unresolved"] == 0 + assert sorted(z["id"] for z in feed.zone_only) == ["z2", "z3"] + assert feed.zone_only[0]["expires"] and feed.zone_only[0]["event"] + assert not any("evil.example" in c[0] for c in http.calls) + + # second refresh: zone cache (24 h) means ZONE_A is not fetched again + weather.alerts._next_attempt = 0.0 + with patch.object(weather.httpx, "AsyncClient", http): + asyncio.run(weather.alerts.get()) + assert sum(1 for c in http.calls if c[0] == ZONE_A) == 1 + + +def test_upstream_500_keeps_last_good_and_reports_error(): + ok = _nws([_alert("a1", "Tornado Warning", SQUARE)]) + with patch.object(weather.httpx, "AsyncClient", ok): + asyncio.run(weather.alerts.get()) + good_at = weather.alerts.fetched_at + weather.alerts._next_attempt = 0.0 + with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.NWS_ALERTS_URL: Resp(status=500)})): + feed = asyncio.run(weather.alerts.get()) + assert [a["id"] for a in feed.alerts] == ["a1"] and feed.fetched_at == good_at + assert "500" in feed.error + assert feed._next_attempt - __import__("time").time() <= weather.RETRY_AFTER_FAILURE_S + 1 + + r = client.get("/weather/alerts") + body = r.json() + assert r.status_code == 200 and body["error"] and len(body["alerts"]) == 1 + assert body["fetched_at"] == good_at and body["credit"] == "National Weather Service" + + +def test_schema_canary_trips_when_no_feature_parses(): + http = FakeHttp({weather.NWS_ALERTS_URL: Resp({"features": [{"props": {}}, {"properties": {"name": "x"}}]})}) + with patch.object(weather.httpx, "AsyncClient", http): + feed = asyncio.run(weather.alerts.get()) + assert feed.alerts == [] and "schema" in feed.error + + +def test_empty_feature_list_is_not_an_error(): + with patch.object(weather.httpx, "AsyncClient", _nws([])): + feed = asyncio.run(weather.alerts.get()) + assert feed.error is None and feed.fetched_at is not None and feed.alerts == [] + + +def test_cache_one_upstream_call_per_ttl_even_when_concurrent(): + http = _nws([_alert("a1", "Tornado Warning", SQUARE)]) + + async def many(): + await asyncio.gather(*(weather.alerts.get() for _ in range(8))) + await weather.alerts.get() + + with patch.object(weather.httpx, "AsyncClient", http): + asyncio.run(many()) + assert sum(1 for c in http.calls if c[0] == weather.NWS_ALERTS_URL) == 1 + + +def test_endpoint_bbox_filter_and_validation(): + far = {"type": "Polygon", "coordinates": [[[-79.0, 42.8], [-78.5, 42.8], [-78.5, 43.0], [-79.0, 42.8]]]} # Buffalo + http = _nws([_alert("near", "Flood Warning", SQUARE), _alert("far", "Lake Effect Snow Warning", far), + _alert("zo", "Heat Advisory", None, [ZONE_B])]) + with patch.object(weather.httpx, "AsyncClient", http): + r = client.get("/weather/alerts", params={"south": 40.9, "west": -74.2, "north": 41.2, "east": -73.4}) + body = r.json() + assert r.status_code == 200 and [a["id"] for a in body["alerts"]] == ["near"] + assert [z["id"] for z in body["zone_only"]] == ["zo"] # never bbox-filtered away + assert len(client.get("/weather/alerts").json()["alerts"]) == 2 + assert client.get("/weather/alerts", params={"south": 41.4, "west": -74, "north": 40.9, "east": -73}).status_code == 400 + assert client.get("/weather/alerts", params={"south": 40.9}).status_code == 400 + + +def _metar(icao, lat, lon, wdir, wspd, wgst=None, obs=None): + return {"icaoId": icao, "name": f"{icao} airport", "lat": lat, "lon": lon, + "obsTime": obs or int(__import__("time").time()) - 600, "reportTime": "2026-10-05T06:48:00.000Z", + "wdir": wdir, "wspd": wspd, "wgst": wgst, "rawOb": f"METAR {icao} ..."} + + +def test_wind_picks_nearest_station_and_caches_per_cell(): + rows = [_metar("KLGA", 40.7794, -73.8803, 310, 6), + _metar("KHPN", 41.067, -73.7076, "VRB", 4, 18), + _metar("KHPN", 41.067, -73.7076, 200, 3, obs=1)] # older duplicate report + http = FakeHttp({weather.METAR_URL: Resp(rows)}) + with patch.object(weather.httpx, "AsyncClient", http): + r = client.get("/weather/wind", params={"lat": 41.03, "lon": -73.76}) # White Plains + body = r.json() + assert r.status_code == 200 and body["available"] is True + w = body["wind"] + assert w["station"] == "KHPN" and w["distance_km"] < 10 + assert w["wind_dir_deg"] is None and w["variable"] is True and w["wind_kt"] == 4 and w["gust_kt"] == 18 + assert w["stale"] is False and body["credit"] + client.get("/weather/wind", params={"lat": 41.04, "lon": -73.77}) # same cell -> cached + assert len(http.calls) == 1 + params, headers = http.calls[0][1], http.calls[0][2] + s, w_, n, e = (float(v) for v in params["bbox"].split(",")) + assert s < 41.03 < n and w_ < -73.76 < e and params["format"] == "json" + assert "User-Agent" in headers + + +def test_wind_unavailable_reasons(): + with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.METAR_URL: Resp(None, status=204)})): + body = client.get("/weather/wind", params={"lat": 30.0, "lon": -50.0}).json() + assert body["available"] is False and "no METAR station" in body["reason"] and body["error"] is None + + weather._wind_cells.clear() + with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.METAR_URL: Resp(status=503)})): + body = client.get("/weather/wind", params={"lat": 41.03, "lon": -73.76}).json() + assert body["available"] is False and "unavailable" in body["reason"] and "503" in body["error"] -- 2.54.0 From bdbde359a1b50b86b34b8f7404ebf4e170438566 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Mon, 5 Oct 2026 02:58:32 -0400 Subject: [PATCH 2/2] Weather review fixes: zone-fetch time budget, prod NWS_USER_AGENT (Refs #237) - 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 --- drb-c2-core/app/internal/weather.py | 14 +++++++++++-- drb-c2-core/tests/test_weather.py | 21 +++++++++++++++++++ .../roles/deploy/templates/c2-core.env.j2 | 5 +++++ infra/ansible/vault.yml.example | 2 ++ 4 files changed, 40 insertions(+), 2 deletions(-) diff --git a/drb-c2-core/app/internal/weather.py b/drb-c2-core/app/internal/weather.py index aa7aae5..3d2e012 100644 --- a/drb-c2-core/app/internal/weather.py +++ b/drb-c2-core/app/internal/weather.py @@ -48,6 +48,11 @@ ZONE_CONCURRENCY = 6 # 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 @@ -61,7 +66,7 @@ _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 and not _warned_ua: + 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} @@ -193,7 +198,12 @@ async def _resolve_zones(client: "httpx.AsyncClient", alerts: List[Dict[str, Any wanted.append(z) if wanted: sem = asyncio.Semaphore(ZONE_CONCURRENCY) - await asyncio.gather(*(_fetch_zone(client, z, sem) for z in wanted[:MAX_ZONE_FETCHES_PER_REFRESH])) + 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 diff --git a/drb-c2-core/tests/test_weather.py b/drb-c2-core/tests/test_weather.py index 478f634..a55f9c8 100644 --- a/drb-c2-core/tests/test_weather.py +++ b/drb-c2-core/tests/test_weather.py @@ -216,3 +216,24 @@ def test_wind_unavailable_reasons(): with patch.object(weather.httpx, "AsyncClient", FakeHttp({weather.METAR_URL: Resp(status=503)})): body = client.get("/weather/wind", params={"lat": 41.03, "lon": -73.76}).json() assert body["available"] is False and "unavailable" in body["reason"] and "503" in body["error"] + + +def test_slow_zone_endpoint_cannot_hold_the_feed_lock(): + """Zone fetches run under the feed lock; a hung zone endpoint must not stall + every /weather/alerts request -- the alert lands in zone_only instead.""" + import time + + class SlowZones(FakeHttp): + async def get(self, url, params=None, headers=None): + if url.startswith(weather.NWS_ZONE_PREFIX): + await asyncio.sleep(5) + return await super().get(url, params, headers) + + http = SlowZones({weather.NWS_ALERTS_URL: Resp({"features": [_alert("z1", "Flood Watch", None, [ZONE_A])]}), + ZONE_A: Resp({"geometry": ZONE_POLY})}) + with patch.object(weather, "ZONE_BUDGET_S", 0.2), patch.object(weather.httpx, "AsyncClient", http): + t0 = time.time() + feed = asyncio.run(weather.alerts.get()) + assert time.time() - t0 < 2 + assert feed.error is None and [z["id"] for z in feed.zone_only] == ["z1"] + assert weather.zones.get(ZONE_A) == (False, None) # not cached as a failure; retried next refresh diff --git a/infra/ansible/roles/deploy/templates/c2-core.env.j2 b/infra/ansible/roles/deploy/templates/c2-core.env.j2 index f1ba13f..8291115 100644 --- a/infra/ansible/roles/deploy/templates/c2-core.env.j2 +++ b/infra/ansible/roles/deploy/templates/c2-core.env.j2 @@ -44,3 +44,8 @@ AGENT_SERVICE_KEY={{ vault_agent_service_key | default('') }} # frontend call to the API would have failed CORS. If the frontend ever moves # to app.{{ domain }}, change this at the same time. CORS_ORIGINS=["https://{{ domain }}"] + +# NWS weather alerts / METAR wind (server-26#237). api.weather.gov requires a +# User-Agent naming the app and a contact; default to the site URL so a vault +# that predates this key still sends an identifiable agent, not the placeholder. +NWS_USER_AGENT={{ vault_nws_user_agent | default('DRB scanner map (https://' ~ domain ~ ')', true) }} diff --git a/infra/ansible/vault.yml.example b/infra/ansible/vault.yml.example index 15d7a00..eacd784 100644 --- a/infra/ansible/vault.yml.example +++ b/infra/ansible/vault.yml.example @@ -60,3 +60,5 @@ vault_firebase_app_id: "" # No GCP key needed — the VM uses Application Default Credentials via the # GCE metadata server. Terraform grants the required IAM roles at apply time. + +vault_nws_user_agent: "" # optional; e.g. "DRB scanner map (ops@yourdomain)" -- blank/absent uses the site URL -- 2.54.0