mangrovedigital/tide-engine-api
0
1"""Wind Sentinel — trailing-window wind advisory + optional storm-flush flag.2 3Additive engine module. Reads NDBC realtime2 and NWS obs history, computes4per-region trailing wind statistics, and produces honest-water advisories.5"""6import json7import os8import math9import time10import tempfile11import urllib.parse12from datetime import datetime, timezone, timedelta13 14from . import sources15from . import stations16 17CONFIG_PATH = os.path.join(os.path.dirname(__file__), "data", "sentinel_regions.v1.json")18 19 20def _iso(epoch):21 return datetime.fromtimestamp(epoch, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")22 23 24def load_config(path=CONFIG_PATH):25 with open(path, "r", encoding="utf-8") as f:26 return json.load(f)27 28 29def region_for_point(lat, lon, cfg=None):30 """Return region key for (lat,lon) or None if out of coverage."""31 cfg = cfg or load_config()32 for key, r in cfg["regions"].items():33 b = r["bbox"]34 if b["south"] <= lat <= b["north"] and b["west"] <= lon <= b["east"]:35 return key36 return None37 38 39# ---------------------------------------------------------------------------40# NDBC realtime2 reader (newest-first)41# ---------------------------------------------------------------------------42 43def _ndbc_row_epoch(parts):44 """YY or YYYY MM DD hh mm -> UTC epoch seconds."""45 try:46 yy, mm, dd, hh, mn = map(int, parts[:5])47 year = yy if yy > 1900 else 2000 + yy48 dt = datetime(year, mm, dd, hh, mn, tzinfo=timezone.utc)49 return int(dt.timestamp())50 except Exception:51 return None52 53 54def _ndbc_float(s):55 try:56 if s in ("MM", "", "99", "999", "9999"):57 return None58 return float(s)59 except Exception:60 return None61 62 63def read_ndbc_realtime2(text):64 """Parse NDBC realtime2 text (newest-first).65 66 Returns {"rows": [...], "hold": None, "latest_epoch": int, "oldest_epoch": int}67 or {"rows": [], "hold": "ordering ambiguous", ...} if the invariant fails.68 69 Each row: {epoch, wdir, wspd_kt, gust_kt, pres_hpa, atmp_c}.70 """71 lines = [l.rstrip() for l in text.splitlines() if l.strip()]72 data_lines = [l for l in lines if not l.startswith("#")]73 if not data_lines:74 return {"rows": [], "hold": "no data", "latest_epoch": None, "oldest_epoch": None}75 76 # Header is the first comment line; determine column positions.77 header = None78 for l in lines:79 if l.startswith("#YY"):80 header = l.lstrip("#").split()81 break82 default_header = ["YY", "MM", "DD", "hh", "mm", "WDIR", "WSPD", "GST",83 "WVHT", "DPD", "APD", "MWD", "PRES", "ATMP", "WTMP",84 "DEWP", "VIS", "PTDY", "TIDE"]85 cols = header or default_header86 87 def col(name):88 try:89 return cols.index(name)90 except ValueError:91 return None92 93 wdir_idx = col("WDIR")94 wspd_idx = col("WSPD")95 gst_idx = col("GST")96 pres_idx = col("PRES")97 atmp_idx = col("ATMP")98 99 rows = []100 for line in data_lines:101 parts = line.split()102 if len(parts) < 5:103 continue104 epoch = _ndbc_row_epoch(parts)105 if epoch is None:106 continue107 row = {"epoch": epoch}108 if wdir_idx is not None and wdir_idx < len(parts):109 row["wdir"] = _ndbc_float(parts[wdir_idx])110 if wspd_idx is not None and wspd_idx < len(parts):111 row["wspd_kt"] = _ndbc_float(parts[wspd_idx])112 if gst_idx is not None and gst_idx < len(parts):113 row["gust_kt"] = _ndbc_float(parts[gst_idx])114 if pres_idx is not None and pres_idx < len(parts):115 row["pres_hpa"] = _ndbc_float(parts[pres_idx])116 if atmp_idx is not None and atmp_idx < len(parts):117 row["atmp_c"] = _ndbc_float(parts[atmp_idx])118 rows.append(row)119 120 if not rows:121 return {"rows": [], "hold": "no data", "latest_epoch": None, "oldest_epoch": None}122 123 latest = rows[0]["epoch"]124 oldest = rows[-1]["epoch"]125 if latest < oldest:126 return {"rows": [], "hold": "ordering ambiguous", "latest_epoch": latest,127 "oldest_epoch": oldest}128 129 return {"rows": rows, "hold": None, "latest_epoch": latest, "oldest_epoch": oldest}130 131 132# ---------------------------------------------------------------------------133# NWS obs history reader134# ---------------------------------------------------------------------------135 136def _mps_to_kt(q):137 if not q or q.get("value") is None:138 return None139 val = float(q["value"])140 unit = (q.get("unitCode") or "").lower()141 if "km_h" in unit:142 return val * 0.539957143 if "m_s" in unit or "m/s" in unit:144 return val * 1.94384145 # Assume knots if no known unit146 return val147 148 149def read_nws_obs_history(data):150 """Parse NWS /stations/{id}/observations GeoJSON history.151 152 Returns {"rows": [...], "hold": None}153 Each row: {epoch, station, wspd_kt, gust_kt, wdir_deg, pres_hpa}.154 Rows are returned ascending by epoch (oldest first).155 """156 rows = []157 for f in data.get("features", []):158 p = f.get("properties", {})159 try:160 ts = p["timestamp"].replace("Z", "+00:00")161 epoch = int(datetime.fromisoformat(ts).timestamp())162 except Exception:163 continue164 station = (f.get("properties") or {}).get("station", "unknown")165 wspd = _mps_to_kt(p.get("windSpeed"))166 gust = _mps_to_kt(p.get("windGust"))167 wdir = p.get("windDirection", {}).get("value")168 slpq = p.get("seaLevelPressure") or p.get("barometricPressure") or {}169 pres = None if slpq.get("value") is None else float(slpq["value"]) / 100.0170 rows.append({171 "epoch": epoch,172 "station": station,173 "wspd_kt": wspd,174 "gust_kt": gust,175 "wdir_deg": None if wdir is None else float(wdir),176 "pres_hpa": pres,177 })178 rows.sort(key=lambda r: r["epoch"])179 return {"rows": rows, "hold": None}180 181 182# ---------------------------------------------------------------------------183# Fetch wrappers184# ---------------------------------------------------------------------------185 186def fetch_ndbc_realtime2(station_id):187 url = f"https://www.ndbc.noaa.gov/data/realtime2/{station_id}.txt"188 try:189 resp = sources._get_with_retry(url, timeout=30)190 return resp.text191 except Exception as exc:192 raise sources.UpstreamError(f"NDBC {station_id} unreachable: {exc}")193 194 195def fetch_nws_obs_history(station_id, start_date, end_date):196 """Fetch NWS observations for station between start_date and end_date (YYYY-MM-DD)."""197 s = datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc).timestamp()198 e = (datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc)199 + timedelta(days=1)).timestamp()200 url = (f"https://api.weather.gov/stations/{station_id}/observations"201 f"?start={datetime.fromtimestamp(s, timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}"202 f"&end={datetime.fromtimestamp(min(e, time.time()), timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}")203 try:204 data, _ = sources._fetch_json(url, f"nws-sentinel|{url}", ttl_seconds=3600)205 return data206 except Exception as exc:207 raise sources.UpstreamError(f"NWS {station_id} unreachable: {exc}")208 209 210# ---------------------------------------------------------------------------211# Wind direction + doctrine classification212# ---------------------------------------------------------------------------213 214_OCTANTS = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]215 216 217def deg_to_octant(deg):218 """Convert wind-from degrees to one of 8 compass octants."""219 if deg is None or not math.isfinite(deg):220 return None221 deg = deg % 360.0222 idx = int(round(deg / 45.0)) % 8223 return _OCTANTS[idx]224 225 226def _doctrine_for_octant(octant, doctrine):227 if octant in doctrine.get("below_chart", []):228 return "below_chart"229 if octant in doctrine.get("above_chart", []):230 return "above_chart"231 return "neutral"232 233 234def _class_label(octants):235 """Collapse consecutive octants into a readable class, e.g. ['NE','E'] -> 'NE-E'."""236 if not octants:237 return "variable"238 uniq = []239 for o in octants:240 if not uniq or o != uniq[-1]:241 uniq.append(o)242 if len(uniq) == 1:243 return uniq[0]244 return "-".join(uniq)245 246 247# ---------------------------------------------------------------------------248# Trailing-window statistics249# ---------------------------------------------------------------------------250 251def _time_weighted_direction(rows):252 """Mean wind direction weighted by time between samples."""253 if not rows:254 return None255 sum_u = sum_v = 0.0256 total_dt = 0.0257 for i in range(len(rows) - 1):258 dt = max(rows[i + 1]["epoch"] - rows[i]["epoch"], 1)259 d = rows[i].get("wdir_deg")260 if d is None:261 continue262 rad = math.radians(d)263 sum_u += math.sin(rad) * dt264 sum_v += math.cos(rad) * dt265 total_dt += dt266 if total_dt <= 0:267 return None268 mean_deg = math.degrees(math.atan2(sum_u / total_dt, sum_v / total_dt))269 if mean_deg < 0:270 mean_deg += 360.0271 return mean_deg272 273 274def _rolling_mean_window(rows, window_s):275 """Max rolling-mean sustained wind over window_s seconds."""276 if not rows:277 return 0.0278 best = 0.0279 j = 0280 for i, r in enumerate(rows):281 while j < i and rows[j]["epoch"] < r["epoch"] - window_s:282 j += 1283 window = rows[j:i + 1]284 vals = [x["wspd_kt"] for x in window if x.get("wspd_kt") is not None]285 if vals:286 best = max(best, sum(vals) / len(vals))287 return best288 289 290def trailing_advisory(rows, now_epoch, doctrine):291 """Compute trailing-window advisory from wind rows (ascending epoch).292 293 Returns dict with keys:294 windows: {6: {max_sustained_kt, max_gust_kt, dominant_octant}, 24: ..., 48: ...}295 fire: bool296 level: 'ADVISORY' or None297 direction_class: str or None298 doctrine_sign: 'below_chart' | 'above_chart' | 'neutral' | None299 ended_hours_ago: float or 'ongoing'300 copy: str301 """302 recent = [r for r in rows if r["epoch"] <= now_epoch and r["epoch"] >= now_epoch - 48 * 3600]303 if not recent:304 return _no_advisory("no recent wind data")305 306 # Maxima per window307 windows = {}308 for h in (6, 24, 48):309 cut = now_epoch - h * 3600310 win = [r for r in recent if r["epoch"] >= cut]311 sustained = [r["wspd_kt"] for r in win if r.get("wspd_kt") is not None]312 gusts = [r["gust_kt"] for r in win if r.get("gust_kt") is not None]313 windows[h] = {314 "max_sustained_kt": round(max(sustained), 1) if sustained else None,315 "max_gust_kt": round(max(gusts), 1) if gusts else None,316 }317 318 # Time-weighted dominant direction across the full 48h window319 dom_deg = _time_weighted_direction(recent)320 dom_octant = deg_to_octant(dom_deg)321 322 # Fire gates323 sustained_6h = _rolling_mean_window(recent, 6 * 3600)324 max_gust_48h = windows[48].get("max_gust_kt")325 fire = (sustained_6h >= 15.0) or (max_gust_48h is not None and max_gust_48h >= 25.0)326 327 if not fire:328 out = _no_advisory("trailing winds below advisory thresholds")329 out["windows"] = windows330 return out331 332 # Determine when the event ended: last epoch in the 48h window where333 # sustained >= 12 kt OR gust >= 20 kt.334 event_rows = [r for r in recent335 if (r.get("wspd_kt") is not None and r["wspd_kt"] >= 12.0)336 or (r.get("gust_kt") is not None and r["gust_kt"] >= 20.0)]337 if not event_rows:338 # Should not happen if fire, but keep safe.339 return _no_advisory("event criteria mismatch")340 341 last_event_epoch = event_rows[-1]["epoch"]342 ongoing = (now_epoch - last_event_epoch) <= 2 * 3600343 if ongoing:344 ended = "ongoing"345 else:346 ended = round((now_epoch - last_event_epoch) / 3600.0, 1)347 348 sign = _doctrine_for_octant(dom_octant, doctrine)349 copy = _advisory_copy(dom_octant, sign, ended)350 351 return {352 "fire": True,353 "level": "ADVISORY",354 "direction_class": _class_label([dom_octant]) if dom_octant else "variable",355 "doctrine_sign": sign,356 "ended_hours_ago": ended,357 "copy": copy,358 "windows": windows,359 "dominant_octant": dom_octant,360 }361 362 363def _no_advisory(reason):364 return {365 "fire": False,366 "level": None,367 "direction_class": None,368 "doctrine_sign": None,369 "ended_hours_ago": None,370 "copy": None,371 "windows": {6: {}, 24: {}, 48: {}},372 "reason": reason,373 }374 375 376# ---------------------------------------------------------------------------377# Storm-flush detector (Collier v1)378# ---------------------------------------------------------------------------379 380def usgs_flush_series(site, param, start_date, end_date):381 """Fetch USGS instantaneous values for a sentinel flush sensor."""382 q = {383 "format": "json",384 "sites": site,385 "parameterCd": param,386 "startDT": start_date,387 "endDT": end_date,388 }389 url = sources.USGS_IV + "?" + urllib.parse.urlencode(q)390 data, _ = sources._fetch_json(url, f"usgs-flush|{url}", ttl_seconds=3600)391 ts_list = data.get("value", {}).get("timeSeries", [])392 if not ts_list:393 raise sources.NoDataError(f"USGS returned no timeSeries for {site}/{param}")394 rows = []395 for r in ts_list[0]["values"][0]["value"]:396 try:397 epoch = sources._parse_iso_to_epoch(r["dateTime"])398 val = float(r["value"])399 except Exception:400 continue401 if val <= -999998:402 continue403 rows.append((epoch, val))404 rows.sort(key=lambda x: x[0])405 return rows406 407 408def _same_phase_baseline(epoch, series, window_days=3):409 """Mean value at the same clock time over the prior window_days."""410 dt = datetime.fromtimestamp(epoch, tz=timezone.utc)411 baseline_values = []412 for d in range(1, window_days + 1):413 target = dt - timedelta(days=d)414 target_epoch = target.timestamp()415 nearest = min(series, key=lambda x: abs(x[0] - target_epoch), default=None)416 if nearest and abs(nearest[0] - target_epoch) < 3600:417 baseline_values.append(nearest[1])418 return sum(baseline_values) / len(baseline_values) if baseline_values else None419 420 421def detect_flush(salinity_series, discharge_series, now_epoch,422 salinity_drop_threshold=3.0, discharge_anomaly_factor=2.0,423 lookback_hours=48, baseline_days=3):424 """Return flush event dict.425 426 Signature: salinity has dropped >= salinity_drop_threshold units within427 the trailing lookback_hours, AND discharge is anomalously outbound versus428 the prior baseline_days same-phase baseline.429 430 Returns {"fired": bool, "since_epoch": int|None, "reason": str, "hold": str|None}.431 """432 if not salinity_series or not discharge_series:433 return {"fired": False, "since_epoch": None, "reason": "no data", "hold": None}434 435 sal_map = dict(salinity_series)436 discharge_map = dict(discharge_series)437 438 # Latest salinity at or before now_epoch439 sal_latest_t = max((t for t in sal_map if t <= now_epoch), default=None)440 if sal_latest_t is None:441 return {"fired": False, "since_epoch": None, "reason": "no salinity data", "hold": None}442 sal_now = sal_map[sal_latest_t]443 444 # Max salinity in trailing window445 window_start = now_epoch - lookback_hours * 3600446 recent_sal = [v for t, v in salinity_series if window_start <= t <= now_epoch]447 if not recent_sal:448 return {"fired": False, "since_epoch": None, "reason": "no recent salinity", "hold": None}449 max_sal = max(recent_sal)450 sal_drop = max_sal - sal_now451 452 if sal_drop < salinity_drop_threshold:453 return {"fired": False, "since_epoch": None,454 "reason": f"salinity drop {sal_drop:.1f} < {salinity_drop_threshold}",455 "hold": None}456 457 # Discharge: latest value at or before now_epoch vs same-phase baseline.458 dis_latest_t = max((t for t in discharge_map if t <= now_epoch), default=None)459 if dis_latest_t is None:460 return {"fired": False, "since_epoch": None, "reason": "no discharge data", "hold": None}461 discharge_now = discharge_map[dis_latest_t]462 baseline = _same_phase_baseline(dis_latest_t, discharge_series, window_days=baseline_days)463 if baseline is None:464 return {"fired": False, "since_epoch": None, "reason": "no discharge baseline", "hold": None}465 466 # Outbound = positive discharge; anomaly = current exceeds baseline.467 anomaly = discharge_now - baseline468 if anomaly <= 0 or (baseline > 0 and anomaly < baseline * (discharge_anomaly_factor - 1)):469 return {"fired": False, "since_epoch": None,470 "reason": f"discharge anomaly {anomaly:.1f} vs baseline {baseline:.1f} not outbound enough",471 "hold": None}472 473 # Find earliest epoch in the trailing window where both thresholds were met.474 since = None475 for t, _ in sorted(salinity_series, key=lambda x: x[0]):476 if t > now_epoch:477 continue478 win = [v for tt, v in salinity_series if t - lookback_hours * 3600 <= tt <= t]479 if not win:480 continue481 drop_at_t = max(win) - sal_map.get(t, max(win))482 if drop_at_t < salinity_drop_threshold:483 continue484 dis_t = max((tt for tt in discharge_map if tt <= t), default=None)485 if dis_t is None:486 continue487 base_t = _same_phase_baseline(dis_t, discharge_series, window_days=baseline_days)488 if base_t is None:489 continue490 if discharge_map[dis_t] - base_t > 0:491 since = t492 break493 494 return {"fired": True, "since_epoch": since,495 "reason": f"salinity drop {sal_drop:.1f}, discharge anomaly {anomaly:.1f} vs baseline {baseline:.1f}",496 "hold": None}497 498 499BANNED_WORDS = ["exact", "perfect", "guaranteed"]500 501 502def _advisory_copy(octant, sign, ended):503 """Honest-water qualitative copy. Never prints a number of feet."""504 if octant in ("NE", "E", "SE"):505 direction_phrase = "northeast-to-east"506 elif octant in ("SW", "W", "NW"):507 direction_phrase = "west-to-southwest"508 elif octant == "N":509 direction_phrase = "northerly"510 elif octant == "S":511 direction_phrase = "southerly"512 else:513 direction_phrase = "strong"514 515 if sign == "below_chart":516 water_phrase = "expect the bay to carry less water than chart on the ebb"517 elif sign == "above_chart":518 water_phrase = "expect the bay to carry more water than chart while the wind piles in"519 else:520 water_phrase = "watch for local shifts in water level relative to chart"521 522 if ended == "ongoing":523 return (f"A {direction_phrase} blow is ongoing; {water_phrase}. "524 f"Plan for wind-driven conditions until it eases.")525 else:526 return (f"A {direction_phrase} blow moved through within the last day; "527 f"{water_phrase}. "528 f"Allow a few hours after the wind eases for the water to settle.")529 530 531# ---------------------------------------------------------------------------532# Endpoint assembly533# ---------------------------------------------------------------------------534 535def _nearest_station(lat, lon, station_list):536 """Return nearest station dict from a list with lat/lon keys."""537 best = None538 bd = float("inf")539 for s in station_list:540 d = stations._mi(lat, lon, s["lat"], s["lon"])541 if d < bd:542 bd = d543 best = s544 return best, bd545 546 547def _fetch_ndbc_text_cached(station_id, ttl_seconds=600):548 """Fetch NDBC realtime2 text with disk caching."""549 url = f"https://www.ndbc.noaa.gov/data/realtime2/{station_id}.txt"550 cache_key = f"ndbc-sentinel|{station_id}"551 try:552 # Try JSON cache wrapper so we can reuse sources._fetch_json logic.553 data, _ = sources._fetch_json(url, cache_key, ttl_seconds=ttl_seconds)554 # _fetch_json parsed as JSON; for text it fails. Fall back to text cache.555 except Exception:556 pass557 558 os.makedirs(sources.CACHE_DIR, exist_ok=True)559 import hashlib560 h = hashlib.sha1(cache_key.encode()).hexdigest()[:16]561 path = os.path.join(sources.CACHE_DIR, f"{h}.txt")562 meta_path = path + ".meta"563 if os.path.exists(path):564 fresh = True565 if ttl_seconds is not None and os.path.exists(meta_path):566 with open(meta_path) as f:567 fetched_at = float(f.read().strip())568 fresh = (time.time() - fetched_at) < ttl_seconds569 if fresh:570 with open(path, "r", encoding="utf-8") as f:571 return f.read()572 resp = sources._get_with_retry(url, timeout=30)573 text = resp.text574 fd, tmp = tempfile.mkstemp(dir=sources.CACHE_DIR, suffix=".tmp")575 try:576 with os.fdopen(fd, "w", encoding="utf-8") as f:577 f.write(text)578 os.replace(tmp, path)579 except BaseException:580 try:581 os.unlink(tmp)582 except OSError:583 pass584 raise585 with open(meta_path, "w") as f:586 f.write(str(time.time()))587 return text588 589 590def _station_point(station_id):591 """Return (lat, lon) for an NWS or NDBC station id (best-effort)."""592 # NWS station lat/lon requires a separate fetch; for speed we skip it here.593 return None, None594 595 596def wind_sentinel_for_point(lat, lon):597 """Build the GET /api/wind_sentinel response dict.598 599 Fetches the nearest NDBC F1 and NWS obs station for the containing region,600 computes a trailing advisory, and returns the documented shape.601 """602 cfg = load_config()603 region_key = region_for_point(lat, lon, cfg)604 issued = _iso(time.time())605 limitations = []606 607 if region_key is None:608 return {609 "status": "ok",610 "region": None,611 "coverage": False,612 "wind_event": None,613 "flush": None,614 "receipts": [],615 "issued_utc": issued,616 "provenance": {"source": "wind sentinel v1", "region": None},617 "limitations": ["Location outside sentinel coverage area."],618 }619 620 region = cfg["regions"][region_key]621 doctrine = region["doctrine"]622 623 receipts = []624 all_rows = []625 hold_reasons = []626 627 for sid in region["ndbc_f1_ids"]:628 try:629 text = _fetch_ndbc_text_cached(sid)630 parsed = read_ndbc_realtime2(text)631 if parsed["hold"]:632 hold_reasons.append(f"NDBC {sid}: {parsed['hold']}")633 receipts.append({"station": sid, "source": "NDBC", "latest": None,634 "status": f"hold:{parsed['hold']}"})635 continue636 rows = [{"epoch": r["epoch"],637 "wspd_kt": r.get("wspd_kt"),638 "gust_kt": r.get("gust_kt"),639 "wdir_deg": r.get("wdir")} for r in parsed["rows"]]640 all_rows.extend(rows)641 receipts.append({"station": sid, "source": "NDBC", "latest": parsed["latest_epoch"],642 "latest_iso": _iso(parsed["latest_epoch"]), "status": "ok",643 "rows": len(rows)})644 except Exception as exc:645 hold_reasons.append(f"NDBC {sid}: {exc}")646 receipts.append({"station": sid, "source": "NDBC", "latest": None,647 "status": f"hold:{exc}"})648 649 for sid in region["nws_obs_stations"]:650 try:651 # Fetch last 48h of obs652 end = datetime.now(tz=timezone.utc)653 start = end - timedelta(hours=48)654 data, _ = sources._fetch_json(655 (f"https://api.weather.gov/stations/{sid}/observations"656 f"?start={start.strftime('%Y-%m-%dT%H:%M:%SZ')}"657 f"&end={end.strftime('%Y-%m-%dT%H:%M:%SZ')}"),658 f"nws-sentinel|{sid}|{start.strftime('%Y%m%d%H')}",659 ttl_seconds=600)660 parsed = read_nws_obs_history(data)661 if parsed["rows"]:662 all_rows.extend(parsed["rows"])663 receipts.append({"station": sid, "source": "NWS",664 "latest": parsed["rows"][-1]["epoch"],665 "latest_iso": _iso(parsed["rows"][-1]["epoch"]),666 "status": "ok", "rows": len(parsed["rows"])})667 else:668 receipts.append({"station": sid, "source": "NWS", "latest": None,669 "status": "hold:no observations"})670 except Exception as exc:671 hold_reasons.append(f"NWS {sid}: {exc}")672 receipts.append({"station": sid, "source": "NWS", "latest": None,673 "status": f"hold:{exc}"})674 675 if not all_rows:676 return {677 "status": "hold:no wind data",678 "region": region_key,679 "coverage": True,680 "wind_event": None,681 "flush": None,682 "receipts": receipts,683 "issued_utc": issued,684 "provenance": {"source": "wind sentinel v1", "region": region_key,685 "stations": region["ndbc_f1_ids"] + region["nws_obs_stations"]},686 "limitations": hold_reasons or ["No usable wind observations for this region."],687 }688 689 all_rows.sort(key=lambda r: r["epoch"])690 adv = trailing_advisory(all_rows, int(time.time()), doctrine)691 692 # Stage 4 flush is HELD: do not return a flush flag in v1.693 flush = None694 limitations.append("Storm-flush detector is HELD for v1; flush flag not active.")695 if hold_reasons:696 limitations.extend(hold_reasons)697 698 return {699 "status": "ok",700 "region": region_key,701 "coverage": True,702 "wind_event": {703 "level": adv["level"],704 "direction_class": adv["direction_class"],705 "doctrine_sign": adv["doctrine_sign"],706 "ended_hours_ago": adv["ended_hours_ago"],707 "copy": adv["copy"],708 "windows": adv["windows"],709 } if adv["fire"] else None,710 "flush": flush,711 "receipts": receipts,712 "issued_utc": issued,713 "provenance": {"source": "wind sentinel v1", "region": region_key,714 "stations": region["ndbc_f1_ids"] + region["nws_obs_stations"]},715 "limitations": limitations,716 }717