From 9b6053365abf5174c6512a8231e67313319d4465 Mon Sep 17 00:00:00 2001 From: malice Date: Tue, 7 Jul 2026 21:43:49 -0600 Subject: [PATCH] feat(region-routing): tag satpass events by observer coordinates so they region-route; add satpass to VALID_TOGGLES (#90) Co-authored-by: Matt Johnson --- work/meshai/coverage_area.py | 48 ++++- work/meshai/notifications/categories.py | 1 + work/tests/test_satpass_region_tagging.py | 247 ++++++++++++++++++++++ 3 files changed, 293 insertions(+), 3 deletions(-) create mode 100644 work/tests/test_satpass_region_tagging.py diff --git a/work/meshai/coverage_area.py b/work/meshai/coverage_area.py index fe9b316..d94b51a 100644 --- a/work/meshai/coverage_area.py +++ b/work/meshai/coverage_area.py @@ -266,14 +266,56 @@ def matching_area_names(geom_json: str | None, areas: list[MonitoringArea]) -> l return result +def observer_region_names(event: Any, areas: list[MonitoringArea]) -> list[str]: + """Region names for an event tagged by its satpass observer coordinates. + + For each observer slug in event.data['observer_list'], look up its lat/lon + via get_observers() and collect every NAMED area whose bbox contains it. + Union, config order, deduped. Returns [] when there's no observer_list, no + resolvable coords, or no observer falls in any named area. Fail-open on any + error (return []). + """ + data = getattr(event, "data", None) or {} + ol = data.get("observer_list") + if not ol or not areas: + return [] + slugs = [s.strip() for s in str(ol).split(",") if s.strip()] + if not slugs: + return [] + try: + from meshai.persistence.observer_locations import get_observers + coords = {o["slug"]: (o["lat"], o["lon"]) for o in get_observers()} + except Exception: + return [] + out: list[str] = [] + for s in slugs: + c = coords.get(s) + if not c: + continue + try: + lat, lon = float(c[0]), float(c[1]) + except (TypeError, ValueError): + continue + for a in areas: + if a.name and a.name not in out and a.south <= lat <= a.north and a.west <= lon <= a.east: + out.append(a.name) + return out + + def event_region_names(event: Any, areas: list[MonitoringArea]) -> list[str]: """Return region names for an event by matching its geometry against named areas. - Convenience wrapper: extracts geometry via _event_geom_json (same chain as - the coverage gate) and delegates to matching_area_names. + Primary path: extracts geometry via _event_geom_json (same chain as the + coverage gate) and delegates to matching_area_names. + Fallback path: when geometry yields no names (e.g. satpass events have no + polygon), tries observer_region_names() which maps satpass observer + coordinates to named coverage areas. Returns [] when no geometry, no named areas, or parse error. """ - return matching_area_names(_event_geom_json(event), areas) + names = matching_area_names(_event_geom_json(event), areas) + if not names: + names = observer_region_names(event, areas) + return names def event_in_areas(event: Any, areas: list[MonitoringArea]) -> bool: diff --git a/work/meshai/notifications/categories.py b/work/meshai/notifications/categories.py index 8566e6a..c7d9cd9 100644 --- a/work/meshai/notifications/categories.py +++ b/work/meshai/notifications/categories.py @@ -32,6 +32,7 @@ VALID_TOGGLES = frozenset({ "avalanche", "seismic", "tracking", + "satpass", }) diff --git a/work/tests/test_satpass_region_tagging.py b/work/tests/test_satpass_region_tagging.py new file mode 100644 index 0000000..db48d73 --- /dev/null +++ b/work/tests/test_satpass_region_tagging.py @@ -0,0 +1,247 @@ +"""Tests for satpass observer-based region tagging (coverage_area.py). + +Covers: +- observer_region_names: single observer in a region, two observers in different + regions (union, config order, deduped), slug not found / outside all areas. +- event_region_names: geometry path takes priority; falls back to observer path + only when geometry yields nothing. +- categories: satpass is now a first-class VALID_TOGGLE; categories_for_toggle + returns ["sat_pass"]. +""" +from __future__ import annotations + +from unittest.mock import patch + +import pytest + +from meshai.coverage_area import ( + MonitoringArea, + event_region_names, + observer_region_names, +) +from meshai.notifications.categories import ( + VALID_TOGGLES, + all_toggles, + categories_for_toggle, +) +from meshai.notifications.events import make_event + + +# --------------------------------------------------------------------------- +# Reference areas — named bboxes +# --------------------------------------------------------------------------- + +SC_IDAHO = MonitoringArea(north=43.0, south=42.0, east=-113.5, west=-115.5, name="SC Idaho") +MAGIC_VALLEY = MonitoringArea(north=43.2, south=42.3, east=-114.0, west=-115.0, name="Magic Valley") +# Unnamed area — must never appear in results +UNNAMED = MonitoringArea(north=44.0, south=43.0, east=-113.0, west=-116.0, name=None) + +AREAS = [SC_IDAHO, MAGIC_VALLEY, UNNAMED] + +# Coordinates inside SC_IDAHO (Twin Falls ~42.56N, -114.47W) +TF_LAT, TF_LON = 42.56, -114.47 + +# Coordinates inside MAGIC_VALLEY but NOT SC_IDAHO +MV_LAT, MV_LON = 42.5, -114.5 # falls in both SC_IDAHO and MAGIC_VALLEY per overlap +# Use a point that is inside MAGIC_VALLEY bbox only, outside SC_IDAHO: +# SC_IDAHO west=-115.5 east=-113.5 magic-valley west=-115.0 east=-114.0 +# A point at lon=-114.8 (between -115.0 and -114.0) and lat=42.7 (inside both bboxes) +# Let's use distinct non-overlapping coords: +# SC_IDAHO 42.0–43.0 N, -115.5–-113.5 W +# MAGIC_VALLEY 42.3–43.2 N, -115.0–-114.0 W +# These boxes overlap. Use a point in SC_IDAHO only (lat=42.1, lon=-115.2): +SC_ONLY_LAT, SC_ONLY_LON = 42.1, -115.2 # south of MAGIC_VALLEY (42.3 min) +# And a point in MAGIC_VALLEY only (lat=43.1, lon=-114.3 — north of SC_IDAHO 43.0 max): +MV_ONLY_LAT, MV_ONLY_LON = 43.1, -114.3 + +# Outside everything +OUT_LAT, OUT_LON = 34.0, -118.0 # Los Angeles + + +def _fake_observers(entries): + """Return a patcher for get_observers that yields `entries`.""" + return patch( + "meshai.coverage_area.observer_region_names.__code__", # won't work — use module path + ) + + +def _patch_observers(entries): + return patch( + "meshai.persistence.observer_locations.get_observers", + return_value=entries, + ) + + +def _make_satpass_event(observer_list_str, **kwargs): + """Build a minimal satpass Event-like object with no geometry.""" + return make_event( + source="satpass", + category="sat_pass", + severity="routine", + title="ISS Pass", + data={"observer_list": observer_list_str}, + **kwargs, + ) + + +# --------------------------------------------------------------------------- +# observer_region_names — core unit tests +# --------------------------------------------------------------------------- + +class TestObserverRegionNames: + def test_single_observer_in_one_region(self): + event = _make_satpass_event("twin-falls") + with _patch_observers([{"slug": "twin-falls", "lat": SC_ONLY_LAT, "lon": SC_ONLY_LON}]): + result = observer_region_names(event, AREAS) + assert result == ["SC Idaho"] + + def test_two_observers_in_different_regions_union_config_order(self): + """Two observers in SC Idaho and Magic Valley (non-overlapping points) + → both region names returned in config order (SC Idaho first, then MV).""" + event = _make_satpass_event("obs-a,obs-b") + fake_obs = [ + {"slug": "obs-a", "lat": SC_ONLY_LAT, "lon": SC_ONLY_LON}, # SC Idaho only + {"slug": "obs-b", "lat": MV_ONLY_LAT, "lon": MV_ONLY_LON}, # Magic Valley only + ] + with _patch_observers(fake_obs): + result = observer_region_names(event, AREAS) + assert result == ["SC Idaho", "Magic Valley"] + + def test_deduplication_same_region_multiple_observers(self): + """Two observers both inside SC Idaho → region appears only once.""" + event = _make_satpass_event("obs-1,obs-2") + fake_obs = [ + {"slug": "obs-1", "lat": SC_ONLY_LAT, "lon": SC_ONLY_LON}, + {"slug": "obs-2", "lat": 42.2, "lon": -115.0}, # also inside SC Idaho + ] + with _patch_observers(fake_obs): + result = observer_region_names(event, AREAS) + assert result.count("SC Idaho") == 1 + + def test_unknown_slug_returns_empty(self): + """Observer slug not in get_observers → [].""" + event = _make_satpass_event("nonexistent-slug") + with _patch_observers([{"slug": "other", "lat": TF_LAT, "lon": TF_LON}]): + result = observer_region_names(event, AREAS) + assert result == [] + + def test_observer_outside_all_areas_returns_empty(self): + """Observer coords outside all configured areas → [].""" + event = _make_satpass_event("la-observer") + with _patch_observers([{"slug": "la-observer", "lat": OUT_LAT, "lon": OUT_LON}]): + result = observer_region_names(event, AREAS) + assert result == [] + + def test_unnamed_area_never_in_result(self): + """Unnamed MonitoringArea (name=None) never contributes to output.""" + event = _make_satpass_event("obs") + # Point inside UNNAMED bbox (lat=43.5, lon=-114.5) — outside SC Idaho / MV: + fake_obs = [{"slug": "obs", "lat": 43.5, "lon": -114.5}] + with _patch_observers(fake_obs): + result = observer_region_names(event, AREAS) + assert result == [] + + def test_no_observer_list_returns_empty(self): + """Event without observer_list key → [].""" + event = make_event( + source="satpass", category="sat_pass", severity="routine", + title="Pass", data={}, + ) + with _patch_observers([]): + result = observer_region_names(event, AREAS) + assert result == [] + + def test_empty_areas_returns_empty(self): + """No configured areas → [].""" + event = _make_satpass_event("obs") + with _patch_observers([{"slug": "obs", "lat": TF_LAT, "lon": TF_LON}]): + result = observer_region_names(event, []) + assert result == [] + + def test_get_observers_exception_returns_empty_failopen(self): + """If get_observers raises, observer_region_names returns [] (fail-open).""" + event = _make_satpass_event("obs") + with patch( + "meshai.persistence.observer_locations.get_observers", + side_effect=RuntimeError("db gone"), + ): + result = observer_region_names(event, AREAS) + assert result == [] + + +# --------------------------------------------------------------------------- +# event_region_names — geometry takes priority; observer fallback for satpass +# --------------------------------------------------------------------------- + +class TestEventRegionNames: + def test_event_with_geometry_uses_geometry_not_observers(self): + """An event with a polygon in SC Idaho must tag via geometry; the + observer fallback must NOT be consulted (and if it were, it would + return a different/no region because the observer slug is absent).""" + geom = { + "type": "Point", + "coordinates": [SC_ONLY_LON, SC_ONLY_LAT], # [lon, lat] + } + event = make_event( + source="satpass", category="sat_pass", severity="routine", + title="Pass", + lat=SC_ONLY_LAT, lon=SC_ONLY_LON, + data={"observer_list": "no-such-slug"}, + ) + # With a lat/lon centroid, _event_geom_json will produce a Point. + # Even though the observer slug won't resolve, the geometry path fires first. + # SC_ONLY_LAT=42.1, SC_ONLY_LON=-115.2 is inside SC_IDAHO bbox. + result = event_region_names(event, AREAS) + assert "SC Idaho" in result + + def test_satpass_event_no_geometry_falls_back_to_observers(self): + """Satpass event with NO lat/lon and NO geometry data → geometry path + yields [], observer fallback kicks in and returns correct region.""" + event = make_event( + source="satpass", category="sat_pass", severity="routine", + title="ISS Pass", + data={"observer_list": "boise-obs"}, + # deliberately no lat/lon + ) + assert event.lat is None + assert event.lon is None + fake_obs = [{"slug": "boise-obs", "lat": SC_ONLY_LAT, "lon": SC_ONLY_LON}] + with _patch_observers(fake_obs): + result = event_region_names(event, AREAS) + assert result == ["SC Idaho"] + + def test_event_with_geometry_in_region_uses_geometry_observers_not_consulted(self): + """When geometry tags a region, the observer path must not run — even if + observers would tag a different region.""" + # Event has a geometry that puts it in SC Idaho + event = make_event( + source="satpass", category="sat_pass", severity="routine", + title="Pass", + lat=SC_ONLY_LAT, lon=SC_ONLY_LON, + data={"observer_list": "mv-obs"}, + ) + # Observer is in Magic Valley (MV_ONLY) — if observer path ran it would + # return Magic Valley. But geometry path should win: SC Idaho. + fake_obs = [{"slug": "mv-obs", "lat": MV_ONLY_LAT, "lon": MV_ONLY_LON}] + with _patch_observers(fake_obs): + result = event_region_names(event, AREAS) + # Geometry (lat/lon centroid at SC_ONLY) fires first → SC Idaho + assert "SC Idaho" in result + # Magic Valley should NOT be in result (observer path not reached) + assert "Magic Valley" not in result + + +# --------------------------------------------------------------------------- +# categories.py — satpass is a first-class VALID_TOGGLE +# --------------------------------------------------------------------------- + +class TestSatpassToggle: + def test_satpass_in_valid_toggles(self): + assert "satpass" in VALID_TOGGLES + + def test_satpass_in_all_toggles(self): + assert "satpass" in all_toggles() + + def test_categories_for_toggle_satpass_returns_sat_pass(self): + result = categories_for_toggle("satpass") + assert result == ["sat_pass"]