app/chokepoints.py

@@ -0,0 +1,62 @@
+"""Static chokepoint preset catalog — one-tap fly-to targets for the map.
+
+Pure data, no upstream calls and no VesselAPI quota spend. ``vesselapi`` is
+``True`` only for Hormuz (the single box the VesselAPI poller already covers);
+every other strait is AISStream-only until a human later spends quota. Never
+call VesselAPI from here.
+
+Bounding boxes are ``minlat,minlon,maxlat,maxlon`` (VesselAPI order) and each
+stays within the ``|dLat|+|dLon| <= 4`` span rule enforced by
+``vesselapi.validate_bbox_span``.
+"""
+
+from __future__ import annotations
+
+# id → preset. ``center`` is ``[lat, lon]`` for Leaflet ``setView``.
+_CHOKEPOINTS: tuple[dict, ...] = (
+    {
+        "id": "hormuz",
+        "title": "Strait of Hormuz",
+        "bbox": "25.5,55.4,27.3,57.2",
+        "center": [26.4, 56.5],
+        "zoom": 9,
+        "vesselapi": True,
+    },
+    {
+        "id": "bab_el_mandeb",
+        "title": "Bab el-Mandeb",
+        "bbox": "12.0,42.8,13.5,44.3",
+        "center": [12.7, 43.4],
+        "zoom": 9,
+        "vesselapi": False,
+    },
+    {
+        "id": "suez",
+        "title": "Suez / N. Red Sea",
+        "bbox": "29.5,32.0,31.0,33.5",
+        "center": [30.0,32.5],
+        "zoom": 9,
+        "vesselapi": False,
+    },
+    {
+        "id": "malacca",
+        "title": "Malacca / Singapore",
+        "bbox": "1.0,103.0,2.5,104.5",
+        "center": [1.3, 103.8],
+        "zoom": 9,
+        "vesselapi": False,
+    },
+    {
+        "id": "taiwan",
+        "title": "Taiwan Strait",
+        "bbox": "23.5,119.0,25.0,120.5",
+        "center": [24.2, 119.8],
+        "zoom": 9,
+        "vesselapi": False,
+    },
+)
+
+
+def chokepoints() -> list[dict]:
+    """Return a fresh copy of the catalog (callers must not mutate the source)."""
+    return [dict(p) for p in _CHOKEPOINTS]

app/live_layers.py

