merge: integrate GFS wind PR #459 into current catalog owners

server/providers/local.js

@@ -19,6 +19,7 @@ import { trackBackfillProxies } from './aircraft/tracks.js';
 import { openAiRealtimeProxy } from './openai.js';
 import { googlePlacesContextProxy } from './places.js';
 import { keySetupEndpoint } from '../standalone/key-setup.js';
+import { windProxy } from './wind.js';
 
 /** Construct the local provider plugins in their established order. */
 function localProviderPlugins() {
@@ -43,6 +44,7 @@ function localProviderPlugins() {
     trackBackfillProxies(),
     openAiRealtimeProxy(),
     googlePlacesContextProxy(),
+    windProxy(),
     keySetupEndpoint(),
   ];
 }

server/providers/wind.js

@@ -0,0 +1,170 @@
+import {
+  GFS_BUCKET,
+  gfsObjectKey,
+  selectLatestGfsCycle,
+} from './wind/catalog.js';
+import {
+  fetchText,
+  fetchRange,
+  parseGfsIdx,
+  windMessageRanges,
+} from './wind/gfs.js';
+import { decodeWindGribMessage } from './wind/decode.js';
+import { resampleWindGrid } from './wind/grid.js';
+
+/**
+ * NOAA GFS 10 m wind proxy.
+ *
+ * Selects the latest available GFS cycle, reads its `.idx` inventory, byte-range
+ * fetches only the UGRD/VGRD 10 m GRIB2 messages, decodes them with ecCodes
+ * (WASM), resamples to a compact grid, and serves a manifest plus a Float32
+ * grid (U values then V values). Keyless; cached per cycle for an hour.
+ *
+ * Routes (mounted at `/api/wind`):
+ *   GET /api/wind/manifest   → manifest JSON
+ *   GET /api/wind/grid/<id>.bin → Float32 U…V payload
+ *   GET /api/wind/status     → manifest without `gridUrl`
+ *
+ * @param {{fetchImpl?: Function, now?: Function, decodeImpl?: Function,
+ *   targetDx?: number, ttlMs?: number}} [options]
+ * @returns {import('vite').Plugin}
+ */
+export function windProxy({
+  fetchImpl = fetch,
+  now = () => Date.now(),
+  decodeImpl = decodeWindGribMessage,
+  targetDx = 1,
+  ttlMs = 3600_000,
+} = {}) {
+  /** @type {?{id: string, idGrid: string, fetchedAt: number, manifest: object, grid: ?object}} */
+  let cached = null;
+  /** @type {?Promise<object>} */
+  let loading = null;
+
+  const sendJson = (res, value, status = 200) => {
+    res.writeHead(status, {
+      'Content-Type': 'application/json',
+      'Cache-Control': 'no-store',
+    });
+    res.end(JSON.stringify(value));
+  };
+
+  const cycleRunIso = (cycle) =>
+    `${cycle.date.slice(0, 4)}-${cycle.date.slice(4, 6)}-${cycle.date.slice(6)}T${String(cycle.hour).padStart(2, '0')}:00:00.000Z`;
+
+  async function refresh() {
+    const cycle = selectLatestGfsCycle(now());
+    const id = `${cycle.date}-${cycle.hour}`;
+    if (cached && cached.id === id && now() - cached.fetchedAt < ttlMs)
+      return cached;
+    if (loading) return loading;
+    loading = (async () => {
+      try {
+        const base = `https://${GFS_BUCKET}.s3.amazonaws.com/${gfsObjectKey(cycle)}`;
+        const index = await fetchText({ url: `${base}.idx`, fetchImpl });
+        const ranges = windMessageRanges(parseGfsIdx(index.toString()));
+        const [uBuffer, vBuffer] = await Promise.all([
+          fetchRange({ url: base, ...ranges.u, fetchImpl }),
+          fetchRange({ url: base, ...ranges.v, fetchImpl }),
+        ]);
+        const [u, v] = await Promise.all([
+          decodeImpl(uBuffer),
+          decodeImpl(vBuffer),
+        ]);
+        const grid = resampleWindGrid({
+          u: u.values,
+          v: v.values,
+          ni: u.ni,
+          nj: u.nj,
+          lo1: u.lo1,
+          la1: u.la1,
+          di: u.di,
+          dj: u.dj,
+          dx: targetDx,
+          dy: targetDx,
+        });
+        const idGrid = `${id}-${targetDx}`;
+        const manifest = {
+          schemaVersion: 1,
+          model: 'gfs',
+          cycle: { ...cycle, forecastHour: 0, runIso: cycleRunIso(cycle) },
+          fetchedAt: now(),
+          level: '10 m above ground',
+          units: 'm/s',
+          grid: {
+            nx: grid.nx,
+            ny: grid.ny,
+            lo1: grid.lo1,
+            la1: grid.la1,
+            dx: grid.dx,
+            dy: grid.dy,
+          },
+          stale: false,
+          unavailable: false,
+          reason: null,
+          gridUrl: `/api/wind/grid/${idGrid}.bin`,
+        };
+        cached = { id, idGrid, fetchedAt: manifest.fetchedAt, manifest, grid };
+        return cached;
+      } catch (error) {
+        if (cached) {
+          cached = {
+            ...cached,
+            manifest: { ...cached.manifest, stale: true, reason: error.message },
+          };
+        } else {
+          cached = {
+            id: null,
+            idGrid: null,
+            fetchedAt: now(),
+            grid: null,
+            manifest: {
+              schemaVersion: 1,
+              model: 'gfs',
+              stale: true,
+              unavailable: true,
+              reason: error.message,
+            },
+          };
+        }
+        return cached;
+      } finally {
+        loading = null;
+      }
+    })();
+    return loading;
+  }
+
+  const handler = async (req, res) => {
+    const path = new URL(req.url, 'http://localhost').pathname;
+    const state = await refresh();
+    if (path === '/status') {
+      const { gridUrl, ...manifest } = state.manifest;
+      return sendJson(res, manifest);
+    }
+    if (path.startsWith('/grid/')) {
+      if (!state.grid || path !== `/grid/${state.idGrid}.bin`)
+        return sendJson(res, { error: 'unknown_grid' }, 404);
+      const payload = Buffer.concat([
+        Buffer.from(state.grid.u.buffer, state.grid.u.byteOffset, state.grid.u.byteLength),
+        Buffer.from(state.grid.v.buffer, state.grid.v.byteOffset, state.grid.v.byteLength),
+      ]);
+      res.writeHead(200, {
+        'Content-Type': 'application/octet-stream',
+        'Cache-Control': 'public, max-age=3600, immutable',
+      });
+      return res.end(payload);
+    }
+    return sendJson(res, state.manifest);
+  };
+
+  return {
+    name: 'wind',
+    configureServer({ middlewares }) {
+      middlewares.use('/api/wind', handler);
+    },
+    configurePreviewServer({ middlewares }) {
+      middlewares.use('/api/wind', handler);
+    },
+  };
+}

