feat(geofence): watch list, filtered hit log, fence snapshot #47

Merged
sirius merged 1 commit from feat/geofence-watch into master 2026-09-01 01:26:31 -04:00
6 changed files with 625 additions and 37 deletions

View file

@ -0,0 +1,24 @@
"""geofence_alerts (geofence_id, created_at DESC) for fence-scoped hit log
Revision ID: 011_geofence_alerts_fence
Revises: 010_bbox_gist
Create Date: 2026-09-01
"""
from alembic import op
revision = "011_geofence_alerts_fence"
down_revision = "010_bbox_gist"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.execute(
"CREATE INDEX IF NOT EXISTS ix_geofence_alerts_fence_created "
"ON geofence_alerts (geofence_id, created_at DESC)"
)
def downgrade() -> None:
op.execute("DROP INDEX IF EXISTS ix_geofence_alerts_fence_created")

View file

@ -319,3 +319,154 @@ async def record_and_notify(
except Exception:
pass
return sent
async def list_alerts(
*,
geofence_id: str | None = None,
since: datetime | None = None,
until: datetime | None = None,
source_kind: str | None = None,
limit: int = 100,
) -> list[dict]:
"""Filterable hit log. Empty list if the DB is down — never raises."""
where = ["TRUE"]
params: dict[str, Any] = {"limit": int(limit)}
if geofence_id:
where.append("geofence_id = CAST(:geofence_id AS uuid)")
params["geofence_id"] = geofence_id
if since is not None:
where.append("created_at >= :since")
params["since"] = since
if until is not None:
where.append("created_at <= :until")
params["until"] = until
if source_kind:
where.append("source_kind = :source_kind")
params["source_kind"] = source_kind
sql = f"""
SELECT id::text, geofence_id::text, source_kind, entity_id,
lat, lon, payload, created_at
FROM geofence_alerts
WHERE {' AND '.join(where)}
ORDER BY created_at DESC
LIMIT :limit
"""
try:
async with async_session() as session:
rows = (await session.execute(text(sql), params)).mappings().all()
out = []
for r in rows:
item = dict(r)
if item.get("created_at") is not None:
item["created_at"] = item["created_at"].isoformat()
out.append(item)
return out
except Exception:
return []
async def get_geofence(gid: str) -> dict | None:
current = next((f for f in _cache if f["id"] == gid), None)
if current is not None:
return current
try:
await refresh_cache()
except Exception:
return None
return next((f for f in _cache if f["id"] == gid), None)
def _marker_from_track(row) -> dict:
from live_layers import to_marker
extra = {"bucket": row["bucket"].isoformat() if row.get("bucket") else None, "dvr": True}
return to_marker(
row["id"], row["lat"], row["lon"],
heading=row.get("heading"), speed=row.get("speed"),
label=row.get("label") or row["id"],
extra=extra,
)
async def _cagg_inside(gid: str, kind: str, bucket: datetime, limit: int = 2000) -> list[dict]:
table = "aircraft_tracks_1min" if kind == "aircraft" else "vessel_tracks_1min"
id_col = "hex" if kind == "aircraft" else "mmsi"
sql = f"""
SELECT {id_col} AS id, lat, lon, heading, speed, label, bucket
FROM {table}
WHERE bucket = :bucket
AND ST_Intersects(
(SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
ST_SetSRID(ST_MakePoint(lon, lat), 4326)
)
LIMIT :limit
"""
try:
async with async_session() as session:
rows = (await session.execute(
text(sql), {"bucket": bucket, "gid": gid, "limit": limit},
)).mappings().all()
return [
_marker_from_track(r)
for r in rows
if r["lat"] is not None and r["lon"] is not None
]
except Exception:
return []
async def _fires_inside(gid: str, ts: datetime, limit: int = 2000) -> list[dict]:
from tracks import minute_bucket
bucket = minute_bucket(ts)
t1 = bucket + timedelta(minutes=1)
sql = """
SELECT latitude, longitude, brightness, confidence, acq_time, satellite,
instrument, bright_ti5, frp, daynight
FROM fires
WHERE acq_time >= :t0 AND acq_time < :t1
AND ST_Intersects(
(SELECT geom FROM geofences WHERE id = CAST(:gid AS uuid)),
ST_SetSRID(ST_MakePoint(longitude, latitude), 4326)
)
LIMIT :limit
"""
try:
async with async_session() as session:
rows = (await session.execute(
text(sql),
{"t0": bucket, "t1": t1, "gid": gid, "limit": limit},
)).mappings().all()
out = []
for r in rows:
item = dict(r)
if item.get("acq_time") is not None:
item["acq_time"] = item["acq_time"].isoformat()
out.append(item)
return out
except Exception:
return []
async def snapshot_at(gid: str, ts: datetime) -> dict | None:
"""Positions inside the fence at time T. None if the fence is missing.
Does not persist or notify. Empty lists if track/fire queries fail.
"""
fence = await get_geofence(gid)
if fence is None:
return None
from tracks import minute_bucket
bucket = minute_bucket(ts)
aircraft = await _cagg_inside(gid, "aircraft", bucket)
vessels = await _cagg_inside(gid, "vessel", bucket)
fires = await _fires_inside(gid, ts)
return {
"geofence_id": gid,
"timestamp": ts.isoformat(),
"aircraft": aircraft,
"vessels": vessels,
"fires": fires,
}

View file

@ -873,7 +873,14 @@ async def trigger_social_ingest(query: str = "", max_items: int = 50):
@app.websocket("/ws/live")
async def live_ws(ws: WebSocket):
"""Viewport-filtered AIS/ADS-B fan-out. Client sends {type:viewport,bbox}."""
"""Viewport-filtered AIS/ADS-B fan-out.
Client JSON:
{"type":"viewport","bbox":"minlon,minlat,maxlon,maxlat"}
{"type":"watch_geofences","ids":["<uuid>", ...]} empty list = none
geofence_alert delivers if the point is in-viewport OR geofence_id is watched.
AIS/ADS-B/fire_aircraft stay viewport-only.
"""
from ws_manager import manager
client_id = str(id(ws))
@ -899,6 +906,10 @@ async def live_ws(ws: WebSocket):
manager.set_viewport(client_id, parse_bbox(str(data["bbox"])))
except ValueError:
continue
elif data.get("type") == "watch_geofences":
ids = data.get("ids") or []
if isinstance(ids, list):
manager.set_watched_geofences(client_id, [str(x) for x in ids])
except WebSocketDisconnect:
pass
finally:
@ -1757,33 +1768,68 @@ async def api_update_geofence(gid: str, payload: GeofenceUpdate):
@app.delete("/api/geofences/{gid}", status_code=204)
async def api_delete_geofence(gid: str):
from geofence import delete_geofence
await delete_geofence(gid)
ok = await delete_geofence(gid)
if not ok:
raise HTTPException(404, "geofence not found")
return None
@app.get("/api/geofence-alerts")
async def api_geofence_alerts(limit: int = Query(100, ge=1, le=500)):
from sqlalchemy import text as sql_text
@app.get("/api/geofences/{gid}/at")
async def api_geofence_at(
gid: str,
timestamp: str = Query(..., description="ISO-8601 instant for the 1-minute DVR bucket"),
):
"""Aircraft/vessels/fires inside this fence at time T. Never writes."""
from geofence import snapshot_at
from tracks import parse_timestamp
try:
async with async_session() as session:
rows = (await session.execute(sql_text(
"""
SELECT id::text, geofence_id::text, source_kind, entity_id,
lat, lon, payload, created_at
FROM geofence_alerts
ORDER BY created_at DESC
LIMIT :limit
"""
), {"limit": limit})).mappings().all()
out = []
for r in rows:
item = dict(r)
if item.get("created_at") is not None:
item["created_at"] = item["created_at"].isoformat()
out.append(item)
return out
ts = parse_timestamp(timestamp)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
if ts is None:
raise HTTPException(422, "timestamp required")
try:
body = await snapshot_at(gid, ts)
except Exception:
return []
body = {
"geofence_id": gid,
"timestamp": ts.isoformat(),
"aircraft": [],
"vessels": [],
"fires": [],
}
if body is None:
raise HTTPException(404, "geofence not found")
return body
@app.get("/api/geofence-alerts")
async def api_geofence_alerts(
geofence_id: UUID | None = Query(None),
since: str | None = Query(None, description="ISO-8601 inclusive lower bound"),
until: str | None = Query(None, description="ISO-8601 inclusive upper bound"),
source_kind: str | None = Query(None, description="firms|ais|adsb"),
limit: int = Query(100, ge=1, le=500),
):
"""Hit log for drawn fences. Not /api/alerts (entity/keyword)."""
from geofence import list_alerts
from tracks import parse_timestamp
if source_kind is not None and source_kind not in ("firms", "ais", "adsb"):
raise HTTPException(422, "source_kind must be one of: firms, ais, adsb")
try:
since_ts = parse_timestamp(since) if since else None
until_ts = parse_timestamp(until) if until else None
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
return await list_alerts(
geofence_id=str(geofence_id) if geofence_id else None,
since=since_ts,
until=until_ts,
source_kind=source_kind,
limit=limit,
)
@app.get("/api/fire-aircraft")

View file

@ -8,10 +8,18 @@ from __future__ import annotations
import asyncio
from typing import Any
from uuid import UUID
BBox = tuple[float, float, float, float] # minlon, minlat, maxlon, maxlat
def _uuid_str(value: object) -> str | None:
try:
return str(UUID(str(value)))
except (ValueError, TypeError, AttributeError):
return None
def point_in_bbox(lon: float, lat: float, bbox: BBox | None) -> bool:
"""True if (lon, lat) sits inside an axis-aligned viewport."""
if bbox is None:
@ -26,6 +34,7 @@ class ConnectionManager:
def __init__(self) -> None:
self._queues: dict[str, asyncio.Queue] = {}
self._viewports: dict[str, BBox] = {}
self._watched: dict[str, set[str]] = {}
def register(self, client_id: str, maxsize: int = 256) -> asyncio.Queue:
q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
@ -35,6 +44,21 @@ class ConnectionManager:
def unregister(self, client_id: str) -> None:
self._queues.pop(client_id, None)
self._viewports.pop(client_id, None)
self._watched.pop(client_id, None)
def set_watched_geofences(self, client_id: str, ids: list[str]) -> None:
"""Watch these fence UUIDs so geofence_alert delivers off-viewport.
Invalid UUIDs are ignored. Empty list = watch none (viewport-only).
"""
if client_id not in self._queues:
return
watched: set[str] = set()
for raw in ids:
uid = _uuid_str(raw)
if uid is not None:
watched.add(uid)
self._watched[client_id] = watched
def set_viewport(self, client_id: str, bbox: BBox) -> None:
if client_id in self._queues:
@ -59,13 +83,21 @@ class ConnectionManager:
) -> int:
"""Enqueue `{type, payload}` for clients whose viewport contains the point.
Drops the oldest queued message if a client's buffer is full so a slow
tab cannot stall ingest. Returns the number of clients that got a copy.
kind=geofence_alert also delivers when payload.geofence_id is in the
client's watch set (even if the point is off-viewport). Other kinds
stay viewport-only. Drops the oldest queued message if a client's
buffer is full. Returns the number of clients that got a copy.
"""
msg = {"type": kind, "payload": payload}
sent = 0
gid = _uuid_str(payload.get("geofence_id")) if kind == "geofence_alert" else None
for client_id, queue in list(self._queues.items()):
if not point_in_bbox(lon, lat, self._viewports.get(client_id)):
in_view = point_in_bbox(lon, lat, self._viewports.get(client_id))
if kind == "geofence_alert":
watching = gid is not None and gid in self._watched.get(client_id, set())
if not in_view and not watching:
continue
elif not in_view:
continue
if queue.full():
try:

