From 2f295b74547fe5e7566ef2cc98680aed23393665 Mon Sep 17 00:00:00 2001 From: Will Shields <136547209+d3mocide@users.noreply.github.com> Date: Sat, 13 Jun 2026 22:20:22 -0700 Subject: [PATCH 1/3] Set GitHub Sponsors username to d3mocide Updated GitHub Sponsors username in FUNDING.yml --- .github/FUNDING.yml | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 .github/FUNDING.yml diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml new file mode 100644 index 00000000..1d1587c0 --- /dev/null +++ b/.github/FUNDING.yml @@ -0,0 +1,15 @@ +# These are supported funding model platforms + +github: d3mocide +patreon: # Replace with a single Patreon username +open_collective: # Replace with a single Open Collective username +ko_fi: # Replace with a single Ko-fi username +tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel +community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry +liberapay: # Replace with a single Liberapay username +issuehunt: # Replace with a single IssueHunt username +lfx_crowdfunding: # Replace with a single LFX Crowdfunding project-name e.g., cloud-foundry +polar: # Replace with a single Polar username +buy_me_a_coffee: # Replace with a single Buy Me a Coffee username +thanks_dev: # Replace with a single thanks.dev username +custom: # Replace with up to 4 custom sponsorship URLs e.g., ['link1', 'link2'] From b303c81aa60a97a0ee6619e7251ac10a46738cbd Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 03:30:11 +0000 Subject: [PATCH 2/3] perf(frontend): pace COT render loop + cut per-frame allocations; fix JS8Call UDP bridge & KiwiSDR auth Rendering (target: smooth with 1000s of COTs): - Adaptive frame pacing in useAnimationLoop: ~30 fps above 800 tracked entities, ~60 fps cap otherwise (no redundant work at 120/144 Hz) - buildEntityLayers: single pass derives all per-layer datasets instead of 5-6 filter/map passes per frame - buildTrailLayers: merged trail/gap-bridge pass + WeakMap path3D cache keyed on update-stable smoothed trails - tak.worker: decode batch 10 -> 64 (fewer main-thread wakeups on the ~11k-message orbital sweeps) - useEntityWorker: maritime localStorage snapshot capped at 750 tracks, 30 s interval, serialized in requestIdleCallback instead of inline on the WebSocket message path JS8Call bridge (terminal previously non-functional): - Bridge now binds UDP 2242 (where JS8Call pushes events per the WSJT-X model) and sends commands to JS8Call's observed datagram source address; the old 2245 listener / fixed-port sends reached nothing - API datagrams use the correct lowercase {type,value,params} envelope; RIG.SET_FREQ uses params.DIAL; MODE.SET_SPEED uses numeric submode - Handles PING/STATION.CALLSIGN/STATION.GRID/RIG.FREQ/MODE.SPEED and reports js8call_connected from real datagram liveness - Removed invented UDPClient2* keys from the JS8Call INI KiwiSDR: - Auth uses plaintext 'SET auth t=kiwi p=' per the reference kiwiclient (the md5 pwd= form is not part of the protocol) on both SND and W/F streams Verification: frontend lint/typecheck/vitest (278 passed); js8call ruff + pytest (26 passed). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01S3R2VyPxQQY1UceaiC5wg6 --- ...ontend-cots-optimization-js8-kiwi-fixes.md | 92 ++++++++++ docker-compose.yml | 2 +- frontend/src/hooks/useAnimationLoop.ts | 25 ++- frontend/src/hooks/useEntityWorker.ts | 34 +++- frontend/src/layers/buildEntityLayers.ts | 117 +++++++----- frontend/src/layers/buildTrailLayers.ts | 66 ++++--- frontend/src/workers/tak.worker.ts | 7 +- js8call/Dockerfile | 9 +- js8call/kiwi_client.py | 37 ++-- js8call/server.py | 168 +++++++++++++++--- js8call/tests/test_kiwi_compatibility.py | 19 +- 11 files changed, 434 insertions(+), 142 deletions(-) create mode 100644 agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md diff --git a/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md b/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md new file mode 100644 index 00000000..403a4b48 --- /dev/null +++ b/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md @@ -0,0 +1,92 @@ +# Frontend COT Rendering Optimization + JS8Call/KiwiSDR Bridge Fixes + +## Issue + +1. **Rendering**: The tactical map needed to stay smooth with thousands of live + COTs. The rAF loop ran uncapped (120/144 Hz on fast displays), recomposed + every deck.gl layer each tick, and `buildEntityLayers`/`buildTrailLayers` + made 5-6 separate `filter`/`map` passes per frame — each allocating + intermediate arrays, plus a fresh path array per trail per frame. +2. **Pipeline**: `useEntityWorker` serialized every maritime entity (with + trails) to `localStorage` inline on the WebSocket message path every 10 s — + a multi-ms main-thread hitch at AIS scale, with no size cap (quota risk). + The TAK worker flushed decode batches every 10 messages, causing excessive + worker→main wakeups during the ~11k-message orbital sweeps. +3. **JS8Call terminal never worked**: the bridge had the UDP API model + inverted. JS8Call (WSJT-X model) binds an ephemeral port and *pushes* + events to the configured "UDP Server" (127.0.0.1:2242 per our INI); the + bridge instead listened on 2245 (where nothing ever arrives — the INI's + `UDPClient2*` keys are not real JS8Call settings) and sent commands to a + fixed port 2242 (where nothing listens). Additionally, every outgoing + datagram used uppercase JSON keys (`{"TYPE": ...}`) while the JS8Call API + requires lowercase (`{"type", "value", "params"}`), `RIG.SET_FREQ` lacked + the required `params.DIAL`, and `MODE.SET_SPEED` sent a string instead of + the numeric submode. +4. **KiwiSDR password nodes never connected**: `kiwi_client.py` sent + `SET auth t=kiwi pwd=`, which is not part of the KiwiSDR protocol. + The reference kiwiclient sends plaintext `SET auth t=kiwi p=`. + The waterfall stream also always authenticated with an empty password. + +## Solution + +- Adaptive frame pacing in the animation loop: ~30 fps when + `entities + satellites > 800`, ~60 fps cap otherwise (skips redundant + 120/144 Hz ticks). dt accumulates across skipped ticks so interpolation is + unaffected. +- Single-pass dataset derivation in `buildEntityLayers` (integrity halos, + altitude stems, tactical halos, selection ring, velocity vectors) and + `buildTrailLayers` (trails + gap bridges), plus a `WeakMap` cache for + smoothed-trail → path3D conversion keyed on the (update-stable) trail array. +- Maritime snapshot: capped at 750 most-recent entities, interval 10 s → 30 s, + serialization moved to `requestIdleCallback`. +- TAK worker batch size 10 → 64 (flush interval unchanged at 50 ms). +- JS8 bridge: binds UDP 2242, records JS8Call's datagram source address as the + command reply address, sends correctly-shaped lowercase-key API messages, + handles `PING`/`STATION.CALLSIGN`/`STATION.GRID`/`RIG.FREQ`/`MODE.SPEED` + responses into a merged `STATION.STATUS` broadcast, reports + `js8call_connected` from actual datagram liveness (60 s window), and pulls + initial station state on first contact. Removed the bogus `UDPClient2*` INI + keys; compose/Dockerfile env updated (`JS8CALL_PORT` → `JS8CALL_UDP_SERVER_PORT`). +- KiwiSDR auth: plaintext `p=` per reference kiwiclient, applied to + both SND and W/F streams. + +## Changes + +- `frontend/src/hooks/useAnimationLoop.ts` — adaptive frame pacing; rAF + scheduled at tick start so paced skips keep the loop alive. +- `frontend/src/layers/buildEntityLayers.ts` — one pass over interpolated + entities builds all five per-layer datasets. +- `frontend/src/layers/buildTrailLayers.ts` — merged trail/gap-bridge pass; + `WeakMap` path3D cache. +- `frontend/src/hooks/useEntityWorker.ts` — sea snapshot cap + idle-time write. +- `frontend/src/workers/tak.worker.ts` — batch size 64. +- `js8call/server.py` — UDP server model fixed (bind 2242, reply-address + routing), JS8Call API message shapes fixed, station-state merge + liveness. +- `js8call/kiwi_client.py` — plaintext password auth on SND and W/F streams. +- `js8call/Dockerfile` — INI cleanup, env var rename. +- `js8call/tests/test_kiwi_compatibility.py` — auth tests updated to the + protocol-correct plaintext form. +- `docker-compose.yml` — `JS8CALL_PORT` (unused) → `JS8CALL_UDP_SERVER_PORT`. + +## Verification + +- `cd frontend && pnpm run lint && pnpm run typecheck && pnpm run test` — + clean; 278/278 tests pass. +- `cd js8call && uv tool run ruff check . && uv run python -m pytest` — clean; + 26/26 tests pass (two tests asserting the incorrect MD5 auth form were + updated to assert the reference-kiwiclient plaintext form). +- JS8Call/KiwiSDR runtime paths require a container rebuild + (`docker compose up -d --build sovereign-js8call`) and a live JS8Call/KiwiSDR + to exercise end-to-end; protocol behavior was verified against the reference + kiwiclient implementation and the JS8Call/WSJT-X UDP API model. + +## Benefits + +- Entity/trail layer construction does ~5x fewer array allocations per frame + and the whole pipeline does bounded work per second regardless of display + refresh rate — steadier frame times with thousands of COTs, less GC churn. +- No more periodic main-thread stalls from maritime cache serialization; the + cache can no longer blow the localStorage quota. +- The JS8 terminal can actually exchange traffic with JS8Call (RX spots, + directed messages, TX, freq/speed control), and password-protected KiwiSDR + nodes can authenticate. diff --git a/docker-compose.yml b/docker-compose.yml index 0031f72b..bf3e7dd9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -360,7 +360,7 @@ services: - KIWI_MODE=${KIWI_MODE:-usb} - MY_GRID=${MY_GRID:-CN85} - JS8CALL_HOST=0.0.0.0 - - JS8CALL_PORT=2442 + - JS8CALL_UDP_SERVER_PORT=2242 - BRIDGE_PORT=8080 - ALLOWED_ORIGINS=${ALLOWED_ORIGINS:-http://localhost,http://127.0.0.1} - AUTH_ENABLED=${AUTH_ENABLED:-true} diff --git a/frontend/src/hooks/useAnimationLoop.ts b/frontend/src/hooks/useAnimationLoop.ts index 640e37e4..11d1e8f2 100644 --- a/frontend/src/hooks/useAnimationLoop.ts +++ b/frontend/src/hooks/useAnimationLoop.ts @@ -525,10 +525,33 @@ export function useAnimationLoop({ }, [filters?.showClausalChains, filters?.clausalLookbackHours]); useEffect(() => { + // Adaptive frame pacing. The per-tick work (interpolation of every track, + // full layer recomposition, GPU attribute re-upload for the entity/trail + // layers) scales linearly with entity count, so at thousands of COTs a + // steady 30 fps is both smoother and far cheaper than a janky + // display-rate loop — and on 120/144 Hz displays an uncapped rAF loop + // does 2-2.5x redundant work even at low counts. + const PACE_ENTITY_THRESHOLD = 800; // above this, pace to ~30 fps + const FRAME_BUDGET_BUSY_MS = 33; // ~30 fps + const FRAME_BUDGET_IDLE_MS = 15; // ~60 fps cap (skip extra 120/144 Hz ticks) + const animate = () => { + // Schedule the next tick first so an early (paced) return keeps the loop alive. + rafRef.current = requestAnimationFrame(animate); + const entities = entitiesRef.current; const now = Date.now(); const rawDt = now - lastFrameTimeRef.current; + + const entityLoad = entities.size + satellitesRef.current.size; + const frameBudget = + entityLoad > PACE_ENTITY_THRESHOLD + ? FRAME_BUDGET_BUSY_MS + : FRAME_BUDGET_IDLE_MS; + // Not yet due: skip all work this tick. dt keeps accumulating, so the + // interpolators see the true elapsed time on the next executed frame. + if (rawDt < frameBudget) return; + const dt = Math.min(rawDt, 100); lastFrameTimeRef.current = now; @@ -820,8 +843,6 @@ export function useAnimationLoop({ onHover: onOverlayHoverRef.current, }); } - - rafRef.current = requestAnimationFrame(animate); }; const rafId = requestAnimationFrame(animate); diff --git a/frontend/src/hooks/useEntityWorker.ts b/frontend/src/hooks/useEntityWorker.ts index 2288c3c0..1a7064e4 100644 --- a/frontend/src/hooks/useEntityWorker.ts +++ b/frontend/src/hooks/useEntityWorker.ts @@ -13,7 +13,10 @@ import { startWorkerProtocol } from "../workers/WorkerProtocol"; const SEA_ENTITY_CACHE_KEY = "tracks:sea:recent"; const SEA_ENTITY_CACHE_TTL_MS = 5 * 60 * 1000; -const SEA_ENTITY_CACHE_WRITE_INTERVAL_MS = 10 * 1000; +const SEA_ENTITY_CACHE_WRITE_INTERVAL_MS = 30 * 1000; +// Cap the snapshot so JSON.stringify stays bounded (and inside the ~5 MB +// localStorage quota) even when thousands of AIS tracks are in memory. +const SEA_ENTITY_CACHE_MAX = 750; /** * Dead-reckoning anchor for an update: the position's source epoch when the @@ -60,6 +63,7 @@ function buildSeaSnapshot(entities: Map): CachedSeaEntity[] { const out: CachedSeaEntity[] = []; entities.forEach((entity) => { if (!isSeaEntity(entity)) return; + if (out.length >= SEA_ENTITY_CACHE_MAX * 2) return; out.push({ uid: entity.uid, lat: entity.lat, @@ -84,9 +88,15 @@ function buildSeaSnapshot(entities: Map): CachedSeaEntity[] { function writeSeaSnapshot(entities: Map): void { try { + let snapshot = buildSeaSnapshot(entities); + if (snapshot.length > SEA_ENTITY_CACHE_MAX) { + snapshot = snapshot + .sort((a, b) => b.lastSeen - a.lastSeen) + .slice(0, SEA_ENTITY_CACHE_MAX); + } const payload = { savedAt: Date.now(), - entities: buildSeaSnapshot(entities), + entities: snapshot, }; localStorage.setItem(SEA_ENTITY_CACHE_KEY, JSON.stringify(payload)); } catch { @@ -94,6 +104,24 @@ function writeSeaSnapshot(entities: Map): void { } } +// Serialize + write during idle time instead of inline on the WebSocket +// message path — with thousands of AIS tracks the JSON.stringify alone is a +// multi-ms main-thread hitch that competes with the render loop. +let seaSnapshotScheduled = false; +function scheduleSeaSnapshotWrite(entities: Map): void { + if (seaSnapshotScheduled) return; + seaSnapshotScheduled = true; + const run = () => { + seaSnapshotScheduled = false; + writeSeaSnapshot(entities); + }; + if (typeof requestIdleCallback === "function") { + requestIdleCallback(run, { timeout: 5000 }); + } else { + setTimeout(run, 250); + } +} + function restoreSeaSnapshot( entitiesRef: MutableRefObject>, knownUidsRef: MutableRefObject>, @@ -526,7 +554,7 @@ export function useEntityWorker({ nowMs - lastSeaCacheWriteRef.current >= SEA_ENTITY_CACHE_WRITE_INTERVAL_MS ) { - writeSeaSnapshot(entitiesRef.current); + scheduleSeaSnapshotWrite(entitiesRef.current); lastSeaCacheWriteRef.current = nowMs; } } diff --git a/frontend/src/layers/buildEntityLayers.ts b/frontend/src/layers/buildEntityLayers.ts index 91b5f1e4..94d86bbb 100644 --- a/frontend/src/layers/buildEntityLayers.ts +++ b/frontend/src/layers/buildEntityLayers.ts @@ -30,15 +30,72 @@ export function buildEntityLayers( ): Layer[] { const layers: Layer[] = []; - // GPS Integrity warning halos — amber ring on aircraft with degraded NIC/NACp (Ingest-04) - const integrityDegraded = interpolated.filter((e) => { - const nic = e.classification?.nic; - const nacp = e.classification?.nacP; - return ( + // Single pass over the interpolated entities. This function runs on every + // paced animation frame with (potentially) thousands of COTs, so the + // per-layer datasets are derived in one walk instead of 5-6 separate + // filter/map passes that each allocate an intermediate array. + const integrityDegraded: CoTEntity[] = []; + const airborne: CoTEntity[] = []; + const haloEntities: CoTEntity[] = []; + const selectedArr: CoTEntity[] = []; + const velocityData: VelocityDatum[] = []; + + for (const d of interpolated) { + const nic = d.classification?.nic; + const nacp = d.classification?.nacP; + if ( (nic !== null && nic !== undefined && nic <= 4) || (nacp !== null && nacp !== undefined && nacp <= 6) - ); - }); + ) { + integrityDegraded.push(d); + } + + if (enable3d && d.altitude > 10) { + airborne.push(d); + } + + const isVessel = d.type.includes("S"); + if (isVessel) { + const cat = d.vesselClassification?.category || ""; + if (cat === "sar" || cat === "military" || cat === "law_enforcement") { + haloEntities.push(d); + } + } else { + const platform = d.classification?.platform || ""; + const affiliation = d.classification?.affiliation || ""; + if ( + platform === "helicopter" || + platform === "drone" || + affiliation === "military" || + affiliation === "government" + ) { + haloEntities.push(d); + } + } + + if (currentSelected && d.uid === currentSelected.uid) { + selectedArr.push(d); + } + + if (velocityVectorsEnabled && d.speed > 0.1) { + const projectionSeconds = 45; + const distMeters = d.speed * projectionSeconds; + const courseRad = ((d.course || 0) * Math.PI) / 180; + const R = 6371000; + const latRad = (d.lat * Math.PI) / 180; + const dLat = (distMeters * Math.cos(courseRad)) / R; + const dLon = (distMeters * Math.sin(courseRad)) / (R * Math.cos(latRad)); + const target: PathPoint3D = [ + d.lon + dLon * (180 / Math.PI), + d.lat + dLat * (180 / Math.PI), + d.altitude || 0, + ]; + velocityData.push({ + path: [[d.lon, d.lat, d.altitude || 0] as PathPoint3D, target], + entity: d, + }); + } + } if (integrityDegraded.length > 0) { layers.push( @@ -74,7 +131,7 @@ export function buildEntityLayers( layers.push( new LineLayer({ id: `altitude-stems-${globeMode ? "globe" : "merc"}`, - data: interpolated.filter((e) => e.altitude > 10), // Only for airborne + data: airborne, // Only for airborne getSourcePosition: (d: CoTEntity) => [d.lon, d.lat, 0], getTargetPosition: (d: CoTEntity) => [d.lon, d.lat, d.altitude], getColor: (d: CoTEntity) => entityColor(d, 80), // Faint line @@ -85,7 +142,7 @@ export function buildEntityLayers( }), new ScatterplotLayer({ id: `ground-shadows-${globeMode ? "globe" : "merc"}`, - data: interpolated.filter((e) => e.altitude > 10), + data: airborne, getPosition: (d: CoTEntity) => [d.lon, d.lat, 0], getRadius: 3, radiusUnits: "pixels" as const, @@ -104,23 +161,7 @@ export function buildEntityLayers( layers.push( new IconLayer({ id: `entity-tactical-halo-${globeMode ? "globe" : "merc"}`, - data: interpolated.filter((d) => { - const isVessel = d.type.includes("S"); - if (isVessel) { - return ["sar", "military", "law_enforcement"].includes( - d.vesselClassification?.category || "", - ); - } else { - return ( - ["helicopter", "drone"].includes( - d.classification?.platform || "", - ) || - ["military", "government"].includes( - d.classification?.affiliation || "", - ) - ); - } - }), + data: haloEntities, getIcon: () => "halo", iconAtlas: ICON_ATLAS.url, iconMapping: ICON_ATLAS.mapping, @@ -296,7 +337,7 @@ export function buildEntityLayers( layers.push( new ScatterplotLayer({ id: `selection-ring-${currentSelected.uid}-${globeMode ? "globe" : "merc"}`, - data: interpolated.filter((e) => e.uid === currentSelected.uid), + data: selectedArr, getPosition: (d: CoTEntity) => [d.lon, d.lat, d.altitude || 0], getRadius: () => { const cycle = (now % 2000) / 2000; // Faster pulse (2s) @@ -327,27 +368,7 @@ export function buildEntityLayers( layers.push( new PathLayer({ id: `velocity-vectors-${globeMode ? "globe" : "merc"}`, - data: interpolated - .filter((e) => e.speed > 0.1) - .map((d) => { - const projectionSeconds = 45; - const distMeters = d.speed * projectionSeconds; - const courseRad = ((d.course || 0) * Math.PI) / 180; - const R = 6371000; - const latRad = (d.lat * Math.PI) / 180; - const dLat = (distMeters * Math.cos(courseRad)) / R; - const dLon = - (distMeters * Math.sin(courseRad)) / (R * Math.cos(latRad)); - const target: PathPoint3D = [ - d.lon + dLon * (180 / Math.PI), - d.lat + dLat * (180 / Math.PI), - d.altitude || 0, - ]; - return { - path: [[d.lon, d.lat, d.altitude || 0] as PathPoint3D, target], - entity: d, - }; - }), + data: velocityData, getPath: (d: VelocityDatum) => d.path, getColor: (d: VelocityDatum) => entityColor(d.entity, 120), getWidth: 2.2, diff --git a/frontend/src/layers/buildTrailLayers.ts b/frontend/src/layers/buildTrailLayers.ts index f2464183..bc95507e 100644 --- a/frontend/src/layers/buildTrailLayers.ts +++ b/frontend/src/layers/buildTrailLayers.ts @@ -20,6 +20,23 @@ interface TrailPathDatum { const toPath3D = (points: number[][]): PathPoint3D[] => points.map((pt) => [pt[0] ?? 0, pt[1] ?? 0, pt[2] ?? 0]); +// Per-trail-array conversion cache. Smoothed trails are regenerated only when +// an entity receives a server update (~1/s), while this builder runs on every +// animation frame for every visible entity — the WeakMap makes the frame cost +// a lookup instead of reallocating a path array per entity per frame. +const path3DCache = new WeakMap(); +const EMPTY_PATH: PathPoint3D[] = []; + +const toPath3DCached = (points: number[][] | undefined): PathPoint3D[] => { + if (!points || points.length === 0) return EMPTY_PATH; + let path = path3DCache.get(points); + if (!path) { + path = toPath3D(points); + path3DCache.set(points, path); + } + return path; +}; + export function buildTrailLayers( interpolated: CoTEntity[], currentSelected: CoTEntity | null, @@ -31,15 +48,33 @@ export function buildTrailLayers( // 1. All History Trails (Global Toggle) // Filter out the selected entity's trail to avoid z-fighting/jaggedness if (historyTailsEnabled) { + // Single pass: derive both the trail dataset and the gap-bridge dataset. + const trailEntities: CoTEntity[] = []; + const gapBridges: GapBridgeDatum[] = []; + for (const d of interpolated) { + if (currentSelected && d.uid === currentSelected.uid) continue; + if (!d.trail || d.trail.length === 0) continue; + + if (d.trail.length >= 2) trailEntities.push(d); + + const last = d.trail[d.trail.length - 1]; + const dist = getDistanceMeters(last[1], last[0], d.lat, d.lon); + if (dist > 5) { + gapBridges.push({ + path: [ + [last[0], last[1], last[2]] as PathPoint3D, + [d.lon, d.lat, d.altitude || 0] as PathPoint3D, + ], + entity: d, + }); + } + } + layers.push( new PathLayer({ id: `all-history-trails-${globeMode ? "globe" : "merc"}`, - data: interpolated.filter( - (e) => - e.trail.length >= 2 && - (!currentSelected || e.uid !== currentSelected.uid), - ), - getPath: (d: CoTEntity) => toPath3D(d.smoothedTrail || []), + data: trailEntities, + getPath: (d: CoTEntity) => toPath3DCached(d.smoothedTrail), getColor: (d: CoTEntity) => { const isShip = d.type.includes("S"); return isShip @@ -63,24 +98,7 @@ export function buildTrailLayers( layers.push( new PathLayer({ id: `history-gap-bridge-${globeMode ? "globe" : "merc"}`, - data: interpolated - .filter((d) => { - if (!d.trail || d.trail.length === 0) return false; - if (currentSelected && d.uid === currentSelected.uid) return false; - const last = d.trail[d.trail.length - 1]; - const dist = getDistanceMeters(last[1], last[0], d.lat, d.lon); - return dist > 5; - }) - .map((d) => { - const last = d.trail![d.trail!.length - 1]; - return { - path: [ - [last[0], last[1], last[2]] as PathPoint3D, - [d.lon, d.lat, d.altitude || 0] as PathPoint3D, - ], - entity: d, - }; - }), + data: gapBridges, getPath: (d: GapBridgeDatum) => d.path, getColor: (d: GapBridgeDatum) => entityColor(d.entity, 180), getWidth: 3.5, diff --git a/frontend/src/workers/tak.worker.ts b/frontend/src/workers/tak.worker.ts index 89bd2caf..f0326ee6 100644 --- a/frontend/src/workers/tak.worker.ts +++ b/frontend/src/workers/tak.worker.ts @@ -4,10 +4,13 @@ import { load, Type } from 'protobufjs'; let takType: Type | null = null; // let processing = false; -// Batching: accumulate decoded entities and flush periodically +// Batching: accumulate decoded entities and flush periodically. +// A larger batch means fewer worker→main postMessage wakeups during dense +// bursts (the orbital sweep alone emits ~11k messages per cycle); latency is +// still bounded by FLUSH_INTERVAL_MS for sparse traffic. let batch: unknown[] = []; let flushTimer: ReturnType | null = null; -const BATCH_SIZE = 10; +const BATCH_SIZE = 64; const FLUSH_INTERVAL_MS = 50; function flushBatch() { diff --git a/js8call/Dockerfile b/js8call/Dockerfile index 06390dac..1b1c26cf 100644 --- a/js8call/Dockerfile +++ b/js8call/Dockerfile @@ -148,7 +148,10 @@ RUN wget -q -O /tmp/JS8Call-2.5.2-x86_64.AppImage "https://github.com/JS8Call-im && chmod +x /tmp/JS8Call-2.5.2-x86_64.AppImage \ && cd /opt && /tmp/JS8Call-2.5.2-x86_64.AppImage --appimage-extract \ && ln -s /opt/squashfs-root/AppRun /usr/bin/js8call \ - && mkdir -p /root/.config && printf "[Main]\n[Configuration]\nTCPEnabled=false\nUDPEnabled=true\nUDPServer=127.0.0.1\nUDPServerPort=2242\nAcceptUDPRequests=true\nUDPClient2=127.0.0.1\nUDPClient2Port=2245\nMyCall=N0CALL\nMyGrid=CN85\n[MultiSettings]\n" > "/root/.config/JS8Call - KiwiSDR-Virtual.ini" \ + # UDPServer/UDPServerPort = where JS8Call PUSHES its API event datagrams + # (PING/RX.SPOT/RX.DIRECTED/...). The FastAPI bridge binds that port and + # replies to the datagram source address for commands (AcceptUDPRequests). + && mkdir -p /root/.config && printf "[Main]\n[Configuration]\nTCPEnabled=false\nUDPEnabled=true\nUDPServer=127.0.0.1\nUDPServerPort=2242\nAcceptUDPRequests=true\nMyCall=N0CALL\nMyGrid=CN85\n[MultiSettings]\n" > "/root/.config/JS8Call - KiwiSDR-Virtual.ini" \ && cp "/root/.config/JS8Call - KiwiSDR-Virtual.ini" /root/.config/JS8Call.ini \ && rm /tmp/JS8Call-2.5.2-x86_64.AppImage && rm -rf /var/lib/apt/lists/* @@ -231,9 +234,9 @@ ENV KIWI_PORT=8073 ENV KIWI_FREQ=14078 ENV KIWI_MODE=usb -# JS8Call API port +# JS8Call UDP API — port the bridge binds to receive JS8Call event datagrams ENV JS8CALL_HOST=127.0.0.1 -ENV JS8CALL_PORT=2442 +ENV JS8CALL_UDP_SERVER_PORT=2242 # Ensure Python logs are unbuffered ENV PYTHONUNBUFFERED=1 diff --git a/js8call/kiwi_client.py b/js8call/kiwi_client.py index 769de0e3..5f3e13c2 100644 --- a/js8call/kiwi_client.py +++ b/js8call/kiwi_client.py @@ -26,13 +26,11 @@ Legacy KiwiSDR: ws://host:port//SND Both formats are tried automatically (modern first). -Authentication: - Open nodes (no password): SET auth t=kiwi p= - Password-protected nodes (modern): SET auth t=kiwi pwd= +Authentication (matches reference kiwiclient): + SET auth t=kiwi p= (p= empty for open public nodes) """ import asyncio -import hashlib import logging import time from typing import Callable, Optional, Dict @@ -110,14 +108,12 @@ def _make_auth_cmd(password: str) -> str: """Build the SET auth command for the given password. - Open nodes (empty password) use the legacy ``p=`` form accepted by all - KiwiSDR versions. Password-protected nodes use ``pwd=`` as required - by current KiwiSDR server code (rx/rx_cmd.cpp). + Matches the reference kiwiclient implementation: the password is sent + in plaintext via ``p=`` (empty for open public nodes). The previous + ``pwd=`` form is not part of the KiwiSDR protocol and caused every + password-protected connection to be rejected. """ - if not password: - return "SET auth t=kiwi p=" - md5 = hashlib.md5(password.encode()).hexdigest() - return f"SET auth t=kiwi pwd={md5}" + return f"SET auth t=kiwi p={password}" # --------------------------------------------------------------------------- @@ -218,8 +214,8 @@ async def connect( ---------- password : Optional KiwiSDR password. Required for password-protected or private nodes; leave empty ("") for open public nodes. - Non-empty passwords are hashed with MD5 per the current - KiwiSDR auth protocol (``SET auth t=kiwi pwd=``). + Sent in plaintext via ``SET auth t=kiwi p=`` per + the KiwiSDR protocol (reference kiwiclient behaviour). """ if not _HAS_WEBSOCKETS: raise RuntimeError("websockets library not installed") @@ -273,7 +269,7 @@ async def connect( }) if self._on_waterfall: - await self._start_waterfall(host, port) + await self._start_waterfall(host, port, password) logger.info("KiwiClient connected: %s:%d @ %.3f kHz %s", host, port, freq_khz, mode) @@ -675,9 +671,8 @@ async def _wrapper(): async def _handshake(self, freq_khz: float, mode: str, password: str = "") -> None: """Execute the KiwiSDR SND handshake sequence. - Auth format follows the current KiwiSDR protocol: - - Open nodes (empty password): ``SET auth t=kiwi p=`` - - Password-protected nodes: ``SET auth t=kiwi pwd=`` + Auth format follows the KiwiSDR protocol (reference kiwiclient): + ``SET auth t=kiwi p=`` with an empty ``p=`` for open nodes. """ self._freq_khz = freq_khz self._mode = mode @@ -782,11 +777,11 @@ async def _keepalive_loop(self) -> None: except Exception as exc: logger.debug("KiwiClient keepalive error: %s", exc) - async def _start_waterfall(self, host: str, port: int) -> None: + async def _start_waterfall(self, host: str, port: int, password: str = "") -> None: """Start the KiwiSDR waterfall stream (W/F). Full W/F handshake per the KiwiSDR protocol: - 1. SET auth t=kiwi p= — authenticate (waterfall always open) + 1. SET auth t=kiwi p= — same credentials as the SND stream 2. SET zoom= cf= — centres waterfall on audio frequency 3. SET maxdb=-10 mindb=-110 — colour scale 4. SET wf_speed=4 — rows/second; must be > 0 to receive frames @@ -820,8 +815,8 @@ async def _start_waterfall(self, host: str, port: int) -> None: raise RuntimeError("Could not connect to waterfall on any URL format") self._wf_ws = ws - # Waterfall auth is always open (public endpoint regardless of node password) - await ws.send("SET auth t=kiwi p=") + # Same auth as the SND stream (password required on protected nodes) + await ws.send(_make_auth_cmd(password)) await ws.send(f"SET zoom={self._zoom} cf={self._freq_khz:.3f}") await ws.send("SET maxdb=-10 mindb=-110") await ws.send("SET wf_speed=4") diff --git a/js8call/server.py b/js8call/server.py index f977378c..71fa55f5 100644 --- a/js8call/server.py +++ b/js8call/server.py @@ -87,8 +87,12 @@ # Configuration (read from environment; Dockerfile sets sensible defaults) # --------------------------------------------------------------------------- JS8CALL_HOST = os.getenv("JS8CALL_HOST", "0.0.0.0") +# JS8Call's UDP API follows the WSJT-X model: JS8Call is the *client*. It +# binds an ephemeral local port and pushes event datagrams to the configured +# "UDP Server" address (our JS8Call.ini sets 127.0.0.1:2242). This bridge is +# therefore the UDP *server*: it must bind 2242 to receive events, and send +# commands back to the source address of JS8Call's own datagrams. JS8CALL_UDP_SERVER_PORT = int(os.getenv("JS8CALL_UDP_SERVER_PORT", "2242")) -JS8CALL_UDP_CLIENT_PORT = int(os.getenv("JS8CALL_UDP_CLIENT_PORT", "2245")) BRIDGE_PORT = int(os.getenv("BRIDGE_PORT", "8080")) MY_GRID = os.getenv("MY_GRID", "CN85") # Operator's Maidenhead locator @@ -198,6 +202,18 @@ async def get_current_user( # Written from the background task (single asyncio thread) – no lock needed. _station_registry: dict[str, dict] = {} +# JS8Call reply routing state. JS8Call binds an ephemeral UDP port and pushes +# events (PING every ~15 s, RX.*, STATION.*) to our bound server port; commands +# must be sent back to that observed source address, not to a fixed port. +_js8_reply_addr: Optional[tuple[str, int]] = None +_js8_last_heard: float = 0.0 +JS8_HEARD_TIMEOUT = 60.0 # seconds without a datagram → considered disconnected + +# Merged local station state (callsign/grid/freq/speed), updated from JS8Call +# responses so partial updates can be broadcast as complete STATION.STATUS +# payloads (the frontend overwrites its grid with whatever arrives). +_station_state: dict = {} + # KiwiSDR subprocess state – managed by _start/_stop_kiwi_pipeline(). # Accessed from both asyncio executor threads and the main thread; guarded by _kiwi_lock. _kiwi_proc: Optional[subprocess.Popen] = None @@ -222,15 +238,43 @@ async def get_current_user( # Utilities # =========================================================================== -def _udp_send(msg: dict) -> None: - """Send a single UDP datagram to the JS8Call API port. Fire-and-forget.""" +def _udp_send(msg_type: str, value="", params: Optional[dict] = None) -> None: + """ + Send a single JS8Call API datagram. Fire-and-forget. + + The JSON envelope uses lowercase keys ({"type", "value", "params"}) per + the JS8Call API. Commands are sent to the source address JS8Call's own + datagrams arrive from — until JS8Call has announced itself (first PING), + there is nowhere to send and the command is dropped with a warning. + """ + if _js8_reply_addr is None: + logger.warning( + "UDP send dropped (%s): JS8Call has not announced itself yet " + "(no datagram received on port %d)", + msg_type, JS8CALL_UDP_SERVER_PORT, + ) + return + msg = {"type": msg_type, "value": value, "params": params or {}} try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as tx: - tx.sendto(json.dumps(msg).encode("utf-8") + b"\n", ("127.0.0.1", JS8CALL_UDP_SERVER_PORT)) + tx.sendto(json.dumps(msg).encode("utf-8") + b"\n", _js8_reply_addr) except Exception as exc: logger.warning("UDP send failed: %s", exc) +def _js8_is_connected() -> bool: + """True if JS8Call has sent us a datagram recently (it PINGs every ~15 s).""" + return ( + _js8_reply_addr is not None + and (time.monotonic() - _js8_last_heard) < JS8_HEARD_TIMEOUT + ) + + +# JS8Call submode integers used by MODE.SET_SPEED / MODE.SPEED +_JS8_SPEED_TO_INT = {"SLOW": 4, "NORMAL": 0, "FAST": 1, "TURBO": 2} +_JS8_INT_TO_SPEED = {v: k for k, v in _JS8_SPEED_TO_INT.items()} + + # =========================================================================== # KiwiSDR Pipeline Management # =========================================================================== @@ -750,34 +794,100 @@ def on_station_status(message: dict) -> None: "timestamp": time.strftime("%H:%M:%SZ", time.gmtime()), "ts_unix": int(time.time()), } + # Keep the merged local state in sync so later partial updates + # (STATION.CALLSIGN, RIG.FREQ, …) broadcast complete payloads. + speed_raw = params.get("SPEED") + _station_state.update({ + k: v + for k, v in { + "callsign": params.get("CALL"), + "grid": params.get("GRID"), + "freq": params.get("FREQ"), + "speed": _JS8_INT_TO_SPEED.get(speed_raw) + if isinstance(speed_raw, int) + else None, + }.items() + if v + }) _enqueue_from_thread(payload) except Exception as exc: logger.warning("on_station_status error: %s", exc) +def _update_station_state(**fields) -> None: + """Merge fields into the local station state and broadcast the result.""" + _station_state.update({k: v for k, v in fields.items() if v is not None}) + payload = { + "type": "STATION.STATUS", + "callsign": _station_state.get("callsign", ""), + "grid": _station_state.get("grid", MY_GRID), + "freq": _station_state.get("freq", 0), + "speed": _station_state.get("speed", ""), + "status": "", + "timestamp": time.strftime("%H:%M:%SZ", time.gmtime()), + "ts_unix": int(time.time()), + } + _enqueue_from_thread(payload) + + class JS8CallUDPProtocol(asyncio.DatagramProtocol): def connection_made(self, transport): self.transport = transport - logger.info("JS8Call UDP API listener active on port %d", JS8CALL_UDP_CLIENT_PORT) + logger.info("JS8Call UDP API listener active on port %d", JS8CALL_UDP_SERVER_PORT) def datagram_received(self, data, addr): + global _js8_reply_addr, _js8_last_heard # Only accept datagrams from localhost — JS8Call runs on the same host. sender_ip = addr[0] if addr else "" if sender_ip not in ("127.0.0.1", "::1"): logger.warning("UDP: rejected datagram from unexpected source %s", sender_ip) return + + # Every datagram (including the periodic PING) tells us where JS8Call's + # socket lives — commands must be sent back to this address. + first_contact = _js8_reply_addr is None + _js8_reply_addr = addr + _js8_last_heard = time.monotonic() + try: line = data.decode("utf-8").strip() if not line: return message = json.loads(line) m_type = message.get("type", "") + params = message.get("params", {}) or {} + value = message.get("value", "") + + if first_contact: + logger.info("JS8Call announced itself from %s (type=%s)", addr, m_type) + # Pull the initial station state now that we can reach it. + _udp_send("STATION.GET_CALLSIGN") + _udp_send("STATION.GET_GRID") + _udp_send("RIG.GET_FREQ") + _udp_send("MODE.GET_SPEED") + if m_type == "RX.DIRECTED": on_rx_directed(message) elif m_type == "RX.SPOT": on_rx_spot(message) elif m_type == "STATION.STATUS": on_station_status(message) + elif m_type == "STATION.CALLSIGN": + _update_station_state(callsign=str(value).strip() or None) + elif m_type == "STATION.GRID": + _update_station_state(grid=str(value).strip() or None) + elif m_type == "RIG.FREQ": + dial = params.get("DIAL") or params.get("FREQ") + if isinstance(dial, (int, float)): + _update_station_state(freq=int(dial)) + elif m_type == "MODE.SPEED": + speed_int = params.get("SPEED") + if isinstance(speed_int, int): + _update_station_state( + speed=_JS8_INT_TO_SPEED.get(speed_int, "NORMAL") + ) + elif m_type == "PING": + pass # heartbeat — reply address / liveness already recorded elif m_type: logger.debug("UDP: ignoring unknown message type %r from %s", m_type, sender_ip) except json.JSONDecodeError as exc: @@ -845,19 +955,19 @@ async def lifespan(app: FastAPI): try: logger.info( "Starting UDP listener on %s:%d (attempt %d/5)...", - JS8CALL_HOST, JS8CALL_UDP_CLIENT_PORT, attempt, + JS8CALL_HOST, JS8CALL_UDP_SERVER_PORT, attempt, ) transport, protocol = await _event_loop.create_datagram_endpoint( lambda: JS8CallUDPProtocol(), - local_addr=(JS8CALL_HOST, JS8CALL_UDP_CLIENT_PORT), + local_addr=(JS8CALL_HOST, JS8CALL_UDP_SERVER_PORT), ) js8_client_udp_transport = transport logger.info( - "UDP listener bound to %s:%d", JS8CALL_HOST, JS8CALL_UDP_CLIENT_PORT + "UDP listener bound to %s:%d", JS8CALL_HOST, JS8CALL_UDP_SERVER_PORT ) break except Exception as exc: - logger.warning("Failed to bind UDP listener (port %d): %s", JS8CALL_UDP_CLIENT_PORT, exc) + logger.warning("Failed to bind UDP listener (port %d): %s", JS8CALL_UDP_SERVER_PORT, exc) if attempt < 5: await asyncio.sleep(2) else: @@ -947,14 +1057,14 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) remote = websocket.client logger.info("WebSocket connected: %s", remote) - # Send immediate simulated connect message - callsign = os.getenv("JS8CALL_CALLSIGN", "N0CALL") - grid = MY_GRID - + # Send immediate connect message with the best station state we have + callsign = _station_state.get("callsign") or os.getenv("JS8CALL_CALLSIGN", "N0CALL") + grid = _station_state.get("grid") or MY_GRID + await websocket.send_json({ "type": "CONNECTED", "message": "JS8Call bridge active", - "js8call_connected": js8_client_udp_transport is not None, + "js8call_connected": _js8_is_connected(), "kiwi_connected": _kiwi_is_running(), "kiwi_host": _kiwi_config.get("host", ""), "kiwi_port": _kiwi_config.get("port", 0), @@ -962,11 +1072,17 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) "kiwi_mode": _kiwi_config.get("mode", ""), "callsign": callsign, "grid": grid, + "speed": _station_state.get("speed", "NORMAL"), "timestamp": time.strftime("%H:%M:%SZ", time.gmtime()), }) - # Ask JS8Call to broadcast its STATUS via UDP immediately - _udp_send({"TYPE": "STATION.GET_STATUS", "VALUE": "", "PARAMS": {}}) + # Ask JS8Call for its current state (responses arrive as UDP datagrams + # typed STATION.CALLSIGN / STATION.GRID / RIG.FREQ / MODE.SPEED). + if _js8_reply_addr is not None: + _udp_send("STATION.GET_CALLSIGN") + _udp_send("STATION.GET_GRID") + _udp_send("RIG.GET_FREQ") + _udp_send("MODE.GET_SPEED") try: # Receive loop – handle commands from the frontend @@ -1000,7 +1116,7 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) tx_target = target.upper() tx_msg = f"{tx_target} {message}" # Forward dynamically to JS8Call UDP port - _udp_send({"TYPE": "TX.SEND_MESSAGE", "VALUE": tx_msg, "PARAMS": {}}) + _udp_send("TX.SEND_MESSAGE", tx_msg) # Echo the sent message back so the UI can display it in the log _enqueue_from_thread({ "type": "TX.SENT", @@ -1025,7 +1141,11 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) "message": f"SET_MODE: invalid mode '{requested_mode}'. Valid: {sorted(_VALID_MODES)}", }) else: - _udp_send({"TYPE": "MODE.SET_SPEED", "VALUE": requested_mode, "PARAMS": {}}) + # JS8Call expects the numeric submode in params.SPEED + _udp_send( + "MODE.SET_SPEED", + params={"SPEED": _JS8_SPEED_TO_INT[requested_mode]}, + ) logger.info("SET_MODE → %s", requested_mode) # ------------------------------------------------------------------ @@ -1038,7 +1158,7 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) await websocket.send_json({"type": "ERROR", "message": "SET_FREQ: freq must be 100 kHz–500 MHz in Hz"}) continue # Forward dynamically to JS8Call UDP port - _udp_send({"TYPE": "RIG.SET_FREQ", "VALUE": freq, "PARAMS": {}}) + _udp_send("RIG.SET_FREQ", params={"DIAL": freq}) # ------------------------------------------------------------------ # Action: GET_STATIONS – force a station list refresh @@ -1103,7 +1223,7 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) }) # Sync JS8Call dial frequency to the KiwiSDR dial frequency # so that decoded message metadata reflects the correct band. - _udp_send({"TYPE": "RIG.SET_FREQ", "VALUE": int(float(freq) * 1000), "PARAMS": {}}) + _udp_send("RIG.SET_FREQ", params={"DIAL": int(float(freq) * 1000)}) except ValueError as exc: await websocket.send_json({"type": "ERROR", "message": f"SET_KIWI validation: {exc}"}) except Exception as exc: @@ -1128,7 +1248,7 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) await _kiwi_native.connect(host, port, float(freq), mode, password=password) # Sync JS8Call dial frequency to the KiwiSDR dial frequency # so that decoded message metadata reflects the correct band. - _udp_send({"TYPE": "RIG.SET_FREQ", "VALUE": int(float(freq) * 1000), "PARAMS": {}}) + _udp_send("RIG.SET_FREQ", params={"DIAL": int(float(freq) * 1000)}) except ValueError as exc: await websocket.send_json({"type": "ERROR", "message": f"SET_KIWI validation: {exc}"}) except Exception as exc: @@ -1395,10 +1515,10 @@ async def ws_js8(websocket: WebSocket, token: str | None = Query(default=None)) }) else: if new_callsign: - _udp_send({"TYPE": "STATION.SET_CALLSIGN", "VALUE": new_callsign, "PARAMS": {}}) + _udp_send("STATION.SET_CALLSIGN", new_callsign) logger.info("SET_STATION callsign → %s", new_callsign) if new_grid: - _udp_send({"TYPE": "STATION.SET_GRID", "VALUE": new_grid, "PARAMS": {}}) + _udp_send("STATION.SET_GRID", new_grid) logger.info("SET_STATION grid → %s", new_grid) # Optimistic echo so the UI updates without waiting for JS8Call confirmation _enqueue_from_thread({ @@ -1626,7 +1746,7 @@ async def get_websdr_nodes( async def health() -> dict: return { "status": "ok", - "js8call_connected": js8_client_udp_transport is not None, + "js8call_connected": _js8_is_connected(), "kiwi_connected": _kiwi_is_running(), } diff --git a/js8call/tests/test_kiwi_compatibility.py b/js8call/tests/test_kiwi_compatibility.py index 909761a1..738a5b5b 100644 --- a/js8call/tests/test_kiwi_compatibility.py +++ b/js8call/tests/test_kiwi_compatibility.py @@ -2,7 +2,7 @@ test_kiwi_compatibility.py — Tests covering KiwiSDR protocol compatibility fixes. Tests: - - MD5 auth command generation (_make_auth_cmd) + - Auth command generation (_make_auth_cmd, plaintext p= per reference kiwiclient) - MODE_FILTERS completeness (all 18 official modes present) - New DSP methods: notch, NR, noise filter, RF attn, passband, mute - New waterfall controls: cmap, aperture @@ -33,20 +33,11 @@ def test_open_node_uses_legacy_p_form(self): cmd = _make_auth_cmd("") self.assertEqual(cmd, "SET auth t=kiwi p=") - def test_password_uses_md5_pwd_form(self): + def test_password_uses_plaintext_p_form(self): + # The KiwiSDR protocol (reference kiwiclient) sends the password in + # plaintext via p= — an MD5 pwd= form is rejected by real nodes. cmd = _make_auth_cmd("secret") - # Must start with modern prefix and contain 32-char hex MD5 - self.assertTrue(cmd.startswith("SET auth t=kiwi pwd="), cmd) - md5_part = cmd.split("pwd=")[1] - self.assertEqual(len(md5_part), 32) - self.assertTrue(all(c in "0123456789abcdef" for c in md5_part), md5_part) - - def test_known_md5_value(self): - import hashlib - pw = "kiwi" - expected = hashlib.md5(pw.encode()).hexdigest() - cmd = _make_auth_cmd(pw) - self.assertIn(expected, cmd) + self.assertEqual(cmd, "SET auth t=kiwi p=secret") # --------------------------------------------------------------------------- From 8710dfeb52cd4078cb9231e1fe93052b013bd98d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 6 Jul 2026 04:38:40 +0000 Subject: [PATCH 3/3] perf: coalesce TAK WebSocket frames + binary attributes for entity icon layer Ingress batching: - BroadcastManager client workers drain their queue and coalesce consecutive proto messages into batch frames (0xbf 0x02 0xbf magic + u32le length-prefixed records; each record is an unmodified legacy frame). Capped at 128 msgs / 60 KB per send; alert JSON ordering preserved; lone messages keep the legacy single frame - Client queue deepened 256 -> 4096 so the ~11k-message orbital sweep is absorbed instead of silently drop-oldest'd for slow clients - Frontend TAK worker decodes both frame formats via a shared batchFraming module (unit-tested on both sides) Rendering: - New EntityIconAttributeCache uploads position/angle/color/size for the 2D entity icon layer as persistent typed arrays filled in one pass per paced frame - deck.gl no longer iterates entity objects for those attributes. Per-frame buffers ping-pong so external-buffer refs change with content; colors/sizes refresh only on membership or selection change, or a 1 s cadence - Picking on the binary layer is index-based; the overlay hover-miss check now keys on info.index so binary picks are not cleared - Globe mode and the no-cache path keep the object-based layers Verification: frontend lint/typecheck/vitest (289 passed); backend ruff + full pytest (164 passed). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01S3R2VyPxQQY1UceaiC5wg6 --- ...ontend-cots-optimization-js8-kiwi-fixes.md | 21 +++ backend/api/services/broadcast.py | 93 +++++++++-- backend/api/tests/test_broadcast_batching.py | 76 +++++++++ frontend/src/hooks/useAnimationLoop.ts | 13 +- frontend/src/layers/buildEntityLayers.ts | 56 +++++++ frontend/src/layers/composition.ts | 8 + .../src/layers/entityIconAttributes.test.ts | 122 ++++++++++++++ frontend/src/layers/entityIconAttributes.ts | 150 ++++++++++++++++++ frontend/src/workers/batchFraming.test.ts | 76 +++++++++ frontend/src/workers/batchFraming.ts | 48 ++++++ frontend/src/workers/tak.worker.ts | 74 +++++---- 11 files changed, 689 insertions(+), 48 deletions(-) create mode 100644 backend/api/tests/test_broadcast_batching.py create mode 100644 frontend/src/layers/entityIconAttributes.test.ts create mode 100644 frontend/src/layers/entityIconAttributes.ts create mode 100644 frontend/src/workers/batchFraming.test.ts create mode 100644 frontend/src/workers/batchFraming.ts diff --git a/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md b/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md index 403a4b48..06ee9e31 100644 --- a/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md +++ b/agent_docs/tasks/2026-07-06-frontend-cots-optimization-js8-kiwi-fixes.md @@ -50,6 +50,27 @@ - KiwiSDR auth: plaintext `p=` per reference kiwiclient, applied to both SND and W/F streams. +## Follow-up (same branch): WS frame batching + binary icon attributes + +- **Coalesced WebSocket frames**: the broadcast service previously sent one + binary frame per Kafka message per client with a 256-deep drop-oldest + queue — the ~11k-message orbital sweep could silently drop most of a slow + client's data. The client worker now drains its queue and coalesces + consecutive proto messages into batch frames (`0xbf 0x02 0xbf` magic + + u32le length-prefixed records, each record an unmodified legacy frame), + capped at 128 messages / 60 KB per send, alert JSON ordering preserved. + Queue deepened to 4096. Single messages still use the legacy frame, and the + frontend worker decodes both formats (`workers/batchFraming.ts`). +- **Binary attributes for the 2D entity icon layer** + (`layers/entityIconAttributes.ts`): position/angle/color/size are uploaded + as persistent typed arrays filled in one pass per paced frame — deck.gl no + longer iterates entity objects for them. Per-frame buffers ping-pong so + external-buffer references change with content; colors/sizes refresh only + on membership/selection change or a 1 s cadence (entityColor tracks + altitude/speed, which drift slowly). Picking is index-based; the overlay + hover-miss check now keys on `info.index` so binary picks aren't cleared. + Globe mode and the no-cache path keep the object-based layers. + ## Changes - `frontend/src/hooks/useAnimationLoop.ts` — adaptive frame pacing; rAF diff --git a/backend/api/services/broadcast.py b/backend/api/services/broadcast.py index 0d417cc2..191a3815 100644 --- a/backend/api/services/broadcast.py +++ b/backend/api/services/broadcast.py @@ -14,8 +14,66 @@ logger = logging.getLogger("SovereignWatch.Broadcast") # Max messages queued per client before we start dropping (oldest dropped first). -# At ~37s orbital cycles emitting 11k messages, 256 gives ~23ms grace before dropping. -_CLIENT_QUEUE_SIZE = 256 +# The client worker drains the queue in coalesced batch frames (up to +# _MAX_BATCH_MSGS per WebSocket send), so even the ~11k-message orbital sweep +# is flushed in a few dozen sends. The deeper queue absorbs that burst for a +# slow client instead of silently dropping most of it (~1.7 MB transient +# worst case at ~150 B/message). +_CLIENT_QUEUE_SIZE = 4096 + +# Coalesced binary frame format: +# [0:3] 0xbf 0x02 0xbf — batch magic +# then per record: uint32 little-endian length + payload +# Each payload is an unmodified single TAK message (with its own +# 0xbf 0x01 0xbf prefix), so the frontend decodes records with the exact +# same code path as legacy single-message frames. Single messages are still +# sent as legacy frames for wire compatibility. +_BATCH_MAGIC = b"\xbf\x02\xbf" +_MAX_BATCH_MSGS = 128 +_MAX_BATCH_BYTES = 60_000 + + +def coalesce_outgoing(items: list) -> list[tuple[str, bytes]]: + """ + Group an ordered mix of proto messages (bytes) and alert tuples + (("alert", json_bytes)) into outgoing WebSocket sends, preserving order. + + Returns a list of ("bytes", frame) / ("text", payload) send instructions. + Consecutive proto messages are coalesced into batch frames capped at + _MAX_BATCH_BYTES; a lone proto message keeps the legacy single frame. + """ + sends: list[tuple[str, bytes]] = [] + pending: list[bytes] = [] + pending_bytes = 0 + + def flush() -> None: + nonlocal pending, pending_bytes + if not pending: + return + if len(pending) == 1: + sends.append(("bytes", pending[0])) + else: + parts = [_BATCH_MAGIC] + for m in pending: + parts.append(len(m).to_bytes(4, "little")) + parts.append(m) + sends.append(("bytes", b"".join(parts))) + pending = [] + pending_bytes = 0 + + for item in items: + if isinstance(item, tuple): + msg_type, data = item + if msg_type == "alert": + flush() + sends.append(("text", data)) + continue + if pending_bytes + len(item) > _MAX_BATCH_BYTES or len(pending) >= _MAX_BATCH_MSGS: + flush() + pending.append(item) + pending_bytes += len(item) + flush() + return sends class BroadcastManager: @@ -238,21 +296,32 @@ async def _consume(self): self._clients.clear() async def _client_worker(self, ws: WebSocket, q: asyncio.Queue): - """Background task per client: dequeue and send, with a generous timeout.""" + """ + Background task per client: drain the queue and send coalesced frames. + + Waiting on the first message then opportunistically draining whatever + else is already queued turns per-message sends into a handful of batch + frames during dense bursts (orbital sweeps), while sparse traffic + still goes out immediately as legacy single frames. + """ try: while True: - msg = await q.get() + first = await q.get() + items = [first] + while len(items) < _MAX_BATCH_MSGS: + try: + items.append(q.get_nowait()) + except asyncio.QueueEmpty: + break + try: - # Handle both TAK proto (bytes) and alert JSON (tuple) - if isinstance(msg, tuple): - msg_type, data = msg - if msg_type == "alert": + for kind, payload in coalesce_outgoing(items): + if kind == "text": await asyncio.wait_for( - ws.send_text(data.decode("utf-8")), timeout=3.0 + ws.send_text(payload.decode("utf-8")), timeout=3.0 ) - else: - # TAK proto (bytes) - await asyncio.wait_for(ws.send_bytes(msg), timeout=3.0) + else: + await asyncio.wait_for(ws.send_bytes(payload), timeout=3.0) except asyncio.TimeoutError: logger.warning("Client send timed out — disconnecting") break diff --git a/backend/api/tests/test_broadcast_batching.py b/backend/api/tests/test_broadcast_batching.py new file mode 100644 index 00000000..6fa2b5f8 --- /dev/null +++ b/backend/api/tests/test_broadcast_batching.py @@ -0,0 +1,76 @@ +"""Tests for the coalesced WebSocket batch framing in services/broadcast.py.""" + +from services.broadcast import ( + _BATCH_MAGIC, + _MAX_BATCH_BYTES, + coalesce_outgoing, +) + +MAGIC = b"\xbf\x01\xbf" + + +def _msg(payload: bytes) -> bytes: + return MAGIC + payload + + +def _parse_batch(frame: bytes) -> list[bytes]: + assert frame[:3] == _BATCH_MAGIC + records = [] + off = 3 + while off < len(frame): + length = int.from_bytes(frame[off : off + 4], "little") + off += 4 + records.append(frame[off : off + length]) + off += length + return records + + +def test_single_message_stays_legacy_frame(): + m = _msg(b"hello") + sends = coalesce_outgoing([m]) + assert sends == [("bytes", m)] + + +def test_multiple_messages_coalesce_into_batch_frame(): + msgs = [_msg(bytes([i]) * 10) for i in range(5)] + sends = coalesce_outgoing(list(msgs)) + assert len(sends) == 1 + kind, frame = sends[0] + assert kind == "bytes" + assert _parse_batch(frame) == msgs + + +def test_alert_preserves_ordering_and_splits_batches(): + a, b, c = _msg(b"a"), _msg(b"b"), _msg(b"c") + alert = ("alert", b'{"type":"alert"}') + sends = coalesce_outgoing([a, b, alert, c]) + assert len(sends) == 3 + kind0, frame0 = sends[0] + assert kind0 == "bytes" + assert _parse_batch(frame0) == [a, b] + assert sends[1] == ("text", b'{"type":"alert"}') + assert sends[2] == ("bytes", c) # lone trailing message → legacy frame + + +def test_batch_respects_byte_cap(): + big = _msg(b"x" * (_MAX_BATCH_BYTES // 2)) + sends = coalesce_outgoing([big, big, big]) + # Three ~30 KB messages cannot fit one 60 KB batch → at least two sends + assert len(sends) >= 2 + reassembled = [] + for kind, frame in sends: + assert kind == "bytes" + if frame[:3] == _BATCH_MAGIC: + reassembled.extend(_parse_batch(frame)) + else: + reassembled.append(frame) + assert reassembled == [big, big, big] + + +def test_only_alerts(): + alert = ("alert", b"{}") + assert coalesce_outgoing([alert, alert]) == [("text", b"{}"), ("text", b"{}")] + + +def test_empty_input(): + assert coalesce_outgoing([]) == [] diff --git a/frontend/src/hooks/useAnimationLoop.ts b/frontend/src/hooks/useAnimationLoop.ts index 11d1e8f2..73622ec6 100644 --- a/frontend/src/hooks/useAnimationLoop.ts +++ b/frontend/src/hooks/useAnimationLoop.ts @@ -14,6 +14,7 @@ import { latLngToCell } from "h3-js"; import { H3CellData } from "../layers/buildH3CoverageLayer"; import { composeAllLayers } from "../layers/composition"; import { LayerCache } from "../layers/layerCache"; +import { EntityIconAttributeCache } from "../layers/entityIconAttributes"; import { CoTEntity, @@ -337,8 +338,11 @@ export function useAnimationLoop({ // Stable overlay hover handler — defined once so setProps doesn't receive a new // function reference every frame. Accesses setters via their refs (always current). + // Layers with binary attributes pick by index with no backing object, so a + // hover only counts as a miss when nothing was picked at all (index -1). const onOverlayHoverRef = useRef((info: PickingInfo) => { - if (!info.object) { + const picked = info.index != null && info.index >= 0; + if (!picked && !info.object) { setHoveredEntityRef.current(null); setHoverPositionRef.current(null); } @@ -356,6 +360,12 @@ export function useAnimationLoop({ const layerCacheRef = useRef(null); if (!layerCacheRef.current) layerCacheRef.current = new LayerCache(); + // Persistent binary-attribute buffers for the 2D entity icon layer — + // per-overlay for the same reason as LayerCache. + const entityIconCacheRef = useRef(null); + if (!entityIconCacheRef.current) + entityIconCacheRef.current = new EntityIconAttributeCache(); + const countryOutageMap = React.useMemo(() => { if (!outagesData || !outagesData.features) return {}; const map: Record> = {}; @@ -835,6 +845,7 @@ export function useAnimationLoop({ firmsData: firmsDataRef.current, darkVesselData: darkVesselDataRef.current, cache: layerCacheRef.current ?? undefined, + entityIconCache: entityIconCacheRef.current ?? undefined, }); if (mapLoadedRef.current && overlayRef.current?.setProps) { diff --git a/frontend/src/layers/buildEntityLayers.ts b/frontend/src/layers/buildEntityLayers.ts index 94d86bbb..0754bed7 100644 --- a/frontend/src/layers/buildEntityLayers.ts +++ b/frontend/src/layers/buildEntityLayers.ts @@ -9,6 +9,7 @@ import { import { CoTEntity } from "../types"; import { entityColor } from "../utils/map/colorUtils"; import { ICON_ATLAS } from "../utils/map/iconAtlas"; +import type { EntityIconAttributeCache } from "./entityIconAttributes"; type PathPoint3D = [number, number, number]; interface VelocityDatum { @@ -27,6 +28,7 @@ export function buildEntityLayers( setHoveredEntity: (entity: CoTEntity | null) => void, setHoverPosition: (pos: { x: number; y: number } | null) => void, selectedEntity: CoTEntity | null, + iconCache?: EntityIconAttributeCache, ): Layer[] { const layers: Layer[] = []; @@ -274,6 +276,60 @@ export function buildEntityLayers( }, }), ); + } else if (iconCache) { + // Standard 2D / 3D pitch mode with precomputed binary attributes: + // position/angle/color/size are uploaded as typed arrays, so deck.gl + // never iterates the entity objects for them. Only the icon-frame + // accessor still runs per instance (it reads a precomputed flag array). + // Picking is index-based: handlers resolve entities via info.index. + const { data: iconData, shipFlags } = iconCache.update( + interpolated, + currentSelected?.uid ?? null, + now, + ); + const entityAt = (info: PickingInfo): CoTEntity | null => + info.index != null && info.index >= 0 && info.index < interpolated.length + ? interpolated[info.index] + : null; + + layers.push( + new IconLayer({ + id: `heading-arrows-merc`, + data: iconData as unknown as CoTEntity[], + getIcon: ((_: unknown, info: { index: number }) => + shipFlags[info.index] ? "vessel" : "aircraft") as unknown as ( + d: CoTEntity, + ) => string, + iconAtlas: ICON_ATLAS.url, + iconMapping: ICON_ATLAS.mapping, + sizeUnits: "pixels" as const, + sizeMinPixels: 18, + billboard: false, + pickable: true, + wrapLongitude: true, + parameters: { depthTest: false, depthBias: 0 }, + onHover: (info: PickingInfo) => { + const entity = entityAt(info); + if (entity) { + setHoveredEntity(entity); + setHoverPosition({ x: info.x, y: info.y }); + } else { + setHoveredEntity(null); + setHoverPosition(null); + } + }, + onClick: (info: PickingInfo) => { + const entity = entityAt(info); + if (entity) { + const newSelection = + selectedEntity?.uid === entity.uid ? null : entity; + onEntitySelect(newSelection); + } else { + onEntitySelect(null); + } + }, + }), + ); } else { // Standard 2D / 3D Pitch Map Mode uses heavily optimized sprites layers.push( diff --git a/frontend/src/layers/composition.ts b/frontend/src/layers/composition.ts index 0ba7f9d6..143bdcc3 100644 --- a/frontend/src/layers/composition.ts +++ b/frontend/src/layers/composition.ts @@ -31,6 +31,7 @@ import { buildTowerLayer } from "./buildTowerLayer"; import { buildTrailLayers } from "./buildTrailLayers"; import { getOrbitalLayers } from "./OrbitalLayer"; import { LayerCache } from "./layerCache"; +import type { EntityIconAttributeCache } from "./entityIconAttributes"; import type { GroundTrackPoint, ISSPosition, SatNOGSStation } from "../types"; import type { H3CellData } from "./buildH3CoverageLayer"; @@ -122,6 +123,12 @@ interface LayerCompositionOptions { * Omitting it disables caching (every group rebuilds on every call). */ cache?: LayerCache; + /** + * Persistent binary-attribute buffers for the 2D entity icon layer. + * Per-overlay instance, same ownership rules as `cache`. Omitting it + * falls back to the object-based icon layer. + */ + entityIconCache?: EntityIconAttributeCache; } export function composeAllLayers(options: LayerCompositionOptions) { @@ -741,6 +748,7 @@ export function composeAllLayers(options: LayerCompositionOptions) { setHoveredEntity, setHoverPosition, currentSelected, + options.entityIconCache, ), ...js8Layers, ]; diff --git a/frontend/src/layers/entityIconAttributes.test.ts b/frontend/src/layers/entityIconAttributes.test.ts new file mode 100644 index 00000000..1a0476f1 --- /dev/null +++ b/frontend/src/layers/entityIconAttributes.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, it } from "vitest"; +import type { CoTEntity } from "../types"; +import { EntityIconAttributeCache } from "./entityIconAttributes"; + +function entity(partial: Partial & { uid: string }): CoTEntity { + return { + lat: 45, + lon: -122, + altitude: 1000, + type: "a-f-A", + course: 90, + speed: 100, + callsign: partial.uid, + lastSeen: Date.now(), + trail: [], + uidHash: 0, + ...partial, + } as CoTEntity; +} + +describe("EntityIconAttributeCache", () => { + it("fills position/angle/color/size buffers parallel to the entity array", () => { + const cache = new EntityIconAttributeCache(); + const entities = [ + entity({ uid: "A", lon: 10, lat: 20, altitude: 300, course: 45 }), + entity({ uid: "B", lon: -30, lat: -40, altitude: 0, course: 180, type: "a-n-S", speed: 5 }), + ]; + const { data, shipFlags } = cache.update(entities, null, 1000); + + expect(data.length).toBe(2); + const pos = data.attributes.getPosition.value; + expect([pos[0], pos[1], pos[2]]).toEqual([10, 20, 300]); + expect([pos[3], pos[4], pos[5]]).toEqual([-30, -40, 0]); + + const angles = data.attributes.getAngle.value; + expect(angles[0]).toBe(-45); + expect(angles[1]).toBe(-180); + + expect(shipFlags[0]).toBe(0); // aircraft + expect(shipFlags[1]).toBe(1); // vessel + + // Colors: 4 bytes per entity, opaque-ish alpha from entityColor default + const colors = data.attributes.getColor.value; + expect(colors[3]).toBe(220); + expect(colors[7]).toBe(220); + + const sizes = data.attributes.getSize.value; + expect(sizes[0]).toBe(32); + expect(sizes[1]).toBe(32); + }); + + it("ping-pongs the per-frame buffers so references change between updates", () => { + const cache = new EntityIconAttributeCache(); + const entities = [entity({ uid: "A" })]; + const first = cache.update(entities, null, 1000); + const second = cache.update(entities, null, 1016); + expect(second.data.attributes.getPosition.value).not.toBe( + first.data.attributes.getPosition.value, + ); + expect(second.data.attributes.getAngle.value).not.toBe( + first.data.attributes.getAngle.value, + ); + }); + + it("keeps color/size buffer references stable between style refreshes", () => { + const cache = new EntityIconAttributeCache(); + const entities = [entity({ uid: "A" })]; + const first = cache.update(entities, null, 1000); + // 16 ms later: same membership, same selection, within the 1 s cadence + const second = cache.update(entities, null, 1016); + expect(second.data.attributes.getColor.value).toBe( + first.data.attributes.getColor.value, + ); + expect(second.data.attributes.getSize.value).toBe( + first.data.attributes.getSize.value, + ); + // Past the cadence → refreshed into the other buffer + const third = cache.update(entities, null, 2100); + expect(third.data.attributes.getColor.value).not.toBe( + second.data.attributes.getColor.value, + ); + }); + + it("refreshes styles immediately when membership changes", () => { + const cache = new EntityIconAttributeCache(); + const first = cache.update([entity({ uid: "A" })], null, 1000); + const second = cache.update( + [entity({ uid: "A" }), entity({ uid: "B", type: "a-n-S" })], + null, + 1016, + ); + expect(second.data.length).toBe(2); + expect(second.data.attributes.getColor.value).not.toBe( + first.data.attributes.getColor.value, + ); + expect(second.shipFlags[1]).toBe(1); + }); + + it("applies the enlarged size to the selected entity on selection change", () => { + const cache = new EntityIconAttributeCache(); + const entities = [entity({ uid: "A" }), entity({ uid: "B" })]; + cache.update(entities, null, 1000); + const { data } = cache.update(entities, "B", 1016); + const sizes = data.attributes.getSize.value; + expect(sizes[0]).toBe(32); + expect(sizes[1]).toBeCloseTo(41.6); + }); + + it("grows capacity and stays correct when the entity count increases", () => { + const cache = new EntityIconAttributeCache(); + cache.update([entity({ uid: "A" })], null, 1000); + const many = Array.from({ length: 100 }, (_, i) => + entity({ uid: `E${i}`, lon: i, lat: -i }), + ); + const { data } = cache.update(many, null, 1016); + expect(data.length).toBe(100); + const pos = data.attributes.getPosition.value; + expect(pos[99 * 3]).toBe(99); + expect(pos[99 * 3 + 1]).toBe(-99); + expect(pos.length).toBeGreaterThanOrEqual(100 * 3); + }); +}); diff --git a/frontend/src/layers/entityIconAttributes.ts b/frontend/src/layers/entityIconAttributes.ts new file mode 100644 index 00000000..66018074 --- /dev/null +++ b/frontend/src/layers/entityIconAttributes.ts @@ -0,0 +1,150 @@ +import type { CoTEntity } from "../types"; +import { entityColor } from "../utils/map/colorUtils"; + +/** + * Persistent binary-attribute cache for the 2D entity icon layer. + * + * With thousands of COTs, letting deck.gl iterate a fresh object array and + * call five accessors per entity per frame dominates the frame budget. This + * cache keeps typed arrays across frames and fills them in one pass: + * + * - positions / angles change every frame → written into ping-pong buffer + * pairs so the external-buffer reference changes whenever content does + * (deck skips re-upload for an unchanged reference). + * - colors / sizes / icon kinds change slowly → recomputed only when the + * entity membership changes, the selection changes, or on a 1 s cadence + * (entityColor tracks altitude/speed, which drift far slower than that). + * + * The returned `data` object is consumed by IconLayer as partial binary data; + * picking then reports indices into the same entity array that filled the + * buffers, so handlers resolve `entities[info.index]`. + */ + +const STYLE_REFRESH_INTERVAL_MS = 1000; +const BASE_ICON_SIZE = 32; +const SELECTED_ICON_SIZE = BASE_ICON_SIZE * 1.3; + +export interface EntityIconBinaryData { + length: number; + attributes: { + getPosition: { value: Float64Array; size: 3 }; + getAngle: { value: Float32Array; size: 1 }; + getColor: { value: Uint8Array; size: 4 }; + getSize: { value: Float32Array; size: 1 }; + }; +} + +export interface EntityIconFrame { + data: EntityIconBinaryData; + /** 1 = vessel icon, 0 = aircraft icon; parallel to the entity array. */ + shipFlags: Uint8Array; +} + +export class EntityIconAttributeCache { + private capacity = 0; + private slot = 0; + private positions: [Float64Array, Float64Array] = [ + new Float64Array(0), + new Float64Array(0), + ]; + private angles: [Float32Array, Float32Array] = [ + new Float32Array(0), + new Float32Array(0), + ]; + private styleSlot = 0; + private colors: [Uint8Array, Uint8Array] = [ + new Uint8Array(0), + new Uint8Array(0), + ]; + private sizes: [Float32Array, Float32Array] = [ + new Float32Array(0), + new Float32Array(0), + ]; + private shipFlags = new Uint8Array(0); + private uids: (string | undefined)[] = []; + private lastCount = -1; + private lastSelectedUid: string | null = null; + private lastStyleRefresh = 0; + + private ensureCapacity(n: number): void { + if (n <= this.capacity) return; + const cap = Math.max(16, Math.ceil(n * 1.5)); + this.positions = [new Float64Array(cap * 3), new Float64Array(cap * 3)]; + this.angles = [new Float32Array(cap), new Float32Array(cap)]; + this.colors = [new Uint8Array(cap * 4), new Uint8Array(cap * 4)]; + this.sizes = [new Float32Array(cap), new Float32Array(cap)]; + this.shipFlags = new Uint8Array(cap); + this.uids = new Array(cap); + this.capacity = cap; + this.lastStyleRefresh = 0; // force a style fill into the new buffers + } + + update( + entities: CoTEntity[], + selectedUid: string | null, + now: number, + ): EntityIconFrame { + const n = entities.length; + this.ensureCapacity(n); + + // Ping-pong the per-frame buffers so their references change with content. + this.slot = 1 - this.slot; + const positions = this.positions[this.slot]; + const angles = this.angles[this.slot]; + + let membershipChanged = n !== this.lastCount; + for (let i = 0; i < n; i++) { + const e = entities[i]; + const base = i * 3; + positions[base] = e.lon; + positions[base + 1] = e.lat; + positions[base + 2] = e.altitude || 0; + angles[i] = -(e.course || 0); + if (this.uids[i] !== e.uid) { + this.uids[i] = e.uid; + membershipChanged = true; + } + } + this.lastCount = n; + + const styleRefresh = + membershipChanged || + selectedUid !== this.lastSelectedUid || + now - this.lastStyleRefresh >= STYLE_REFRESH_INTERVAL_MS; + + if (styleRefresh) { + this.styleSlot = 1 - this.styleSlot; + const colors = this.colors[this.styleSlot]; + const sizes = this.sizes[this.styleSlot]; + for (let i = 0; i < n; i++) { + const e = entities[i]; + const [r, g, b, a] = entityColor(e); + const base = i * 4; + colors[base] = r; + colors[base + 1] = g; + colors[base + 2] = b; + colors[base + 3] = a; + sizes[i] = + selectedUid && e.uid === selectedUid + ? SELECTED_ICON_SIZE + : BASE_ICON_SIZE; + this.shipFlags[i] = e.type.includes("S") ? 1 : 0; + } + this.lastStyleRefresh = now; + this.lastSelectedUid = selectedUid; + } + + return { + data: { + length: n, + attributes: { + getPosition: { value: positions, size: 3 }, + getAngle: { value: angles, size: 1 }, + getColor: { value: this.colors[this.styleSlot], size: 4 }, + getSize: { value: this.sizes[this.styleSlot], size: 1 }, + }, + }, + shipFlags: this.shipFlags, + }; + } +} diff --git a/frontend/src/workers/batchFraming.test.ts b/frontend/src/workers/batchFraming.test.ts new file mode 100644 index 00000000..d52ae7ad --- /dev/null +++ b/frontend/src/workers/batchFraming.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it } from "vitest"; +import { + BATCH_MAGIC, + isBatchFrame, + isTakFrame, + splitBatchFrame, + TAK_MAGIC, +} from "./batchFraming"; + +function takMessage(payload: number[]): Uint8Array { + return new Uint8Array([...TAK_MAGIC, ...payload]); +} + +function batchFrame(records: Uint8Array[]): Uint8Array { + let total = 3; + for (const r of records) total += 4 + r.length; + const out = new Uint8Array(total); + out.set(BATCH_MAGIC, 0); + let offset = 3; + const view = new DataView(out.buffer); + for (const r of records) { + view.setUint32(offset, r.length, true); + offset += 4; + out.set(r, offset); + offset += r.length; + } + return out; +} + +describe("batchFraming", () => { + it("identifies legacy TAK frames and batch frames", () => { + expect(isTakFrame(takMessage([1, 2, 3]))).toBe(true); + expect(isBatchFrame(takMessage([1, 2, 3]))).toBe(false); + expect(isBatchFrame(batchFrame([takMessage([1])]))).toBe(true); + expect(isTakFrame(batchFrame([takMessage([1])]))).toBe(false); + expect(isTakFrame(new Uint8Array([]))).toBe(false); + expect(isBatchFrame(new Uint8Array([0xbf]))).toBe(false); + }); + + it("splits a batch frame into its original records", () => { + const a = takMessage([10, 11]); + const b = takMessage([20]); + const c = takMessage([30, 31, 32]); + const records = splitBatchFrame(batchFrame([a, b, c])); + expect(records.map((r) => Array.from(r))).toEqual([ + Array.from(a), + Array.from(b), + Array.from(c), + ]); + }); + + it("handles a batch frame arriving in a non-zero-offset view", () => { + const a = takMessage([1, 2, 3, 4]); + const frame = batchFrame([a]); + // Simulate a subarray view into a larger buffer + const padded = new Uint8Array(frame.length + 8); + padded.set(frame, 8); + const view = padded.subarray(8); + const records = splitBatchFrame(view); + expect(records.length).toBe(1); + expect(Array.from(records[0])).toEqual(Array.from(a)); + }); + + it("drops a truncated trailing record without throwing", () => { + const a = takMessage([1, 2]); + const frame = batchFrame([a, takMessage([3, 4, 5, 6])]); + const truncated = frame.subarray(0, frame.length - 3); + const records = splitBatchFrame(truncated); + expect(records.length).toBe(1); + expect(Array.from(records[0])).toEqual(Array.from(a)); + }); + + it("returns no records for an empty batch frame", () => { + expect(splitBatchFrame(batchFrame([]))).toEqual([]); + }); +}); diff --git a/frontend/src/workers/batchFraming.ts b/frontend/src/workers/batchFraming.ts new file mode 100644 index 00000000..2da5f53b --- /dev/null +++ b/frontend/src/workers/batchFraming.ts @@ -0,0 +1,48 @@ +/** + * Wire framing for the /api/tracks/live WebSocket. + * + * Legacy frame: 0xbf 0x01 0xbf + one TAK protobuf payload. + * Batch frame: 0xbf 0x02 0xbf + repeated records of + * [uint32 little-endian length + payload], where each payload + * is itself a complete legacy frame (magic included) so single + * and batched messages share the same decode path. + */ + +export const TAK_MAGIC = [0xbf, 0x01, 0xbf] as const; +export const BATCH_MAGIC = [0xbf, 0x02, 0xbf] as const; + +export function isBatchFrame(buffer: Uint8Array): boolean { + return ( + buffer.length >= 3 && + buffer[0] === BATCH_MAGIC[0] && + buffer[1] === BATCH_MAGIC[1] && + buffer[2] === BATCH_MAGIC[2] + ); +} + +export function isTakFrame(buffer: Uint8Array): boolean { + return ( + buffer.length >= 3 && + buffer[0] === TAK_MAGIC[0] && + buffer[1] === TAK_MAGIC[1] && + buffer[2] === TAK_MAGIC[2] + ); +} + +/** + * Split a batch frame into its record payloads (zero-copy subarray views). + * Malformed tails (truncated length or payload) are dropped silently. + */ +export function splitBatchFrame(buffer: Uint8Array): Uint8Array[] { + const records: Uint8Array[] = []; + const view = new DataView(buffer.buffer, buffer.byteOffset, buffer.byteLength); + let offset = 3; + while (offset + 4 <= buffer.length) { + const length = view.getUint32(offset, true); + offset += 4; + if (length === 0 || offset + length > buffer.length) break; + records.push(buffer.subarray(offset, offset + length)); + offset += length; + } + return records; +} diff --git a/frontend/src/workers/tak.worker.ts b/frontend/src/workers/tak.worker.ts index f0326ee6..b71e0f36 100644 --- a/frontend/src/workers/tak.worker.ts +++ b/frontend/src/workers/tak.worker.ts @@ -1,4 +1,5 @@ import { load, Type } from 'protobufjs'; +import { isBatchFrame, isTakFrame, splitBatchFrame } from './batchFraming'; // --- State --- let takType: Type | null = null; @@ -51,48 +52,51 @@ self.onmessage = async (e: MessageEvent) => { if (type === 'decode_batch') { if (!takType) return; - // Payload is Array or just ArrayBuffer - // We expect raw bytes. const buffer = new Uint8Array(payload); - // 1. Check Magic Bytes (Simple Check) - if (buffer[0] === 0xbf && buffer[1] === 0x01 && buffer[2] === 0xbf) { - try { - // Skip 3 bytes? Or does the proto include them? - // Usually protocol wrappers strip headers before proto decoding. - // If the proto IS the payload after magic bytes: - const cleanBuffer = buffer.subarray(3); - - const message = takType.decode(cleanBuffer); - - // Convert to plain object - const object = takType.toObject(message, { - longs: Number, - enums: String, - bytes: String, - }); - - // BUG-018: Removed hex debug computation (Array.from().map().join()) - // that ran on every decoded message in production. Raw hex is - // a debug/inspection artifact and not consumed by any UI feature. - - // Return Parsed Data - // Optimization: In real world, we would write to a SharedArrayBuffer here. - // For FE-05 MVP, we just return the object. - batch.push(object); - if (batch.length >= BATCH_SIZE) { - flushBatch(); - } else if (!flushTimer) { - flushTimer = setTimeout(flushBatch, FLUSH_INTERVAL_MS); - } - - } catch (parseErr) { - console.error("TAK Parse Error:", parseErr); + // Coalesced frame: repeated [u32le length + single-message payload] + if (isBatchFrame(buffer)) { + for (const record of splitBatchFrame(buffer)) { + decodeOne(record); } + return; } + + // Legacy frame: exactly one magic-prefixed TAK message + decodeOne(buffer); } }; +function decodeOne(buffer: Uint8Array): void { + if (!takType || !isTakFrame(buffer)) return; + try { + // The proto payload follows the 3-byte magic header. + const cleanBuffer = buffer.subarray(3); + + const message = takType.decode(cleanBuffer); + + // Convert to plain object + const object = takType.toObject(message, { + longs: Number, + enums: String, + bytes: String, + }); + + // BUG-018: Removed hex debug computation (Array.from().map().join()) + // that ran on every decoded message in production. Raw hex is + // a debug/inspection artifact and not consumed by any UI feature. + + batch.push(object); + if (batch.length >= BATCH_SIZE) { + flushBatch(); + } else if (!flushTimer) { + flushTimer = setTimeout(flushBatch, FLUSH_INTERVAL_MS); + } + } catch (parseErr) { + console.error("TAK Parse Error:", parseErr); + } +} + function str(err: unknown): string { return err instanceof Error ? err.message : String(err); }