server/providers/wind/catalog.js

@@ -0,0 +1,16 @@
+export const GFS_BUCKET = 'noaa-gfs-bdp-pds';
+
+/** Build the public GFS object key. */
+export function gfsObjectKey({ date, hour, forecastHour = 0 }) {
+  const hh = String(hour).padStart(2, '0');
+  return `gfs.${date}/${hh}/atmos/gfs.t${hh}z.pgrb2.0p25.f${String(forecastHour).padStart(3, '0')}`;
+}
+
+/** Select the latest GFS cycle that should be available. */
+export function selectLatestGfsCycle(nowMs, { availabilityLagMs = 5 * 3600_000 } = {}) {
+  const d = new Date(nowMs - availabilityLagMs);
+  const date = new Date(Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate()));
+  const hour = Math.floor(d.getUTCHours() / 6) * 6;
+  date.setUTCHours(hour);
+  return { date: date.toISOString().slice(0, 10).replaceAll('-', ''), hour };
+}

server/providers/wind/decode.js

@@ -0,0 +1,63 @@
+import os from 'node:os';
+import path from 'node:path';
+import { promises as fs } from 'node:fs';
+import { randomUUID } from 'node:crypto';
+
+/**
+ * Lazily created ecCodes WASM instance. The `@meri-imperiumi/eccodes-wasm`
+ * module is only imported on first use so unit tests can inject a fake
+ * decoder without loading the WASM (which requires Node >= 24).
+ * @type {?Promise<object>}
+ */
+let eccodesPromise = null;
+
+/** @returns {Promise<object>} The shared ecCodes WASM instance. */
+function loadEccodes() {
+  eccodesPromise ??= (async () => {
+    const module = await import('@meri-imperiumi/eccodes-wasm');
+    return module.createEccodes();
+  })();
+  return eccodesPromise;
+}
+
+/**
+ * Decode one GRIB2 wind message into its grid metadata and values.
+ *
+ * The WASM decoder reads from its own mounted filesystem, so the message is
+ * written to a temp file under `mountDir` and removed in `finally`.
+ *
+ * @param {Buffer} buffer - One GRIB2 message (already byte-range extracted).
+ * @param {{ eccodesModule?: object, mountDir?: string }} [options]
+ * @returns {Promise<{ni: number, nj: number, lo1: number, la1: number,
+ *   di: number, dj: number, values: ArrayLike<number>, shortName: string,
+ *   level: number, units: string}>}
+ */
+export async function decodeWindGribMessage(
+  buffer,
+  { eccodesModule, mountDir = os.tmpdir() } = {},
+) {
+  const eccodes = eccodesModule ?? (await loadEccodes());
+  const name = `wind-${randomUUID()}.grib2`;
+  const file = path.join(mountDir, name);
+  await fs.writeFile(file, buffer);
+  let handle = null;
+  try {
+    eccodes.mountFilesystem(mountDir);
+    handle = eccodes.openGrib(file);
+    return {
+      ni: handle.getLong('Ni'),
+      nj: handle.getLong('Nj'),
+      lo1: handle.getDouble('longitudeOfFirstGridPointInDegrees'),
+      la1: handle.getDouble('latitudeOfFirstGridPointInDegrees'),
+      di: handle.getDouble('iDirectionIncrementInDegrees'),
+      dj: handle.getDouble('jDirectionIncrementInDegrees'),
+      values: handle.getDoubleArray('values'),
+      shortName: handle.getString('shortName'),
+      level: handle.getLong('level'),
+      units: handle.getString('units'),
+    };
+  } finally {
+    if (handle) handle.delete();
+    await fs.unlink(file).catch(() => {});
+  }
+}