View file

@ -51,23 +51,33 @@ def test_matching_geofences_only_active_hits():
assert matching_geofences(-122.4, 37.7, fences) == []
FENCE_ID = "11111111-1111-1111-1111-111111111111"
NC_VIEW = (-80.0, 35.0, -78.0, 36.0)
SF_VIEW = (-123.0, 37.0, -121.0, 38.0)
def _alert_payload(gid=FENCE_ID):
return {
"geofence_id": gid,
"geofence_name": "NC",
"source_kind": "ais",
"entity_id": "366123456",
"lat": 35.5,
"lon": -79.0,
}
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))
mgr.set_viewport("nc", NC_VIEW)
mgr.set_viewport("sf", SF_VIEW)
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)
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 1
msg = q_nc.get_nowait()
assert msg["type"] == "geofence_alert"
@ -77,6 +87,109 @@ def test_geofence_alert_fans_out_only_to_viewport_clients():
asyncio.run(run())
def test_off_viewport_watch_receives_geofence_alert():
mgr = ConnectionManager()
q_sf = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
mgr.set_watched_geofences("sf", [FENCE_ID])
async def run():
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 1
msg = q_sf.get_nowait()
assert msg["type"] == "geofence_alert"
assert msg["payload"]["geofence_id"] == FENCE_ID
asyncio.run(run())
def test_off_viewport_without_watch_does_not_receive_geofence_alert():
mgr = ConnectionManager()
q_sf = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
async def run():
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 0
assert q_sf.empty()
asyncio.run(run())
def test_on_viewport_receives_geofence_alert_without_watch():
mgr = ConnectionManager()
q_nc = mgr.register("nc")
mgr.set_viewport("nc", NC_VIEW)
async def run():
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 1
assert q_nc.get_nowait()["type"] == "geofence_alert"
asyncio.run(run())
def test_ais_stays_viewport_only_even_when_watching():
mgr = ConnectionManager()
q_sf = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
mgr.set_watched_geofences("sf", [FENCE_ID])
async def run():
n = await mgr.publish_point("ais", {"id": "366123456"}, lat=35.5, lon=-79.0)
assert n == 0
assert q_sf.empty()
asyncio.run(run())
def test_invalid_watch_uuids_ignored_empty_list_clears():
mgr = ConnectionManager()
q = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
mgr.set_watched_geofences("sf", ["not-a-uuid", FENCE_ID, "also-bad"])
async def run():
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 1
q.get_nowait()
mgr.set_watched_geofences("sf", [])
n2 = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n2 == 0
assert q.empty()
asyncio.run(run())
def test_unregister_clears_watched_geofences():
mgr = ConnectionManager()
q = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
mgr.set_watched_geofences("sf", [FENCE_ID])
mgr.unregister("sf")
q2 = mgr.register("sf")
mgr.set_viewport("sf", SF_VIEW)
async def run():
n = await mgr.publish_point(
"geofence_alert", _alert_payload(), lat=35.5, lon=-79.0,
)
assert n == 0
assert q2.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
@ -134,3 +247,117 @@ def test_record_and_notify_queries_postgis_when_cache_empty(monkeypatch):
inserts = [p for p in executed if isinstance(p, dict)]
assert inserts and inserts[0]["source_kind"] == "firms"
assert "commit" in executed
def test_list_alerts_sql_filters(monkeypatch):
captured: dict = {}
class FakeResult:
def mappings(self):
return self
def all(self):
return []
class FakeSession:
async def execute(self, stmt, params=None):
captured["sql"] = str(stmt)
captured["params"] = params
return FakeResult()
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
monkeypatch.setattr(geofence, "async_session", FakeSession)
from datetime import datetime, timezone
since = datetime(2026, 8, 28, tzinfo=timezone.utc)
until = datetime(2026, 8, 29, tzinfo=timezone.utc)
async def run():
return await geofence.list_alerts(
geofence_id=FENCE_ID, since=since, until=until,
source_kind="firms", limit=5,
)
assert asyncio.run(run()) == []
sql = captured["sql"].lower()
assert "geofence_id" in sql
assert "created_at >=" in sql
assert "created_at <=" in sql
assert "source_kind" in sql
assert captured["params"]["geofence_id"] == FENCE_ID
assert captured["params"]["source_kind"] == "firms"
assert captured["params"]["limit"] == 5
def test_alembic_fence_created_index_exists():
from pathlib import Path
text = Path(__file__).resolve().parent.parent.joinpath(
"alembic/versions/011_geofence_alerts_fence.py",
).read_text()
assert "ix_geofence_alerts_fence_created" in text
assert "010_bbox_gist" in text
def test_snapshot_at_404_when_fence_missing(monkeypatch):
geofence._cache.clear()
async def boom():
raise RuntimeError("db down")
monkeypatch.setattr(geofence, "refresh_cache", boom)
async def run():
from datetime import datetime, timezone
return await geofence.snapshot_at(
FENCE_ID, datetime(2026, 8, 28, 12, 4, tzinfo=timezone.utc),
)
assert asyncio.run(run()) is None
def test_snapshot_queries_st_intersects(monkeypatch):
geofence._cache[:] = [{
"id": FENCE_ID, "name": "NC", "geojson": NC_BOX, "active": True,
}]
sqls: list[str] = []
class FakeResult:
def mappings(self):
return self
def all(self):
return []
class FakeSession:
async def execute(self, stmt, params=None):
sqls.append(str(stmt))
return FakeResult()
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
monkeypatch.setattr(geofence, "async_session", FakeSession)
async def run():
from datetime import datetime, timezone
return await geofence.snapshot_at(
FENCE_ID, datetime(2026, 8, 28, 12, 4, 30, tzinfo=timezone.utc),
)
body = asyncio.run(run())
assert body["aircraft"] == []
assert body["vessels"] == []
assert body["fires"] == []
blob = "\n".join(sqls).lower()
assert "st_intersects" in blob
assert "aircraft_tracks_1min" in blob
assert "vessel_tracks_1min" in blob
assert "from fires" in blob

