From 32320635dc508215d4a1ea10061cc2f03c5aceb2 Mon Sep 17 00:00:00 2001 From: Sirius DevOps Date: Thu, 27 Aug 2026 21:46:37 -0400 Subject: [PATCH] feat: AIS stream follows the map viewport The Vessels layer now retunes the server-side AISStream subscription to the client viewport instead of a static AISSTREAM_BBOX. The frontend POSTs its quantized viewport box to /api/vessels/subscribe on moveend; the ais_stream worker coalesces and applies it at the service's 1 subscription/s cap, then last-known positions for the new area arrive within a couple of seconds (the frontend does one follow-up fetch after retuning). Bounds the in-memory vessel store across regions. Key stays server-side. --- app/ais_stream.py | 56 +++++++++++++++++++++++++++++++++-- app/live_layers.py | 11 +++++++ app/main.py | 30 ++++++++++++++++++- app/schemas.py | 10 +++++++ app/static/index.html | 46 ++++++++++++++++++++++++++-- tests/test_api_live_layers.py | 38 ++++++++++++++++++++++++ 6 files changed, 185 insertions(+), 6 deletions(-) diff --git a/app/ais_stream.py b/app/ais_stream.py index 0c27cae..18fe70b 100644 --- a/app/ais_stream.py +++ b/app/ais_stream.py @@ -14,6 +14,7 @@ import json import logging import os import random +import time from config import AISSTREAM_API_KEY, AISSTREAM_BBOX from keystore import get_api_key @@ -29,6 +30,36 @@ FILTER_TYPES = [ "ShipStaticData", ] +# ── Viewport-following ───────────────────────────────────────────────────── +# The frontend POSTs its current viewport box to /api/vessels/subscribe; the +# worker retunes the AISStream subscription to it (throttled to 1/s, the +# service's subscription-update cap). ``None`` keeps the env AISSTREAM_BBOX +# default. Last writer wins; the key never reaches the browser. +_desired_boxes: list[list[list[float]]] | None = None +_bbox_guard = asyncio.Lock() + + +async def request_viewport_bbox( + minlon: float, minlat: float, maxlon: float, maxlat: float +) -> None: + """Retune the live subscription to a viewport box (lon/lat input order).""" + global _desired_boxes + box = [[minlat, minlon], [maxlat, maxlon]] # AISStream wants [lat, lon] corners + async with _bbox_guard: + _desired_boxes = [box] + + +async def reset_viewport_bbox() -> None: + """Fall back to the env AISSTREAM_BBOX default.""" + global _desired_boxes + async with _bbox_guard: + _desired_boxes = None + + +async def _take_desired_boxes() -> list[list[list[float]]] | None: + async with _bbox_guard: + return _desired_boxes + def _parse_boxes(raw: str) -> list[list[list[float]]]: """Env format: minlat,minlon,maxlat,maxlon[; ...]. AIS wants [[lat,lon],[lat,lon]].""" @@ -65,7 +96,7 @@ async def run_ais_worker() -> None: ) await asyncio.sleep(60) continue - boxes = _parse_boxes(AISSTREAM_BBOX) + boxes = await _take_desired_boxes() or _parse_boxes(AISSTREAM_BBOX) try: async with websockets.connect( WS_URL, @@ -82,7 +113,28 @@ async def run_ais_worker() -> None: await ws.send(json.dumps(sub)) logger.info("AISStream subscribed (%d bbox(es))", len(boxes)) backoff = 2.0 - async for raw in ws: + last_submit = time.monotonic() + while True: + # Follow the client viewport: coalesce to the latest request + # and honor AISStream's 1 subscription-update/s cap. + desired = await _take_desired_boxes() + if ( + desired is not None + and desired != boxes + and time.monotonic() - last_submit >= 1.0 + ): + boxes = desired + sub["BoundingBoxes"] = boxes + await ws.send(json.dumps(sub)) + last_submit = time.monotonic() + logger.info( + "AISStream re-subscribed to viewport (%d bbox)", + len(boxes), + ) + try: + raw = await asyncio.wait_for(ws.recv(), timeout=0.25) + except asyncio.TimeoutError: + continue if isinstance(raw, bytes): raw = raw.decode("utf-8", errors="replace") try: diff --git a/app/live_layers.py b/app/live_layers.py index c2ea170..5daf056 100644 --- a/app/live_layers.py +++ b/app/live_layers.py @@ -66,6 +66,9 @@ _QUANT = 0.25 # degrees — pan jitter inside a cell reuses the TTL entry # Last-known AIS positions (MMSI -> marker). Filled by ais_stream worker. vessel_last_known: dict[str, dict] = {} vessel_lock = asyncio.Lock() +# Viewport-following accumulates vessels across every region visited in a +# session — keep the in-memory store bounded (oldest entries evicted). +_MAX_VESSELS = 6000 def overlay_catalog() -> dict: @@ -620,6 +623,14 @@ async def upsert_vessel(marker: dict) -> None: ), "seen_at": datetime.now(timezone.utc).isoformat(), } + if len(vessel_last_known) > _MAX_VESSELS: + excess = len(vessel_last_known) - int(_MAX_VESSELS * 0.9) + oldest = sorted( + vessel_last_known, + key=lambda k: vessel_last_known[k].get("seen_at", ""), + )[:excess] + for k in oldest: + vessel_last_known.pop(k, None) def _wfigs_params(bbox: str | None, *, offset_m: float = 250.0) -> dict: diff --git a/app/main.py b/app/main.py index 7da2f84..421fbfc 100644 --- a/app/main.py +++ b/app/main.py @@ -40,7 +40,7 @@ from schemas import ( FeedSourceCreate, FeedSourceOut, KeyOut, KeyValueIn, SearchResult, SentimentSummary, SourceType, - SearchQuery, TimelinePoint, + SearchQuery, TimelinePoint, VesselBboxUpdate, ) from ingestor import ingest_event, fetch_and_process from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals @@ -1176,6 +1176,34 @@ async def list_vessels( raise HTTPException(422, str(exc)) from exc +@app.post("/api/vessels/subscribe") +async def vessels_subscribe(payload: VesselBboxUpdate): + """Retune the server-side AISStream subscription to the client viewport. + + The stream follows the map: after a moveend the frontend posts its bbox, + the worker re-subscribes (≤1/s upstream), and last-known vessels for the + new area start arriving within a second or two. Empty/null bbox resets to + the env AISSTREAM_BBOX default. The API key never reaches the browser. + """ + raw = (payload.bbox or "").strip() + if not raw: + from ais_stream import reset_viewport_bbox + await reset_viewport_bbox() + return {"ok": True, "bbox": None} + try: + minlon, minlat, maxlon, maxlat = parse_bbox(raw) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + if not (-180 <= minlon <= 180 and -180 <= maxlon <= 180 + and -90 <= minlat <= 90 and -90 <= maxlat <= 90): + raise HTTPException(422, "bbox coordinates out of range") + if minlon >= maxlon or minlat >= maxlat: + raise HTTPException(422, "bbox must have min < max") + from ais_stream import request_viewport_bbox + await request_viewport_bbox(minlon, minlat, maxlon, maxlat) + return {"ok": True, "bbox": raw} + + @app.get("/api/fire-incidents") async def list_fire_incidents( bbox: str | None = Query(None), diff --git a/app/schemas.py b/app/schemas.py index 90d0f67..f49d3a7 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -300,3 +300,13 @@ class DashboardSummary(BaseModel): sentiment: SentimentSummary top_entities: list[EntityOut] + +class VesselBboxUpdate(BaseModel): + """Retune the server-side AISStream subscription to a client viewport box. + + ``bbox`` is "minlon,minlat,maxlon,maxlat" (Leaflet order). ``None``/empty + resets to the env AISSTREAM_BBOX default. + """ + + bbox: str | None = None + diff --git a/app/static/index.html b/app/static/index.html index 5074875..0d8e730 100644 --- a/app/static/index.html +++ b/app/static/index.html @@ -1652,6 +1652,8 @@ let overlayReq = {ac:0, trains:0, vessels:0, alerts:0, perim:0, incidents:0, sto let moveDebounce = null; let overlayAbort = null; let lastCell = ''; +let lastVesselSubBox = ''; // last viewport box sent to the AIS stream +let vesselRefollowTimer = null; // one follow-up fetch after a retune function bboxCell() { if (!map) return ''; return currentBBox().split(',').map(n => Number(n).toFixed(2)).join(',') + '@' + map.getZoom(); @@ -1917,6 +1919,17 @@ function currentBBox() { } return `${west.toFixed(4)},${south.toFixed(4)},${east.toFixed(4)},${north.toFixed(4)}`; } +function quantizeBBox(bbox, decimals = 1) { + // Round outward (west/south floor, east/north ceil) so the box always + // covers the viewport; a coarse box keeps AIS stream retunes sparse. + const [w, s, e, n] = bbox.split(',').map(Number); + const p = Math.pow(10, decimals); + const wq = Math.floor(w * p) / p; + const sq = Math.floor(s * p) / p; + const eq = Math.ceil(e * p) / p; + const nq = Math.ceil(n * p) / p; + return [wq, sq, eq, nq].map(v => v.toFixed(decimals)).join(','); +} function toggleLayerPanel() { const panel = document.getElementById('layer-panel'); panel.classList.toggle('collapsed'); @@ -2576,8 +2589,13 @@ async function loadTrains() { } async function toggleVessels() { vesselsOn = document.getElementById('lp-vessels-on').checked; - if (vesselsOn) await loadVessels(); - else vesselsGroup = dropLayer(vesselsGroup); + if (vesselsOn) { + lastVesselSubBox = ''; + await loadVessels(); + } else { + vesselsGroup = dropLayer(vesselsGroup); + clearTimeout(vesselRefollowTimer); + } } async function loadVessels() { if (!map) return; @@ -2586,8 +2604,25 @@ async function loadVessels() { return; } const req = ++overlayReq.vessels; + const bb = currentBBox(); + // Retune the server-side AIS stream to this viewport. Skip whole-world + // boxes (zoom out / dateline wrap) so the free tier is not flooded; the + // worker coalesces retunes and caps them at 1/s upstream. + const [bw, bs, be, bn] = bb.split(',').map(Number); + const isWorld = (be - bw) >= 300 || bb.startsWith('-180.0000,-85.0000,180.0000,85.0000'); + const subKey = isWorld ? '' : quantizeBBox(bb); + let retuned = false; + if (subKey && subKey !== lastVesselSubBox) { + lastVesselSubBox = subKey; + retuned = true; + fetch(`${API}/api/vessels/subscribe`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ bbox: subKey }) + }).catch(() => {}); + } try { - const r = await overlayFetch(`${API}/api/vessels?bbox=${currentBBox()}`); + const r = await overlayFetch(`${API}/api/vessels?bbox=${bb}`); const pts = await r.json(); if (req !== overlayReq.vessels) return; vesselsGroup = renderPoints(vesselsGroup, Array.isArray(pts) ? pts : [], p => { @@ -2596,6 +2631,11 @@ async function loadVessels() { }, true, 'vessel'); document.getElementById('lp-vessels-count').textContent = (pts.length || 0).toLocaleString(); addExtraAttrib('AISStream'); + // First positions for a new zone arrive ~1-3s after the retune. + if (retuned) { + clearTimeout(vesselRefollowTimer); + vesselRefollowTimer = setTimeout(() => { if (vesselsOn) loadVessels(); }, 2200); + } } catch (e) { if (isAbort(e)) return; console.error('Vessels load failed', e); diff --git a/tests/test_api_live_layers.py b/tests/test_api_live_layers.py index ec17da8..83f200c 100644 --- a/tests/test_api_live_layers.py +++ b/tests/test_api_live_layers.py @@ -16,6 +16,12 @@ async def _get(path: str) -> httpx.Response: return await client.get(path) +async def _post(path: str, payload: dict | None) -> httpx.Response: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url=BASE) as client: + return await client.post(path, json=payload) + + def test_map_layers_includes_overlays(): body = asyncio.run(_get("/api/map/layers")).json() assert "layers" in body @@ -37,3 +43,35 @@ def test_vessels_empty_without_ais_key(): assert resp.status_code == 200 assert resp.json() == [] assert "max-age" in (resp.headers.get("cache-control") or "").lower() + + +def _desired_boxes(): + import asyncio as _a + from ais_stream import _take_desired_boxes + return _a.run(_take_desired_boxes()) + + +def test_vessels_subscribe_sets_viewport_box(): + assert _desired_boxes() is None + resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": "-70,40,-60,45"})) + assert resp.status_code == 200 + body = resp.json() + assert body["ok"] is True and body["bbox"] == "-70,40,-60,45" + # AISStream corner order: [[lat, lon], [lat, lon]] (southwest, northeast). + assert _desired_boxes() == [[[40.0, -70.0], [45.0, -60.0]]] + + +def test_vessels_subscribe_empty_resets(): + assert asyncio.run(_post("/api/vessels/subscribe", {"bbox": "-70,40,-60,45"})).status_code == 200 + resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": ""})) + assert resp.status_code == 200 + assert resp.json()["bbox"] is None + assert _desired_boxes() is None + + +def test_vessels_subscribe_rejects_bad_bbox(): + for bad in ("1,2,3", "a,b,c,d", "20,30,10,40", "0,0,0,200"): + resp = asyncio.run(_post("/api/vessels/subscribe", {"bbox": bad})) + assert resp.status_code == 422, bad + # Explicit null bbox is the "reset to env default" path (still 200). + assert asyncio.run(_post("/api/vessels/subscribe", {"bbox": None})).status_code == 200