server/providers/wind/gfs.js

@@ -0,0 +1,89 @@
+/**
+ * Parse a GFS `.idx` inventory into ordered message records.
+ *
+ * Each line is `msgNumber:byteOffset:d=YYYYMMDDHH:VAR:LEVEL:...:`. Malformed
+ * lines are skipped; a payload with no valid lines is rejected so callers never
+ * treat an HTML error page as an empty inventory.
+ *
+ * @param {string} text - Raw `.idx` body.
+ * @returns {{ messages: Array<{index: number, offset: number, variable: string, level: string}> }}
+ */
+export function parseGfsIdx(text) {
+  const messages = [];
+  for (const line of String(text).split(/\r?\n/)) {
+    if (!line) continue;
+    const parts = line.split(':');
+    if (
+      parts.length < 6 ||
+      !/^\d+$/.test(parts[0]) ||
+      !/^\d+$/.test(parts[1]) ||
+      !parts[3] ||
+      !parts[4]
+    )
+      continue;
+    messages.push({
+      index: Number(parts[0]),
+      offset: Number(parts[1]),
+      variable: parts[3],
+      level: parts[4],
+    });
+  }
+  if (!messages.length) throw new Error('malformed GFS index');
+  return { messages };
+}
+
+/**
+ * Byte ranges for the two 10 m wind components. A message runs from its own
+ * offset to one byte before the next message in file order.
+ *
+ * @param {{messages: Array<object>}} parsed - {@link parseGfsIdx} result.
+ * @param {{level?: string}} [options]
+ * @returns {{u: {start: number, end: number}, v: {start: number, end: number}}}
+ */
+export function windMessageRanges(parsed, { level = '10 m above ground' } = {}) {
+  const range = (variable) => {
+    const index = parsed.messages.findIndex(
+      (message) => message.variable === variable && message.level === level,
+    );
+    const message = parsed.messages[index];
+    const next = parsed.messages[index + 1];
+    if (!message || !next) throw new Error(`missing ${variable} at ${level}`);
+    return { start: message.offset, end: next.offset - 1 };
+  };
+  return { u: range('UGRD'), v: range('VGRD') };
+}
+
+/** Fetch a bounded text response as a Buffer. */
+export async function fetchText({ url, fetchImpl = fetch, signal }) {
+  const response = await fetchImpl(url, { signal });
+  if (!response.ok) throw new Error(`HTTP ${response.status}`);
+  return Buffer.from(await response.arrayBuffer());
+}
+
+/**
+ * Fetch one bounded byte range. Rejects oversized ranges, non-206 responses
+ * (except a whole-object 200 starting at zero), and short bodies.
+ */
+export async function fetchRange({
+  url,
+  start,
+  end,
+  fetchImpl = fetch,
+  signal,
+  maxBytes = 8 * 1024 * 1024,
+}) {
+  if (end < start || end - start + 1 > maxBytes)
+    throw new Error('range too large');
+  const response = await fetchImpl(url, {
+    signal,
+    headers: { Range: `bytes=${start}-${end}` },
+  });
+  const validStatus =
+    response.status === 206 || (response.status === 200 && start === 0);
+  if (!validStatus || !response.ok)
+    throw new Error(`invalid range response ${response.status}`);
+  const buffer = Buffer.from(await response.arrayBuffer());
+  if (buffer.length !== end - start + 1)
+    throw new Error('invalid range length');
+  return buffer;
+}

