feat(coverage): per-adapter override — exclude adapters from the coverage bbox (#64)

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 <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-06 10:11:33 -06:00 committed by GitHub
commit a3e5a7518f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 131 additions and 2 deletions

View file

@ -807,6 +807,7 @@ class Coverage:
bbox: list = field(default_factory=list) # [west, south, east, north]; empty = not configured bbox: list = field(default_factory=list) # [west, south, east, north]; empty = not configured
enabled: bool = True # master switch for deriving adapter scope from bbox 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 @dataclass

View file

@ -39,6 +39,7 @@ class EnvironmentalStore:
region_anchors: list = None, region_anchors: list = None,
event_bus: Optional["EventBus"] = None, event_bus: Optional["EventBus"] = None,
coverage_bbox: list = None, coverage_bbox: list = None,
coverage_excluded: list = None,
): ):
self._adapters = {} # name -> adapter instance self._adapters = {} # name -> adapter instance
self._failed_adapters = {} # name -> last_error string self._failed_adapters = {} # name -> last_error string
@ -49,6 +50,7 @@ class EnvironmentalStore:
self._mesh_zones = config.nws_zones or [] self._mesh_zones = config.nws_zones or []
self._region_anchors = region_anchors or [] self._region_anchors = region_anchors or []
self._coverage_bbox = coverage_bbox or [] self._coverage_bbox = coverage_bbox or []
self._coverage_excluded = set(coverage_excluded or [])
# ── Received-delta gate (NATIVE-only) ──────────────────────────── # ── Received-delta gate (NATIVE-only) ────────────────────────────
# The model the operator demanded: a native adapter broadcasts an item # The model the operator demanded: a native adapter broadcasts an item
@ -148,7 +150,14 @@ class EnvironmentalStore:
self._failed_adapters[name] = err_msg self._failed_adapters[name] = err_msg
def _coverage_for(self, adapter: str): 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 from meshai import coverage as _cov
return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native") return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native")

View file

@ -402,13 +402,16 @@ class MeshAI:
# Native satpass: seed observer_locations. When a valid coverage bbox is # Native satpass: seed observer_locations. When a valid coverage bbox is
# configured, seed a single derived observer at the bbox centroid instead # configured, seed a single derived observer at the bbox centroid instead
# of the hand-listed config observers (coverage wins; config is fallback). # 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: try:
from meshai.persistence.observer_locations import seed_observers_from_config from meshai.persistence.observer_locations import seed_observers_from_config
from meshai import coverage as _cov from meshai import coverage as _cov
from types import SimpleNamespace from types import SimpleNamespace
_satpass_excluded = "satpass" in self.config.coverage.excluded_adapters
_sat_scope = _cov.resolve_adapter_coverage( _sat_scope = _cov.resolve_adapter_coverage(
"satpass", "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", "native",
) )
if _sat_scope is not None: if _sat_scope is not None:
@ -637,6 +640,7 @@ class MeshAI:
self.env_store = EnvironmentalStore( self.env_store = EnvironmentalStore(
config=env_cfg, region_anchors=region_anchors, config=env_cfg, region_anchors=region_anchors,
coverage_bbox=coverage_bbox, event_bus=self.event_bus, 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)") logger.info(f"Environmental feeds enabled ({len(self.env_store._adapters)} adapters)")
else: else:

View file

@ -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()