Martino Ferrari
284 lines
10 KiB
JavaScript
284 lines
10 KiB
JavaScript
'use strict';
|
||
/* ════════════════════════════════════════════════════════════════
|
||
Web Worker – buffer management, binary parsing, LTTB
|
||
════════════════════════════════════════════════════════════════ */
|
||
|
||
const TEMPORAL_CAP = 600_000;
|
||
const DEFAULT_CAP = 10_000;
|
||
|
||
// Circular buffers: key → {t:Float64Array, v:Float64Array, head, size, cap}
|
||
const buffers = {};
|
||
|
||
function makeBuffer(cap) {
|
||
return { t: new Float64Array(cap), v: new Float64Array(cap), head: 0, size: 0, cap };
|
||
}
|
||
function pushBuffer(buf, t, v) {
|
||
buf.t[buf.head] = t; buf.v[buf.head] = v;
|
||
buf.head = (buf.head + 1) % buf.cap;
|
||
if (buf.size < buf.cap) buf.size++;
|
||
}
|
||
|
||
// ─── Binary frame parser ─────────────────────────────────────────────
|
||
// Format (little-endian):
|
||
// uint8 version (1)
|
||
// uint8 sourceIdLen
|
||
// UTF-8 sourceId
|
||
// uint32 numSignals
|
||
// for each signal:
|
||
// uint16 keyLen
|
||
// UTF-8 key (relative to source)
|
||
// uint32 pairCount N
|
||
// float64[N] t values
|
||
// float64[N] v values
|
||
function parseBinaryFrame(buf) {
|
||
const dv = new DataView(buf);
|
||
let off = 0;
|
||
|
||
if (dv.getUint8(off) !== 1) { console.warn('[worker] bad binary version'); return; }
|
||
off += 1;
|
||
|
||
const srcIdLen = dv.getUint8(off); off += 1;
|
||
const srcId = new TextDecoder().decode(new Uint8Array(buf, off, srcIdLen));
|
||
off += srcIdLen;
|
||
|
||
const prefix = srcId + ':';
|
||
const numSigs = dv.getUint32(off, true); off += 4;
|
||
|
||
for (let s = 0; s < numSigs; s++) {
|
||
const keyLen = dv.getUint16(off, true); off += 2;
|
||
const key = new TextDecoder().decode(new Uint8Array(buf, off, keyLen));
|
||
off += keyLen;
|
||
|
||
const fullKey = prefix + key;
|
||
const n = dv.getUint32(off, true); off += 4;
|
||
|
||
let bufObj = buffers[fullKey];
|
||
if (!bufObj) {
|
||
// Auto-create buffer with reasonable capacity
|
||
const cap = n > 100 ? TEMPORAL_CAP : DEFAULT_CAP;
|
||
bufObj = makeBuffer(cap);
|
||
buffers[fullKey] = bufObj;
|
||
}
|
||
|
||
// Read t values
|
||
for (let i = 0; i < n; i++) {
|
||
const t = dv.getFloat64(off, true); off += 8;
|
||
const v = dv.getFloat64(off + n * 8, true); // v array starts after t array
|
||
pushBuffer(bufObj, t, v);
|
||
}
|
||
off += n * 8; // skip v array (already read inline above)
|
||
}
|
||
}
|
||
|
||
// ─── Range slice from circular buffer ────────────────────────────────
|
||
function getBufferSliceRange(bufObj, t0, t1) {
|
||
const { cap, size, head } = bufObj;
|
||
if (size === 0) return { t: new Float64Array(0), v: new Float64Array(0) };
|
||
const start = (size === cap) ? head : 0;
|
||
const physAt = k => (start + k) % cap;
|
||
|
||
let lo = 0, hi = size;
|
||
while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] < t0) lo = m + 1; else hi = m; }
|
||
const kStart = lo;
|
||
lo = kStart; hi = size;
|
||
while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] <= t1) lo = m + 1; else hi = m; }
|
||
const kEnd = lo, len = kEnd - kStart;
|
||
if (len <= 0) return { t: new Float64Array(0), v: new Float64Array(0) };
|
||
|
||
const outT = new Float64Array(len), outV = new Float64Array(len);
|
||
const physStart = physAt(kStart), tail = cap - physStart;
|
||
if (tail >= len) {
|
||
outT.set(bufObj.t.subarray(physStart, physStart + len));
|
||
outV.set(bufObj.v.subarray(physStart, physStart + len));
|
||
} else {
|
||
outT.set(bufObj.t.subarray(physStart, physStart + tail));
|
||
outT.set(bufObj.t.subarray(0, len - tail), tail);
|
||
outV.set(bufObj.v.subarray(physStart, physStart + tail));
|
||
outV.set(bufObj.v.subarray(0, len - tail), tail);
|
||
}
|
||
return { t: outT, v: outV };
|
||
}
|
||
|
||
// ─── LTTB decimation ─────────────────────────────────────────────────
|
||
function lttb(t, v, threshold) {
|
||
const len = t.length;
|
||
if (len <= threshold || threshold < 3) return { t, v };
|
||
const outT = new Float64Array(threshold), outV = new Float64Array(threshold);
|
||
outT[0] = t[0]; outV[0] = v[0];
|
||
outT[threshold - 1] = t[len - 1]; outV[threshold - 1] = v[len - 1];
|
||
const every = (len - 2) / (threshold - 2);
|
||
let a = 0;
|
||
for (let i = 0; i < threshold - 2; i++) {
|
||
const avgS = Math.floor((i + 1) * every) + 1, avgE = Math.min(Math.floor((i + 2) * every) + 1, len);
|
||
let avgT = 0, avgV = 0, n = 0;
|
||
for (let j = avgS; j < avgE; j++) { avgT += t[j]; avgV += v[j]; n++; }
|
||
if (n) { avgT /= n; avgV /= n; }
|
||
const rS = Math.floor(i * every) + 1, rE = Math.min(Math.floor((i + 1) * every) + 1, len);
|
||
let maxA = -1, next = rS;
|
||
const aT = t[a], aV = v[a];
|
||
for (let j = rS; j < rE; j++) {
|
||
const area = Math.abs((aT - avgT) * (v[j] - aV) - (aT - t[j]) * (avgV - aV));
|
||
if (area > maxA) { maxA = area; next = j; }
|
||
}
|
||
outT[i + 1] = t[next]; outV[i + 1] = v[next]; a = next;
|
||
}
|
||
return { t: outT, v: outV };
|
||
}
|
||
|
||
// ─── Linear resampling ───────────────────────────────────────────────
|
||
function resampleLinear(tSrc, vSrc, tDst) {
|
||
const n = tDst.length;
|
||
const out = new Float64Array(n);
|
||
if (tSrc.length === 0) return out;
|
||
if (tSrc.length === 1) { out.fill(vSrc[0]); return out; }
|
||
let j = 0;
|
||
for (let i = 0; i < n; i++) {
|
||
const td = tDst[i];
|
||
while (j < tSrc.length - 2 && tSrc[j + 1] < td) j++;
|
||
if (td <= tSrc[0]) { out[i] = vSrc[0]; }
|
||
else if (td >= tSrc[tSrc.length - 1]) { out[i] = vSrc[vSrc.length - 1]; }
|
||
else {
|
||
const t0 = tSrc[j], t1 = tSrc[j + 1];
|
||
const frac = (td - t0) / (t1 - t0);
|
||
out[i] = vSrc[j] + frac * (vSrc[j + 1] - vSrc[j]);
|
||
}
|
||
}
|
||
return out;
|
||
}
|
||
|
||
// ─── Master time grid selection ──────────────────────────────────────
|
||
// samplingRates: key → rate (Hz), provided by main thread on init
|
||
const samplingRates = {};
|
||
|
||
function pickMasterKey(keys) {
|
||
let bestKey = keys[0], bestRate = -1;
|
||
for (const k of keys) {
|
||
const rate = samplingRates[k] || 0;
|
||
if (rate > bestRate) { bestRate = rate; bestKey = k; }
|
||
}
|
||
return bestKey;
|
||
}
|
||
|
||
// ─── Build uPlot-compatible data arrays ──────────────────────────────
|
||
function buildRenderData(keys, t0, t1, targetPts) {
|
||
if (!keys || keys.length === 0) return [new Float64Array(0)];
|
||
|
||
const slices = {};
|
||
let masterKey = pickMasterKey(keys), masterCount = -1;
|
||
|
||
for (const key of keys) {
|
||
const bufObj = buffers[key];
|
||
if (!bufObj || bufObj.size === 0) continue;
|
||
const sl = getBufferSliceRange(bufObj, t0, t1);
|
||
slices[key] = sl;
|
||
if (sl.t.length > masterCount) { masterCount = sl.t.length; masterKey = key; }
|
||
}
|
||
|
||
const masterRaw = slices[masterKey];
|
||
if (!masterRaw || masterRaw.t.length === 0)
|
||
return [new Float64Array(0), ...keys.map(() => new Float64Array(0))];
|
||
|
||
const dec = lttb(masterRaw.t, masterRaw.v, targetPts);
|
||
const sharedT = dec.t;
|
||
const yArrays = [];
|
||
|
||
for (const key of keys) {
|
||
if (key === masterKey) { yArrays.push(dec.v); continue; }
|
||
const sl = slices[key];
|
||
if (!sl || sl.t.length === 0) { yArrays.push(new Float64Array(sharedT.length)); continue; }
|
||
yArrays.push(resampleLinear(sl.t, sl.v, sharedT));
|
||
}
|
||
|
||
const result = [sharedT, ...yArrays];
|
||
// Transfer ownership of the Float64Arrays to main thread
|
||
const transferList = result.map(a => a.buffer);
|
||
return { data: result, transfer: transferList };
|
||
}
|
||
|
||
// ─── Message handler ─────────────────────────────────────────────────
|
||
self.onmessage = function(e) {
|
||
const msg = e.data;
|
||
|
||
switch (msg.type) {
|
||
case 'initSignals': {
|
||
// {signals: [{key, cap}]}
|
||
const sigs = msg.signals || [];
|
||
sigs.forEach(s => {
|
||
if (!buffers[s.key]) {
|
||
buffers[s.key] = makeBuffer(s.cap || DEFAULT_CAP);
|
||
}
|
||
if (s.samplingRate !== undefined) {
|
||
samplingRates[s.key] = s.samplingRate;
|
||
}
|
||
});
|
||
break;
|
||
}
|
||
|
||
case 'binaryData': {
|
||
// {buffer: ArrayBuffer} — transferred from main thread
|
||
parseBinaryFrame(msg.buffer);
|
||
self.postMessage({ type: 'dataReady' });
|
||
break;
|
||
}
|
||
|
||
case 'requestData': {
|
||
// {id, t0, t1, targetPts, keys}
|
||
const { id, t0, t1, targetPts, keys } = msg;
|
||
const { data, transfer } = buildRenderData(keys, t0, t1, targetPts);
|
||
self.postMessage({ type: 'renderData', id, data }, transfer);
|
||
break;
|
||
}
|
||
|
||
case 'clearSource': {
|
||
const prefix = msg.prefix;
|
||
Object.keys(buffers).forEach(k => {
|
||
if (k.startsWith(prefix)) delete buffers[k];
|
||
});
|
||
Object.keys(samplingRates).forEach(k => {
|
||
if (k.startsWith(prefix)) delete samplingRates[k];
|
||
});
|
||
break;
|
||
}
|
||
|
||
case 'getBufferNow': {
|
||
// Returns newest timestamp across given keys
|
||
const keys = msg.keys || [];
|
||
let latest = -Infinity;
|
||
keys.forEach(key => {
|
||
const bufObj = buffers[key];
|
||
if (bufObj && bufObj.size > 0) {
|
||
const t = bufObj.t[(bufObj.head - 1 + bufObj.cap) % bufObj.cap];
|
||
if (t > latest) latest = t;
|
||
}
|
||
});
|
||
self.postMessage({ type: 'bufferNow', id: msg.id, now: isFinite(latest) ? latest : null });
|
||
break;
|
||
}
|
||
|
||
case 'getBufferForTrig': {
|
||
// Returns full buffer contents for a single key (used for trigger check)
|
||
const key = msg.key;
|
||
const bufObj = buffers[key];
|
||
if (!bufObj || bufObj.size === 0) {
|
||
self.postMessage({ type: 'trigBuf', id: msg.id, key, size: 0 });
|
||
break;
|
||
}
|
||
// Copy out all data
|
||
const { cap, size, head } = bufObj;
|
||
const start = (size === cap) ? head : 0;
|
||
const t = new Float64Array(size), v = new Float64Array(size);
|
||
const physAt = k => (start + k) % cap;
|
||
for (let i = 0; i < size; i++) {
|
||
const p = physAt(i);
|
||
t[i] = bufObj.t[p];
|
||
v[i] = bufObj.v[p];
|
||
}
|
||
self.postMessage({
|
||
type: 'trigBuf', id: msg.id, key, size,
|
||
t, v
|
||
}, [t.buffer, v.buffer]);
|
||
break;
|
||
}
|
||
}
|
||
};
|