diff --git a/.env.example b/.env.example index 9e99921..50cf113 100644 --- a/.env.example +++ b/.env.example @@ -59,6 +59,22 @@ FIRMS_INTERVAL=900 # Set to 0 to disable the fire loop entirely. INGEST_FIRES=1 +# ── VesselAPI (quota-capped REST AIS poller — free tier 150 calls/mo) ───── +# AISStream keeps US coasts live; VesselAPI fills the Middle East blind spot. +# The poller idles when VESSELAPI_API_KEY is unset. Never called from the map +# pans (GET /api/vessels serves the shared last-known cache only). +VESSELAPI_API_KEY= +# Bounding box(es) as minlat,minlon,maxlat,maxlon (lat/lon order). Semicolon- +# separated for multiple boxes. Default = Strait of Hormuz (span 3.6 ≤ 4° cap). +VESSELAPI_BBOX=25.5,55.4,27.3,57.2 +# Poll cadence in seconds (17280 = 4.8h → 5 polls/day = 150/mo). +VESSELAPI_INTERVAL=17280 +# Local hard cap on successful 2xx calls per UTC day (persisted in Postgres). +VESSELAPI_MAX_CALLS_PER_DAY=5 +# 1 = run the poller inside the dashboard process (default); ingester off. +VESSELAPI_IN_APP=1 +VESSELAPI_IN_INGEST=0 + # ── API keys (managed from the dashboard UI) ────────────────────────────── # Keys such as NOUS_API_KEY and TELEGRAM_TOKEN are stored in the Postgres # `api_keys` table and managed from the dashboard's "Keys" tab diff --git a/app/config.py b/app/config.py index be6ffec..8b8fd82 100644 --- a/app/config.py +++ b/app/config.py @@ -81,3 +81,19 @@ AISSTREAM_BBOX = os.getenv("AISSTREAM_BBOX", "24,-125,50,-66") # without the ingest profile). Set 0 if the ingester owns the only connection. AISSTREAM_IN_APP = os.getenv("AISSTREAM_IN_APP", "1").lower() in ("1", "true", "yes") AISSTREAM_IN_INGEST = os.getenv("AISSTREAM_IN_INGEST", "0").lower() in ("1", "true", "yes") + +# VesselAPI (quota-capped REST AIS poller — free tier 150 calls/mo). +# AISStream keeps US coasts; VesselAPI fills the Middle East blind spot. The +# poller idles when VESSELAPI_API_KEY is unset (never from GET /api/vessels). +VESSELAPI_API_KEY = os.getenv("VESSELAPI_API_KEY", "") +# Bounding box(es) as minlat,minlon,maxlat,maxlon — note lat/lon order (same as +# AISSTREAM_BBOX). Semicolon-separated for multiple boxes. Default: Strait of +# Hormuz (|dLat|+|dLon| = 3.6 ≤ 4° span cap). VesselAPI 400s any box over 4°. +VESSELAPI_BBOX = os.getenv("VESSELAPI_BBOX", "25.5,55.4,27.3,57.2") +# Poll cadence in seconds. 17280 = 4.8h → 5 polls/day (150/mo free tier). +VESSELAPI_INTERVAL = int(os.getenv("VESSELAPI_INTERVAL", "17280")) +# Local hard cap on successful 2xx calls per UTC day (persisted in Postgres). +VESSELAPI_MAX_CALLS_PER_DAY = int(os.getenv("VESSELAPI_MAX_CALLS_PER_DAY", "5")) +# Run the VesselAPI poller inside the dashboard process (default on, like AIS). +VESSELAPI_IN_APP = os.getenv("VESSELAPI_IN_APP", "1").lower() in ("1", "true", "yes") +VESSELAPI_IN_INGEST = os.getenv("VESSELAPI_IN_INGEST", "0").lower() in ("1", "true", "yes") diff --git a/app/keystore.py b/app/keystore.py index 41ea782..c196e8e 100644 --- a/app/keystore.py +++ b/app/keystore.py @@ -72,6 +72,11 @@ KEY_REGISTRY: dict[str, dict] = { "pattern": r"^.{8,}$", "example": "key from https://aisstream.io/account (GitHub login)", }, + "VESSELAPI_API_KEY": { + "description": "VesselAPI REST AIS (150 calls/mo free; server-side poller).", + "pattern": r"^.{8,}$", + "example": "Bearer token from https://dashboard.vesselapi.com/", + }, "OPENSKY_CLIENT_ID": { "description": "OpenSky OAuth client id — optional ADS-B fallback (unused until enabled).", "example": "client id from opensky-network.org account", diff --git a/app/main.py b/app/main.py index 80c5133..7131fc4 100644 --- a/app/main.py +++ b/app/main.py @@ -71,18 +71,26 @@ async def _lifespan(app: FastAPI): await refresh_cache() except Exception: pass - from config import AISSTREAM_IN_APP + from config import AISSTREAM_IN_APP, VESSELAPI_IN_APP ais_task = None + vesselapi_task = None adsb_task = None if AISSTREAM_IN_APP: from ais_stream import run_ais_worker ais_task = asyncio.create_task(run_ais_worker()) + if VESSELAPI_IN_APP: + from vesselapi import run_vesselapi_worker + vesselapi_task = asyncio.create_task(run_vesselapi_worker()) adsb_task = asyncio.create_task(_adsb_refresh_loop()) yield if ais_task is not None: ais_task.cancel() + if vesselapi_task is not None: + vesselapi_task.cancel() if adsb_task is not None: adsb_task.cancel() + from vesselapi import close_client + await close_client() await close_http() @@ -1464,7 +1472,11 @@ async def list_vessels( limit: int = Query(2000, ge=1, le=5000), timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"), ): - """AIS last-known from the server-side AISStream worker. Empty without a key.""" + """AIS last-known from the AISStream worker and/or the VesselAPI poller. + + Empty without either key / until the first successful poll. VesselAPI + positions upsert into the same store (extra.src = "vesselapi"). + """ if bbox: _parse_bbox_query(bbox) try: diff --git a/app/run_ingester.py b/app/run_ingester.py index 293a2e5..5d94389 100644 --- a/app/run_ingester.py +++ b/app/run_ingester.py @@ -24,7 +24,7 @@ import sys sys.path.insert(0, sys_path) -from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST # noqa: E402 +from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET, AISSTREAM_IN_INGEST, VESSELAPI_IN_INGEST # noqa: E402 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_eonet, ingest_cisa_kev # noqa: E402 from fire_sources import ingest_fires # noqa: E402 from ingestor import ingest_event, start_nats_consumer # noqa: E402 @@ -134,6 +134,9 @@ async def main() -> None: if AISSTREAM_IN_INGEST: from ais_stream import run_ais_worker # noqa: E402 tasks.append(asyncio.create_task(run_ais_worker())) + if VESSELAPI_IN_INGEST: + from vesselapi import run_vesselapi_worker # noqa: E402 + tasks.append(asyncio.create_task(run_vesselapi_worker())) await asyncio.gather(producer_loop(), consumer_loop(), *tasks) diff --git a/app/vesselapi.py b/app/vesselapi.py new file mode 100644 index 0000000..266e3e9 --- /dev/null +++ b/app/vesselapi.py @@ -0,0 +1,443 @@ +"""VesselAPI REST poller — quota-capped AIS fallback (free tier 150 calls/mo). + +AISStream (WebSocket) keeps US coasts live; VesselAPI fills the Middle East +blind spot (Strait of Hormuz default box). This worker polls the REST +``GET /v1/location/vessels/bounding-box`` endpoint at most +``VESSELAPI_MAX_CALLS_PER_DAY`` (default 5) *successful 2xx* calls per UTC day +and upserts the results into the shared ``vessel_last_known`` store. + +Idle (no crash) when VESSELAPI_API_KEY is unset. Never called from the GET +/api/vessels path — map pans must not hit upstream. One request per poll, +``pagination.limit=50``, never follow ``nextToken``, never send ``filter.sat``. +""" + +from __future__ import annotations + +import asyncio +import calendar +import logging +import os +from datetime import date, datetime, timezone + +import httpx +from sqlalchemy import Column, Date, DateTime, Integer, Table, func, select, text + +from config import ( + OSINT_USER_AGENT, + VESSELAPI_API_KEY, + VESSELAPI_BBOX, + VESSELAPI_INTERVAL, + VESSELAPI_MAX_CALLS_PER_DAY, +) +from database import async_session, engine, metadata +from live_layers import to_marker, upsert_vessel + +logger = logging.getLogger("osint.vesselapi") + +BASE_URL = "https://api.vesselapi.com/v1" +ENDPOINT = f"{BASE_URL}/location/vessels/bounding-box" +MAX_SPAN_DEG = 4.0 # |dLat| + |dLon| — VesselAPI 400s above this. +PAGE_LIMIT = 50 # pagination.limit; never follow nextToken on the free tier. + +_client: httpx.AsyncClient | None = None +_client_lock = asyncio.Lock() + + +# ── Box parsing / span validation ───────────────────────────────────────── + +class BboxError(ValueError): + """A configured VesselAPI box violates the 4° span rule or is malformed.""" + + +def validate_bbox_span( + minlat: float, minlon: float, maxlat: float, maxlon: float, +) -> None: + """Reject boxes VesselAPI would 400 on (span > 4°, bad order, bad range).""" + if not (-90 <= minlat <= 90 and -90 <= maxlat <= 90 + and -180 <= minlon <= 180 and -180 <= maxlon <= 180): + raise BboxError("coordinates out of range") + if minlat >= maxlat or minlon >= maxlon: + raise BboxError("bbox must have min < max on both axes") + dlat = abs(maxlat - minlat) + dlon = abs(maxlon - minlon) + if dlat + dlon > MAX_SPAN_DEG: + raise BboxError( + f"span |dLat|+|dLon| = {dlat + dlon:.2f}° exceeds {MAX_SPAN_DEG}° cap" + ) + + +def parse_boxes(raw: str) -> list[tuple[float, float, float, float]]: + """Env format: ``minlat,minlon,maxlat,maxlon[; ...]`` (lat/lon order).""" + out: list[tuple[float, float, float, float]] = [] + for chunk in (raw or "").split(";"): + parts = [p.strip() for p in chunk.split(",") if p.strip()] + if len(parts) != 4: + continue + try: + minlat = float(parts[0]) + minlon = float(parts[1]) + maxlat = float(parts[2]) + maxlon = float(parts[3]) + except ValueError: + continue + out.append((minlat, minlon, maxlat, maxlon)) + return out + + +def parse_boxes_validated(raw: str) -> list[tuple[float, float, float, float]]: + """Parse boxes, log + skip any that violate the span/order/range rules.""" + valid: list[tuple[float, float, float, float]] = [] + for box in parse_boxes(raw): + try: + validate_bbox_span(*box) + valid.append(box) + except BboxError as exc: + logger.warning("VesselAPI bbox %r skipped: %s", box, exc) + return valid + + +# ── Position → marker transform ─────────────────────────────────────────── + +def _f(value: object) -> float | None: + if value is None or value == "": + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _s(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def transform_vesselapi_position(obj: dict | None) -> dict | None: + """Map one VesselAPI position object to the shared marker contract. + + Returns None for glitch rows, missing MMSI, or missing coordinates. + """ + if not obj or not isinstance(obj, dict): + return None + if obj.get("suspected_glitch") is True: + return None + mmsi = obj.get("mmsi") + if mmsi is None: + return None + lat = _f(obj.get("latitude")) + lon = _f(obj.get("longitude")) + if lat is None or lon is None: + return None + mmsi_s = str(mmsi) + name = _s(obj.get("vessel_name") or obj.get("name")) + heading = _f(obj.get("heading")) + if heading is None: + heading = _f(obj.get("cog")) + sog = _f(obj.get("sog")) + extra: dict = { + "src": "vesselapi", + "mmsi": mmsi_s, + "cog": obj.get("cog"), + "sog": obj.get("sog"), + "navstat": obj.get("nav_status"), + } + imo = obj.get("imo") + if imo: + extra["imo"] = imo + dest = _s(obj.get("dest") or obj.get("destination")) + if dest: + extra["dest"] = dest + ts = obj.get("timestamp") or obj.get("processed_timestamp") + if ts: + extra["timestamp"] = ts + return to_marker( + mmsi_s, lat, lon, + heading=heading, + speed=sog, + label=name or mmsi_s, + extra=extra, + ) + + +def transform_vesselapi_payload(payload: dict | None) -> list[dict]: + """Flatten a bounding-box response ``{vessels: [...]}`` to markers.""" + if not payload or not isinstance(payload, dict): + return [] + rows = payload.get("vessels") or [] + out = [] + for row in rows: + marker = transform_vesselapi_position(row) + if marker: + out.append(marker) + return out + + +# ── Durable daily quota (Postgres, survives restarts) ───────────────────── +# Mirrors keystore.api_keys: lazy CREATE TABLE IF NOT EXISTS, no alembic fork. + +vesselapi_quota = Table( + "vesselapi_quota", + metadata, + Column("day", Date, primary_key=True), + Column("calls", Integer, nullable=False, server_default="0"), + Column("remaining", Integer, nullable=True), + Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False), +) + +_CREATE_QUOTA_SQL = text( + """ + CREATE TABLE IF NOT EXISTS vesselapi_quota ( + day DATE PRIMARY KEY, + calls INTEGER NOT NULL DEFAULT 0, + remaining INTEGER, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() + ) + """ +) + +_quota_lock = asyncio.Lock() +_quota_ensured = False + + +async def ensure_quota_table() -> None: + global _quota_ensured + if _quota_ensured: + return + async with _quota_lock: + if _quota_ensured: + return + async with engine.begin() as conn: + await conn.execute(_CREATE_QUOTA_SQL) + _quota_ensured = True + + +class PgQuotaStore: + """Postgres-backed daily call counter. Injected for tests.""" + + async def calls_today(self, day: date) -> int: + await ensure_quota_table() + async with async_session() as session: + row = (await session.execute( + select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day) + )).scalar() + return int(row) if row else 0 + + async def remaining_today(self, day: date) -> int | None: + await ensure_quota_table() + async with async_session() as session: + row = (await session.execute( + select(vesselapi_quota.c.remaining).where(vesselapi_quota.c.day == day) + )).scalar() + return int(row) if row is not None else None + + async def bump(self, day: date, remaining: int | None) -> int: + await ensure_quota_table() + now = datetime.now(timezone.utc) + async with async_session() as session: + existing = (await session.execute( + select(vesselapi_quota.c.calls).where(vesselapi_quota.c.day == day) + )).scalar() + if existing is None: + await session.execute( + vesselapi_quota.insert().values( + day=day, calls=1, remaining=remaining, updated_at=now, + ) + ) + else: + await session.execute( + vesselapi_quota.update() + .where(vesselapi_quota.c.day == day) + .values( + calls=vesselapi_quota.c.calls + 1, + remaining=remaining, + updated_at=now, + ) + ) + await session.commit() + return (int(existing) if existing else 0) + 1 + + +# ── Budget / scheduling (pure, unit-testable) ───────────────────────────── + +def days_left_in_month(now: datetime) -> int: + """UTC days remaining in the current month, inclusive of today.""" + _, last = calendar.monthrange(now.year, now.month) + return last - now.day + 1 + + +def budget_allows( + calls_today: int, + remaining: int | None, + days_left: int, + max_per_day: int, +) -> bool: + """True if another poll is permitted today. + + Local hard cap: fewer than ``max_per_day`` successful calls today. + Monthly floor: if ``X-RateLimit-Remaining`` is known, keep at least + ``max_per_day * days_left`` in reserve for the rest of the month. + """ + if calls_today >= max_per_day: + return False + if remaining is not None and remaining <= max_per_day * days_left: + return False + return True + + +def choose_box( + boxes: list[tuple[float, float, float, float]], + calls_today: int, + max_per_day: int, +) -> int: + """Index into ``boxes`` for the next poll. + + Prefer refreshing the first (primary) box rather than spraying one call + across every region — round-robin only when the remaining daily budget is + enough to cover all boxes. + """ + if len(boxes) <= 1: + return 0 + budget_left = max_per_day - calls_today + if budget_left >= len(boxes): + return calls_today % len(boxes) + return 0 + + +# ── HTTP / poll ─────────────────────────────────────────────────────────── + +async def _resolve_key() -> str: + from keystore import get_api_key + return ( + os.getenv("VESSELAPI_API_KEY") + or VESSELAPI_API_KEY + or (await get_api_key("VESSELAPI_API_KEY")) + or "" + ).strip() + + +async def _get_client() -> httpx.AsyncClient: + global _client + if _client is None: + async with _client_lock: + if _client is None: + _client = httpx.AsyncClient( + timeout=httpx.Timeout(15.0, connect=5.0), + follow_redirects=True, + headers={"User-Agent": OSINT_USER_AGENT, "Accept": "application/json"}, + limits=httpx.Limits(max_connections=1, max_keepalive_connections=1), + ) + return _client + + +async def close_client() -> None: + global _client + if _client is not None: + await _client.aclose() + _client = None + + +def _int_header(value: str | None) -> int | None: + if value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +async def poll_once(store, boxes: list[tuple[float, float, float, float]], key: str) -> bool: + """One quota-checked poll. Returns True if a successful 2xx was made. + + Only successful 2xx responses count against the monthly quota; 4xx/5xx/429 + are skipped without retry-storming (Retry-After respected by simply + sleeping the interval). + """ + now = datetime.now(timezone.utc) + today = now.date() + calls = await store.calls_today(today) + remaining = await store.remaining_today(today) + days_left = days_left_in_month(now) + if not budget_allows(calls, remaining, days_left, VESSELAPI_MAX_CALLS_PER_DAY): + logger.info( + "VesselAPI quota reached (calls_today=%d, remaining=%s, days_left=%d) — skip poll", + calls, remaining, days_left, + ) + return False + + idx = choose_box(boxes, calls, VESSELAPI_MAX_CALLS_PER_DAY) + minlat, minlon, maxlat, maxlon = boxes[idx] + client = await _get_client() + params = { + "filter.latBottom": str(minlat), + "filter.latTop": str(maxlat), + "filter.lonLeft": str(minlon), + "filter.lonRight": str(maxlon), + "pagination.limit": str(PAGE_LIMIT), + } + headers = {"Authorization": f"Bearer {key}"} + try: + resp = await client.get(ENDPOINT, params=params, headers=headers) + except httpx.HTTPError as exc: + logger.warning("VesselAPI request failed: %s", exc) + return False + + if resp.status_code == 429: + logger.warning( + "VesselAPI rate-limited (Retry-After=%s) — skip poll", + resp.headers.get("Retry-After"), + ) + return False + if resp.status_code >= 400: + logger.warning("VesselAPI HTTP %d — not counted against quota", resp.status_code) + return False + + # 2xx success — counts against the monthly quota. + remaining = _int_header(resp.headers.get("X-RateLimit-Remaining")) + calls = await store.bump(today, remaining) + try: + data = resp.json() + except ValueError: + logger.warning("VesselAPI 2xx with non-JSON body — counted but ignored") + return True + markers = transform_vesselapi_payload(data) + for m in markers: + await upsert_vessel(m) + logger.info( + "VesselAPI poll OK: %d vessels (remaining=%s, calls_today=%d)", + len(markers), remaining, calls, + ) + return True + + +# ── Worker loop ─────────────────────────────────────────────────────────── + +async def run_vesselapi_worker(store: PgQuotaStore | None = None) -> None: + """Long-lived poll loop. Idle when the key is unset; never crashes the app.""" + if store is None: + store = PgQuotaStore() + boxes = parse_boxes_validated(VESSELAPI_BBOX) + if not boxes: + logger.warning( + "VESSELAPI_BBOX has no valid boxes (span ≤ %.1f°) — poller idle", MAX_SPAN_DEG, + ) + while True: + try: + if not boxes: + await asyncio.sleep(VESSELAPI_INTERVAL) + continue + key = await _resolve_key() + if not key: + logger.warning( + "VESSELAPI_API_KEY not set — VesselAPI poller idle. " + "Create a free key at https://dashboard.vesselapi.com/" + ) + await asyncio.sleep(VESSELAPI_INTERVAL) + continue + await poll_once(store, boxes, key) + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 — keep the loop alive across transient failures + logger.exception("VesselAPI poll error") + await asyncio.sleep(VESSELAPI_INTERVAL) diff --git a/docker-compose.yml b/docker-compose.yml index 00fd71c..f908975 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -97,6 +97,11 @@ services: AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-} AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66} AISSTREAM_IN_INGEST: ${AISSTREAM_IN_INGEST:-0} + VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-} + VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2} + VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280} + VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5} + VESSELAPI_IN_INGEST: ${VESSELAPI_IN_INGEST:-0} command: ["python", "app/run_ingester.py"] entrypoint: ["python", "app/run_ingester.py"] @@ -131,6 +136,11 @@ services: AISSTREAM_API_KEY: ${AISSTREAM_API_KEY:-} AISSTREAM_BBOX: ${AISSTREAM_BBOX:-24,-125,50,-66} AISSTREAM_IN_APP: ${AISSTREAM_IN_APP:-1} + VESSELAPI_API_KEY: ${VESSELAPI_API_KEY:-} + VESSELAPI_BBOX: ${VESSELAPI_BBOX:-25.5,55.4,27.3,57.2} + VESSELAPI_INTERVAL: ${VESSELAPI_INTERVAL:-17280} + VESSELAPI_MAX_CALLS_PER_DAY: ${VESSELAPI_MAX_CALLS_PER_DAY:-5} + VESSELAPI_IN_APP: ${VESSELAPI_IN_APP:-1} ports: - "127.0.0.1:8000:8000" deploy: diff --git a/tests/test_vesselapi.py b/tests/test_vesselapi.py new file mode 100644 index 0000000..5891ca3 --- /dev/null +++ b/tests/test_vesselapi.py @@ -0,0 +1,298 @@ +"""Unit tests for the VesselAPI poller (no network, no DB). + +Covers box span validation, position → marker mapping, glitch skipping, +and the daily-quota gate (6th 2xx attempt skipped). ``poll_once`` is driven +with an in-memory fake quota store + fake HTTP client; ``upsert_vessel``'s +DB/WS side effects are monkeypatched to no-ops so markers can be asserted in +``vessel_last_known``. +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +import pytest + +import vesselapi +from vesselapi import ( + BboxError, + budget_allows, + choose_box, + days_left_in_month, + parse_boxes, + parse_boxes_validated, + poll_once, + transform_vesselapi_payload, + transform_vesselapi_position, + validate_bbox_span, +) +from live_layers import vessel_last_known + + +# ── Fake quota store (in-memory, injected) ──────────────────────────────── + +class FakeStore: + def __init__(self, calls: int = 0, remaining: int | None = None): + self.calls = calls + self.remaining = remaining + self.bumps = 0 + + async def calls_today(self, day): + return self.calls + + async def remaining_today(self, day): + return self.remaining + + async def bump(self, day, remaining): + self.calls += 1 + self.remaining = remaining + self.bumps += 1 + return self.calls + + +# ── Fake HTTP client ─────────────────────────────────────────────────────── + +class FakeResp: + def __init__(self, status: int = 200, headers: dict | None = None, data: dict | None = None): + self.status_code = status + self.headers = headers or {} + self._data = data or {} + + def json(self): + return self._data + + +class FakeClient: + def __init__(self, *responses: FakeResp): + self.responses = list(responses) + self.calls: list[tuple] = [] + + async def get(self, url, params=None, headers=None): + self.calls.append((url, params, headers)) + return self.responses.pop(0) + + async def aclose(self): + pass + + +def _sample_payload() -> dict: + return { + "vessels": [ + { + "mmsi": 422050100, + "imo": 9321483, + "vessel_name": "HORMUZ STAR", + "latitude": 26.5, + "longitude": 56.3, + "cog": 88.0, + "sog": 12.4, + "heading": 90, + "nav_status": 0, + "timestamp": "2026-08-29T12:00:00Z", + "suspected_glitch": False, + }, + { + "mmsi": 422050101, + "vessel_name": "GLITCHY", + "latitude": 26.6, + "longitude": 56.4, + "cog": 45.0, + "sog": 5.0, + "heading": 45, + "nav_status": 5, + "suspected_glitch": True, + }, + {"mmsi": 422050102, "vessel_name": "NOFIX"}, # no coords → drop + ], + "nextToken": "deadbeef", + } + + +# ── Box span / parsing ───────────────────────────────────────────────────── + +def test_validate_bbox_span_accepts_hormuz(): + validate_bbox_span(25.5, 55.4, 27.3, 57.2) # span 3.6 — no raise + + +def test_validate_bbox_span_rejects_conus(): + with pytest.raises(BboxError): + validate_bbox_span(24.0, -125.0, 50.0, -66.0) # span 85 + + +def test_validate_bbox_span_rejects_marine_regions_gazetteer_box(): + # 25.2732–27.3713 N, 55.1647–57.3419 E → span 4.28 > 4.0. + with pytest.raises(BboxError): + validate_bbox_span(25.2732, 55.1647, 27.3713, 57.3419) + + +def test_validate_bbox_span_rejects_inverted_axes(): + with pytest.raises(BboxError): + validate_bbox_span(27.0, 55.0, 25.0, 57.0) + + +def test_parse_boxes_semicolon_and_skip_malformed(): + boxes = parse_boxes("25.5,55.4,27.3,57.2; 10,20,11,21; garbage") + assert boxes == [(25.5, 55.4, 27.3, 57.2), (10.0, 20.0, 11.0, 21.0)] + + +def test_parse_boxes_validated_skips_over_span(): + # Second box is CONUS-sized → dropped, first kept. + valid = parse_boxes_validated("25.5,55.4,27.3,57.2;24,-125,50,-66") + assert valid == [(25.5, 55.4, 27.3, 57.2)] + + +# ── Position → marker mapping ───────────────────────────────────────────── + +def test_transform_position_maps_shared_marker_contract(): + m = transform_vesselapi_position({ + "mmsi": 422050100, "imo": 9321483, "vessel_name": "HORMUZ STAR", + "latitude": 26.5, "longitude": 56.3, "heading": 90, "cog": 88.0, + "sog": 12.4, "nav_status": 0, "timestamp": "2026-08-29T12:00:00Z", + "suspected_glitch": False, + }) + assert m is not None + assert m["id"] == "422050100" + assert m["lat"] == 26.5 + assert m["lon"] == 56.3 + assert m["label"] == "HORMUZ STAR" + assert m["heading"] == 90 + assert m["speed"] == 12.4 + assert m["extra"]["src"] == "vesselapi" + assert m["extra"]["mmsi"] == "422050100" + assert m["extra"]["imo"] == 9321483 + assert m["extra"]["navstat"] == 0 + assert m["extra"]["cog"] == 88.0 + assert m["extra"]["sog"] == 12.4 + assert m["extra"]["timestamp"] == "2026-08-29T12:00:00Z" + + +def test_transform_position_heading_falls_back_to_cog(): + m = transform_vesselapi_position({ + "mmsi": 123456789, "vessel_name": "X", "latitude": 1.0, "longitude": 2.0, + "heading": None, "cog": 123.4, "sog": 5.0, + }) + assert m["heading"] == 123.4 + + +def test_transform_position_skips_glitch(): + assert transform_vesselapi_position({ + "mmsi": 123456789, "latitude": 1.0, "longitude": 2.0, + "suspected_glitch": True, + }) is None + + +def test_transform_position_skips_missing_coords(): + assert transform_vesselapi_position({"mmsi": 123456789, "vessel_name": "NOFIX"}) is None + + +def test_transform_payload_skips_glitch_and_nofix_rows(): + rows = transform_vesselapi_payload(_sample_payload()) + assert [r["id"] for r in rows] == ["422050100"] + + +# ── Quota budget / scheduling ───────────────────────────────────────────── + +def test_days_left_in_month(): + assert days_left_in_month(datetime(2026, 8, 29, tzinfo=timezone.utc)) == 3 + + +def test_budget_allows_local_daily_cap(): + # 5 calls already made → 6th is blocked regardless of remaining. + assert budget_allows(5, remaining=1000, days_left=3, max_per_day=5) is False + + +def test_budget_allows_monthly_floor(): + # remaining 14 ≤ 5*3=15 → skip; 16 > 15 → allow. + assert budget_allows(2, remaining=14, days_left=3, max_per_day=5) is False + assert budget_allows(2, remaining=16, days_left=3, max_per_day=5) is True + + +def test_budget_allows_unknown_remaining(): + assert budget_allows(2, remaining=None, days_left=3, max_per_day=5) is True + + +def test_choose_box_prefers_primary_when_budget_tight(): + boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)] + # 4 calls made, 1 left → always box 0. + assert choose_box(boxes, 4, max_per_day=5) == 0 + + +def test_choose_box_round_robins_when_budget_covers_all(): + boxes = [(1, 1, 2, 2), (3, 3, 4, 4), (5, 5, 6, 6)] + # 0 calls made, 5 left ≥ 3 boxes → round-robin. + assert choose_box(boxes, 0, max_per_day=5) == 0 + assert choose_box(boxes, 1, max_per_day=5) == 1 + assert choose_box(boxes, 2, max_per_day=5) == 2 + + +# ── poll_once integration (fake store + fake client) ───────────────────── + +def _patch_side_effects(monkeypatch): + async def _noop(*a, **k): + return None + + monkeypatch.setattr("tracks.record_position", _noop) + monkeypatch.setattr("geofence.record_and_notify", _noop) + + +def test_poll_once_lands_markers_in_vessel_last_known(monkeypatch): + _patch_side_effects(monkeypatch) + vessel_last_known.clear() + client = FakeClient( + FakeResp(200, {"X-RateLimit-Remaining": "140"}, _sample_payload()), + ) + monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client)) + store = FakeStore() + + ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key")) + + assert ok is True + assert store.calls == 1 + assert "422050100" in vessel_last_known + assert vessel_last_known["422050100"]["extra"]["src"] == "vesselapi" + assert "422050101" not in vessel_last_known # glitch skipped + # One HTTP call, bounding-box params, no sat / no nextToken follow. + assert len(client.calls) == 1 + _url, params, headers = client.calls[0] + assert params["filter.latBottom"] == "25.5" + assert params["filter.latTop"] == "27.3" + assert params["filter.lonLeft"] == "55.4" + assert params["filter.lonRight"] == "57.2" + assert params["pagination.limit"] == "50" + assert "sat" not in params + assert headers["Authorization"] == "Bearer test-key" + + +def test_poll_once_sixth_2xx_is_skipped_with_zero_http(monkeypatch): + client = FakeClient() + monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client)) + # 5 successful calls already today → 6th poll makes no HTTP request. + store = FakeStore(calls=5, remaining=1000) + + ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key")) + + assert ok is False + assert store.bumps == 0 + assert client.calls == [] + + +def test_poll_once_4xx_not_counted_and_no_upsert(monkeypatch): + _patch_side_effects(monkeypatch) + vessel_last_known.clear() + client = FakeClient(FakeResp(400, {}, {"error": {}})) + monkeypatch.setattr(vesselapi, "_get_client", _make_get_client(client)) + store = FakeStore() + + ok = asyncio.run(poll_once(store, [(25.5, 55.4, 27.3, 57.2)], "test-key")) + + assert ok is False + assert store.bumps == 0 # 4xx does not count against quota + assert vessel_last_known == {} + + +def _make_get_client(client): + async def _get_client(): + return client + + return _get_client