server/providers/wind/grid.js

@@ -0,0 +1,51 @@
+/**
+ * Resample a GRIB wind grid into a compact equirectangular grid.
+ *
+ * Source grids are north-to-south (`la1` at the top, `dj` increasing toward the
+ * south) and start at `lo1`. Longitude wraps across the seam; latitude clamps.
+ *
+ * @param {{u: ArrayLike<number>, v: ArrayLike<number>, ni: number, nj: number,
+ *   lo1: number, la1: number, di: number, dj: number, dx?: number, dy?: number}} input
+ * @returns {{nx: number, ny: number, lo1: number, la1: number, dx: number,
+ *   dy: number, u: Float32Array, v: Float32Array}}
+ */
+export function resampleWindGrid({
+  u,
+  v,
+  ni,
+  nj,
+  lo1,
+  la1,
+  di,
+  dj,
+  dx = 1,
+  dy = 1,
+}) {
+  const nx = Math.round(360 / dx);
+  const ny = Math.round(180 / dy) + 1;
+  const outU = new Float32Array(nx * ny);
+  const outV = new Float32Array(nx * ny);
+  const sample = (source, lon, lat) => {
+    const x = ((lon - lo1) / di + ni) % ni;
+    const y = Math.max(0, Math.min(nj - 1, (la1 - lat) / dj));
+    const x0 = Math.floor(x);
+    const x1 = (x0 + 1) % ni;
+    const y0 = Math.floor(y);
+    const y1 = Math.min(nj - 1, y0 + 1);
+    const fx = x - x0;
+    const fy = y - y0;
+    const top = source[y0 * ni + x0] * (1 - fx) + source[y0 * ni + x1] * fx;
+    const bottom = source[y1 * ni + x0] * (1 - fx) + source[y1 * ni + x1] * fx;
+    return top * (1 - fy) + bottom * fy;
+  };
+  for (let y = 0; y < ny; y += 1) {
+    for (let x = 0; x < nx; x += 1) {
+      const index = y * nx + x;
+      const lon = x * dx;
+      const lat = 90 - y * dy;
+      outU[index] = sample(u, lon, lat);
+      outV[index] = sample(v, lon, lat);
+    }
+  }
+  return { nx, ny, lo1: 0, la1: 90, dx, dy, u: outU, v: outV };
+}