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.
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""Viewport-filtered in-memory pub/sub — no Redis."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from ws_manager import ConnectionManager, point_in_bbox
|
|
|
|
|
|
def test_point_in_bbox_inclusive():
|
|
box = (-80.0, 35.0, -78.0, 36.0)
|
|
assert point_in_bbox(-79.0, 35.5, box) is True
|
|
assert point_in_bbox(-80.0, 35.0, box) is True
|
|
assert point_in_bbox(-77.0, 35.5, box) is False
|
|
assert point_in_bbox(-79.0, 34.0, box) is False
|
|
|
|
|
|
def test_missing_viewport_does_not_get_firehose():
|
|
mgr = ConnectionManager()
|
|
q = mgr.register("tailnet-a")
|
|
|
|
async def run():
|
|
n = await mgr.publish_point("ais", {"id": "mmsi-1"}, lat=35.5, lon=-79.0)
|
|
assert n == 0
|
|
assert q.empty()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_fanout_only_to_intersecting_viewports():
|
|
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():
|
|
n = await mgr.publish_point(
|
|
"ais", {"id": "mmsi-1", "lat": 35.5, "lon": -79.0},
|
|
lat=35.5, lon=-79.0,
|
|
)
|
|
assert n == 1
|
|
msg = q_nc.get_nowait()
|
|
assert msg["type"] == "ais"
|
|
assert msg["payload"]["id"] == "mmsi-1"
|
|
assert q_sf.empty()
|
|
|
|
n2 = await mgr.publish_point(
|
|
"adsb", {"id": "a1b2c3"}, lat=37.7, lon=-122.4,
|
|
)
|
|
assert n2 == 1
|
|
msg2 = q_sf.get_nowait()
|
|
assert msg2["type"] == "adsb"
|
|
assert q_nc.empty()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_disconnect_stops_fanout():
|
|
mgr = ConnectionManager()
|
|
q = mgr.register("gone")
|
|
mgr.set_viewport("gone", (-180.0, -90.0, 180.0, 90.0))
|
|
mgr.unregister("gone")
|
|
|
|
async def run():
|
|
n = await mgr.publish_point("ais", {"id": "x"}, lat=0.0, lon=0.0)
|
|
assert n == 0
|
|
assert q.empty()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_upsert_vessel_fans_out_to_intersecting_client(monkeypatch):
|
|
from live_layers import upsert_vessel, vessel_last_known
|
|
from ws_manager import manager
|
|
|
|
vessel_last_known.clear()
|
|
manager._queues.clear()
|
|
manager._viewports.clear()
|
|
q = manager.register("nc")
|
|
manager.set_viewport("nc", (-80.0, 35.0, -78.0, 36.0))
|
|
|
|
async def run():
|
|
await upsert_vessel({"id": "366123456", "lat": 35.2, "lon": -79.1, "label": "TEST"})
|
|
msg = q.get_nowait()
|
|
assert msg["type"] == "ais"
|
|
assert msg["payload"]["id"] == "366123456"
|
|
|
|
asyncio.run(run())
|
|
manager.unregister("nc")
|
|
vessel_last_known.clear()
|