Merge pull request #920 from bilawalsidhu/feat/area-vessel-requests

feat: ask for vessels by area

scripts/package-boundaries.json

@@ -871,6 +871,8 @@
       "src/layers/vessels/evidence.js",
       "src/layers/vessels/index.js",
       "src/layers/vessels/ingestion.js",
+      "src/layers/vessels/cameraArea.js",
+      "src/layers/vessels/viewArea.js",
       "src/layers/vessels/lifecycle.js",
       "src/layers/vessels/policy.js",
       "src/layers/vessels/queries.js",
@@ -1724,7 +1726,9 @@
       "src/layers/vessels/cards.js",
       "src/layers/vessels/evidence.js",
       "src/layers/vessels/index.js",
+      "src/layers/vessels/cameraArea.js",
       "src/layers/vessels/ingestion.js",
+      "src/layers/vessels/viewArea.js",
       "src/layers/vessels/lifecycle.js",
       "src/layers/vessels/policy.js",
       "src/layers/vessels/queries.js",
@@ -2394,7 +2398,9 @@
       "src/layers/vessels/cards.js",
       "src/layers/vessels/evidence.js",
       "src/layers/vessels/index.js",
+      "src/layers/vessels/cameraArea.js",
       "src/layers/vessels/ingestion.js",
+      "src/layers/vessels/viewArea.js",
       "src/layers/vessels/lifecycle.js",
       "src/layers/vessels/policy.js",
       "src/layers/vessels/queries.js",
@@ -2743,6 +2749,7 @@
     "exports": ["./layers/vessels/ingestion"],
     "modules": [
       "src/layers/vessels/ingestion.js",
+      "src/layers/vessels/viewArea.js",
       "src/layers/vessels/recordPolicy.js"
     ],
     "external": []

src/layers/vessels/cameraArea.js

@@ -0,0 +1,33 @@
+import * as Cesium from 'cesium';
+import { viewAreaRadiusKm } from './viewArea.js';
+
+/**
+ * The area a viewer is looking at, as `{ lat, lon, radiusKm }`: centered on
+ * the point in the middle of the screen (or below the camera when the middle
+ * shows sky), sized by the camera's height. Null for a view that wants every
+ * vessel.
+ */
+export function vesselViewArea(viewer) {
+  const camera = viewer?.camera;
+  const scene = viewer?.scene;
+  const height = camera?.positionCartographic?.height;
+  const radiusKm = viewAreaRadiusKm(height);
+  if (radiusKm == null) return null;
+  let center = camera.positionCartographic;
+  const canvas = scene?.canvas;
+  if (canvas?.clientWidth && canvas?.clientHeight) {
+    try {
+      const hit = camera.pickEllipsoid(
+        new Cesium.Cartesian2(canvas.clientWidth / 2, canvas.clientHeight / 2),
+        scene.globe?.ellipsoid,
+      );
+      if (hit) center = Cesium.Cartographic.fromCartesian(hit);
+    } catch {
+      /* the camera position stands in */
+    }
+  }
+  const lat = Cesium.Math.toDegrees(center.latitude);
+  const lon = Cesium.Math.toDegrees(center.longitude);
+  if (!Number.isFinite(lat) || !Number.isFinite(lon)) return null;
+  return { lat, lon, radiusKm };
+}

src/layers/vessels/index.js

@@ -1,6 +1,7 @@
 import { createVesselState } from './state.js';
 import { createLifecycle } from './lifecycle.js';
 import { createIngestion } from './ingestion.js';
+import { vesselViewArea } from './cameraArea.js';
 import { createVesselSnapshotRenderer } from './snapshotRenderer.js';
 import { createRendering } from './rendering.js';
 import { createSelection } from './selection.js';
