mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-05 19:04:02 +00:00
style(wind): adopt automatically discovered runtime formatting
This commit is contained in:
1 parent
a5dad07cc4
commit
9ae79cf34a
13 files changed
+421
-97
No files matched your search
+99
-22
@@ -3,14 +3,33 @@ import { fetchIfsWind } from './wind/ifs.js';
|
||||
import { decodeWindGribMessage } from './wind/decode.js';
|
||||
|
||||
/** Serve bounded, single-flight, per-model forecast snapshots from fixed providers. */
|
||||
export function windProxy({ fetchImpl = fetch, now = () => Date.now(), decodeImpl = decodeWindGribMessage, targetDx = 1, ttlMs = 3600_000, timeoutMs = 40_000, models = { gfs: fetchGfsWind, ifs: fetchIfsWind } } = {}) {
|
||||
export function windProxy({
|
||||
fetchImpl = fetch,
|
||||
now = () => Date.now(),
|
||||
decodeImpl = decodeWindGribMessage,
|
||||
targetDx = 1,
|
||||
ttlMs = 3600_000,
|
||||
timeoutMs = 40_000,
|
||||
models = { gfs: fetchGfsWind, ifs: fetchIfsWind },
|
||||
} = {}) {
|
||||
const caches = new Map();
|
||||
const loadings = new Map();
|
||||
const grids = new Map();
|
||||
const attempts = new Map();
|
||||
const unavailable = (model) => ({ manifest: { model, schemaVersion: 1, unavailable: true, stale: true, reason: 'Wind upstream unavailable' } });
|
||||
const unavailable = (model) => ({
|
||||
manifest: {
|
||||
model,
|
||||
schemaVersion: 1,
|
||||
unavailable: true,
|
||||
stale: true,
|
||||
reason: 'Wind upstream unavailable',
|
||||
},
|
||||
});
|
||||
const sendJson = (res, value, status = 200) => {
|
||||
res.writeHead(status, { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' });
|
||||
res.writeHead(status, {
|
||||
'Content-Type': 'application/json',
|
||||
'Cache-Control': 'no-store',
|
||||
});
|
||||
res.end(JSON.stringify(value));
|
||||
};
|
||||
function refresh(model) {
|
||||
@@ -21,15 +40,36 @@ export function windProxy({ fetchImpl = fetch, now = () => Date.now(), decodeImp
|
||||
operation.promise = (async () => {
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
const value = await models[model]({ fetchImpl, now, decodeImpl, targetDx, signal: controller.signal });
|
||||
const value = await models[model]({
|
||||
fetchImpl,
|
||||
now,
|
||||
decodeImpl,
|
||||
targetDx,
|
||||
signal: controller.signal,
|
||||
});
|
||||
controller.signal.throwIfAborted();
|
||||
const { grid, cycle } = value;
|
||||
if (!grid.u.every(Number.isFinite) || !grid.v.every(Number.isFinite)) throw new Error('Invalid wind grid');
|
||||
if (!grid.u.every(Number.isFinite) || !grid.v.every(Number.isFinite))
|
||||
throw new Error('Invalid wind grid');
|
||||
const id = `${model}-${cycle.date}-${cycle.hour}-f${cycle.forecastHour || 0}-${targetDx}`;
|
||||
const manifest = {
|
||||
schemaVersion: 1, model, cycle, fetchedAt: now(), level: value.level, units: value.units,
|
||||
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,
|
||||
schemaVersion: 1,
|
||||
model,
|
||||
cycle,
|
||||
fetchedAt: now(),
|
||||
level: value.level,
|
||||
units: value.units,
|
||||
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/${id}.bin?model=${model}`,
|
||||
};
|
||||
const state = { id, grid, manifest, fetchedAt: now() };
|
||||
@@ -47,7 +87,11 @@ export function windProxy({ fetchImpl = fetch, now = () => Date.now(), decodeImp
|
||||
}
|
||||
const old = caches.get(model);
|
||||
if (old) {
|
||||
old.manifest = { ...old.manifest, stale: true, reason: 'Wind upstream unavailable' };
|
||||
old.manifest = {
|
||||
...old.manifest,
|
||||
stale: true,
|
||||
reason: 'Wind upstream unavailable',
|
||||
};
|
||||
return old;
|
||||
}
|
||||
return unavailable(model);
|
||||
@@ -62,38 +106,71 @@ export function windProxy({ fetchImpl = fetch, now = () => Date.now(), decodeImp
|
||||
const handler = async (req, res) => {
|
||||
const url = new URL(req.url, 'http://localhost');
|
||||
const model = url.searchParams.get('model') || 'gfs';
|
||||
if (req.method !== 'GET') return sendJson(res, { error: 'method_not_allowed' }, 405);
|
||||
if (!['gfs', 'ifs'].includes(model) || !Object.hasOwn(models, model)) return sendJson(res, { error: 'unknown_model' }, 400);
|
||||
if (req.method !== 'GET')
|
||||
return sendJson(res, { error: 'method_not_allowed' }, 405);
|
||||
if (!['gfs', 'ifs'].includes(model) || !Object.hasOwn(models, model))
|
||||
return sendJson(res, { error: 'unknown_model' }, 400);
|
||||
if (url.pathname.startsWith('/grid/')) {
|
||||
const id = url.pathname.slice(6).replace(/\.bin$/, '');
|
||||
const state = url.pathname === `/grid/${id}.bin` && grids.get(model)?.get(id);
|
||||
const state =
|
||||
url.pathname === `/grid/${id}.bin` && grids.get(model)?.get(id);
|
||||
if (!state) return sendJson(res, { error: 'unknown_grid' }, 404);
|
||||
const bytes = 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' });
|
||||
const bytes = 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(bytes);
|
||||
}
|
||||
if (!['/', '/manifest', '/status'].includes(url.pathname)) return sendJson(res, { error: 'not_found' }, 404);
|
||||
if (!['/', '/manifest', '/status'].includes(url.pathname))
|
||||
return sendJson(res, { error: 'not_found' }, 404);
|
||||
let state = caches.get(model);
|
||||
if (!state || now() - state.fetchedAt >= ttlMs) {
|
||||
let operation = loadings.get(model);
|
||||
if (!operation && now() - (attempts.get(model) ?? -Infinity) >= 60_000) operation = refresh(model);
|
||||
if (!operation && now() - (attempts.get(model) ?? -Infinity) >= 60_000)
|
||||
operation = refresh(model);
|
||||
if (operation) {
|
||||
let disconnected = false;
|
||||
operation.waiters += 1;
|
||||
const close = () => { disconnected = true; if (--operation.waiters === 0) operation.controller.abort(); };
|
||||
const close = () => {
|
||||
disconnected = true;
|
||||
if (--operation.waiters === 0) operation.controller.abort();
|
||||
};
|
||||
res.once?.('close', close);
|
||||
try { state = await operation.promise; }
|
||||
finally { res.removeListener?.('close', close); if (!disconnected) operation.waiters -= 1; }
|
||||
try {
|
||||
state = await operation.promise;
|
||||
} finally {
|
||||
res.removeListener?.('close', close);
|
||||
if (!disconnected) operation.waiters -= 1;
|
||||
}
|
||||
if (disconnected) return;
|
||||
}
|
||||
}
|
||||
const manifest = state?.manifest || unavailable(model).manifest;
|
||||
if (url.pathname === '/status') { const { gridUrl, ...status } = manifest; return sendJson(res, status); }
|
||||
if (url.pathname === '/status') {
|
||||
const { gridUrl, ...status } = manifest;
|
||||
return sendJson(res, status);
|
||||
}
|
||||
return sendJson(res, manifest);
|
||||
};
|
||||
return {
|
||||
name: 'wind',
|
||||
configureServer({ middlewares }) { middlewares.use('/api/wind', handler); },
|
||||
configurePreviewServer({ middlewares }) { middlewares.use('/api/wind', handler); },
|
||||
configureServer({ middlewares }) {
|
||||
middlewares.use('/api/wind', handler);
|
||||
},
|
||||
configurePreviewServer({ middlewares }) {
|
||||
middlewares.use('/api/wind', handler);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -7,9 +7,14 @@ export function gfsObjectKey({ date, hour, forecastHour = 0 }) {
|
||||
}
|
||||
|
||||
/** Select the latest GFS cycle that should be available. */
|
||||
export function selectLatestGfsCycle(nowMs, { availabilityLagMs = 5 * 3600_000 } = {}) {
|
||||
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 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 };
|
||||
|
||||
@@ -1,17 +1,20 @@
|
||||
/** Lazy shared ecCodes WASM instance, created only for an active wind request. */
|
||||
let eccodesPromise = null;
|
||||
function loadEccodes() {
|
||||
eccodesPromise ??= import('@meri-imperiumi/eccodes-wasm').then(module => module.createEccodes()).catch(error => {
|
||||
eccodesPromise = null;
|
||||
throw error;
|
||||
});
|
||||
eccodesPromise ??= import('@meri-imperiumi/eccodes-wasm')
|
||||
.then((module) => module.createEccodes())
|
||||
.catch((error) => {
|
||||
eccodesPromise = null;
|
||||
throw error;
|
||||
});
|
||||
return eccodesPromise;
|
||||
}
|
||||
|
||||
/** Decode one bounded message in WASM memory without mounting the host filesystem. */
|
||||
export async function decodeWindGribMessage(buffer, { eccodesModule } = {}) {
|
||||
if (buffer.byteLength > 8 * 1024 * 1024) throw new Error('Wind GRIB exceeds byte budget');
|
||||
const eccodes = eccodesModule ?? await loadEccodes();
|
||||
if (buffer.byteLength > 8 * 1024 * 1024)
|
||||
throw new Error('Wind GRIB exceeds byte budget');
|
||||
const eccodes = eccodesModule ?? (await loadEccodes());
|
||||
// Everything after this await is synchronous. A shared virtual file is safe
|
||||
// across U/V and model requests and its storage is released after each decode.
|
||||
const file = '/gev-wind.grib2';
|
||||
@@ -21,9 +24,17 @@ export async function decodeWindGribMessage(buffer, { eccodesModule } = {}) {
|
||||
handle = eccodes.openGrib(file);
|
||||
const ni = handle.getLong('Ni');
|
||||
const nj = handle.getLong('Nj');
|
||||
if (!Number.isInteger(ni) || !Number.isInteger(nj) || ni < 1 || nj < 1 || ni * nj > 2_000_000) throw new Error('Invalid wind GRIB geometry');
|
||||
if (
|
||||
!Number.isInteger(ni) ||
|
||||
!Number.isInteger(nj) ||
|
||||
ni < 1 ||
|
||||
nj < 1 ||
|
||||
ni * nj > 2_000_000
|
||||
)
|
||||
throw new Error('Invalid wind GRIB geometry');
|
||||
return {
|
||||
ni, nj,
|
||||
ni,
|
||||
nj,
|
||||
lo1: handle.getDouble('longitudeOfFirstGridPointInDegrees'),
|
||||
la1: handle.getDouble('latitudeOfFirstGridPointInDegrees'),
|
||||
di: handle.getDouble('iDirectionIncrementInDegrees'),
|
||||
|
||||
@@ -41,7 +41,10 @@ export function parseGfsIdx(text) {
|
||||
* @param {{level?: string}} [options]
|
||||
* @returns {{u: {start: number, end: number}, v: {start: number, end: number}}}
|
||||
*/
|
||||
export function windMessageRanges(parsed, { level = '10 m above ground' } = {}) {
|
||||
export function windMessageRanges(
|
||||
parsed,
|
||||
{ level = '10 m above ground' } = {},
|
||||
) {
|
||||
const range = (variable) => {
|
||||
const index = parsed.messages.findIndex(
|
||||
(message) => message.variable === variable && message.level === level,
|
||||
@@ -57,7 +60,10 @@ export function windMessageRanges(parsed, { level = '10 m above ground' } = {})
|
||||
/** Fetch a bounded text response as a Buffer. */
|
||||
export async function fetchText({ url, fetchImpl = fetch, signal }) {
|
||||
const response = await fetchImpl(url, { signal, redirect: 'error' });
|
||||
if (!response.ok) { await response.body?.cancel(); throw new Error('Wind upstream unavailable'); }
|
||||
if (!response.ok) {
|
||||
await response.body?.cancel();
|
||||
throw new Error('Wind upstream unavailable');
|
||||
}
|
||||
return Buffer.from(await readWindBody(response, 2 * 1024 * 1024, signal));
|
||||
}
|
||||
|
||||
@@ -73,7 +79,13 @@ export async function fetchRange({
|
||||
signal,
|
||||
maxBytes = 8 * 1024 * 1024,
|
||||
}) {
|
||||
if (!Number.isSafeInteger(start) || !Number.isSafeInteger(end) || start < 0 || end < start || end - start + 1 > maxBytes)
|
||||
if (
|
||||
!Number.isSafeInteger(start) ||
|
||||
!Number.isSafeInteger(end) ||
|
||||
start < 0 ||
|
||||
end < start ||
|
||||
end - start + 1 > maxBytes
|
||||
)
|
||||
throw new Error('range too large');
|
||||
const response = await fetchImpl(url, {
|
||||
signal,
|
||||
@@ -82,19 +94,35 @@ export async function fetchRange({
|
||||
});
|
||||
const validStatus =
|
||||
response.status === 206 || (response.status === 200 && start === 0);
|
||||
if (!validStatus || !response.ok) { await response.body?.cancel(); throw new Error('Wind upstream unavailable'); }
|
||||
const buffer = Buffer.from(await readWindBody(response, end - start + 1, signal));
|
||||
if (!validStatus || !response.ok) {
|
||||
await response.body?.cancel();
|
||||
throw new Error('Wind upstream unavailable');
|
||||
}
|
||||
const buffer = Buffer.from(
|
||||
await readWindBody(response, end - start + 1, signal),
|
||||
);
|
||||
if (buffer.length !== end - start + 1)
|
||||
throw new Error('invalid range length');
|
||||
return buffer;
|
||||
}
|
||||
|
||||
import { GFS_BUCKET, gfsObjectKey, selectLatestGfsCycle, nearestGfsStep } from './catalog.js';
|
||||
import {
|
||||
GFS_BUCKET,
|
||||
gfsObjectKey,
|
||||
selectLatestGfsCycle,
|
||||
nearestGfsStep,
|
||||
} from './catalog.js';
|
||||
import { decodeWindGribMessage } from './decode.js';
|
||||
import { resampleWindGrid } from './grid.js';
|
||||
|
||||
/** Fetch and decode the GFS 10 m wind field valid closest to now. */
|
||||
export async function fetchGfsWind({ fetchImpl = fetch, now = () => Date.now(), targetDx = 1, decodeImpl = decodeWindGribMessage, signal } = {}) {
|
||||
export async function fetchGfsWind({
|
||||
fetchImpl = fetch,
|
||||
now = () => Date.now(),
|
||||
targetDx = 1,
|
||||
decodeImpl = decodeWindGribMessage,
|
||||
signal,
|
||||
} = {}) {
|
||||
const nowMs = now();
|
||||
const cycle = selectLatestGfsCycle(nowMs);
|
||||
const runMs = Date.UTC(
|
||||
@@ -114,8 +142,24 @@ export async function fetchGfsWind({ fetchImpl = fetch, now = () => Date.now(),
|
||||
fetchRange({ url: base, ...ranges.v, fetchImpl, signal }),
|
||||
]);
|
||||
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 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 runIso = new Date(runMs).toISOString();
|
||||
const validIso = new Date(runMs + forecastHour * 3600_000).toISOString();
|
||||
return { cycle: { ...cycle, forecastHour, runIso, validIso }, level: '10 m above ground', units: 'm/s', grid };
|
||||
return {
|
||||
cycle: { ...cycle, forecastHour, runIso, validIso },
|
||||
level: '10 m above ground',
|
||||
units: 'm/s',
|
||||
grid,
|
||||
};
|
||||
}
|
||||
@@ -21,7 +21,22 @@ export function resampleWindGrid({
|
||||
dx = 1,
|
||||
dy = 1,
|
||||
}) {
|
||||
if (!Number.isInteger(ni) || !Number.isInteger(nj) || ni < 1 || nj < 1 || ni * nj > 2_000_000 || u?.length !== ni * nj || v?.length !== ni * nj || ![lo1, la1, di, dj, dx, dy].every(Number.isFinite) || di <= 0 || dj <= 0 || dx < 0.25 || dy < 0.25 || dx > 180 || dy > 180)
|
||||
if (
|
||||
!Number.isInteger(ni) ||
|
||||
!Number.isInteger(nj) ||
|
||||
ni < 1 ||
|
||||
nj < 1 ||
|
||||
ni * nj > 2_000_000 ||
|
||||
u?.length !== ni * nj ||
|
||||
v?.length !== ni * nj ||
|
||||
![lo1, la1, di, dj, dx, dy].every(Number.isFinite) ||
|
||||
di <= 0 ||
|
||||
dj <= 0 ||
|
||||
dx < 0.25 ||
|
||||
dy < 0.25 ||
|
||||
dx > 180 ||
|
||||
dy > 180
|
||||
)
|
||||
throw new Error('Invalid wind grid geometry');
|
||||
const nx = Math.round(360 / dx);
|
||||
const ny = Math.round(180 / dy) + 1;
|
||||
|
||||
@@ -89,7 +89,8 @@ export function ifsWindRanges(entries) {
|
||||
export function nearestIfsStep(hours, cycleHour) {
|
||||
const max = cycleHour === 6 || cycleHour === 18 ? 144 : 360;
|
||||
const value = Math.max(0, Number.isFinite(hours) ? hours : 0);
|
||||
const step = value <= 144 ? Math.round(value / 3) * 3 : Math.round(value / 6) * 6;
|
||||
const step =
|
||||
value <= 144 ? Math.round(value / 3) * 3 : Math.round(value / 6) * 6;
|
||||
return Math.max(0, Math.min(max, step));
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user