From 8dddf88119ab8469bd654d4187139d98a1e58f1b Mon Sep 17 00:00:00 2001 From: malice Date: Mon, 6 Jul 2026 10:07:23 -0600 Subject: [PATCH] feat(coverage): wire hydro + satpass to the coverage bbox (native adapters complete) (#63) Phase 2d: hydro discovers gauges via the USGS IV bBox query across the coverage box (thresholds stay NWPS-cached; unrated sites never emit) instead of a hand-listed site set; satpass seeds a single centroid observer derived from the box. Completes wiring all native env adapters to the universal coverage bbox. Coverage governs when set; each adapter's config is fallback. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/env/store.py | 2 +- work/meshai/env/usgs.py | 52 ++++++-- work/meshai/main.py | 22 +++- work/tests/test_coverage_wiring_2d.py | 176 ++++++++++++++++++++++++++ 4 files changed, 237 insertions(+), 15 deletions(-) create mode 100644 work/tests/test_coverage_wiring_2d.py diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 42c860c..c784e30 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -89,7 +89,7 @@ class EnvironmentalStore: self._register_adapter("avalanche", config.avalanche, ".avalanche", "AvalancheAdapter", lambda cfg: (cfg, self._coverage_for("avalanche"))) self._register_adapter("usgs", config.usgs, ".usgs", "USGSStreamsAdapter", - lambda cfg: (cfg,)) + lambda cfg: (cfg, self._coverage_for("usgs"))) self._register_adapter("usgs_quake", config.usgs_quake, ".usgs_quake", "USGSQuakeAdapter", lambda cfg: (cfg, self._coverage_for("usgs_quake"))) self._register_adapter("traffic", config.traffic, ".traffic", "TomTomTrafficAdapter", diff --git a/work/meshai/env/usgs.py b/work/meshai/env/usgs.py index 841b648..538ee0e 100644 --- a/work/meshai/env/usgs.py +++ b/work/meshai/env/usgs.py @@ -35,8 +35,9 @@ class USGSStreamsAdapter: BASE_URL = "https://waterservices.usgs.gov/nwis/iv/" NWPS_BASE_URL = "https://api.water.noaa.gov/nwps/v1/gauges" - def __init__(self, config: "USGSConfig"): + def __init__(self, config: "USGSConfig", coverage: dict = None): self._sites = config.sites or [] + self._coverage_bbox = coverage["bbox"] if coverage is not None else None self._tick_interval = max(config.tick_seconds or 900, MIN_TICK_SECONDS) self._flood_thresholds = getattr(config, "flood_thresholds", {}) or {} self._last_tick = 0.0 @@ -61,8 +62,8 @@ class USGSStreamsAdapter: """ now = time.time() - # No sites configured - if not self._sites: + # No sites configured and no coverage bbox to discover by + if not self._sites and self._coverage_bbox is None: return False # Check tick interval @@ -245,6 +246,36 @@ class USGSStreamsAdapter: return result + def _build_iv_params(self, site_ids: list) -> dict: + """Build USGS IV API params — bBox-first when coverage is configured. + + When a coverage bbox is set, the query uses USGS IV bBox discovery + (west,south,east,north) and ignores preset site IDs. When no bbox is + configured, falls back to an explicit sites= list. Shared between + _fetch() and tests so the param logic is directly verifiable. + """ + if self._coverage_bbox is not None: + west, south, east, north = self._coverage_bbox + width = east - west + height = north - south + if width * height > 25: + logger.warning( + "USGS bBox %s too large (>25 sq-deg); readings may be rejected", + self._coverage_bbox, + ) + return { + "format": "json", + "bBox": ",".join(str(c) for c in self._coverage_bbox), + "parameterCd": "00060,00065", # Streamflow (cfs) and Gage height (ft) + "siteStatus": "active", + } + return { + "format": "json", + "sites": ",".join(site_ids), + "parameterCd": "00060,00065", # Streamflow (cfs) and Gage height (ft) + "siteStatus": "active", + } + def _fetch(self) -> bool: """Fetch instantaneous values from USGS Water Services. @@ -252,15 +283,10 @@ class USGSStreamsAdapter: True if data changed """ site_ids = self._get_site_ids() - if not site_ids: + if not site_ids and self._coverage_bbox is None: return False - params = { - "format": "json", - "sites": ",".join(site_ids), - "parameterCd": "00060,00065", # Streamflow (cfs) and Gage height (ft) - "siteStatus": "active", - } + params = self._build_iv_params(site_ids) url = f"{self.BASE_URL}?{urlencode(params)}" @@ -433,7 +459,11 @@ class USGSStreamsAdapter: self._is_loaded = True if changed: - logger.info(f"USGS streams updated: {len(new_events)} readings from {len(site_ids)} sites") + if self._coverage_bbox is not None: + logger.info("USGS streams updated: %d readings (bbox query)", len(new_events)) + else: + logger.info("USGS streams updated: %d readings from %d sites", + len(new_events), len(site_ids)) return changed diff --git a/work/meshai/main.py b/work/meshai/main.py index cc96513..fc8792e 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -399,11 +399,27 @@ class MeshAI: except Exception: logger.exception("persistence init_db failed at startup") - # Native satpass: seed observer_locations from SatpassConfig.observers - # (config is the source of truth; the table is the predictor's store). + # Native satpass: seed observer_locations. When a valid coverage bbox is + # configured, seed a single derived observer at the bbox centroid instead + # of the hand-listed config observers (coverage wins; config is fallback). try: from meshai.persistence.observer_locations import seed_observers_from_config - seed_observers_from_config(self.config.environmental.satpass) + from meshai import coverage as _cov + from types import SimpleNamespace + _sat_scope = _cov.resolve_adapter_coverage( + "satpass", + (self.config.coverage.bbox if self.config.coverage.enabled else []), + "native", + ) + if _sat_scope is not None: + lat, lon = _sat_scope["centroid"] + observers_to_seed = [ + {"slug": "coverage_center", "name": "Coverage Center", + "lat": lat, "lon": lon, "alt_m": 0.0} + ] + seed_observers_from_config(SimpleNamespace(observers=observers_to_seed)) + else: + seed_observers_from_config(self.config.environmental.satpass) except Exception: logger.exception("observer_locations seed failed at startup") diff --git a/work/tests/test_coverage_wiring_2d.py b/work/tests/test_coverage_wiring_2d.py new file mode 100644 index 0000000..baf3849 --- /dev/null +++ b/work/tests/test_coverage_wiring_2d.py @@ -0,0 +1,176 @@ +"""Tests for Phase 2d coverage-bbox wiring: hydro (usgs) + satpass. + +Verifies that: +1. usgs (hydro): _coverage_bbox is set from coverage dict; _build_iv_params uses + bBox when coverage is present, sites= when coverage is None; >25 sq-deg warns. +2. satpass: resolve_adapter_coverage returns a centroid dict; derived observer + matches the shape expected by seed_observers_from_config; empty bbox → None. +""" + +from __future__ import annotations + +import logging +from unittest.mock import MagicMock + +import pytest + +from meshai.coverage import resolve_adapter_coverage + +# --------------------------------------------------------------------------- +# Shared reference bbox (Magic Valley / south-central Idaho) +# --------------------------------------------------------------------------- +IDAHO_BOX = [-116.5, 42.0, -112.0, 44.0] + + +def _cov(adapter_name: str, bbox: list = None) -> dict | None: + return resolve_adapter_coverage(adapter_name, bbox or IDAHO_BOX, "native") + + +# =========================================================================== +# hydro (USGSStreamsAdapter) +# =========================================================================== + +def _usgs_cfg(sites=None): + cfg = MagicMock() + cfg.sites = sites if sites is not None else [] + cfg.tick_seconds = 900 + cfg.flood_thresholds = {} + return cfg + + +def test_usgs_coverage_bbox_set(): + """When coverage is provided, _coverage_bbox is set from coverage['bbox'].""" + from meshai.env.usgs import USGSStreamsAdapter + cov = _cov("usgs") + assert cov is not None + assert "bbox" in cov + cfg = _usgs_cfg() + adapter = USGSStreamsAdapter(cfg, coverage=cov) + assert adapter._coverage_bbox == cov["bbox"] + assert adapter._coverage_bbox == IDAHO_BOX + + +def test_usgs_fallback_no_coverage(): + """When coverage=None, _coverage_bbox is None and sites are used.""" + from meshai.env.usgs import USGSStreamsAdapter + cfg = _usgs_cfg(sites=["13090500"]) + adapter = USGSStreamsAdapter(cfg, coverage=None) + assert adapter._coverage_bbox is None + assert adapter._sites == ["13090500"] + + +def test_usgs_build_iv_params_with_coverage(): + """_build_iv_params uses bBox (not sites) when coverage is set.""" + from meshai.env.usgs import USGSStreamsAdapter + cov = _cov("usgs") + cfg = _usgs_cfg() + adapter = USGSStreamsAdapter(cfg, coverage=cov) + params = adapter._build_iv_params([]) + assert "bBox" in params + assert params["bBox"] == "-116.5,42.0,-112.0,44.0" + assert "sites" not in params + # Shared IV fields are always present + assert params["format"] == "json" + assert "parameterCd" in params + assert "siteStatus" in params + + +def test_usgs_build_iv_params_without_coverage(): + """_build_iv_params uses sites= (not bBox) when coverage is None.""" + from meshai.env.usgs import USGSStreamsAdapter + cfg = _usgs_cfg(sites=["13090500", "13095500"]) + adapter = USGSStreamsAdapter(cfg, coverage=None) + params = adapter._build_iv_params(["13090500", "13095500"]) + assert "sites" in params + assert params["sites"] == "13090500,13095500" + assert "bBox" not in params + + +def test_usgs_large_bbox_warns_but_does_not_crash(caplog): + """A bBox whose width*height > 25 sq-deg logs a warning but still builds params.""" + from meshai.env.usgs import USGSStreamsAdapter + # 10 deg wide × 10 deg tall = 100 sq-deg (well over the 25 limit) + large_bbox = [-120.0, 35.0, -110.0, 45.0] + cfg = _usgs_cfg() + adapter = USGSStreamsAdapter(cfg, coverage={"bbox": large_bbox}) + with caplog.at_level(logging.WARNING, logger="meshai.env.usgs"): + params = adapter._build_iv_params([]) + assert "too large" in caplog.text + # Params are still returned despite the warning + assert "bBox" in params + assert "sites" not in params + + +def test_usgs_small_bbox_no_warning(caplog): + """A bBox whose width*height <= 25 sq-deg does not log a warning.""" + from meshai.env.usgs import USGSStreamsAdapter + # IDAHO_BOX: (112.0-116.5) * (44.0-42.0) = 4.5 * 2 = 9 sq-deg — well under 25 + cfg = _usgs_cfg() + adapter = USGSStreamsAdapter(cfg, coverage={"bbox": IDAHO_BOX}) + with caplog.at_level(logging.WARNING, logger="meshai.env.usgs"): + params = adapter._build_iv_params([]) + assert "too large" not in caplog.text + assert "bBox" in params + + +# =========================================================================== +# satpass — observer derivation logic +# =========================================================================== + +def test_satpass_coverage_centroid_correct(): + """resolve_adapter_coverage('satpass', bbox) returns the bbox centroid.""" + cov = _cov("satpass") + assert cov is not None + assert "centroid" in cov + lat, lon = cov["centroid"] + # centroid of IDAHO_BOX = [W=-116.5, S=42.0, E=-112.0, N=44.0] + # lat_center = (42.0 + 44.0) / 2 = 43.0 + # lon_center = (-116.5 + -112.0) / 2 = -114.25 + assert lat == pytest.approx(43.0) + assert lon == pytest.approx(-114.25) + + +def test_satpass_derived_observer_shape(): + """Derived coverage observer has the exact keys seed_observers_from_config expects.""" + cov = _cov("satpass") + assert cov is not None + lat, lon = cov["centroid"] + derived = [{"slug": "coverage_center", "name": "Coverage Center", + "lat": lat, "lon": lon, "alt_m": 0.0}] + obs = derived[0] + # seed_observers_from_config reads: slug, name, lat, lon, alt_m (optional) + assert obs["slug"] == "coverage_center" + assert obs["name"] == "Coverage Center" + assert obs["lat"] == pytest.approx(43.0) + assert obs["lon"] == pytest.approx(-114.25) + assert obs["alt_m"] == 0.0 + + +def test_satpass_empty_bbox_returns_none(): + """When coverage_bbox is empty, resolve returns None → config fallback.""" + cov = resolve_adapter_coverage("satpass", [], "native") + assert cov is None + + +def test_satpass_central_feed_returns_none(): + """Central-fed satpass always returns None (Central governs coverage).""" + cov = resolve_adapter_coverage("satpass", IDAHO_BOX, "central") + assert cov is None + + +def test_satpass_coverage_disabled_fallback(): + """When coverage is not enabled (empty bbox passed), config observers are used.""" + # Simulate main.py's: (config.coverage.bbox if config.coverage.enabled else []) + effective_bbox = [] # coverage.enabled is False + cov = resolve_adapter_coverage("satpass", effective_bbox, "native") + assert cov is None + # When cov is None the fallback path in main.py uses config.environmental.satpass + config_observers = [ + {"slug": "home", "name": "Home Base", "lat": 42.56, "lon": -114.47, "alt_m": 1000} + ] + # Verify: when cov is None, observers_to_seed == config.satpass.observers + observers_to_seed = config_observers if cov is None else [ + {"slug": "coverage_center", "name": "Coverage Center", + "lat": 0.0, "lon": 0.0, "alt_m": 0.0} + ] + assert observers_to_seed == config_observers