@@ -36,6 +37,7 @@ export function createVesselLayer({ source, services, options = {} } = {}) {
     feed: vesselState.state.feed,
     readSource: () => vesselState._source,
     readViewer: () => vesselState.state.viewer,
+    readArea: vesselViewArea,
     getRowLimit: parts.rendering.renderRowLimit,
     readCount: () => vesselState.state.records.all.length,
     applyRows: parts.snapshots.reconcileVessels,

src/layers/vessels/ingestion.js

@@ -1,10 +1,12 @@
 import { AIS_FIRST_CONNECT_LABEL } from './recordPolicy.js';
+import { viewAreaMovedEnough } from './viewArea.js';
 
 /** Own source requests and classified feed state through explicit operations. */
 export function createIngestion({
   feed,
   readSource,
   readViewer,
+  readArea = () => null,
   getRowLimit,
   readCount,
   applyRows,
@@ -34,8 +36,11 @@ export function createIngestion({
               AbortSignal.timeout(10000),
             ])
           : requestController.signal;
+      // Ask for the vessels in view; a wide view asks for every vessel.
+      const area = readArea(viewer);
+      feed.lastArea = area;
       const snapshot = await readSource().getSnapshot(
-        { maxRows: getRowLimit() },
+        { maxRows: getRowLimit(), ...(area ? { area } : {}) },
         { signal },
       );
       if (!ownsAisRequest(requestController, requestSessionId)) return;
@@ -169,6 +174,15 @@ export function createIngestion({
           : new Date(record.observedAtMs).toISOString(),
     };
   }
+  /** Ask again when the view moved away from the last area asked for. */
+  function refreshIfMoved(viewer) {
+    if (!feed.enabled || feed.loading) return Promise.resolve();
+    const active = viewer || readViewer();
+    if (!viewAreaMovedEnough(feed.lastArea ?? null, readArea(active)))
+      return Promise.resolve();
+    return loadLivePositions(active);
+  }
+
   const methods = {
     update(viewer) {
       if (!feed.enabled) return Promise.resolve();
@@ -178,6 +192,7 @@ export function createIngestion({
 
   return {
     loadLivePositions,
+    refreshIfMoved,
     ownsAisRequest,
     applyAisFeedSnapshot,
     vesselDisplayRow,
@@ -209,5 +224,9 @@ export function createVesselFeed() {
     firstConnectDeadline: null,
     firstConnectTimer: null,
     abort: null,
+    /** The area last asked for, null for every vessel. */
+    lastArea: null,
+    /** Stops listening for camera moves. */
+    removeViewListener: null,
   };
 }

src/layers/vessels/ingestion.test.mjs

@@ -104,3 +104,46 @@ test('vessel ingestion converts source units once and retains warm records on a
   assert.equal(probe.feed.loading, false);
   assert.equal(probe.feed.abort, null);
 });
+
+test('vessels are asked for by the area in view, and again only when the view moves away', async () => {
+  const asked = [];
+  let area = { lat: 37.8, lon: -122.4, radiusKm: 40 };
+  const feed = createVesselFeed();
+  feed.enabled = true;
+  feed.sessionId = 1;
+  const ingestion = createIngestion({
+    feed,
+    readSource: () => ({
+      getSnapshot: async (query) => {
+        asked.push(query);
+        return { records: [], source: 'Feed' };
+      },
+    }),
+    readViewer: () => ({}),
+    readArea: () => area,
+    getRowLimit: () => 500,
+    readCount: () => 0,
+    now: () => 0,
+    setSourceLabel: () => {},
+    applyRows: () => {},
+    classifySnapshot: () => ({
+      acceptedRows: [],
+      acceptedRowCount: 0,
+      rawRowCount: 0,
+    }),
+    isDefinitiveTransportFailure: () => false,
+    isGraceEligibleTransport: () => false,
+    markUnavailable: () => {},
+    settleFirstConnect: () => {},
+  });
+  await ingestion.methods.update();
+  assert.deepEqual(asked, [{ maxRows: 500, area }]);
+  await ingestion.refreshIfMoved();
+  assert.equal(asked.length, 1);
+  area = { ...area, lat: 38.2 };
+  await ingestion.refreshIfMoved();
+  assert.deepEqual(asked.at(-1), { maxRows: 500, area });
+  area = null;
+  await ingestion.refreshIfMoved();
+  assert.deepEqual(asked.at(-1), { maxRows: 500 });
+});

src/layers/vessels/lifecycle.js

@@ -196,10 +196,18 @@ export function createLifecycle({
         state.records.byMmsi.has(pickedId),
       );
       restoreSpriteOrderOnEnable('ais', activeViewer);
+      // Vessels are asked for by view, so ask again when the view moves away.
+      removeViewListener();
+      const moveEnd = activeViewer?.camera?.moveEnd;
+      if (typeof moveEnd?.addEventListener === 'function')
+        state.feed.removeViewListener = moveEnd.addEventListener(() => {
+          components.ingestion.refreshIfMoved(activeViewer);
+        });
       return components.ingestion.loadLivePositions(activeViewer);
     },
 
     disable() {
+      removeViewListener();
       state.feed.enabled = false;
       invalidateAisSession();
       releaseContinuousRender('ais-vessels');
@@ -218,6 +226,7 @@ export function createLifecycle({
     },
 
     destroy(viewer) {
+      removeViewListener();
       const activeViewer = viewer || state.viewer;
       invalidateAisSession();
       releaseContinuousRender('ais-vessels'); // direct-destroy path (perf wave 2 fix)
@@ -241,6 +250,12 @@ export function createLifecycle({
     },
   };
 
+  function removeViewListener() {
+    const remove = state.feed.removeViewListener;
+    state.feed.removeViewListener = null;
+    if (typeof remove === 'function') remove();
+  }
+
   return {
     clearFirstConnectTimer,
     invalidateAisSession,

src/layers/vessels/viewArea.js

@@ -0,0 +1,31 @@
+/** Smallest area asked for around the view, in kilometres. */
+export const VIEW_AREA_MIN_RADIUS_KM = 10;
+/** Largest area asked for; a wider view asks for every vessel instead. */
+export const VIEW_AREA_MAX_RADIUS_KM = 450;
+
+/**
+ * The radius worth asking for from a camera this high, in kilometres: about
+ * the height, so the area covers what a camera looking down shows. Null when
+ * the view is wide enough that every vessel is wanted.
+ */
+export function viewAreaRadiusKm(heightM) {
+  if (!Number.isFinite(heightM)) return null;
+  const radiusKm = heightM / 1000;
+  if (radiusKm > VIEW_AREA_MAX_RADIUS_KM) return null;
+  return Math.max(VIEW_AREA_MIN_RADIUS_KM, radiusKm);
+}
+
+/**
+ * Whether a view moved far enough from the last area asked for to ask
+ * again: its radius changed by a quarter, or its center moved a quarter of
+ * the radius. Switching between an area and every vessel always counts.
+ */
+export function viewAreaMovedEnough(previous, next) {
+  if (!previous || !next) return previous !== next;
+  if (Math.abs(next.radiusKm - previous.radiusKm) >= previous.radiusKm * 0.25)
+    return true;
+  const lat = ((previous.lat + next.lat) / 2) * (Math.PI / 180);
+  const dLat = (next.lat - previous.lat) * 111.32;
+  const dLon = (next.lon - previous.lon) * 111.32 * Math.cos(lat);
+  return Math.hypot(dLat, dLon) >= previous.radiusKm * 0.25;
+}

src/layers/vessels/viewArea.test.mjs

@@ -0,0 +1,30 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {
+  VIEW_AREA_MAX_RADIUS_KM,
+  VIEW_AREA_MIN_RADIUS_KM,
+  viewAreaMovedEnough,
+  viewAreaRadiusKm,
+} from './viewArea.js';
+
+test('the area asked for follows the camera height, and a wide view asks for every vessel', () => {
+  assert.equal(viewAreaRadiusKm(2_000), VIEW_AREA_MIN_RADIUS_KM);
+  assert.equal(viewAreaRadiusKm(80_000), 80);
+  assert.equal(
+    viewAreaRadiusKm(VIEW_AREA_MAX_RADIUS_KM * 1000),
+    VIEW_AREA_MAX_RADIUS_KM,
+  );
+  assert.equal(viewAreaRadiusKm(VIEW_AREA_MAX_RADIUS_KM * 1000 + 1), null);
+  assert.equal(viewAreaRadiusKm(Number.NaN), null);
+});
+
+test('a view asks again only after moving or zooming a quarter of its radius', () => {
+  const here = { lat: 37.8, lon: -122.4, radiusKm: 40 };
+  assert.equal(viewAreaMovedEnough(null, here), true);
+  assert.equal(viewAreaMovedEnough(here, null), true);
+  assert.equal(viewAreaMovedEnough(null, null), false);
+  assert.equal(viewAreaMovedEnough(here, { ...here, lat: 37.85 }), false);
+  assert.equal(viewAreaMovedEnough(here, { ...here, lat: 37.9 }), true);
+  assert.equal(viewAreaMovedEnough(here, { ...here, radiusKm: 45 }), false);
+  assert.equal(viewAreaMovedEnough(here, { ...here, radiusKm: 55 }), true);
+});

src/sources/live/standalone.js

@@ -172,9 +172,20 @@ export function createAisStreamSource({
 } = {}) {
   return {
     label: 'AISStream',
-    async getSnapshot({ maxRows = 12000 } = {}, { signal } = {}) {
+    async getSnapshot({ maxRows = 12000, area } = {}, { signal } = {}) {
       const url = new URL(apiUrl, origin());
       url.searchParams.set('maxRows', String(maxRows));
+      // An area asks for the vessels around a point; servers that hold every
+      // vessel may ignore it, since callers still keep only what they need.
+      if (
+        Number.isFinite(area?.lat) &&
+        Number.isFinite(area?.lon) &&
+        Number.isFinite(area?.radiusKm)
+      ) {
+        url.searchParams.set('lat', area.lat.toFixed(5));
+        url.searchParams.set('lon', area.lon.toFixed(5));
+        url.searchParams.set('radius_km', area.radiusKm.toFixed(1));
+      }
       const { response, payload } = await readResponse(
         fetchImpl,
         url.toString(),

src/tools/queries/maritime.js

@@ -6,6 +6,7 @@ import {
   AREA_SCHEMA,
   areaCenter,
   areaContains,
+  areaRadiusKm,
   distanceKm,
   resolveArea,
 } from '../area.js';
@@ -48,12 +49,29 @@ function vesselRow(record, from) {
 }
 
 /**
- * Every vessel the source retains. The source's default is the newest few
- * thousand, which can leave out a vessel the server still holds; asking for
- * more than any server keeps lets the server return all of them.
+ * Every vessel the source retains, or with `area`, every vessel around it.
+ * The source's default is the newest few thousand, which can leave out a
+ * vessel the server still holds; asking for more than any server keeps lets
+ * the server return all of them. The area lets a server answer for that
+ * place alone; callers still keep only the vessels inside it.
  */
-function readVessels(services, signal) {
-  return services.vessels.getSnapshot({ maxRows: ALL_VESSELS }, { signal });
+function readVessels(services, signal, area) {
+  const center = area ? areaCenter(area) : null;
+  return services.vessels.getSnapshot(
+    {
+      maxRows: ALL_VESSELS,
+      ...(center
+        ? {
+            area: {
+              lat: center.lat,
+              lon: center.lon,
+              radiusKm: areaRadiusKm(area),
+            },
+          }
+        : {}),
+    },
+    { signal },
+  );
 }
 
 function snapshotInfo(snapshot) {
@@ -90,7 +108,7 @@ export const vesselsInArea = defineTool({
   async run(args, { services, signal }) {
     const area = await resolveArea(args.area, { services, signal });
     const center = areaCenter(area);
-    const snapshot = await readVessels(services, signal);
+    const snapshot = await readVessels(services, signal, area);
     const wanted = args.type?.toLowerCase();
     const rows = snapshot.records
       .filter(

src/tools/queries/maritime.test.mjs

@@ -201,3 +201,26 @@ test('vessel searches ask for every vessel the server retains', async () => {
   for (const maxRows of requested)
     assert.ok(Number(maxRows) >= AISSTREAM_CACHE_MAX, maxRows);
 });
+
+test('an area search names its area to the server; a vessel lookup asks for every vessel', async () => {
+  const requested = [];
+  const source = createAisStreamSource({
+    origin: () => 'http://localhost',
+    fetchImpl: async (url) => {
+      requested.push(new URL(url).searchParams);
+      return Response.json({ rows: [], source: 'AISStream', status: 'live' });
+    },
+  });
+  const catalog = composeCatalog({
+    tools: coreTools,
+    services: { vessels: source },
+  });
+  await catalog.call('vessels_in_area', { area: bay });
+  await catalog.call('find_vessel', { mmsi: '366999712' });
+  const [area, lookup] = requested;
+  assert.equal(area.get('lat'), '37.78000');
+  assert.equal(area.get('lon'), '-122.40000');
+  assert.equal(area.get('radius_km'), '20.0');
+  assert.equal(lookup.has('lat'), false);
+  assert.equal(lookup.has('radius_km'), false);
+});