5ea9a4e879
Phase 1: in-memory ConnectionManager viewport fan-out, 500ms map debounce, cachetools TTLCache, background masscan/ffmpeg, compose memory caps. Phase 2: PostGIS geofences + ST_Intersects alerts, Timescale 1-min CAGGs and timestamp playback, FIRMS/WFIGS x firefighting ADS-B within 20 miles. No Redis/Kafka/Celery.
136 lines
3.7 KiB
Python
136 lines
3.7 KiB
Python
"""Geofence hit detection and WS alert routing (no Redis)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import geofence
|
|
from geofence import (
|
|
matching_geofences,
|
|
point_in_geojson,
|
|
validate_polygon_geojson,
|
|
)
|
|
from ws_manager import ConnectionManager
|
|
|
|
|
|
NC_BOX = {
|
|
"type": "Polygon",
|
|
"coordinates": [[
|
|
[-80.0, 35.0],
|
|
[-78.0, 35.0],
|
|
[-78.0, 36.0],
|
|
[-80.0, 36.0],
|
|
[-80.0, 35.0],
|
|
]],
|
|
}
|
|
|
|
|
|
def test_point_inside_polygon_hits():
|
|
assert point_in_geojson(-79.0, 35.5, NC_BOX) is True
|
|
|
|
|
|
def test_point_outside_polygon_misses():
|
|
assert point_in_geojson(-122.4, 37.7, NC_BOX) is False
|
|
|
|
|
|
def test_validate_rejects_non_polygon():
|
|
try:
|
|
validate_polygon_geojson({"type": "Point", "coordinates": [-79.0, 35.5]})
|
|
assert False, "expected ValueError"
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
def test_matching_geofences_only_active_hits():
|
|
fences = [
|
|
{"id": "a", "name": "NC", "active": True, "geojson": NC_BOX},
|
|
{"id": "b", "name": "off", "active": False, "geojson": NC_BOX},
|
|
]
|
|
hits = matching_geofences(-79.0, 35.5, fences)
|
|
assert [h["id"] for h in hits] == ["a"]
|
|
assert matching_geofences(-122.4, 37.7, fences) == []
|
|
|
|
|
|
def test_geofence_alert_fans_out_only_to_viewport_clients():
|
|
mgr = ConnectionManager()
|
|
q_nc = mgr.register("nc")
|
|
q_sf = mgr.register("sf")
|
|
mgr.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
|
|
mgr.set_viewport("sf", (-123.0, 37.0, -121.0, 38.0))
|
|
|
|
async def run():
|
|
payload = {
|
|
"geofence_id": "a",
|
|
"geofence_name": "NC",
|
|
"source_kind": "ais",
|
|
"entity_id": "366123456",
|
|
"lat": 35.5,
|
|
"lon": -79.0,
|
|
}
|
|
n = await mgr.publish_point("geofence_alert", payload, lat=35.5, lon=-79.0)
|
|
assert n == 1
|
|
msg = q_nc.get_nowait()
|
|
assert msg["type"] == "geofence_alert"
|
|
assert msg["payload"]["entity_id"] == "366123456"
|
|
assert q_sf.empty()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
|
|
"""FIRMS ingest in the ingester has an empty in-process cache — still ST_Intersects."""
|
|
import geofence
|
|
|
|
geofence._cache.clear()
|
|
geofence._recent_hits.clear()
|
|
st_called = []
|
|
|
|
async def fake_st(lon, lat):
|
|
st_called.append((lon, lat))
|
|
return [{
|
|
"id": "11111111-1111-1111-1111-111111111111",
|
|
"name": "NC",
|
|
"geojson": NC_BOX,
|
|
"active": True,
|
|
}]
|
|
|
|
monkeypatch.setattr(geofence, "st_intersects", fake_st)
|
|
|
|
executed: list = []
|
|
|
|
class FakeSession:
|
|
async def execute(self, stmt, params=None):
|
|
executed.append(params or {})
|
|
return None
|
|
|
|
async def commit(self):
|
|
executed.append("commit")
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
monkeypatch.setattr(geofence, "async_session", FakeSession)
|
|
|
|
async def run():
|
|
from ws_manager import manager
|
|
q = manager.register("nc")
|
|
manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
|
|
n = await geofence.record_and_notify(
|
|
source_kind="firms", entity_id="35.5,-79.0,N",
|
|
lat=35.5, lon=-79.0, payload={"satellite": "N"},
|
|
)
|
|
msg = None if q.empty() else q.get_nowait()
|
|
manager.unregister("nc")
|
|
return n, msg
|
|
|
|
n, msg = asyncio.run(run())
|
|
assert st_called == [(-79.0, 35.5)]
|
|
assert n == 1
|
|
assert msg["type"] == "geofence_alert"
|
|
assert msg["payload"]["source_kind"] == "firms"
|
|
inserts = [p for p in executed if isinstance(p, dict)]
|
|
assert inserts and inserts[0]["source_kind"] == "firms"
|
|
assert "commit" in executed
|