103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
|
"""AISStream WebSocket worker — server-side only.
|
||
|
|
|
||
|
|
aisstream.io forbids browser clients. Connect from the FastAPI/ingest
|
||
|
|
process, upsert last-known positions, and expose them via GET /api/vessels.
|
||
|
|
|
||
|
|
Idle (no crash) when AISSTREAM_API_KEY is unset. Reconnect with jittered
|
||
|
|
backoff and resend the full subscription within 3 seconds of each connect.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
import random
|
||
|
|
|
||
|
|
from config import AISSTREAM_API_KEY, AISSTREAM_BBOX
|
||
|
|
from keystore import get_api_key
|
||
|
|
from live_layers import transform_ais_frame, upsert_vessel
|
||
|
|
|
||
|
|
logger = logging.getLogger("osint.aisstream")
|
||
|
|
|
||
|
|
WS_URL = "wss://stream.aisstream.io/v0/stream"
|
||
|
|
FILTER_TYPES = [
|
||
|
|
"PositionReport",
|
||
|
|
"StandardClassBPositionReport",
|
||
|
|
"ExtendedClassBPositionReport",
|
||
|
|
"ShipStaticData",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_boxes(raw: str) -> list[list[list[float]]]:
|
||
|
|
"""Env format: minlat,minlon,maxlat,maxlon[; ...]. AIS wants [[lat,lon],[lat,lon]]."""
|
||
|
|
boxes = []
|
||
|
|
for chunk in (raw or "").split(";"):
|
||
|
|
parts = [p.strip() for p in chunk.split(",") if p.strip()]
|
||
|
|
if len(parts) != 4:
|
||
|
|
continue
|
||
|
|
minlat, minlon, maxlat, maxlon = (float(p) for p in parts)
|
||
|
|
boxes.append([[minlat, minlon], [maxlat, maxlon]])
|
||
|
|
return boxes or [[[24.0, -125.0], [50.0, -66.0]]]
|
||
|
|
|
||
|
|
|
||
|
|
async def _resolve_key() -> str:
|
||
|
|
return (os.getenv("AISSTREAM_API_KEY") or AISSTREAM_API_KEY
|
||
|
|
or (await get_api_key("AISSTREAM_API_KEY")) or "").strip()
|
||
|
|
|
||
|
|
|
||
|
|
async def run_ais_worker() -> None:
|
||
|
|
"""Long-lived reconnect loop. Safe to spawn as an asyncio task."""
|
||
|
|
try:
|
||
|
|
import websockets
|
||
|
|
except ImportError:
|
||
|
|
logger.warning("websockets package not installed — AIS worker disabled")
|
||
|
|
return
|
||
|
|
|
||
|
|
backoff = 2.0
|
||
|
|
while True:
|
||
|
|
key = await _resolve_key()
|
||
|
|
if not key:
|
||
|
|
logger.warning(
|
||
|
|
"AISSTREAM_API_KEY not set — AIS ingest idle. "
|
||
|
|
"Create a free key at https://aisstream.io/account"
|
||
|
|
)
|
||
|
|
await asyncio.sleep(60)
|
||
|
|
continue
|
||
|
|
boxes = _parse_boxes(AISSTREAM_BBOX)
|
||
|
|
try:
|
||
|
|
async with websockets.connect(
|
||
|
|
WS_URL,
|
||
|
|
max_size=2 ** 22,
|
||
|
|
ping_interval=20,
|
||
|
|
ping_timeout=20,
|
||
|
|
compression="deflate",
|
||
|
|
) as ws:
|
||
|
|
sub = {
|
||
|
|
"APIKey": key,
|
||
|
|
"BoundingBoxes": boxes,
|
||
|
|
"FilterMessageTypes": FILTER_TYPES,
|
||
|
|
}
|
||
|
|
await ws.send(json.dumps(sub))
|
||
|
|
logger.info("AISStream subscribed (%d bbox(es))", len(boxes))
|
||
|
|
backoff = 2.0
|
||
|
|
async for raw in ws:
|
||
|
|
if isinstance(raw, bytes):
|
||
|
|
raw = raw.decode("utf-8", errors="replace")
|
||
|
|
try:
|
||
|
|
frame = json.loads(raw)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
continue
|
||
|
|
marker = transform_ais_frame(frame)
|
||
|
|
if marker:
|
||
|
|
await upsert_vessel(marker)
|
||
|
|
except asyncio.CancelledError:
|
||
|
|
raise
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
logger.exception("AISStream disconnected")
|
||
|
|
delay = backoff + random.uniform(0, 1.5)
|
||
|
|
logger.info("AISStream reconnect in %.1fs", delay)
|
||
|
|
await asyncio.sleep(delay)
|
||
|
|
backoff = min(60.0, backoff * 1.7)
|