View file

@ -17,6 +17,9 @@ async def _req(method: str, path: str, **kw) -> httpx.Response:
return await client.request(method, path, **kw)
FENCE_ID = "11111111-1111-1111-1111-111111111111"
def test_geofence_post_rejects_point():
resp = asyncio.run(_req(
"POST", "/api/geofences",
@ -25,6 +28,111 @@ def test_geofence_post_rejects_point():
assert resp.status_code == 422
def test_delete_geofence_404_when_missing(monkeypatch):
async def missing(_gid: str) -> bool:
return False
monkeypatch.setattr("geofence.delete_geofence", missing)
resp = asyncio.run(_req("DELETE", f"/api/geofences/{FENCE_ID}"))
assert resp.status_code == 404
def test_geofence_alerts_passes_filters(monkeypatch):
seen = {}
async def fake_list(**kwargs):
seen.update(kwargs)
return [{"id": "a", "geofence_id": FENCE_ID, "source_kind": "ais"}]
monkeypatch.setattr("geofence.list_alerts", fake_list)
resp = asyncio.run(_req(
"GET", "/api/geofence-alerts",
params={
"geofence_id": FENCE_ID,
"since": "2026-08-28T00:00:00Z",
"until": "2026-08-29T00:00:00Z",
"source_kind": "ais",
"limit": 10,
},
))
assert resp.status_code == 200
assert resp.json()[0]["source_kind"] == "ais"
assert seen["geofence_id"] == FENCE_ID
assert seen["source_kind"] == "ais"
assert seen["limit"] == 10
assert seen["since"] is not None
assert seen["until"] is not None
def test_geofence_alerts_rejects_bad_source_kind():
resp = asyncio.run(_req(
"GET", "/api/geofence-alerts", params={"source_kind": "camera"},
))
assert resp.status_code == 422
def test_geofence_at_404_when_missing(monkeypatch):
async def no_snap(gid: str, ts):
return None
monkeypatch.setattr("geofence.snapshot_at", no_snap)
resp = asyncio.run(_req(
"GET", f"/api/geofences/{FENCE_ID}/at",
params={"timestamp": "2026-08-28T12:04:00Z"},
))
assert resp.status_code == 404
def test_geofence_at_empty_lists_when_db_down(monkeypatch):
async def empty_snap(gid: str, ts):
return {
"geofence_id": gid,
"timestamp": ts.isoformat(),
"aircraft": [],
"vessels": [],
"fires": [],
}
monkeypatch.setattr("geofence.snapshot_at", empty_snap)
resp = asyncio.run(_req(
"GET", f"/api/geofences/{FENCE_ID}/at",
params={"timestamp": "2026-08-28T12:04:00Z"},
))
assert resp.status_code == 200
body = resp.json()
assert body["geofence_id"] == FENCE_ID
assert body["aircraft"] == []
assert body["vessels"] == []
assert body["fires"] == []
assert "timestamp" in body
def test_geofence_at_does_not_notify(monkeypatch):
called = {"notify": 0}
async def empty_snap(gid: str, ts):
return {
"geofence_id": gid,
"timestamp": ts.isoformat(),
"aircraft": [],
"vessels": [],
"fires": [],
}
async def boom(**_kw):
called["notify"] += 1
raise AssertionError("GET /at must not record_and_notify")
monkeypatch.setattr("geofence.snapshot_at", empty_snap)
monkeypatch.setattr("geofence.record_and_notify", boom)
resp = asyncio.run(_req(
"GET", f"/api/geofences/{FENCE_ID}/at",
params={"timestamp": "2026-08-28T12:04:00Z"},
))
assert resp.status_code == 200
assert called["notify"] == 0
def test_geofences_list_does_not_collide_with_alerts():
resp = asyncio.run(_req("GET", "/api/geofences"))
assert resp.status_code == 200