-
- North
- {{ errors.north }} {% endif %}
-
-
- South
- {{ errors.south }} {% endif %}
-
-
- East
- {{ errors.east }} {% endif %}
-
-
- West
- {{ errors.west }} {% endif %}
-
+
-
-
- Save Monitoring Area
+
Drag the rectangle's corners on the map to set the bounds.
+
Create area
+
Cancel
+{% if areas %}
+
+
+ Name Bounds (N, S, E, W)
+ {% for a in areas %}
+
+ {{ a.name }}
+ {{ "%.3f"|format(a.north) }}, {{ "%.3f"|format(a.south) }}, {{ "%.3f"|format(a.east) }}, {{ "%.3f"|format(a.west) }}
+ Edit
+
+
+
+ {% endfor %}
+
+
+
+{% endif %}
+
+
+
+
+
+
{% endblock %}
diff --git a/src/central/monitoring_area.py b/src/central/monitoring_area.py
index c52dc39..723fe9c 100644
--- a/src/central/monitoring_area.py
+++ b/src/central/monitoring_area.py
@@ -80,31 +80,49 @@ def build_geom_json(geo_data: dict[str, Any] | None) -> str | None:
return None
-def classify_geom(geom_json: str | None, area: MonitoringArea | None) -> str:
- """Classify a GeoJSON string against the monitoring area.
+def classify_geom_areas(
+ geom_json: str | None, areas: list[MonitoringArea]
+) -> str:
+ """Classify a GeoJSON string against a list of monitoring areas (set union).
Returns one of:
'null-geom' -- no geometry; always kept (SWPC trio, .removed tombstones)
- 'no-area' -- no monitoring area configured; keep everything
- 'in-bounds' -- geometry intersects the area; keep
- 'out-of-bounds' -- geometry lies entirely outside the area; drop
+ 'no-area' -- empty area list; keep everything (pre-feature default)
+ 'in-bounds' -- geometry intersects AT LEAST ONE area; keep
+ 'out-of-bounds' -- geometry lies entirely outside every area; drop
'invalid-geom' -- geometry could not be evaluated; kept (fail-open) + warn
+ v0.14.0 generalizes the single-bbox ``classify_geom``: an event is kept if it
+ intersects ANY configured area. Areas may overlap. An empty list means "no
+ area configured" and keeps everything, exactly as a NULL single bbox did.
+
Uses ``intersects()`` so border-straddlers and points-on-edge are kept
(matches PostGIS ``ST_Intersects``). The filter must never drop an event
because of a parse failure -- when in doubt, keep it.
"""
if geom_json is None:
return "null-geom"
- if area is None:
+ if not areas:
return "no-area"
try:
geom = shape(json.loads(geom_json))
- return "in-bounds" if geom.intersects(area.as_box()) else "out-of-bounds"
+ for area in areas:
+ if geom.intersects(area.as_box()):
+ return "in-bounds"
+ return "out-of-bounds"
except Exception:
return "invalid-geom"
+def classify_geom(geom_json: str | None, area: MonitoringArea | None) -> str:
+ """Single-area back-compat shim over :func:`classify_geom_areas` (v0.14.0).
+
+ Preserves the pre-v0.14.0 signature and verdicts byte-for-byte: ``None`` area
+ -> ``'no-area'`` (an empty list), a present area -> a one-element list.
+ """
+ return classify_geom_areas(geom_json, [area] if area is not None else [])
+
+
async def load_monitoring_area(conn: asyncpg.Connection) -> MonitoringArea | None:
"""Read the system monitoring area from ``config.system``.
@@ -128,3 +146,26 @@ async def load_monitoring_area(conn: asyncpg.Connection) -> MonitoringArea | Non
west=row["monitor_west"],
)
return None
+
+
+async def load_monitoring_areas(conn: asyncpg.Connection) -> list[MonitoringArea]:
+ """Read the system monitoring areas from ``config.monitoring_areas`` (v0.14.0).
+
+ Returns a list of every configured area, ordered by name for stable logging
+ and rendering. An empty list (no rows) means "no area configured" -> the
+ filter keeps everything (see :func:`classify_geom_areas`).
+
+ Column names (north/south/east/west) match the ``MonitoringArea`` fields so
+ the row maps straight onto the dataclass. Callers own pool acquisition and
+ any exception handling (archive/supervisor keep the last-known value on a
+ read failure and warn).
+ """
+ rows = await conn.fetch(
+ "SELECT north, south, east, west FROM config.monitoring_areas ORDER BY name"
+ )
+ return [
+ MonitoringArea(
+ north=r["north"], south=r["south"], east=r["east"], west=r["west"]
+ )
+ for r in rows
+ ]
diff --git a/src/central/supervisor.py b/src/central/supervisor.py
index 7c56b54..407f0c2 100644
--- a/src/central/supervisor.py
+++ b/src/central/supervisor.py
@@ -35,7 +35,7 @@ from central.monitoring_area import (
MONITORING_AREA_REFRESH_S,
MonitoringArea,
build_geom_json,
- classify_geom,
+ classify_geom_areas,
)
from central.stream_manager import StreamManager
from central.streams import STREAMS as STREAM_REGISTRY
@@ -243,9 +243,21 @@ class Supervisor:
# ACK-but-don't-insert behavior at the supervisor->NATS hop so
# subscribers (meshai, Navi) never see out-of-area events. Refreshed
# every MONITORING_AREA_REFRESH_S from config.system.
- self._monitoring_area: MonitoringArea | None = None
+ self._monitoring_areas: list[MonitoringArea] = []
self._dropped_publish: dict[str, int] = {}
+ @property
+ def _monitoring_area(self) -> MonitoringArea | None:
+ """Back-compat single-area accessor (pre-v0.14.0): the first area, or None.
+
+ v0.14.0 holds a list internally; this property keeps single-area callers
+ and tests working. Setting it wraps the value into a one-element list."""
+ return self._monitoring_areas[0] if self._monitoring_areas else None
+
+ @_monitoring_area.setter
+ def _monitoring_area(self, area: MonitoringArea | None) -> None:
+ self._monitoring_areas = [area] if area is not None else []
+
async def connect(self) -> None:
"""Connect to NATS."""
self._nc = await nats.connect(self._nats_url)
@@ -257,19 +269,22 @@ class Supervisor:
# on failure keep the last value and warn -- never block startup over a
# config read.
try:
- self._monitoring_area = await self._config_store.get_monitoring_area()
+ self._monitoring_areas = await self._config_store.get_monitoring_areas()
except Exception as e:
logger.warning(
- "Could not load monitoring area at startup; publish filter no-ops until refresh",
+ "Could not load monitoring areas at startup; publish filter no-ops until refresh",
extra={"error": str(e)},
)
- area = self._monitoring_area
logger.info(
- "Publish-time monitoring area loaded",
- extra={"monitoring_area": (
- {"north": area.north, "south": area.south,
- "east": area.east, "west": area.west} if area else None
- )},
+ "Publish-time monitoring areas loaded",
+ extra={
+ "monitoring_areas": len(self._monitoring_areas),
+ "bounds": [
+ {"north": a.north, "south": a.south,
+ "east": a.east, "west": a.west}
+ for a in self._monitoring_areas
+ ],
+ },
)
async def disconnect(self) -> None:
@@ -313,7 +328,7 @@ class Supervisor:
)
except asyncio.TimeoutError:
try:
- self._monitoring_area = await self._config_store.get_monitoring_area()
+ self._monitoring_areas = await self._config_store.get_monitoring_areas()
except Exception as e:
logger.warning(
"Could not refresh monitoring area; keeping previous value",
@@ -419,7 +434,7 @@ class Supervisor:
geom_json = build_geom_json(
event.geo.model_dump() if event.geo else None
)
- verdict = classify_geom(geom_json, self._monitoring_area)
+ verdict = classify_geom_areas(geom_json, self._monitoring_areas)
if verdict == "out-of-bounds":
self._dropped_publish[state.name] = (
self._dropped_publish.get(state.name, 0) + 1
diff --git a/tests/test_archive_bbox_filter.py b/tests/test_archive_bbox_filter.py
index be25e68..e4a51ea 100644
--- a/tests/test_archive_bbox_filter.py
+++ b/tests/test_archive_bbox_filter.py
@@ -80,3 +80,40 @@ class TestProcessMessageFilter:
await c._process_message(msg, conn)
conn.execute.assert_awaited_once()
assert c._dropped == {}
+
+
+# NYC metro box -- disjoint from IDAHO, for set-union (v0.14.0) coverage.
+NYC_BOX = MonitoringArea(north=41.0, south=40.3, east=-73.5, west=-74.5)
+
+
+class TestProcessMessageMultiArea:
+ """v0.14.0: archive keeps an event if it intersects ANY configured area."""
+
+ @pytest.mark.asyncio
+ async def test_dropped_when_outside_the_only_configured_area(self):
+ # Boise event, but only the NYC area is configured -> dropped.
+ c = ArchiveConsumer("nats://x", "postgresql://x")
+ c._monitoring_areas = [NYC_BOX]
+ conn = AsyncMock()
+ await c._process_message(_make_msg(_envelope("nws", -114.0, 43.5)), conn)
+ conn.execute.assert_not_called()
+ assert c._dropped == {"nws": 1}
+
+ @pytest.mark.asyncio
+ async def test_kept_when_inside_any_of_several_areas(self):
+ # Same Boise event, but IDAHO is also configured -> kept via union.
+ c = ArchiveConsumer("nats://x", "postgresql://x")
+ c._monitoring_areas = [NYC_BOX, IDAHO]
+ conn = AsyncMock()
+ await c._process_message(_make_msg(_envelope("nws", -114.0, 43.5)), conn)
+ conn.execute.assert_awaited_once()
+ assert c._dropped == {}
+
+ @pytest.mark.asyncio
+ async def test_empty_list_keeps_everything(self):
+ c = ArchiveConsumer("nats://x", "postgresql://x")
+ c._monitoring_areas = []
+ conn = AsyncMock()
+ await c._process_message(_make_msg(_envelope("wzdx", -74.0, 40.7)), conn)
+ conn.execute.assert_awaited_once()
+ assert c._dropped == {}
diff --git a/tests/test_gui_monitoring_area.py b/tests/test_gui_monitoring_area.py
new file mode 100644
index 0000000..d981623
--- /dev/null
+++ b/tests/test_gui_monitoring_area.py
@@ -0,0 +1,146 @@
+"""v0.14.0 monitoring-areas GUI routes: list / create / update / delete.
+
+Server-rendered forms (matching the rest of the GUI), so these call the route
+handlers directly with a mock pool + request and assert status + side effects.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from asyncpg.exceptions import UniqueViolationError
+from fastapi.responses import RedirectResponse
+
+from central.gui.routes import (
+ monitoring_area_create,
+ monitoring_area_delete,
+ monitoring_area_list,
+ monitoring_area_update,
+)
+
+_AREA_ROW = {"id": 1, "name": "treasure_valley",
+ "north": 44.0, "south": 43.0, "east": -115.5, "west": -116.5}
+
+
+class _Tmpl:
+ """Stand-in for Jinja templates -- echoes status + context for assertions."""
+ def TemplateResponse(self, **kw):
+ return SimpleNamespace(
+ status_code=kw.get("status_code", 200), context=kw["context"])
+
+
+def _conn(*, fetch=None, fetchrow=None, execute_error=None):
+ c = MagicMock()
+ c.fetch = AsyncMock(return_value=fetch if fetch is not None else [])
+ c.fetchrow = AsyncMock(return_value=fetchrow)
+ c.execute = AsyncMock(side_effect=execute_error)
+ return c
+
+
+def _pool(conn):
+ pool = MagicMock()
+ cm = MagicMock()
+ cm.__aenter__ = AsyncMock(return_value=conn)
+ cm.__aexit__ = AsyncMock(return_value=False)
+ pool.acquire = MagicMock(return_value=cm)
+ return pool
+
+
+def _req(form=None):
+ r = MagicMock()
+ r.state.csrf_token = "tok"
+ r.state.operator = SimpleNamespace(id=1, username="admin")
+
+ async def _form():
+ return form or {}
+ r.form = _form
+ return r
+
+
+def _form(**over):
+ base = {"csrf_token": "tok", "name": "magic_valley",
+ "north": "43.0", "south": "42.3", "east": "-113.4", "west": "-114.9"}
+ base.update(over)
+ return base
+
+
+def _patches(conn):
+ return (
+ patch("central.gui.routes.get_pool", return_value=_pool(conn)),
+ patch("central.gui.routes._get_templates", return_value=_Tmpl()),
+ patch("central.gui.routes.write_audit", new=AsyncMock()),
+ )
+
+
+@pytest.mark.asyncio
+class TestList:
+ async def test_renders_areas(self):
+ conn = _conn(fetch=[_AREA_ROW], fetchrow=None)
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_list(_req())
+ assert res.status_code == 200
+ assert res.context["areas"] == [_AREA_ROW]
+
+
+@pytest.mark.asyncio
+class TestCreate:
+ async def test_valid_redirects_and_inserts(self):
+ conn = _conn()
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_create(_req(_form()))
+ assert isinstance(res, RedirectResponse) and res.status_code == 302
+ assert "INSERT INTO config.monitoring_areas" in conn.execute.call_args[0][0]
+
+ async def test_invalid_name_rerenders_no_insert(self):
+ conn = _conn()
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_create(_req(_form(name="")))
+ assert res.status_code == 200 and res.context["error"]
+ conn.execute.assert_not_called()
+
+ async def test_inverted_bounds_rerenders(self):
+ conn = _conn()
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_create(_req(_form(north="42.0", south="43.0")))
+ assert res.status_code == 200 and res.context["error"]
+ conn.execute.assert_not_called()
+
+ async def test_duplicate_name_rerenders(self):
+ conn = _conn(execute_error=UniqueViolationError("dup"))
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_create(_req(_form()))
+ assert res.status_code == 200 and "already exists" in res.context["error"]
+
+
+@pytest.mark.asyncio
+class TestUpdate:
+ async def test_valid_redirects_and_updates(self):
+ conn = _conn(fetchrow=_AREA_ROW)
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_update(_req(_form()), 1)
+ assert isinstance(res, RedirectResponse) and res.status_code == 302
+ assert "UPDATE config.monitoring_areas" in conn.execute.call_args[0][0]
+
+ async def test_missing_id_returns_404(self):
+ conn = _conn(fetchrow=None)
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_update(_req(_form()), 999)
+ assert res.status_code == 404
+
+
+@pytest.mark.asyncio
+class TestDelete:
+ async def test_redirects_and_deletes(self):
+ conn = _conn(fetchrow=_AREA_ROW)
+ p1, p2, p3 = _patches(conn)
+ with p1, p2, p3:
+ res = await monitoring_area_delete(_req({"csrf_token": "tok"}), 1)
+ assert isinstance(res, RedirectResponse) and res.status_code == 302
+ assert "DELETE FROM config.monitoring_areas" in conn.execute.call_args[0][0]
diff --git a/tests/test_migration_042.py b/tests/test_migration_042.py
new file mode 100644
index 0000000..3a1902f
--- /dev/null
+++ b/tests/test_migration_042.py
@@ -0,0 +1,40 @@
+"""v0.14.0 migration 042: single config.system bbox -> config.monitoring_areas.
+
+The suite has no live Postgres (it runs identically as zvx or central), so this
+asserts the migration's shape statically: it creates the table, enforces the
+union-filter invariants, seeds 'default' from the existing config.system bbox to
+preserve current bounds, and -- deliberately for v0.14.0 -- does NOT drop the old
+columns (that lands in v0.14.1).
+"""
+
+from pathlib import Path
+
+_SQL = Path("sql/migrations/042_monitoring_area_to_multi_areas.sql").read_text()
+_NORM = " ".join(_SQL.split()) # whitespace-insensitive matching
+
+
+def test_creates_monitoring_areas_table():
+ assert "CREATE TABLE IF NOT EXISTS config.monitoring_areas" in _NORM
+
+
+def test_name_is_unique():
+ assert "name TEXT NOT NULL UNIQUE" in _NORM
+
+
+def test_check_constraints_enforce_bbox_ordering():
+ assert "CHECK (north > south)" in _NORM
+ assert "CHECK (east > west)" in _NORM
+
+
+def test_seeds_default_from_config_system_preserving_bounds():
+ assert "INSERT INTO config.monitoring_areas" in _NORM
+ assert "FROM config.system" in _NORM
+ assert "'default'" in _NORM
+ # Idempotent re-run / already-seeded installs must not error or duplicate.
+ assert "ON CONFLICT (name) DO NOTHING" in _NORM
+
+
+def test_does_not_drop_old_columns_in_v0_14_0():
+ upper = _NORM.upper()
+ assert "DROP COLUMN" not in upper
+ assert "DROP TABLE" not in upper
diff --git a/tests/test_monitoring_area.py b/tests/test_monitoring_area.py
index abbfae8..39f8e14 100644
--- a/tests/test_monitoring_area.py
+++ b/tests/test_monitoring_area.py
@@ -20,10 +20,14 @@ from central.monitoring_area import (
MonitoringArea,
build_geom_json,
classify_geom,
+ classify_geom_areas,
load_monitoring_area,
+ load_monitoring_areas,
)
IDAHO = MonitoringArea(north=44.5, south=41.8, east=-111.0, west=-117.5)
+# A second, disjoint area to exercise set-union semantics (the NYC metro box).
+NYC_BOX = MonitoringArea(north=41.0, south=40.3, east=-73.5, west=-74.5)
def _pt(lon, lat):
@@ -117,6 +121,55 @@ class TestClassifyGeom:
assert classify_geom(json.dumps({"type": "Nonsense"}), IDAHO) == "invalid-geom"
+class TestClassifyGeomAreas:
+ """v0.14.0 set-union: keep if the geometry intersects ANY area."""
+
+ def test_empty_list_keeps_everything(self):
+ assert classify_geom_areas(_pt(-74.0, 40.7), []) == "no-area"
+
+ def test_null_geom_always_kept(self):
+ assert classify_geom_areas(None, [IDAHO]) == "null-geom"
+ assert classify_geom_areas(None, []) == "null-geom"
+
+ def test_single_area_matches_single_bbox_behavior(self):
+ # Same verdicts as the legacy single-area classify_geom for a 1-list.
+ assert classify_geom_areas(_pt(-114.0, 43.5), [IDAHO]) == "in-bounds"
+ assert classify_geom_areas(_pt(-74.0, 40.7), [IDAHO]) == "out-of-bounds"
+
+ def test_kept_if_in_either_area(self):
+ # Boise -> IDAHO; NYC -> NYC_BOX. Union keeps both.
+ assert classify_geom_areas(_pt(-114.0, 43.5), [IDAHO, NYC_BOX]) == "in-bounds"
+ assert classify_geom_areas(_pt(-74.0, 40.7), [IDAHO, NYC_BOX]) == "in-bounds"
+
+ def test_dropped_only_when_outside_every_area(self):
+ # London is in neither box.
+ assert classify_geom_areas(_pt(-0.13, 51.5), [IDAHO, NYC_BOX]) == "out-of-bounds"
+
+ def test_invalid_geom_fails_open(self):
+ assert classify_geom_areas("{bad json", [IDAHO]) == "invalid-geom"
+
+ def test_overlapping_areas_still_in_bounds(self):
+ overlap = MonitoringArea(north=44.0, south=42.0, east=-112.0, west=-118.0)
+ assert classify_geom_areas(_pt(-114.0, 43.0), [IDAHO, overlap]) == "in-bounds"
+
+
+@pytest.mark.asyncio
+class TestLoadMonitoringAreas:
+ async def test_returns_all_rows_as_areas(self):
+ conn = MagicMock()
+ conn.fetch = AsyncMock(return_value=[
+ {"north": 44.5, "south": 41.8, "east": -111.0, "west": -117.5},
+ {"north": 41.0, "south": 40.3, "east": -73.5, "west": -74.5},
+ ])
+ areas = await load_monitoring_areas(conn)
+ assert areas == [IDAHO, NYC_BOX]
+
+ async def test_empty_table_returns_empty_list(self):
+ conn = MagicMock()
+ conn.fetch = AsyncMock(return_value=[])
+ assert await load_monitoring_areas(conn) == []
+
+
@pytest.mark.asyncio
class TestLoadMonitoringArea:
async def test_returns_area_when_all_columns_set(self):
diff --git a/tests/test_supervisor_publish_filter.py b/tests/test_supervisor_publish_filter.py
index bb58145..0a936ac 100644
--- a/tests/test_supervisor_publish_filter.py
+++ b/tests/test_supervisor_publish_filter.py
@@ -71,6 +71,9 @@ def sup_factory():
store.set_adapter_last_error = AsyncMock()
store.get_api_key = AsyncMock(return_value=None)
store.get_monitoring_area = AsyncMock(return_value=area)
+ store.get_monitoring_areas = AsyncMock(
+ return_value=[area] if area is not None else []
+ )
config_source = MagicMock()
config_source.get_enrichment_config = AsyncMock(return_value=EnrichmentConfig())
sup = sup_mod.Supervisor(
@@ -173,7 +176,7 @@ async def test_refresh_loop_reloads_area_and_logs_summary(
sup = sup_factory(None)
sup._dropped_publish = {"mock": 7}
monkeypatch.setattr(sup_mod, "MONITORING_AREA_REFRESH_S", 0.05)
- sup._config_store.get_monitoring_area = AsyncMock(return_value=IDAHO)
+ sup._config_store.get_monitoring_areas = AsyncMock(return_value=[IDAHO])
with caplog.at_level(logging.INFO):
task = asyncio.create_task(sup._refresh_monitoring_area_loop())
await asyncio.sleep(0.15)