f6c1cfc454
Off-viewport clients that send {type:watch_geofences,ids} on /ws/live
still receive geofence_alert; AIS/ADS-B stay viewport-only.
GET /api/geofence-alerts accepts geofence_id/since/until/source_kind.
GET /api/geofences/{id}/at returns CAGG+FIRMS inside the fence at T
(404 if missing, empty lists if DB down). DELETE missing fences 404s.
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""In-memory WebSocket pub/sub with viewport filtering.
|
|
|
|
Zero extra deps. Ingest workers publish AIS/ADS-B points; only clients whose
|
|
current map bbox contains the point receive the payload. No Redis/Kafka.
|
|
"""
|
|
|
|
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:
|
|
return False
|
|
minlon, minlat, maxlon, maxlat = bbox
|
|
return minlon <= lon <= maxlon and minlat <= lat <= maxlat
|
|
|
|
|
|
class ConnectionManager:
|
|
"""Maps Tailscale/browser clients → viewport bbox + per-client queue."""
|
|
|
|
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)
|
|
self._queues[client_id] = q
|
|
return q
|
|
|
|
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:
|
|
self._viewports[client_id] = bbox
|
|
|
|
def viewport_of(self, client_id: str) -> BBox | None:
|
|
return self._viewports.get(client_id)
|
|
|
|
def viewports(self) -> list[BBox]:
|
|
return list(self._viewports.values())
|
|
|
|
def has_clients(self) -> bool:
|
|
return bool(self._queues)
|
|
|
|
async def publish_point(
|
|
self,
|
|
kind: str,
|
|
payload: dict[str, Any],
|
|
*,
|
|
lat: float,
|
|
lon: float,
|
|
) -> int:
|
|
"""Enqueue `{type, payload}` for clients whose viewport contains the point.
|
|
|
|
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()):
|
|
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:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
try:
|
|
queue.put_nowait(msg)
|
|
except asyncio.QueueFull:
|
|
continue
|
|
sent += 1
|
|
return sent
|
|
|
|
|
|
manager = ConnectionManager()
|