@@ -957,10 +957,16 @@ async def fetch_trains(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dic
     return rows[:limit]
 
 
-async def fetch_vessels(bbox: str | None, limit: int = DEFAULT_LIMIT) -> list[dict]:
+async def fetch_vessels(
+    bbox: str | None,
+    limit: int = DEFAULT_LIMIT,
+    src: str | None = None,
+) -> list[dict]:
     async with vessel_lock:
         rows = [dict(v) for v in vessel_last_known.values()
                 if v.get("lat") is not None and v.get("lon") is not None]
+    if src and src != "all":
+        rows = [r for r in rows if (r.get("extra") or {}).get("src") == src]
     if bbox:
         minlon, minlat, maxlon, maxlat = parse_bbox(bbox)
         return filter_points_bbox(rows, minlon, minlat, maxlon, maxlat, limit)

app/main.py

@@ -1400,6 +1400,17 @@ async def map_layers():
     return {"layers": MAP_LAYERS, "overlays": overlay_catalog()}
 
 
+@app.get("/api/map/chokepoints")
+async def map_chokepoints():
+    """Static one-tap fly-to presets (Strait of Hormuz, Bab el-Mandeb, …).
+
+    Pure catalog — no upstream calls and no VesselAPI quota spend. ``vesselapi``
+    is True only for Hormuz (the box the VesselAPI poller already covers).
+    """
+    from chokepoints import chokepoints
+    return {"chokepoints": chokepoints()}
+
+
 def _upstream_or_502(exc: Exception, name: str) -> NoReturn:
     logger.warning("live_layer_upstream_failed", layer=name, error=str(exc))
     raise HTTPException(502, f"{name} upstream unavailable: {exc}") from exc
@@ -1499,13 +1510,18 @@ async def list_vessels(
     bbox: str | None = Query(None, description="minlon,minlat,maxlon,maxlat"),
     limit: int = Query(2000, ge=1, le=5000),
     timestamp: str | None = Query(None, description="ISO time — DVR 1-min tracks instead of live"),
+    src: str | None = Query(None, description="aisstream|vesselapi|all (default all)"),
 ):
     """AIS last-known — union of two independent providers.
 
     AISStream (extra.src="aisstream", live US-coast WebSocket) and VesselAPI
     (extra.src="vesselapi", Strait of Hormuz 5×/day poll) both upsert into the
     same store. Empty without either key / until the first successful poll.
+    ``src`` filters the union to one provider (default ``all``) so a Hormuz
+    view can skip the ~5k CONUS AISStream rows.
     """
+    if src is not None and src not in ("aisstream", "vesselapi", "all"):
+        raise HTTPException(422, "src must be one of: aisstream, vesselapi, all")
     if bbox:
         _parse_bbox_query(bbox)
     try:
@@ -1513,7 +1529,7 @@ async def list_vessels(
         ts = parse_timestamp(timestamp)
         if ts is not None:
             return overlay_json(await fetch_positions_at("vessel", ts, bbox, limit), 5)
-        return overlay_json(await fetch_vessels(bbox, limit), 5)
+        return overlay_json(await fetch_vessels(bbox, limit, src=src), 5)
     except ValueError as exc:
         raise HTTPException(422, str(exc)) from exc
 

tests/test_chokepoints.py

@@ -0,0 +1,122 @@
+"""Tests for the chokepoint preset catalog + vessels ``src=`` filter.
+
+- Span: every catalog box passes VesselAPI's ``|dLat|+|dLon| <= 4`` validator.
+- Catalog: ``GET /api/map/chokepoints`` returns 200 with the documented shape.
+- Vessels filter: ``GET /api/vessels?src=`` narrows the union store by provider.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+
+from chokepoints import chokepoints
+from live_layers import fetch_vessels, vessel_last_known
+from main import app
+from vesselapi import validate_bbox_span
+
+BASE = "http://test"
+
+
+async def _get(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+# ── Span validation (VesselAPI rule) ──────────────────────────────────────
+
+def test_all_catalog_boxes_within_span() -> None:
+    for preset in chokepoints():
+        minlat, minlon, maxlat, maxlon = (float(p) for p in preset["bbox"].split(","))
+        dlat = abs(maxlat - minlat)
+        dlon = abs(maxlon - minlon)
+        assert dlat + dlon <= 4.0, preset["id"]
+        validate_bbox_span(minlat, minlon, maxlat, maxlon)  # no raise
+
+
+# ── Catalog API contract ──────────────────────────────────────────────────
+
+def test_chokepoints_catalog_shape() -> None:
+    resp = asyncio.run(_get("/api/map/chokepoints"))
+    assert resp.status_code == 200
+    body = resp.json()
+    assert set(body) == {"chokepoints"}
+    rows = body["chokepoints"]
+    assert [r["id"] for r in rows] == [
+        "hormuz", "bab_el_mandeb", "suez", "malacca", "taiwan",
+    ]
+    for r in rows:
+        assert set(r) == {"id", "title", "bbox", "center", "zoom", "vesselapi"}
+        assert isinstance(r["center"], list) and len(r["center"]) == 2
+        assert r["zoom"] == 9
+        assert isinstance(r["vesselapi"], bool)
+        # bbox is minlat,minlon,maxlat,maxlon
+        minlat, minlon, maxlat, maxlon = (float(p) for p in r["bbox"].split(","))
+        assert minlat < maxlat and minlon < maxlon
+
+
+def test_only_hormuz_is_vesselapi() -> None:
+    rows = chokepoints()
+    by_id = {r["id"]: r for r in rows}
+    assert by_id["hormuz"]["vesselapi"] is True
+    for cid in ("bab_el_mandeb", "suez", "malacca", "taiwan"):
+        assert by_id[cid]["vesselapi"] is False
+
+
+# ── Vessels src= filter (mocked store) ────────────────────────────────────
+
+def _seed_store() -> None:
+    vessel_last_known.clear()
+    vessel_last_known["422050100"] = {
+        "id": "422050100", "lat": 26.5, "lon": 56.3, "label": "HORMUZ STAR",
+        "extra": {"src": "vesselapi", "mmsi": "422050100"},
+    }
+    vessel_last_known["366001230"] = {
+        "id": "366001230", "lat": 35.0, "lon": -79.0, "label": "CONUS SHIP",
+        "extra": {"src": "aisstream", "mmsi": "366001230"},
+    }
+    vessel_last_known["366001231"] = {
+        "id": "366001231", "lat": 36.0, "lon": -78.0, "label": "CONUS SHIP 2",
+        "extra": {"src": "aisstream", "mmsi": "366001231"},
+    }
+
+
+def test_fetch_vessels_src_filters() -> None:
+    _seed_store()
+    assert {v["id"] for v in asyncio.run(fetch_vessels(None, src="vesselapi"))} == {"422050100"}
+    assert {v["id"] for v in asyncio.run(fetch_vessels(None, src="aisstream"))} == {
+        "366001230", "366001231",
+    }
+    assert len(asyncio.run(fetch_vessels(None, src="all"))) == 3
+    assert len(asyncio.run(fetch_vessels(None))) == 3  # default all
+
+
+def test_vessels_src_query_param(monkeypatch) -> None:
+    _seed_store()
+
+    async def _fake_fetch(bbox, limit, src=None):
+        rows = [
+            {"id": k, **{kk: v[kk] for kk in ("lat", "lon", "label", "extra")}}
+            for k, v in vessel_last_known.items()
+        ]
+        if src and src != "all":
+            rows = [r for r in rows if (r.get("extra") or {}).get("src") == src]
+        return rows
+
+    monkeypatch.setattr("main.fetch_vessels", _fake_fetch)
+
+    body = asyncio.run(_get("/api/vessels?src=vesselapi")).json()
+    assert [r["id"] for r in body] == ["422050100"]
+
+    body = asyncio.run(_get("/api/vessels?src=aisstream")).json()
+    assert {r["id"] for r in body} == {"366001230", "366001231"}
+
+    body = asyncio.run(_get("/api/vessels?src=all")).json()
+    assert len(body) == 3
+
+
+def test_vessels_src_rejects_bad_value() -> None:
+    resp = asyncio.run(_get("/api/vessels?src=marine-traffic"))
+    assert resp.status_code == 422