From a3e5a7518f03d2b38337b11062f741f465b6ef8c Mon Sep 17 00:00:00 2001 From: malice Date: Mon, 6 Jul 2026 10:11:33 -0600 Subject: [PATCH] =?UTF-8?q?feat(coverage):=20per-adapter=20override=20?= =?UTF-8?q?=E2=80=94=20exclude=20adapters=20from=20the=20coverage=20bbox?= =?UTF-8?q?=20(#64)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds coverage.excluded_adapters: adapters listed here ignore the universal coverage bbox and use their own config (the "advanced override", native only). _coverage_for() returns None for excluded adapters; satpass seeding respects it too. Backend for the GUI override toggles. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/config.py | 1 + work/meshai/env/store.py | 11 ++- work/meshai/main.py | 6 +- work/tests/test_coverage_overrides.py | 115 ++++++++++++++++++++++++++ 4 files changed, 131 insertions(+), 2 deletions(-) create mode 100644 work/tests/test_coverage_overrides.py diff --git a/work/meshai/config.py b/work/meshai/config.py index c182374..5dc3ae3 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -807,6 +807,7 @@ class Coverage: bbox: list = field(default_factory=list) # [west, south, east, north]; empty = not configured enabled: bool = True # master switch for deriving adapter scope from bbox + excluded_adapters: list = field(default_factory=list) # adapters that OPT OUT of coverage bbox and use their own config @dataclass diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index c784e30..e1c9e93 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -39,6 +39,7 @@ class EnvironmentalStore: region_anchors: list = None, event_bus: Optional["EventBus"] = None, coverage_bbox: list = None, + coverage_excluded: list = None, ): self._adapters = {} # name -> adapter instance self._failed_adapters = {} # name -> last_error string @@ -49,6 +50,7 @@ class EnvironmentalStore: self._mesh_zones = config.nws_zones or [] self._region_anchors = region_anchors or [] self._coverage_bbox = coverage_bbox or [] + self._coverage_excluded = set(coverage_excluded or []) # ── Received-delta gate (NATIVE-only) ──────────────────────────── # The model the operator demanded: a native adapter broadcasts an item @@ -148,7 +150,14 @@ class EnvironmentalStore: self._failed_adapters[name] = err_msg def _coverage_for(self, adapter: str): - """Derived coverage scope for a NATIVE adapter, or None to use its own config (override/fallback).""" + """Derived coverage scope for a NATIVE adapter, or None to use its own config (override/fallback). + + Returns None immediately for any adapter listed in coverage_excluded, so + that adapter falls back to its own config fields exactly as if no coverage + bbox were set (the "advanced override" escape hatch). + """ + if adapter in self._coverage_excluded: + return None from meshai import coverage as _cov return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native") diff --git a/work/meshai/main.py b/work/meshai/main.py index fc8792e..795889f 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -402,13 +402,16 @@ class MeshAI: # 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). + # Exception: if "satpass" is in coverage.excluded_adapters, treat as if + # no coverage bbox is set — fall back to the adapter's own config observers. try: from meshai.persistence.observer_locations import seed_observers_from_config from meshai import coverage as _cov from types import SimpleNamespace + _satpass_excluded = "satpass" in self.config.coverage.excluded_adapters _sat_scope = _cov.resolve_adapter_coverage( "satpass", - (self.config.coverage.bbox if self.config.coverage.enabled else []), + (self.config.coverage.bbox if (self.config.coverage.enabled and not _satpass_excluded) else []), "native", ) if _sat_scope is not None: @@ -637,6 +640,7 @@ class MeshAI: self.env_store = EnvironmentalStore( config=env_cfg, region_anchors=region_anchors, coverage_bbox=coverage_bbox, event_bus=self.event_bus, + coverage_excluded=cov.excluded_adapters, ) logger.info(f"Environmental feeds enabled ({len(self.env_store._adapters)} adapters)") else: diff --git a/work/tests/test_coverage_overrides.py b/work/tests/test_coverage_overrides.py new file mode 100644 index 0000000..dfda5e4 --- /dev/null +++ b/work/tests/test_coverage_overrides.py @@ -0,0 +1,115 @@ +"""Tests for Phase 3a coverage override (per-adapter excluded_adapters escape hatch). + +Verifies: +1. Coverage.excluded_adapters defaults to []. +2. A store built with coverage_bbox + coverage_excluded=["traffic"]: + - _coverage_for("traffic") returns None (excluded). + - _coverage_for("usgs_quake") returns a dict (non-excluded still derives). +3. With coverage_excluded=[]: _coverage_for("traffic") returns a dict. +""" + +from __future__ import annotations + +from dataclasses import field +from unittest.mock import MagicMock + +import pytest + +# Shared reference bbox (Magic Valley / south-central Idaho) +IDAHO_BOX = [-116.5, 42.0, -112.0, 44.0] + + +# --------------------------------------------------------------------------- +# Helpers — minimal EnvironmentalConfig mock (mirrors test_coverage_wiring.py) +# --------------------------------------------------------------------------- + +def _make_env_cfg(): + """Minimal EnvironmentalConfig mock with all adapters disabled.""" + env_cfg = MagicMock() + for attr in ("nws", "swpc", "ducting", "fires", "avalanche", "usgs", + "usgs_quake", "traffic", "roads511", "wzdx", "satpass", "firms"): + sub = MagicMock() + sub.enabled = False + sub.feed_source = "central" + setattr(env_cfg, attr, sub) + env_cfg.nws_zones = [] + return env_cfg + + +# --------------------------------------------------------------------------- +# Coverage dataclass +# --------------------------------------------------------------------------- + +def test_coverage_excluded_adapters_default(): + """Coverage.excluded_adapters defaults to an empty list.""" + from meshai.config import Coverage + cov = Coverage() + assert cov.excluded_adapters == [] + + +# --------------------------------------------------------------------------- +# EnvironmentalStore exclusion gate +# --------------------------------------------------------------------------- + +def test_coverage_for_excluded_adapter_returns_none(): + """_coverage_for returns None for an adapter in coverage_excluded.""" + from meshai.env.store import EnvironmentalStore + env_cfg = _make_env_cfg() + store = EnvironmentalStore( + config=env_cfg, + coverage_bbox=IDAHO_BOX, + coverage_excluded=["traffic"], + ) + result = store._coverage_for("traffic") + assert result is None + + +def test_coverage_for_non_excluded_adapter_returns_dict(): + """_coverage_for returns a dict for adapters NOT in coverage_excluded.""" + from meshai.env.store import EnvironmentalStore + env_cfg = _make_env_cfg() + store = EnvironmentalStore( + config=env_cfg, + coverage_bbox=IDAHO_BOX, + coverage_excluded=["traffic"], + ) + result = store._coverage_for("usgs_quake") + assert isinstance(result, dict) + assert "bbox" in result + assert result["bbox"] == IDAHO_BOX + + +def test_coverage_for_empty_excluded_list_returns_dict(): + """With coverage_excluded=[], _coverage_for returns a dict for any adapter that supports coverage.""" + from meshai.env.store import EnvironmentalStore + env_cfg = _make_env_cfg() + store = EnvironmentalStore( + config=env_cfg, + coverage_bbox=IDAHO_BOX, + coverage_excluded=[], + ) + result = store._coverage_for("traffic") + # traffic adapter IS wired to coverage (returns a dict when bbox is set) + assert isinstance(result, dict) + + +def test_coverage_excluded_set_stored_as_set(): + """_coverage_excluded is stored as a set for O(1) lookup.""" + from meshai.env.store import EnvironmentalStore + env_cfg = _make_env_cfg() + store = EnvironmentalStore( + config=env_cfg, + coverage_bbox=IDAHO_BOX, + coverage_excluded=["traffic", "satpass"], + ) + assert isinstance(store._coverage_excluded, set) + assert "traffic" in store._coverage_excluded + assert "satpass" in store._coverage_excluded + + +def test_coverage_excluded_none_defaults_to_empty_set(): + """When coverage_excluded is not passed, _coverage_excluded is an empty set.""" + from meshai.env.store import EnvironmentalStore + env_cfg = _make_env_cfg() + store = EnvironmentalStore(config=env_cfg, coverage_bbox=IDAHO_BOX) + assert store._coverage_excluded == set()