From 80c5a09f1523bc1aa88638a3014f695e908114a1 Mon Sep 17 00:00:00 2001 From: Sameh Khamis Date: Sun, 13 Sep 2026 08:58:59 -0700 Subject: [PATCH] Format live source modules and adopt their formatting checks --- scripts/format-scope.json | 8 +- src/sources/live/aircraft.js | 160 +++++++++++++++++++++------ src/sources/live/contract.js | 59 ++++++++-- src/sources/live/contract.test.mjs | 171 ++++++++++++++++++++++++----- src/sources/live/index.js | 20 +++- src/sources/live/standalone.js | 164 +++++++++++++++++++++------ src/sources/live/vessels.js | 84 ++++++++++---- 7 files changed, 537 insertions(+), 129 deletions(-) diff --git a/scripts/format-scope.json b/scripts/format-scope.json index a6b088d31..0c12835de 100644 --- a/scripts/format-scope.json +++ b/scripts/format-scope.json @@ -248,5 +248,11 @@ "src/ui/styles/cockpit.css", "src/ui/styles/command-dock-compact.css", "src/app/stateChannel.js", - "src/app/stateChannel.test.mjs" + "src/app/stateChannel.test.mjs", + "src/sources/live/aircraft.js", + "src/sources/live/contract.js", + "src/sources/live/contract.test.mjs", + "src/sources/live/index.js", + "src/sources/live/standalone.js", + "src/sources/live/vessels.js" ] diff --git a/src/sources/live/aircraft.js b/src/sources/live/aircraft.js index 0b7d4e609..0c7614021 100644 --- a/src/sources/live/aircraft.js +++ b/src/sources/live/aircraft.js @@ -1,4 +1,10 @@ -import { admitRecords, cleanText, coordinates, epoch, finite } from './contract.js'; +import { + admitRecords, + cleanText, + coordinates, + epoch, + finite, +} from './contract.js'; /** Normalize the OpenSky state-vector route, including its regional fallback. */ export function normalizeOpenSkyAircraft(row) { @@ -6,71 +12,155 @@ export function normalizeOpenSkyAircraft(row) { const id = cleanText(row[0]).toLowerCase(); if (!id || !coordinates(row[6], row[5])) return null; return { - id, reference: id, latitude: row[6], longitude: row[5], - callsign: cleanText(row[1]), originCountry: cleanText(row[2]), - positionTimeMs: epoch(row[3], 1000), contactTimeMs: epoch(row[4], 1000), - baroAltitudeM: finite(row[7]), ellipsoidAltitudeM: finite(row[13]), - onGround: row[8] === true, speedMps: finite(row[9]), courseDeg: finite(row[10]), - verticalRateMps: finite(row[11]), category: finite(row[17]), - typeCode: null, registration: null, operator: null, + id, + reference: id, + latitude: row[6], + longitude: row[5], + callsign: cleanText(row[1]), + originCountry: cleanText(row[2]), + positionTimeMs: epoch(row[3], 1000), + contactTimeMs: epoch(row[4], 1000), + baroAltitudeM: finite(row[7]), + ellipsoidAltitudeM: finite(row[13]), + onGround: row[8] === true, + speedMps: finite(row[9]), + courseDeg: finite(row[10]), + verticalRateMps: finite(row[11]), + category: finite(row[17]), + typeCode: null, + registration: null, + operator: null, }; } /** Normalize readsb observations. The caller supplies the source snapshot epoch. */ export function normalizeReadsbAircraft(row, snapshotTimeMs) { const id = cleanText(row?.hex).toLowerCase(); - const latitude = finite(row?.lat), longitude = finite(row?.lon); + const latitude = finite(row?.lat), + longitude = finite(row?.lon); if (!id || !coordinates(latitude, longitude)) return null; - const baroFt = finite(row.alt_baro), geoFt = finite(row.alt_geom); - const speedKts = finite(row.gs), rateFtMin = finite(row.baro_rate); - const seen = finite(row.seen), seenPos = finite(row.seen_pos); + const baroFt = finite(row.alt_baro), + geoFt = finite(row.alt_geom); + const speedKts = finite(row.gs), + rateFtMin = finite(row.baro_rate); + const seen = finite(row.seen), + seenPos = finite(row.seen_pos); return { - id, reference: id, latitude, longitude, - callsign: cleanText(row.flight), originCountry: null, - positionTimeMs: snapshotTimeMs == null ? null : snapshotTimeMs - (seenPos ?? 0) * 1000, - contactTimeMs: snapshotTimeMs == null || seen == null ? null : snapshotTimeMs - seen * 1000, + id, + reference: id, + latitude, + longitude, + callsign: cleanText(row.flight), + originCountry: null, + positionTimeMs: + snapshotTimeMs == null ? null : snapshotTimeMs - (seenPos ?? 0) * 1000, + contactTimeMs: + snapshotTimeMs == null || seen == null + ? null + : snapshotTimeMs - seen * 1000, baroAltitudeM: baroFt == null ? null : baroFt * 0.3048, ellipsoidAltitudeM: geoFt == null ? null : geoFt * 0.3048, onGround: cleanText(row.alt_baro).toLowerCase() === 'ground', speedMps: speedKts == null ? null : speedKts * 0.514444, - courseDeg: finite(row.track), verticalRateMps: rateFtMin == null ? null : rateFtMin * 0.00508, - category: row.category ?? null, typeCode: cleanText(row.t), - registration: cleanText(row.r), operator: cleanText(row.ownOp), + courseDeg: finite(row.track), + verticalRateMps: rateFtMin == null ? null : rateFtMin * 0.00508, + category: row.category ?? null, + typeCode: cleanText(row.t), + registration: cleanText(row.r), + operator: cleanText(row.ownOp), }; } -export function openSkySnapshot(payload, { source = 'OpenSky Network', coverage = 'worldwide upstream snapshot', now = Date.now(), stale = false } = {}) { - const admitted = admitRecords(payload?.states, normalizeOpenSkyAircraft, 'OpenSky'); +export function openSkySnapshot( + payload, + { + source = 'OpenSky Network', + coverage = 'worldwide upstream snapshot', + now = Date.now(), + stale = false, + } = {}, +) { + const admitted = admitRecords( + payload?.states, + normalizeOpenSkyAircraft, + 'OpenSky', + ); const observedAtMs = epoch(payload?.time, 1000); const ageMs = observedAtMs == null ? null : Math.max(0, now - observedAtMs); - return { ...admitted, source, coverage, observedAtMs, ageMs, + return { + ...admitted, + source, + coverage, + observedAtMs, + ageMs, stale: stale || (ageMs != null && ageMs > 120000), - freshness: observedAtMs == null ? 'unknown' : (stale || ageMs > 120000) ? 'stale' : 'current', + freshness: + observedAtMs == null + ? 'unknown' + : stale || ageMs > 120000 + ? 'stale' + : 'current', }; } -export function readsbSnapshot(payload, { observedAtMs, source = 'adsb.lol', coverage = 'military upstream snapshot', now = Date.now(), stale = false } = {}) { - const admitted = admitRecords(payload?.ac, row => normalizeReadsbAircraft(row, observedAtMs), 'adsb.lol'); +export function readsbSnapshot( + payload, + { + observedAtMs, + source = 'adsb.lol', + coverage = 'military upstream snapshot', + now = Date.now(), + stale = false, + } = {}, +) { + const admitted = admitRecords( + payload?.ac, + (row) => normalizeReadsbAircraft(row, observedAtMs), + 'adsb.lol', + ); const ageMs = observedAtMs == null ? null : Math.max(0, now - observedAtMs); - return { ...admitted, source, coverage, observedAtMs, ageMs, stale, + return { + ...admitted, + source, + coverage, + observedAtMs, + ageMs, + stale, freshness: observedAtMs == null ? 'unknown' : stale ? 'stale' : 'current', }; } /** Track altitude remains barometric; renderers choose a visual ground fallback. */ -export function normalizeAircraftTrack(rows, { baseTimeMs = 0, readsb = false } = {}) { +export function normalizeAircraftTrack( + rows, + { baseTimeMs = 0, readsb = false } = {}, +) { if (!Array.isArray(rows)) return []; - return rows.flatMap(row => { + return rows.flatMap((row) => { if (!Array.isArray(row)) return []; - const latitude = finite(row[1]), longitude = finite(row[2]); + const latitude = finite(row[1]), + longitude = finite(row[2]); const offset = finite(row[0]); const observedAtMs = offset == null ? null : baseTimeMs + offset * 1000; - if (!coordinates(latitude, longitude) || observedAtMs == null || observedAtMs <= 0) return []; + if ( + !coordinates(latitude, longitude) || + observedAtMs == null || + observedAtMs <= 0 + ) + return []; const altitude = finite(row[3]); - return [{ latitude, longitude, observedAtMs, - baroAltitudeM: altitude == null ? null : altitude * (readsb ? 0.3048 : 1), - ellipsoidAltitudeM: null, - onGround: readsb ? cleanText(row[3]).toLowerCase() === 'ground' : row[5] === true, - }]; + return [ + { + latitude, + longitude, + observedAtMs, + baroAltitudeM: + altitude == null ? null : altitude * (readsb ? 0.3048 : 1), + ellipsoidAltitudeM: null, + onGround: readsb + ? cleanText(row[3]).toLowerCase() === 'ground' + : row[5] === true, + }, + ]; }); } diff --git a/src/sources/live/contract.js b/src/sources/live/contract.js index a37e9d670..a1169f1f2 100644 --- a/src/sources/live/contract.js +++ b/src/sources/live/contract.js @@ -10,7 +10,11 @@ * Records contain observation data only, with no scene objects or transport data. */ export class LiveSourceError extends Error { - constructor(code, message, { status = null, retryAfterMs = 20000, source = null } = {}) { + constructor( + code, + message, + { status = null, retryAfterMs = 20000, source = null } = {}, + ) { super(message); this.name = 'LiveSourceError'; this.code = code; @@ -36,13 +40,18 @@ export function cleanText(value) { } export function coordinates(latitude, longitude) { - return Number.isFinite(latitude) && Math.abs(latitude) <= 90 - && Number.isFinite(longitude) && Math.abs(longitude) <= 180; + return ( + Number.isFinite(latitude) && + Math.abs(latitude) <= 90 && + Number.isFinite(longitude) && + Math.abs(longitude) <= 180 + ); } /** Admit a snapshot atomically; an all-invalid nonempty feed is unavailable. */ export function admitRecords(rows, normalize, label) { - if (!Array.isArray(rows)) throw new LiveSourceError('malformed', `Malformed ${label} response`); + if (!Array.isArray(rows)) + throw new LiveSourceError('malformed', `Malformed ${label} response`); const records = []; const ids = new Set(); for (const row of rows) { @@ -55,11 +64,20 @@ export function admitRecords(rows, normalize, label) { if (rows.length && !records.length) { throw new LiveSourceError('malformed', `Malformed ${label} aircraft rows`); } - return { records, complete: records.length === rows.length, rejectedCount: rows.length - records.length }; + return { + records, + complete: records.length === rows.length, + rejectedCount: rows.length - records.length, + }; } /** Cancellation is checked after body parsing even when a transport ignores it. */ -export async function readResponse(fetchImpl, url, { signal, ...init } = {}, source = 'Live source') { +export async function readResponse( + fetchImpl, + url, + { signal, ...init } = {}, + source = 'Live source', +) { signal?.throwIfAborted(); let response; try { @@ -68,18 +86,35 @@ export async function readResponse(fetchImpl, url, { signal, ...init } = {}, sou } catch (error) { signal?.throwIfAborted(); if (error?.name === 'AbortError') throw error; - throw new LiveSourceError('unavailable', `${source} network error`, { source }); + throw new LiveSourceError('unavailable', `${source} network error`, { + source, + }); } let payload = null; - try { payload = await response.json(); } catch { /* classified below */ } + try { + payload = await response.json(); + } catch { + /* classified below */ + } signal?.throwIfAborted(); return { response, payload }; } export function httpError(response, source) { const status = response.status; - const code = status === 429 ? 'limited' : status === 401 || status === 403 ? 'denied' : 'unavailable'; - return new LiveSourceError(code, status === 429 ? `${source} rate limited` : `${source} HTTP ${status}`, { - status, source, retryAfterMs: code === 'limited' || code === 'denied' ? 45000 : 20000, - }); + const code = + status === 429 + ? 'limited' + : status === 401 || status === 403 + ? 'denied' + : 'unavailable'; + return new LiveSourceError( + code, + status === 429 ? `${source} rate limited` : `${source} HTTP ${status}`, + { + status, + source, + retryAfterMs: code === 'limited' || code === 'denied' ? 45000 : 20000, + }, + ); } diff --git a/src/sources/live/contract.test.mjs b/src/sources/live/contract.test.mjs index 7ef8d7201..461b1d6c6 100644 --- a/src/sources/live/contract.test.mjs +++ b/src/sources/live/contract.test.mjs @@ -1,13 +1,40 @@ import test from 'node:test'; import assert from 'node:assert/strict'; -import { createOpenSkySource, createAdsbLolSource, createAisStreamSource, normalizeReadsbAircraft, normalizeVesselObservation, openSkySnapshot, readsbSnapshot } from './index.js'; +import { + createOpenSkySource, + createAdsbLolSource, + createAisStreamSource, + normalizeReadsbAircraft, + normalizeVesselObservation, + openSkySnapshot, + readsbSnapshot, +} from './index.js'; const now = 1800000000000; -const aircraft = ['AbC123', 'TEST', 'Example', now / 1000 - 15, now / 1000 - 2, -97, 30, 1000, false, 100, 250, -2, null, 1040]; -const response = (payload, headers = {}, status = 200) => Response.json(payload, { headers, status }); +const aircraft = [ + 'AbC123', + 'TEST', + 'Example', + now / 1000 - 15, + now / 1000 - 2, + -97, + 30, + 1000, + false, + 100, + 250, + -2, + null, + 1040, +]; +const response = (payload, headers = {}, status = 200) => + Response.json(payload, { headers, status }); test('civil observations retain identity, separate altitude datums and source epochs', () => { - const snapshot = openSkySnapshot({ time: now / 1000 - 130, states: [aircraft] }, { now, source: 'Test source', coverage: 'regional' }); + const snapshot = openSkySnapshot( + { time: now / 1000 - 130, states: [aircraft] }, + { now, source: 'Test source', coverage: 'regional' }, + ); const row = snapshot.records[0]; assert.equal(row.id, 'abc123'); assert.equal(row.baroAltitudeM, 1000); @@ -17,40 +44,77 @@ test('civil observations retain identity, separate altitude datums and source ep assert.equal(snapshot.ageMs, 130000); assert.equal(snapshot.freshness, 'stale'); assert.equal(snapshot.coverage, 'regional'); - const missing = openSkySnapshot({ states: [[...aircraft.slice(0, 3), null, null, ...aircraft.slice(5)]] }, { now }); + const missing = openSkySnapshot( + { states: [[...aircraft.slice(0, 3), null, null, ...aircraft.slice(5)]] }, + { now }, + ); assert.equal(missing.observedAtMs, null); assert.equal(missing.freshness, 'unknown'); assert.equal(missing.records[0].positionTimeMs, null); }); test('mixed-invalid and duplicate admissions are explicitly incomplete; all-invalid rejects', () => { - const invalid = [...aircraft]; invalid[5] = 181; + const invalid = [...aircraft]; + invalid[5] = 181; const partial = openSkySnapshot({ states: [aircraft, invalid, aircraft] }); assert.equal(partial.records.length, 1); assert.equal(partial.complete, false); assert.equal(partial.rejectedCount, 2); - assert.throws(() => openSkySnapshot({ states: [invalid] }), { code: 'malformed' }); + assert.throws(() => openSkySnapshot({ states: [invalid] }), { + code: 'malformed', + }); assert.throws(() => openSkySnapshot({}), { code: 'malformed' }); assert.equal(openSkySnapshot({ states: [] }).complete, true); }); test('readsb units and position/contact ages are normalized once against the source epoch', () => { - const row = normalizeReadsbAircraft({ hex: 'abc123', lat: '30', lon: '-97', alt_baro: 10000, alt_geom: 10100, gs: 120, baro_rate: -600, seen: 2, seen_pos: 12 }, now); + const row = normalizeReadsbAircraft( + { + hex: 'abc123', + lat: '30', + lon: '-97', + alt_baro: 10000, + alt_geom: 10100, + gs: 120, + baro_rate: -600, + seen: 2, + seen_pos: 12, + }, + now, + ); assert.equal(row.baroAltitudeM, 3048); assert.equal(row.ellipsoidAltitudeM, 3078.48); assert.equal(row.speedMps, 120 * 0.514444); assert.equal(row.verticalRateMps, -600 * 0.00508); assert.equal(row.positionTimeMs, now - 12000); assert.equal(row.contactTimeMs, now - 2000); - const ground = normalizeReadsbAircraft({ hex: 'abc123', lat: 30, lon: -97, alt_baro: 'ground' }, null); + const ground = normalizeReadsbAircraft( + { hex: 'abc123', lat: 30, lon: -97, alt_baro: 'ground' }, + null, + ); assert.equal(ground.onGround, true); assert.equal(ground.baroAltitudeM, null); assert.equal(ground.positionTimeMs, null); - assert.equal(readsbSnapshot({ ac: [] }, { observedAtMs: null }).freshness, 'unknown'); + assert.equal( + readsbSnapshot({ ac: [] }, { observedAtMs: null }).freshness, + 'unknown', + ); }); test('vessel identity, opaque reference and heading/course remain distinct, unknowns stay null', () => { - const row = normalizeVesselObservation({ mmsi: '123456789', lat: 30, lon: -97, speed: 10, heading: 45, course: 60, last_position_epoch: now / 1000, extra: 'discard' }, 'opaque-reference'); + const row = normalizeVesselObservation( + { + mmsi: '123456789', + lat: 30, + lon: -97, + speed: 10, + heading: 45, + course: 60, + last_position_epoch: now / 1000, + extra: 'discard', + }, + 'opaque-reference', + ); assert.equal(row.id, '123456789'); assert.equal(row.reference, 'opaque-reference'); assert.equal(row.headingDeg, 45); @@ -59,23 +123,44 @@ test('vessel identity, opaque reference and heading/course remain distinct, unkn assert.equal(row.altitudeDatum, 'sea-surface'); assert.equal(row.observedAtMs, now); assert.equal('extra' in row, false); - assert.equal(normalizeVesselObservation({ mmsi: 'x', lat: null, lon: null }), null); + assert.equal( + normalizeVesselObservation({ mmsi: 'x', lat: null, lon: null }), + null, + ); }); test('construction is inert; adapters preserve routes, viewport query, cache epoch and history datum', async () => { const requests = []; const fetchImpl = async (url, init) => { requests.push({ url, init }); - if (url.startsWith('/api/opensky?')) return response({ time: now / 1000, states: [aircraft] }, { 'x-flight-source': 'adsb.lol', 'x-flight-coverage': '250 nm regional' }); - if (url.startsWith('/api/opensky-track')) return response({ path: [[now / 1000 - 30, 30, -97, 1000, 45, false]] }); - if (url === '/api/adsblol/mil') return response({ ac: [{ hex: 'abc123', lat: 30, lon: -97, seen_pos: 15 }] }, { 'x-ads-b-cache-age-ms': '60000', 'x-ads-b-cache': 'STALE' }); - return response({ timestamp: now / 1000 - 60, trace: [[10, 30, -97, 1000]] }); + if (url.startsWith('/api/opensky?')) + return response( + { time: now / 1000, states: [aircraft] }, + { + 'x-flight-source': 'adsb.lol', + 'x-flight-coverage': '250 nm regional', + }, + ); + if (url.startsWith('/api/opensky-track')) + return response({ path: [[now / 1000 - 30, 30, -97, 1000, 45, false]] }); + if (url === '/api/adsblol/mil') + return response( + { ac: [{ hex: 'abc123', lat: 30, lon: -97, seen_pos: 15 }] }, + { 'x-ads-b-cache-age-ms': '60000', 'x-ads-b-cache': 'STALE' }, + ); + return response({ + timestamp: now / 1000 - 60, + trace: [[10, 30, -97, 1000]], + }); }; const civil = createOpenSkySource({ fetchImpl, now: () => now }); const military = createAdsbLolSource({ fetchImpl, now: () => now }); assert.equal(requests.length, 0); const signal = new AbortController().signal; - const snapshot = await civil.getSnapshot({ latitude: 30.123456, longitude: -97 }, { signal }); + const snapshot = await civil.getSnapshot( + { latitude: 30.123456, longitude: -97 }, + { signal }, + ); assert.equal(requests[0].url, '/api/opensky?lat=30.1235&lon=-97.0000'); assert.equal(requests[0].init.signal, signal); assert.equal(snapshot.source, 'adsb.lol'); @@ -95,9 +180,17 @@ test('construction is inert; adapters preserve routes, viewport query, cache epo test('cancellation after slow body parsing rejects even with a transport that ignores abort', async () => { let release; const controller = new AbortController(); - const source = createOpenSkySource({ fetchImpl: async () => ({ ok: true, json: () => new Promise(resolve => { release = resolve; }) }) }); + const source = createOpenSkySource({ + fetchImpl: async () => ({ + ok: true, + json: () => + new Promise((resolve) => { + release = resolve; + }), + }), + }); const pending = source.getSnapshot({}, { signal: controller.signal }); - await new Promise(resolve => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); controller.abort(); release({ states: [aircraft] }); await assert.rejects(pending, { name: 'AbortError' }); @@ -106,8 +199,13 @@ test('cancellation after slow body parsing rejects even with a transport that ig test('denials and outages never start another source and do not echo arbitrary response bodies', async () => { for (const status of [401, 403, 429, 502]) { let calls = 0; - const source = createOpenSkySource({ fetchImpl: async () => { calls++; return response({ error: 'sensitive upstream detail' }, {}, status); } }); - await assert.rejects(source.getSnapshot(), error => { + const source = createOpenSkySource({ + fetchImpl: async () => { + calls++; + return response({ error: 'sensitive upstream detail' }, {}, status); + }, + }); + await assert.rejects(source.getSnapshot(), (error) => { assert.equal(error.status, status); assert.equal(error.message.includes('sensitive'), false); assert.ok(error.retryAfterMs >= 20000); @@ -118,15 +216,34 @@ test('denials and outages never start another source and do not echo arbitrary r }); test('AIS reports limited received coverage and keeps connection state separate from positions', async () => { - const source = createAisStreamSource({ origin: () => 'http://example.test', fetchImpl: async url => { - if (url.includes('/track?')) return response({ samples: [{ lat: 30, lon: -97, t: now / 1000 }] }); - assert.equal(url, 'http://example.test/api/ais-live?maxRows=500'); - return response({ status: 'reconnecting', refreshing: true, rows: [{ mmsi: '123456789', lat: 30, lon: -97, last_position_UTC: new Date(now).toISOString() }], newestPositionAt: new Date(now).toISOString() }); - } }); + const source = createAisStreamSource({ + origin: () => 'http://example.test', + fetchImpl: async (url) => { + if (url.includes('/track?')) + return response({ samples: [{ lat: 30, lon: -97, t: now / 1000 }] }); + assert.equal(url, 'http://example.test/api/ais-live?maxRows=500'); + return response({ + status: 'reconnecting', + refreshing: true, + rows: [ + { + mmsi: '123456789', + lat: 30, + lon: -97, + last_position_UTC: new Date(now).toISOString(), + }, + ], + newestPositionAt: new Date(now).toISOString(), + }); + }, + }); const snapshot = await source.getSnapshot({ maxRows: 500 }); assert.equal(snapshot.complete, false); assert.equal(snapshot.stale, true); assert.equal(snapshot.observedAtMs, now); assert.equal(snapshot.transportStatus, 'reconnecting'); - assert.equal((await source.getTrack('123456789')).records[0].observedAtMs, now); + assert.equal( + (await source.getTrack('123456789')).records[0].observedAtMs, + now, + ); }); diff --git a/src/sources/live/index.js b/src/sources/live/index.js index 7047208e8..d19890111 100644 --- a/src/sources/live/index.js +++ b/src/sources/live/index.js @@ -1,4 +1,18 @@ export { LiveSourceError } from './contract.js'; -export { normalizeOpenSkyAircraft, normalizeReadsbAircraft, normalizeAircraftTrack, openSkySnapshot, readsbSnapshot } from './aircraft.js'; -export { normalizeVesselObservation, normalizeVesselTrack, vesselSnapshot } from './vessels.js'; -export { createOpenSkySource, createAdsbLolSource, createAisStreamSource } from './standalone.js'; +export { + normalizeOpenSkyAircraft, + normalizeReadsbAircraft, + normalizeAircraftTrack, + openSkySnapshot, + readsbSnapshot, +} from './aircraft.js'; +export { + normalizeVesselObservation, + normalizeVesselTrack, + vesselSnapshot, +} from './vessels.js'; +export { + createOpenSkySource, + createAdsbLolSource, + createAisStreamSource, +} from './standalone.js'; diff --git a/src/sources/live/standalone.js b/src/sources/live/standalone.js index a2f72b154..32885d466 100644 --- a/src/sources/live/standalone.js +++ b/src/sources/live/standalone.js @@ -1,5 +1,15 @@ -import { epoch, finite, httpError, LiveSourceError, readResponse } from './contract.js'; -import { normalizeAircraftTrack, openSkySnapshot, readsbSnapshot } from './aircraft.js'; +import { + epoch, + finite, + httpError, + LiveSourceError, + readResponse, +} from './contract.js'; +import { + normalizeAircraftTrack, + openSkySnapshot, + readsbSnapshot, +} from './aircraft.js'; import { normalizeVesselTrack, vesselSnapshot } from './vessels.js'; const defaultFetch = (...args) => globalThis.fetch(...args); @@ -7,9 +17,16 @@ const header = (response, name) => response.headers?.get?.(name); function openSkyError(response) { const error = httpError(response, 'OpenSky'); - const mode = String(header(response, 'x-opensky-auth-mode-used') || header(response, 'x-opensky-auth') || '').toLowerCase(); - const reason = String(header(response, 'x-opensky-auth-reason') || '').toLowerCase(); - if (response.status === 429 && (!mode || mode === 'anon')) error.message = 'OpenSky rate limited (anonymous)'; + const mode = String( + header(response, 'x-opensky-auth-mode-used') || + header(response, 'x-opensky-auth') || + '', + ).toLowerCase(); + const reason = String( + header(response, 'x-opensky-auth-reason') || '', + ).toLowerCase(); + if (response.status === 429 && (!mode || mode === 'anon')) + error.message = 'OpenSky rate limited (anonymous)'; if (response.status === 401 || response.status === 403) { const reasons = { oauth_invalid_or_missing: 'OpenSky OAuth client missing/invalid', @@ -17,80 +34,165 @@ function openSkyError(response) { basic_invalid_credentials: 'OpenSky username/password rejected', missing_basic_creds: 'OpenSky auth missing', missing_oauth_and_basic_creds: 'OpenSky auth missing', - auth_required: 'OpenSky auth required', forced_anonymous: 'OpenSky auth required', + auth_required: 'OpenSky auth required', + forced_anonymous: 'OpenSky auth required', }; - error.message = reasons[reason] || (/^(oauth_|basic_)/.test(reason) ? 'OpenSky auth invalid' : mode === 'anon' ? 'OpenSky auth required' : 'OpenSky auth failed'); + error.message = + reasons[reason] || + (/^(oauth_|basic_)/.test(reason) + ? 'OpenSky auth invalid' + : mode === 'anon' + ? 'OpenSky auth required' + : 'OpenSky auth failed'); } return error; } /** Existing same-origin aircraft routes; no request starts during construction. */ -export function createOpenSkySource({ fetchImpl = defaultFetch, now = () => Date.now() } = {}) { +export function createOpenSkySource({ + fetchImpl = defaultFetch, + now = () => Date.now(), +} = {}) { return { async getSnapshot(query = {}, { signal } = {}) { const params = new URLSearchParams(); if (Number.isFinite(query.latitude) && Number.isFinite(query.longitude)) { - params.set('lat', query.latitude.toFixed(4)); params.set('lon', query.longitude.toFixed(4)); + params.set('lat', query.latitude.toFixed(4)); + params.set('lon', query.longitude.toFixed(4)); } - const { response, payload } = await readResponse(fetchImpl, `/api/opensky${params.size ? '?' + params : ''}`, { signal }, 'OpenSky'); + const { response, payload } = await readResponse( + fetchImpl, + `/api/opensky${params.size ? '?' + params : ''}`, + { signal }, + 'OpenSky', + ); if (!response.ok) throw openSkyError(response); - return { ...openSkySnapshot(payload, { - source: header(response, 'x-flight-source') || 'OpenSky Network', - coverage: header(response, 'x-flight-coverage') || 'worldwide upstream snapshot', now: now(), - }), status: response.status }; + return { + ...openSkySnapshot(payload, { + source: header(response, 'x-flight-source') || 'OpenSky Network', + coverage: + header(response, 'x-flight-coverage') || + 'worldwide upstream snapshot', + now: now(), + }), + status: response.status, + }; }, async getTrack(reference, { signal } = {}) { - const { response, payload } = await readResponse(fetchImpl, '/api/opensky-track?icao24=' + encodeURIComponent(reference), { signal }, 'OpenSky'); + const { response, payload } = await readResponse( + fetchImpl, + '/api/opensky-track?icao24=' + encodeURIComponent(reference), + { signal }, + 'OpenSky', + ); if (!response.ok) throw httpError(response, 'OpenSky'); - return { records: normalizeAircraftTrack(payload?.path), complete: false }; + return { + records: normalizeAircraftTrack(payload?.path), + complete: false, + }; }, async getEnrichment(query, { signal } = {}) { - if (!['type', 'route'].includes(query.kind)) throw new LiveSourceError('unsupported', 'Enrichment unavailable'); - const { response, payload } = await readResponse(fetchImpl, `/api/adsbdb/${query.kind}/${encodeURIComponent(query.id)}`, { signal }, 'adsbdb'); + if (!['type', 'route'].includes(query.kind)) + throw new LiveSourceError('unsupported', 'Enrichment unavailable'); + const { response, payload } = await readResponse( + fetchImpl, + `/api/adsbdb/${query.kind}/${encodeURIComponent(query.id)}`, + { signal }, + 'adsbdb', + ); if (!response.ok) throw httpError(response, 'adsbdb'); return payload; }, }; } -export function createAdsbLolSource({ fetchImpl = defaultFetch, now = () => Date.now() } = {}) { +export function createAdsbLolSource({ + fetchImpl = defaultFetch, + now = () => Date.now(), +} = {}) { return { async getSnapshot(_query = {}, { signal } = {}) { - const { response, payload } = await readResponse(fetchImpl, '/api/adsblol/mil', { signal }, 'adsb.lol'); + const { response, payload } = await readResponse( + fetchImpl, + '/api/adsblol/mil', + { signal }, + 'adsb.lol', + ); if (!response.ok) throw httpError(response, 'adsb.lol'); const age = finite(header(response, 'x-ads-b-cache-age-ms')); - return { ...readsbSnapshot(payload, { - observedAtMs: now() - (age != null && age > 0 ? age : 0), now: now(), - stale: header(response, 'x-ads-b-cache') === 'STALE', - }), status: response.status }; + return { + ...readsbSnapshot(payload, { + observedAtMs: now() - (age != null && age > 0 ? age : 0), + now: now(), + stale: header(response, 'x-ads-b-cache') === 'STALE', + }), + status: response.status, + }; }, async getTrack(reference, { signal } = {}) { - const { response, payload } = await readResponse(fetchImpl, '/api/adsblol/trace?hex=' + encodeURIComponent(reference), { signal }, 'adsb.lol'); + const { response, payload } = await readResponse( + fetchImpl, + '/api/adsblol/trace?hex=' + encodeURIComponent(reference), + { signal }, + 'adsb.lol', + ); if (!response.ok) throw httpError(response, 'adsb.lol'); const baseTimeMs = epoch(payload?.timestamp, 1000); - return { records: baseTimeMs == null ? [] : normalizeAircraftTrack(payload?.trace, { baseTimeMs, readsb: true }), complete: false }; + return { + records: + baseTimeMs == null + ? [] + : normalizeAircraftTrack(payload?.trace, { + baseTimeMs, + readsb: true, + }), + complete: false, + }; }, }; } -export function createAisStreamSource({ fetchImpl = defaultFetch, apiUrl = '/api/ais-live', origin = () => globalThis.location?.origin || 'http://localhost' } = {}) { +export function createAisStreamSource({ + fetchImpl = defaultFetch, + apiUrl = '/api/ais-live', + origin = () => globalThis.location?.origin || 'http://localhost', +} = {}) { return { async getSnapshot({ maxRows = 12000 } = {}, { signal } = {}) { const url = new URL(apiUrl, origin()); url.searchParams.set('maxRows', String(maxRows)); - const { response, payload } = await readResponse(fetchImpl, url.toString(), { signal, cache: 'no-store' }, 'AIS live'); + const { response, payload } = await readResponse( + fetchImpl, + url.toString(), + { signal, cache: 'no-store' }, + 'AIS live', + ); if (!response.ok) { const error = httpError(response, 'AIS live'); - const reasons = { 'missing-key': 'AISSTREAM_API_KEY not set', 'auth-failed': 'API key rejected — check AISSTREAM_API_KEY', unsupported: 'live feed unsupported', error: 'feed down', closed: 'feed disconnected' }; + const reasons = { + 'missing-key': 'AISSTREAM_API_KEY not set', + 'auth-failed': 'API key rejected — check AISSTREAM_API_KEY', + unsupported: 'live feed unsupported', + error: 'feed down', + closed: 'feed disconnected', + }; error.message = reasons[payload?.status] || error.message; throw error; } return { ...vesselSnapshot(payload), status: response.status }; }, async getTrack(reference, { signal } = {}) { - const { response, payload } = await readResponse(fetchImpl, '/api/ais-live/track?mmsi=' + encodeURIComponent(reference), { signal }, 'AIS live'); + const { response, payload } = await readResponse( + fetchImpl, + '/api/ais-live/track?mmsi=' + encodeURIComponent(reference), + { signal }, + 'AIS live', + ); if (!response.ok) throw httpError(response, 'AIS live'); - return { records: normalizeVesselTrack(payload?.samples), complete: false }; + return { + records: normalizeVesselTrack(payload?.samples), + complete: false, + }; }, }; } diff --git a/src/sources/live/vessels.js b/src/sources/live/vessels.js index 03c7a8b1d..352e9fac0 100644 --- a/src/sources/live/vessels.js +++ b/src/sources/live/vessels.js @@ -3,49 +3,93 @@ import { coordinates, epoch, finite, LiveSourceError } from './contract.js'; /** AIS positions are sea-surface observations; heading and course are distinct. */ export function normalizeVesselObservation(row, reference = null) { const id = String(row?.mmsi || row?.input_identifier || '').trim(); - const latitude = finite(row?.lat), longitude = finite(row?.lon); + const latitude = finite(row?.lat), + longitude = finite(row?.lon); if (!id || !coordinates(latitude, longitude)) return null; const speedKts = finite(row.speed); return { - id, reference: reference ?? id, latitude, longitude, + id, + reference: reference ?? id, + latitude, + longitude, name: String(row.name || row.input_name || id), - imo: String(row.imo || ''), type: String(row.type_specific || row.type || ''), + imo: String(row.imo || ''), + type: String(row.type_specific || row.type || ''), destination: String(row.destination || ''), speedMps: speedKts == null ? null : speedKts * 0.514444, - courseDeg: finite(row.course), headingDeg: finite(row.heading), - observedAtMs: epoch(row.last_position_epoch, 1000) - ?? epoch(Date.parse(row.last_position_UTC)), + courseDeg: finite(row.course), + headingDeg: finite(row.heading), + observedAtMs: + epoch(row.last_position_epoch, 1000) ?? + epoch(Date.parse(row.last_position_UTC)), altitudeDatum: 'sea-surface', }; } -export function vesselSnapshot(payload, { source = 'AISStream', coverage = 'received AIS positions', referenceFor = row => String(row.mmsi || row.input_identifier || '') } = {}) { - if (!Array.isArray(payload?.rows)) throw new LiveSourceError('malformed', 'Malformed vessel response'); +export function vesselSnapshot( + payload, + { + source = 'AISStream', + coverage = 'received AIS positions', + referenceFor = (row) => String(row.mmsi || row.input_identifier || ''), + } = {}, +) { + if (!Array.isArray(payload?.rows)) + throw new LiveSourceError('malformed', 'Malformed vessel response'); const rows = payload.rows; - const records = [], ids = new Set(); + const records = [], + ids = new Set(); for (const row of rows) { const record = normalizeVesselObservation(row, referenceFor(row)); - if (record && !ids.has(record.id)) { records.push(record); ids.add(record.id); } + if (record && !ids.has(record.id)) { + records.push(record); + ids.add(record.id); + } } - const observedAtMs = epoch(payload?.newestPositionAt) ?? epoch(Date.parse(payload?.newestPositionAt)) - ?? records.reduce((newest, row) => Math.max(newest || 0, row.observedAtMs || 0) || null, null); + const observedAtMs = + epoch(payload?.newestPositionAt) ?? + epoch(Date.parse(payload?.newestPositionAt)) ?? + records.reduce( + (newest, row) => Math.max(newest || 0, row.observedAtMs || 0) || null, + null, + ); return { - records, source, coverage, complete: false, rejectedCount: rows.length - records.length, - observedAtMs, freshness: payload?.refreshing ? 'stale' : observedAtMs == null ? 'unknown' : 'current', + records, + source, + coverage, + complete: false, + rejectedCount: rows.length - records.length, + observedAtMs, + freshness: payload?.refreshing + ? 'stale' + : observedAtMs == null + ? 'unknown' + : 'current', stale: Boolean(payload?.refreshing), // Connection state is distinct from receipt of a usable position. - transportStatus: typeof payload?.status === 'string' ? payload.status : null, + transportStatus: + typeof payload?.status === 'string' ? payload.status : null, lastMessageAt: payload?.lastMessageAt ?? null, - nextAttemptAt: finite(payload?.nextAttemptAt), silentForMs: finite(payload?.silentForMs), - reconnectAttempt: finite(payload?.reconnectAttempt), rawRowCount: rows.length, + nextAttemptAt: finite(payload?.nextAttemptAt), + silentForMs: finite(payload?.silentForMs), + reconnectAttempt: finite(payload?.reconnectAttempt), + rawRowCount: rows.length, }; } export function normalizeVesselTrack(samples) { if (!Array.isArray(samples)) return []; - return samples.flatMap(sample => { - const latitude = finite(sample?.lat), longitude = finite(sample?.lon); + return samples.flatMap((sample) => { + const latitude = finite(sample?.lat), + longitude = finite(sample?.lon); if (!coordinates(latitude, longitude)) return []; - return [{ latitude, longitude, observedAtMs: epoch(sample?.t, 1000), altitudeDatum: 'sea-surface' }]; + return [ + { + latitude, + longitude, + observedAtMs: epoch(sample?.t, 1000), + altitudeDatum: 'sea-surface', + }, + ]; }); }