From 5bb7471f6464e0279c3b03197bcde66dbfd9da84 Mon Sep 17 00:00:00 2001 From: Bill Bai Date: Tue, 2 Jun 2026 23:22:37 -0500 Subject: [PATCH 01/11] feat(web): streaming columnar parse foundation (Feature A) Add web/column_store.js (JS port of python store.ColumnStore) holding ~26 analysis channels as packed Float32Array columns + Float64Array timestamps instead of 590K record objects (~1.6 GB -> ~55 MB on the 7-day file). Add parser.js: - parseTrendColumnar: one-shot ArrayBuffer -> Transferable typed-array columns - parseTrendColumnarStream: chunked Blob.slice path so the full 438 MB ArrayBuffer is never resident; record-aligned 8 MB chunks - decodeColumnarSlice / allocColumns helpers Wire parser_worker.js to support parse-stream (Blob) and columnar parse, transferring the typed-array buffers back zero-copy. Legacy parseTrendBin path retained for small-file/CSV flows and existing tests. Tests: web/tests/columnar.test.js asserts record count + value parity with the legacy parse, streaming==one-shot, bounded per-chunk reads, and ColumnStore round-trips. Web suite 82 -> 87 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- web/column_store.js | 135 +++++++++++++++++++++++++++ web/parser.js | 129 ++++++++++++++++++++++++++ web/parser_worker.js | 93 +++++++++++++------ web/tests/columnar.test.js | 184 +++++++++++++++++++++++++++++++++++++ 4 files changed, 512 insertions(+), 29 deletions(-) create mode 100644 web/column_store.js create mode 100644 web/tests/columnar.test.js diff --git a/web/column_store.js b/web/column_store.js new file mode 100644 index 0000000..a5de3c7 --- /dev/null +++ b/web/column_store.js @@ -0,0 +1,135 @@ +// Columnar session store — JS port of python/src/fluke_3540/store.py. +// +// The legacy web path materialised every record as a {index, startMs, endMs, +// floats:Float32Array(180)} object. For a week-long capture (~590 K records) +// that is ~1.6 GB of heap. This ColumnStore keeps only the ~24 analysis +// channels the event / snapshot / insight / stats engines actually read, each +// as a packed Float32Array, plus a Float64Array of start/end millisecond +// timestamps. That is ~55 MB for a full week instead of >1 GB, and the typed +// arrays are Transferable so the worker hands them back with zero copy. +// +// STORE_COLUMNS mirrors python store.STORE_COLUMNS exactly so the two analysis +// paths read identical channels. + +export const STORE_COLUMNS = Object.freeze([ + // Per-phase L-N voltage min/max/avg + 'V_LN_a_min_V', 'V_LN_b_min_V', 'V_LN_c_min_V', + 'V_LN_a_max_V', 'V_LN_b_max_V', 'V_LN_c_max_V', + 'V_LN_a_avg_V', 'V_LN_b_avg_V', 'V_LN_c_avg_V', + // Per-phase current max + avg + 'I_a_max_A', 'I_b_max_A', 'I_c_max_A', + 'I_a_avg_A', 'I_b_avg_A', 'I_c_avg_A', + // Line frequency + 'freq_avg_Hz', + // Power / apparent / reactive / power-factor totals + 'P_total_avg_W', 'S_total_avg_VA', 'Q_total_avg_VAR', 'PF_total_avg', + // Per-row energy (per-bucket kWh roll-ups) + 'Wh_total', + // THD per phase (IEEE 519) — V and I, avg only + 'V_THD_pct_a_avg', 'V_THD_pct_b_avg', 'V_THD_pct_c_avg', + 'I_THD_pct_a_avg', 'I_THD_pct_b_avg', 'I_THD_pct_c_avg', +]); + +/** + * Resolve STORE_COLUMNS to spec float indices, once. + * @param {object} spec parsed field_map.json + * @returns {Map} name -> spec float index + */ +export function resolveStoreIndices(spec) { + const nameToIdx = new Map(spec.fields.map((f) => [f.name, f.index])); + const out = new Map(); + for (const name of STORE_COLUMNS) { + const idx = nameToIdx.get(name); + if (idx === undefined) { + throw new Error(`Store column ${name} missing from spec/field_map.json`); + } + out.set(name, idx); + } + return out; +} + +export class ColumnStore { + /** + * @param {number} n record count (used to pre-size typed arrays) + */ + constructor(n = 0) { + this.n = n; + this.cols = {}; + for (const name of STORE_COLUMNS) this.cols[name] = new Float32Array(n); + this.startMs = new Float64Array(n); + this.endMs = new Float64Array(n); + } + + /** Packed Float32Array column for `name` (no copy). */ + col(name) { + const c = this.cols[name]; + if (c === undefined) { + throw new Error( + `Column ${name} is not retained in the ColumnStore. ` + + `Retained columns: ${STORE_COLUMNS.join(', ')}` + ); + } + return c; + } + + start(i) { return this.startMs[i]; } + end(i) { return this.endMs[i]; } + + get firstStartMs() { return this.n ? this.startMs[0] : null; } + get lastEndMs() { return this.n ? this.endMs[this.n - 1] : null; } + + /** + * Lazily yield a lightweight record view ({index, startMs, endMs, floats}) + * for each record. The `floats` proxy only carries the retained columns at + * their spec index; reads of non-retained indices return 0. Used to feed the + * existing record-array consumers (events.js, snapshots.js, insights.js) + * without holding 590 K full record objects. + * + * @param {object} spec + * @returns {Array} array of record-shaped views + */ + toRecords(spec) { + const idxByName = resolveStoreIndices(spec); + const dataFloats = spec.data_floats; + const colArrays = STORE_COLUMNS.map((name) => [idxByName.get(name), this.cols[name]]); + const out = new Array(this.n); + for (let i = 0; i < this.n; i++) { + const floats = new Float32Array(dataFloats); + for (const [idx, arr] of colArrays) floats[idx] = arr[i]; + out[i] = { index: i, startMs: this.startMs[i], endMs: this.endMs[i], floats }; + } + return out; + } + + /** + * Reconstruct a ColumnStore from a worker "done-columnar" payload: + * { recordCount, columns: {name: Float32Array}, startMs, endMs }. + */ + static fromTransfer(payload) { + const store = Object.create(ColumnStore.prototype); + store.n = payload.recordCount; + store.cols = payload.columns; + store.startMs = payload.startMs; + store.endMs = payload.endMs; + return store; + } + + /** + * Build a ColumnStore from an array of record objects (legacy small-file / + * test path). Mirrors python ColumnStore.from_records. + * @param {Array<{startMs:number, endMs:number, floats:ArrayLike}>} records + * @param {object} spec + */ + static fromRecords(records, spec) { + const idxByName = resolveStoreIndices(spec); + const store = new ColumnStore(records.length); + const colArrays = STORE_COLUMNS.map((name) => [idxByName.get(name), store.cols[name]]); + for (let i = 0; i < records.length; i++) { + const r = records[i]; + for (const [idx, arr] of colArrays) arr[i] = r.floats[idx]; + store.startMs[i] = r.startMs; + store.endMs[i] = r.endMs; + } + return store; + } +} diff --git a/web/parser.js b/web/parser.js index 2c00f78..125c2ac 100644 --- a/web/parser.js +++ b/web/parser.js @@ -179,6 +179,135 @@ export function parseTrendBin(arrayBuffer, spec, opts = {}) { return { records }; } +// --- Streaming columnar parse (Feature A: ~1.6 GB → ~55 MB) ---------------- +// +// Instead of holding one 180-float record object per second, decode straight +// into packed Float32Array columns (only the analysis channels) + Float64Array +// timestamp arrays. The full ArrayBuffer is processed record-aligned; callers +// that stream a Blob (parser_worker.js) feed chunks here so the whole 438 MB +// is never resident at once. + +import { STORE_COLUMNS, resolveStoreIndices } from './column_store.js'; + +/** + * Allocate the column arrays for a columnar parse of `recordCount` records. + * @param {number} recordCount + * @returns {{columns: Object, startMs: Float64Array, endMs: Float64Array}} + */ +export function allocColumns(recordCount) { + const columns = {}; + for (const name of STORE_COLUMNS) columns[name] = new Float32Array(recordCount); + return { + columns, + startMs: new Float64Array(recordCount), + endMs: new Float64Array(recordCount), + }; +} + +/** + * Decode a record-aligned slice of `arrayBuffer` into pre-allocated column + * arrays, starting at output record index `outBase`. Used by both the one-shot + * and the chunked-streaming columnar parsers. + * + * @param {ArrayBuffer} arrayBuffer a buffer whose length is a multiple of recordSize + * @param {object} idx buildIndex() result + * @param {{columns, startMs, endMs}} sink pre-allocated output + * @param {Map} storeIdx STORE_COLUMNS name -> float index + * @param {Set|null} flip reverse-CT indices to negate, or null + * @param {number} outBase output record offset + * @returns {number} number of records decoded from this slice + */ +function decodeColumnarSlice(arrayBuffer, idx, sink, storeIdx, flip, outBase) { + const view = new DataView(arrayBuffer); + const recs = Math.floor(view.byteLength / idx.recordSize); + const magic = idx.recordMagic; + // Pre-resolve the (column-array, floatIndex, flipFlag) tuples once. + const plan = STORE_COLUMNS.map((name) => { + const fi = storeIdx.get(name); + return [sink.columns[name], fi, flip ? flip.has(fi) : false]; + }); + for (let r = 0; r < recs; r++) { + const offset = r * idx.recordSize; + for (let m = 0; m < magic.length; m++) { + if (view.getUint8(offset + m) !== magic[m]) { + throw new FlukeBinaryError( + `Bad magic at record ${outBase + r} (offset 0x${offset.toString(16)})` + ); + } + } + const startHi = BigInt(view.getUint32(offset + 4, true)); + const startLo = BigInt(view.getUint32(offset + 8, true)); + const endHi = BigInt(view.getUint32(offset + 12, true)); + const endLo = BigInt(view.getUint32(offset + 16, true)); + const out = outBase + r; + sink.startMs[out] = filetimeToUnixMs((startHi << 32n) | startLo); + sink.endMs[out] = filetimeToUnixMs((endHi << 32n) | endLo); + const floatsBase = offset + idx.headerBytes; + for (const [arr, fi, doFlip] of plan) { + let v = view.getFloat32(floatsBase + fi * 4, true); + if (doFlip) v = -v; + arr[out] = v; + } + } + return recs; +} + +/** + * One-shot columnar parse of a full trend.bin ArrayBuffer. Returns Transferable + * typed-array columns instead of 590 K record objects. + * + * @param {ArrayBuffer} arrayBuffer + * @param {object} spec + * @param {{reverseCts?: boolean|string[], onProgress?: (done:number,total:number)=>void}} [opts] + * @returns {{recordCount:number, columns:Object, startMs:Float64Array, endMs:Float64Array}} + */ +export function parseTrendColumnar(arrayBuffer, spec, opts = {}) { + const { reverseCts = false, onProgress = null } = opts; + const idx = buildIndex(spec, { reverseCts }); + const flip = reverseCts ? idx.reverseCtsIndices : null; + const storeIdx = resolveStoreIndices(spec); + const total = Math.floor(arrayBuffer.byteLength / idx.recordSize); + const sink = allocColumns(total); + decodeColumnarSlice(arrayBuffer, idx, sink, storeIdx, flip, 0); + if (onProgress) onProgress(total, total); + return { recordCount: total, ...sink }; +} + +/** + * Streaming columnar parse: read a Blob/File in record-aligned chunks so the + * full ArrayBuffer is never resident. `readSlice(start, end)` must return a + * Promise for the byte range [start, end) — in the browser that's + * `blob.slice(start, end).arrayBuffer()`. + * + * @param {{size:number, readSlice:(start:number,end:number)=>Promise}} source + * @param {object} spec + * @param {{reverseCts?: boolean|string[], chunkBytes?: number, onProgress?: (done:number,total:number)=>void}} [opts] + * @returns {Promise<{recordCount:number, columns, startMs, endMs}>} + */ +export async function parseTrendColumnarStream(source, spec, opts = {}) { + const { reverseCts = false, chunkBytes = 8 * 1024 * 1024, onProgress = null } = opts; + const idx = buildIndex(spec, { reverseCts }); + const flip = reverseCts ? idx.reverseCtsIndices : null; + const storeIdx = resolveStoreIndices(spec); + const recordSize = idx.recordSize; + const total = Math.floor(source.size / recordSize); + const sink = allocColumns(total); + // Round the chunk down to a whole number of records so slices stay aligned. + const recsPerChunk = Math.max(1, Math.floor(chunkBytes / recordSize)); + const bytesPerChunk = recsPerChunk * recordSize; + let done = 0; + for (let start = 0; done < total; start += bytesPerChunk) { + const remaining = total - done; + const recsThisChunk = Math.min(recsPerChunk, remaining); + const end = start + recsThisChunk * recordSize; + const buf = await source.readSlice(start, end); + decodeColumnarSlice(buf, idx, sink, storeIdx, flip, done); + done += recsThisChunk; + if (onProgress) onProgress(done, total); + } + return { recordCount: total, ...sink }; +} + /** * Convert a {records} parse result to a labelled-row generator (one object per * record, keyed by field name). Convenience for downstream code; the raw diff --git a/web/parser_worker.js b/web/parser_worker.js index f162c15..ee2a375 100644 --- a/web/parser_worker.js +++ b/web/parser_worker.js @@ -1,39 +1,74 @@ -// Web Worker wrapper around parser.js. Keeps the 55 MB parse off the main -// thread so the UI stays responsive. +// Web Worker wrapper around parser.js. Keeps the parse off the main thread so +// the UI stays responsive. // // Message protocol: -// in: { type: 'parse', spec, arrayBuffer, reverseCts } -// out: { type: 'progress', done, total } (~once per 1000 records) -// out: { type: 'done', records, recordCount } +// in: { type: 'parse', spec, arrayBuffer, reverseCts } (legacy small-file) +// in: { type: 'parse-stream', spec, blob, reverseCts } (Feature A: streaming columnar) +// out: { type: 'progress', done, total } +// out: { type: 'done', records, recordCount } (legacy record objects) +// out: { type: 'done-columnar', recordCount, columns, startMs, endMs } // out: { type: 'error', message, stack } -import { parseTrendBin } from './parser.js'; +import { parseTrendBin, parseTrendColumnar, parseTrendColumnarStream } from './parser.js'; +import { STORE_COLUMNS } from './column_store.js'; -self.onmessage = (event) => { - const msg = event.data; - if (msg?.type !== 'parse') { - self.postMessage({ - type: 'error', - message: `Unknown message type: ${msg?.type}`, - }); - return; +// Collect the Transferable typed-array buffers from a columnar payload so they +// move (zero-copy) back to the main thread instead of being structured-cloned. +function columnarTransferList(payload) { + const list = [payload.startMs.buffer, payload.endMs.buffer]; + for (const name of STORE_COLUMNS) { + const c = payload.columns[name]; + if (c && c.buffer) list.push(c.buffer); } + return list; +} + +self.onmessage = async (event) => { + const msg = event.data; try { - const { spec, arrayBuffer, reverseCts = false } = msg; - const result = parseTrendBin(arrayBuffer, spec, { - reverseCts, - onProgress: (done, total) => { - self.postMessage({ type: 'progress', done, total }); - }, - }); - // Transfer the underlying Float32Array buffers back to main thread - // (records are kept structured-cloneable; no transferable wrapping for now - // since structured cloning is fast enough for ~75 K records). - self.postMessage({ - type: 'done', - records: result.records, - recordCount: result.records.length, - }); + if (msg?.type === 'parse-stream') { + const { spec, blob, reverseCts = false } = msg; + const source = { + size: blob.size, + readSlice: (start, end) => blob.slice(start, end).arrayBuffer(), + }; + const payload = await parseTrendColumnarStream(source, spec, { + reverseCts, + onProgress: (done, total) => { + self.postMessage({ type: 'progress', done, total }); + }, + }); + self.postMessage({ type: 'done-columnar', ...payload }, + columnarTransferList(payload)); + return; + } + if (msg?.type === 'parse') { + const { spec, arrayBuffer, reverseCts = false, columnar = false } = msg; + if (columnar) { + const payload = parseTrendColumnar(arrayBuffer, spec, { + reverseCts, + onProgress: (done, total) => { + self.postMessage({ type: 'progress', done, total }); + }, + }); + self.postMessage({ type: 'done-columnar', ...payload }, + columnarTransferList(payload)); + return; + } + const result = parseTrendBin(arrayBuffer, spec, { + reverseCts, + onProgress: (done, total) => { + self.postMessage({ type: 'progress', done, total }); + }, + }); + self.postMessage({ + type: 'done', + records: result.records, + recordCount: result.records.length, + }); + return; + } + self.postMessage({ type: 'error', message: `Unknown message type: ${msg?.type}` }); } catch (err) { self.postMessage({ type: 'error', diff --git a/web/tests/columnar.test.js b/web/tests/columnar.test.js new file mode 100644 index 0000000..3c68dbd --- /dev/null +++ b/web/tests/columnar.test.js @@ -0,0 +1,184 @@ +// Feature A — streaming/columnar parse tests. +// +// Verifies: +// 1. parseTrendColumnar produces the correct record count + correct values +// against the shared 10-record synthetic fixture (parity with parser.py). +// 2. parseTrendColumnarStream (chunked Blob.slice path) yields identical +// columns to the one-shot path AND keeps peak heap bounded — it never +// holds the whole buffer or 590 K record objects. +// 3. ColumnStore.fromRecords / toRecords round-trips the retained channels. +import { strict as assert } from 'node:assert'; +import { readFileSync, writeFileSync, rmSync, mkdtempSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import { dirname, resolve, join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { test } from 'node:test'; + +import { + parseTrendBin, + parseTrendColumnar, + parseTrendColumnarStream, +} from '../parser.js'; +import { ColumnStore, STORE_COLUMNS } from '../column_store.js'; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const repoRoot = resolve(__dirname, '..', '..'); +const SPEC_PATH = resolve(repoRoot, 'spec', 'field_map.json'); +const FIXTURE_PATH = resolve( + repoRoot, 'python', 'tests', 'fixtures', 'synthetic_trend.bin' +); +const spec = JSON.parse(readFileSync(SPEC_PATH, 'utf8')); + +function loadFixture() { + const buf = readFileSync(FIXTURE_PATH); + return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength); +} + +// Build a large healthy trend.bin in a temp file (mirrors conftest.build_large_trend). +function buildLargeTrend(path, count) { + const recordSize = spec.record_size; + const headerBytes = spec.header_bytes; + const dataFloats = spec.data_floats; + const magic = new Uint8Array(spec.record_magic); + const nameToIdx = new Map(spec.fields.map((f) => [f.name, f.index])); + // Healthy 277V / 60Hz / 100A / 50kW profile. + const healthy = new Float32Array(dataFloats); + for (const ph of ['a', 'b', 'c']) { + for (const st of ['min', 'max', 'avg']) { + healthy[nameToIdx.get(`V_LN_${ph}_${st}_V`)] = 277.0; + healthy[nameToIdx.get(`I_${ph}_${st}_A`)] = 100.0; + } + } + healthy[nameToIdx.get('freq_avg_Hz')] = 60.0; + healthy[nameToIdx.get('P_total_avg_W')] = 50000.0; + healthy[nameToIdx.get('S_total_avg_VA')] = 52000.0; + healthy[nameToIdx.get('PF_total_avg')] = 0.96; + + const FILETIME_EPOCH_DIFF_MS = 11644473600000n; + const base = Date.UTC(2024, 0, 13, 22, 0, 0); + const buf = Buffer.alloc(recordSize * count); + for (let n = 0; n < count; n++) { + const o = n * recordSize; + for (let m = 0; m < magic.length; m++) buf[o + m] = magic[m]; + const startMs = BigInt(base + n * 1000); + const endMs = BigInt(base + (n + 1) * 1000); + const startFt = (startMs + FILETIME_EPOCH_DIFF_MS) * 10000n; + const endFt = (endMs + FILETIME_EPOCH_DIFF_MS) * 10000n; + buf.writeUInt32LE(Number(startFt >> 32n & 0xffffffffn), o + 4); + buf.writeUInt32LE(Number(startFt & 0xffffffffn), o + 8); + buf.writeUInt32LE(Number(endFt >> 32n & 0xffffffffn), o + 12); + buf.writeUInt32LE(Number(endFt & 0xffffffffn), o + 16); + for (let i = 0; i < dataFloats; i++) { + buf.writeFloatLE(healthy[i], o + headerBytes + i * 4); + } + } + writeFileSync(path, buf); +} + +test('parseTrendColumnar: record count + retained channels match legacy parse', () => { + const ab = loadFixture(); + const legacy = parseTrendBin(ab, spec); + const col = parseTrendColumnar(ab, spec); + assert.equal(col.recordCount, legacy.records.length); + const nameToIdx = new Map(spec.fields.map((f) => [f.name, f.index])); + for (const name of STORE_COLUMNS) { + const fi = nameToIdx.get(name); + const arr = col.columns[name]; + for (let n = 0; n < col.recordCount; n++) { + assert.equal(arr[n], legacy.records[n].floats[fi], + `${name}[${n}] mismatch`); + } + } + // Timestamps match. + for (let n = 0; n < col.recordCount; n++) { + assert.equal(col.startMs[n], legacy.records[n].startMs); + assert.equal(col.endMs[n], legacy.records[n].endMs); + } +}); + +test('parseTrendColumnar: reverseCts negates retained signed columns', () => { + const ab = loadFixture(); + const plain = parseTrendColumnar(ab, spec, { reverseCts: false }); + const flipped = parseTrendColumnar(ab, spec, { reverseCts: true }); + // P_total_avg_W is a reverse-CT column; PF too. V/I/S are not. + assert.equal(flipped.columns.P_total_avg_W[0], -plain.columns.P_total_avg_W[0]); + assert.equal(flipped.columns.PF_total_avg[0], -plain.columns.PF_total_avg[0]); + assert.equal(flipped.columns.V_LN_a_avg_V[0], plain.columns.V_LN_a_avg_V[0]); + assert.equal(flipped.columns.I_a_avg_A[0], plain.columns.I_a_avg_A[0]); +}); + +test('parseTrendColumnarStream: identical to one-shot, bounded chunk reads', async () => { + const dir = mkdtempSync(join(tmpdir(), 'fluke-col-')); + const path = join(dir, 'trend.bin'); + const COUNT = 20000; // ~14.9 MB — forces multiple 8 MB chunks + try { + buildLargeTrend(path, COUNT); + const buf = readFileSync(path); + const ab = buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength); + const oneShot = parseTrendColumnar(ab, spec); + assert.equal(oneShot.recordCount, COUNT); + + // Streaming source backed by the file — tracks the largest slice read so we + // can assert no single read approaches the whole-file size. + let maxSliceBytes = 0; + let totalSliceBytes = 0; + const source = { + size: buf.length, + readSlice: async (start, end) => { + const sliceBytes = end - start; + maxSliceBytes = Math.max(maxSliceBytes, sliceBytes); + totalSliceBytes += sliceBytes; + const sub = buf.subarray(start, end); + return sub.buffer.slice(sub.byteOffset, sub.byteOffset + sub.byteLength); + }, + }; + let lastDone = 0; + const streamed = await parseTrendColumnarStream(source, spec, { + chunkBytes: 8 * 1024 * 1024, + onProgress: (done) => { lastDone = done; }, + }); + assert.equal(streamed.recordCount, COUNT); + assert.equal(lastDone, COUNT); + // No chunk read more than ~8 MB + one record (record-aligned rounding). + assert.ok(maxSliceBytes <= 8 * 1024 * 1024 + spec.record_size, + `max slice ${maxSliceBytes} should stay near the 8 MB chunk size`); + assert.ok(maxSliceBytes < buf.length, + 'streaming must never read the whole file in one slice'); + // Every byte read exactly once. + assert.equal(totalSliceBytes, COUNT * spec.record_size); + // Columns identical to the one-shot decode. + for (const name of STORE_COLUMNS) { + assert.deepEqual(streamed.columns[name], oneShot.columns[name], + `column ${name} mismatch between streamed and one-shot`); + } + assert.deepEqual(streamed.startMs, oneShot.startMs); + assert.deepEqual(streamed.endMs, oneShot.endMs); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('ColumnStore.fromRecords / toRecords round-trips retained channels', () => { + const ab = loadFixture(); + const { records } = parseTrendBin(ab, spec); + const store = ColumnStore.fromRecords(records, spec); + assert.equal(store.n, records.length); + const nameToIdx = new Map(spec.fields.map((f) => [f.name, f.index])); + const fi = nameToIdx.get('P_total_avg_W'); + for (let i = 0; i < store.n; i++) { + assert.equal(store.col('P_total_avg_W')[i], records[i].floats[fi]); + } + const back = store.toRecords(spec); + assert.equal(back.length, records.length); + assert.equal(back[3].floats[fi], records[3].floats[fi]); + assert.equal(back[3].startMs, records[3].startMs); +}); + +test('ColumnStore.fromTransfer wraps a worker payload', () => { + const ab = loadFixture(); + const payload = parseTrendColumnar(ab, spec); + const store = ColumnStore.fromTransfer(payload); + assert.equal(store.n, payload.recordCount); + assert.equal(store.col('V_LN_a_avg_V')[0], payload.columns.V_LN_a_avg_V[0]); + assert.equal(store.firstStartMs, payload.startMs[0]); +}); From c3b92c4cfe87f59bf66e09f19de9305d0b7d5bd0 Mon Sep 17 00:00:00 2001 From: Bill Bai Date: Tue, 2 Jun 2026 23:32:01 -0500 Subject: [PATCH 02/11] feat(web): wire streaming columnar parse end-to-end (Feature A) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit App now streams the dropped File through the worker (parse-stream), holds a resident ColumnStore (~62 MB) instead of 590 K record objects, and runs event / snapshot / insight detection + charts + range + tariff straight off the store. currentRecords stays null on the streaming path; exports materialise a transient records array via recordsForExport() and drop it. Make the analysis engines store-aware via web/column_source.js: - events.js, snapshots.js, insights.js accept a ColumnStore OR a records array - plots.js buildPlotData reads columns from a store (charts memory-bounded) - range_select.js renderRangeSelector + tariff.js computeCost read the store Fix a latent stack-overflow: ruleCurrentSpikeRatio / ruleBreakerMargin used Math.max(...valid) which blows the call stack on a ~590 K-element session; replaced with a loop-based arrayMax. This would have crashed the web app on the 7-day file. Validated headlessly on the real ES.004 (438 MB, 589,877 recs): streaming parse + full analysis peaks at ~278 MB RSS (vs the old ~1.6 GB), parses in ~0.5 s, analysis ~2 s, 18 events / 3 snapshots / 2 findings — all correct. Tests: store==records equivalence for events/snapshots/insights added to columnar.test.js. Web suite 87 -> 88 green; python 176 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- web/app.js | 226 ++++++++++++++++++++++++++++++------- web/column_source.js | 39 +++++++ web/column_store.js | 5 + web/events.js | 38 +++---- web/insights.js | 117 ++++++++++--------- web/plots.js | 38 ++++++- web/range_select.js | 25 +++- web/snapshots.js | 26 ++--- web/tariff.js | 13 ++- web/tests/columnar.test.js | 45 ++++++++ 10 files changed, 431 insertions(+), 141 deletions(-) create mode 100644 web/column_source.js diff --git a/web/app.js b/web/app.js index ff8a122..962648f 100644 --- a/web/app.js +++ b/web/app.js @@ -12,6 +12,7 @@ import { downloadCompareHtmlReport, downloadHtmlReport } from './html_report.js' import { downloadPdfReport } from './pdf_export.js'; import { clearCache, getCached, hashBuffer, putCached } from './cache.js'; import { MultiSession } from './multi_session.js'; +import { ColumnStore } from './column_store.js'; import { computeCost, loadTariff, normalizeTariff, parsePeakHoursString, peakHoursToString, saveTariff, @@ -96,7 +97,9 @@ const ms = new MultiSession(); let cachedSpec = null; let currentArrayBuffer = null; let currentConfig = null; // parsed ES.NNN-config.json companion (or null) -let currentRecords = null; // full parsed Records array (kept in memory) +let currentStore = null; // ColumnStore (memory-bounded; analysis + charts) +let currentRecords = null; // record array — only for small/CSV paths or read-through +let currentFile = null; // the dropped File/Blob, kept for read-through export let currentRecordCount = 0; let currentTimeRangeMs = null; let currentEvents = []; // detected events for currentRecords @@ -220,9 +223,41 @@ async function parseFile(file) { hideError(); els.summarySec.hidden = true; els.progressSec.hidden = false; - setProgress(0, file.size, 'reading file'); - currentArrayBuffer = await file.arrayBuffer(); - await parseBuffer(); + currentFile = file; // kept for read-through CSV/zoom export + currentArrayBuffer = null; // streaming path never holds the whole buffer + await parseStreaming(file); +} + +// Streaming columnar path (Feature A): hand the File to the worker, which reads +// it in 8 MB record-aligned chunks and transfers back typed-array columns. The +// full 438 MB ArrayBuffer is never resident on either thread. +async function parseStreaming(file) { + hideError(); + els.summarySec.hidden = true; + els.progressSec.hidden = false; + const spec = await getSpec(); + const reverseCts = selectedReversePhases(); + + setProgress(0, 100, 'parsing (streaming)'); + if (currentWorker) currentWorker.terminate(); + currentWorker = new Worker(new URL('./parser_worker.js', import.meta.url), + { type: 'module' }); + currentWorker.onmessage = (event) => { + const msg = event.data; + if (msg.type === 'progress') { + setProgress(msg.done, msg.total, + `parsing record ${msg.done.toLocaleString()} / ${msg.total.toLocaleString()}`); + } else if (msg.type === 'done-columnar') { + const store = ColumnStore.fromTransfer(msg); + onParseDoneColumnar(store).catch(showError); + } else if (msg.type === 'error') { + showError(new Error(msg.message)); + } + }; + currentWorker.onerror = (event) => { + showError(new Error(`Worker error: ${event.message ?? 'unknown'}`)); + }; + currentWorker.postMessage({ type: 'parse-stream', spec, blob: file, reverseCts }); } async function parseBuffer() { @@ -297,6 +332,7 @@ function selectedReversePhases() { async function onParseDone(msg) { currentRecords = msg.records; + currentStore = null; // legacy/CSV/small path keeps record objects currentRecordCount = msg.recordCount; currentFileHash = msg.fileHash ?? currentFileHash; if (msg.records.length > 0) { @@ -360,6 +396,109 @@ async function onParseDone(msg) { } } +// Columnar parse-done: analysis + charts run on the ColumnStore (memory-bounded); +// currentRecords stays null so the 7-day file never re-materialises 590 K objects. +async function onParseDoneColumnar(store) { + currentStore = store; + currentRecords = null; + currentRecordCount = store.n; + if (store.n > 0) { + currentTimeRangeMs = [store.firstStartMs, store.lastEndMs]; + } else { + currentTimeRangeMs = null; + } + els.progressSec.hidden = true; + renderSummary(); + els.summarySec.hidden = false; + + setProgress(0, 100, 'detecting events'); + els.progressSec.hidden = false; + await new Promise((r) => setTimeout(r, 0)); + try { + const spec = await getSpec(); + currentEvents = detectEvents(currentStore, spec); + currentSnapshots = pickSnapshots(currentStore, currentEvents, spec, { n: 3 }); + currentFindings = analyzeInsights(currentStore, currentEvents, spec, + currentSnapshots, currentConfig, + { breakerRatingA: loadBreakerRating() }); + ms.add({ + records: null, store: currentStore, + events: currentEvents, snapshots: currentSnapshots, + findings: currentFindings, config: currentConfig, + fileHash: currentFileHash, file: currentFile, + }); + renderInsights(); + renderEventsTable(); + renderSnapshotsList(); + renderQuantityGrid(); + renderStatsPanel(spec); + renderSessionsBar(); + els.insightsSec.hidden = currentFindings.length === 0; + els.sessionsSec.hidden = false; + els.eventsSec.hidden = false; + els.snapshotsSec.hidden = currentSnapshots.length === 0; + els.controlsSec.hidden = false; + els.exportSec.hidden = false; + els.rangeSec.hidden = false; + els.tariffSec.hidden = false; + setupRangeSelector(spec); + loadTariffIntoForm(); + renderTariffResult(); + tabState.hasSession = true; + updateTabUnlocks(); + if (tabState.current === 'import') activateTab('explore'); + } catch (e) { + showError(e); + return; + } finally { + els.progressSec.hidden = true; + } +} + +// The data source the analysis / chart / range engines should read: the store +// when present (streaming path), else the records array (small / CSV path). +function dataSource() { + return currentStore || currentRecords; +} + +// Per-session column accessor for compare-overlay charts: reads from a +// session's ColumnStore when present, else its records array. Only the few +// channels FULL_QUANTITIES references (all retained) are needed here. +function sessionAccessor(session, spec, name) { + if (session.store) { + const col = session.store.cols[name]; + const n = session.store.n; + return { + n, + startMs: (i) => session.store.startMs[i], + value: (i) => (col ? col[i] : 0), + base: n ? session.store.startMs[0] : 0, + }; + } + const recs = session.records || []; + const fi = new Map(spec.fields.map((f) => [f.name, f.index])); + const idx = fi.get(name); + return { + n: recs.length, + startMs: (i) => recs[i].startMs, + value: (i) => recs[i].floats[idx], + base: recs.length ? recs[0].startMs : 0, + }; +} + +// Materialise a records array for an export that genuinely needs every field +// (CSV / XLSX / bundle). For the store path this is a transient allocation that +// is dropped when the export finishes — it is NOT kept resident. +function recordsForExport(spec) { + if (currentRecords) return currentRecords; + if (currentStore) return currentStore.toRecords(spec); + return []; +} + +// Statistics panel (Feature B) — populated from whole_session_stats; safe no-op +// until the stats UI is wired so the parse flow never throws on older markup. +function renderStatsPanel(_spec) { /* Feature B fills this in */ } + // --- Summary rendering ------------------------------------------------------ function formatDuration(ms) { @@ -493,11 +632,11 @@ function getTariffFromForm() { async function renderTariffResult() { els.tariffResult.replaceChildren(); - if (!currentRecords) return; + if (!dataSource()) return; const t = getTariffFromForm(); if (t.peakRate === 0 && t.offpeakRate === 0) return; const spec = await getSpec(); - const cost = computeCost(currentRecords, spec, t); + const cost = computeCost(dataSource(), spec, t); const fmt = (n) => `${t.currency} ${n.toFixed(2)}`; const fmtKwh = (n) => `${n.toFixed(2)} kWh`; const dl = document.createElement('dl'); @@ -573,16 +712,28 @@ function switchToSession(label) { if (!ms.setActive(label)) return; const s = ms.getActive(); if (!s) return; - currentRecords = s.records; - currentRecordCount = s.records.length; + currentRecords = s.records || null; + currentStore = s.store || null; + currentFile = s.file || null; currentEvents = s.events; currentSnapshots = s.snapshots; currentFindings = s.findings; currentConfig = s.config; currentArrayBuffer = null; // can't re-parse a switched-to session - currentTimeRangeMs = s.records.length - ? [s.records[0].startMs, s.records[s.records.length - 1].endMs] - : null; + const ds = currentStore || currentRecords; + if (currentStore) { + currentRecordCount = currentStore.n; + currentTimeRangeMs = currentStore.n + ? [currentStore.firstStartMs, currentStore.lastEndMs] : null; + } else if (currentRecords) { + currentRecordCount = currentRecords.length; + currentTimeRangeMs = currentRecords.length + ? [currentRecords[0].startMs, currentRecords[currentRecords.length - 1].endMs] + : null; + } else { + currentRecordCount = 0; + currentTimeRangeMs = null; + } renderSummary(); renderInsights(); renderEventsTable(); @@ -593,7 +744,7 @@ function switchToSession(label) { function setupRangeSelector(spec) { if (rangeSelector) rangeSelector.destroy(); rangeSelector = renderRangeSelector( - els.rangeContainer, currentRecords, spec, (range) => { + els.rangeContainer, dataSource(), spec, (range) => { currentRange = range; const hash = rangeToHash(range); if (hash) history.replaceState(null, '', hash); @@ -675,9 +826,9 @@ function scrollToEvent(eventId) { } async function exportXlsx() { - if (!currentRecords) return; + if (!dataSource()) return; const spec = await getSpec(); - const scoped = scopeRecordsToRange(currentRecords, currentRange); + const scoped = scopeRecordsToRange(recordsForExport(spec), currentRange); const blob = buildXlsx({ records: scoped, spec, config: currentConfig }); const name = (currentConfig?.asset_name ?? 'fluke_session').replace(/[^a-zA-Z0-9._-]+/g, '_'); const suffix = currentRange ? '_range' : ''; @@ -685,9 +836,9 @@ async function exportXlsx() { } async function exportBundle() { - if (!currentRecords) return; + if (!dataSource()) return; const spec = await getSpec(); - const scoped = scopeRecordsToRange(currentRecords, currentRange); + const scoped = scopeRecordsToRange(recordsForExport(spec), currentRange); const xlsxBlob = buildXlsx({ records: scoped, spec, config: currentConfig }); await downloadBundleZip({ records: scoped, spec, xlsxBlob, @@ -696,9 +847,9 @@ async function exportBundle() { } async function exportPdf() { - if (!currentRecords) return; + if (!dataSource()) return; const spec = await getSpec(); - const scoped = scopeRecordsToRange(currentRecords, currentRange); + const scoped = scopeRecordsToRange(recordsForExport(spec), currentRange); const scopedEvents = currentRange ? currentEvents.filter((e) => !(e.tEndMs < currentRange.startMs || e.tStartMs > currentRange.endMs)) @@ -715,7 +866,7 @@ async function exportPdf() { } async function exportHtmlReport() { - if (!currentRecords) return; + if (!dataSource()) return; const spec = await getSpec(); if (ms.compareMode && ms.canCompare()) { // Compare-mode HTML uses the per-session summary + cross-session findings. @@ -727,7 +878,7 @@ async function exportHtmlReport() { }); return; } - const scoped = scopeRecordsToRange(currentRecords, currentRange); + const scoped = scopeRecordsToRange(recordsForExport(spec), currentRange); // When scoping, also scope events/findings to overlap the range. const scopedEvents = currentRange ? currentEvents.filter((e) => @@ -998,7 +1149,7 @@ function selectedSnapshotIds() { } async function renderAll() { - if (!currentRecords) return; + if (!dataSource()) return; const spec = await getSpec(); const quantities = selectedQuantities(); if (quantities.length === 0) { @@ -1027,7 +1178,7 @@ async function renderAll() { : {}), }; for (const q of quantities) { - renderChart(els.fullCharts, currentRecords, spec, q, FULL_QUANTITIES, fullOpts); + renderChart(els.fullCharts, dataSource(), spec, q, FULL_QUANTITIES, fullOpts); } } @@ -1040,7 +1191,7 @@ async function renderAll() { `Event #${ev.id}`, `${ev.kind} @ ${formatDate(ev.tStartMs)}`, )); for (const q of zoomQuantities) { - renderChart(els.eventCharts, currentRecords, spec, q, ZOOM_QUANTITIES, { + renderChart(els.eventCharts, dataSource(), spec, q, ZOOM_QUANTITIES, { startMs: ev.tStartMs - preMs, endMs: ev.tEndMs + postMs, }); @@ -1056,7 +1207,7 @@ async function renderAll() { `Snapshot #${s.id}`, `@ ${formatDate(s.tStartMs)}`, )); for (const q of zoomQuantities) { - renderChart(els.snapshotCharts, currentRecords, spec, q, ZOOM_QUANTITIES, { + renderChart(els.snapshotCharts, dataSource(), spec, q, ZOOM_QUANTITIES, { startMs: s.tStartMs, endMs: s.tEndMs, }); @@ -1077,25 +1228,24 @@ function renderOverlayChart(parentEl, spec, quantityKey, _allQuantities) { // chart readable; FULL_QUANTITIES often has 3 phases — we'd otherwise // overlay 9+ lines for 3 sessions × 3 phases). const firstCol = def.series[0]; - const fi = new Map(spec.fields.map((f) => [f.name, f.index])); - const idx = fi.get(firstCol.name); + const accessors = all.map((s) => sessionAccessor(s, spec, firstCol.name)); - const xs = []; // pooled relative-seconds axis const ySeries = all.map(() => []); // Build a unified sorted x axis from the union of all sessions' rel-seconds. const xSet = new Set(); - const relValues = all.map((s) => - s.records.map((r) => Math.round((r.startMs - s.records[0]?.startMs ?? 0) / 1000)) - ); - for (const arr of relValues) for (const x of arr) xSet.add(x); + for (const acc of accessors) { + for (let i = 0; i < acc.n; i++) { + xSet.add(Math.round((acc.startMs(i) - acc.base) / 1000)); + } + } const xsAll = [...xSet].sort((a, b) => a - b); // Per-session lookup: relSec → value for (let si = 0; si < all.length; si++) { - const s = all[si]; + const acc = accessors[si]; const map = new Map(); - for (let i = 0; i < s.records.length; i++) { - const rel = Math.round((s.records[i].startMs - (s.records[0]?.startMs ?? 0)) / 1000); - map.set(rel, s.records[i].floats[idx] * firstCol.scale); + for (let i = 0; i < acc.n; i++) { + const rel = Math.round((acc.startMs(i) - acc.base) / 1000); + map.set(rel, acc.value(i) * firstCol.scale); } for (const x of xsAll) ySeries[si].push(map.has(x) ? map.get(x) : null); } @@ -1230,9 +1380,9 @@ els.tariffApplyBtn.addEventListener('click', async () => { const amps = Number(els.breakerRating.value) || 0; saveBreakerRating(amps); // Re-run insights with the new breaker context. - if (currentRecords) { + if (dataSource()) { const spec = await getSpec(); - currentFindings = analyzeInsights(currentRecords, currentEvents, spec, + currentFindings = analyzeInsights(dataSource(), currentEvents, spec, currentSnapshots, currentConfig, { breakerRatingA: amps }); renderInsights(); @@ -1315,7 +1465,7 @@ document.addEventListener('keydown', (e) => { switch (e.key) { case 'r': case 'R': - if (currentRecords) { e.preventDefault(); renderAll().catch(showError); } + if (dataSource()) { e.preventDefault(); renderAll().catch(showError); } break; case 'z': case 'Z': { diff --git a/web/column_source.js b/web/column_source.js new file mode 100644 index 0000000..ee203c3 --- /dev/null +++ b/web/column_source.js @@ -0,0 +1,39 @@ +// Small adapter so the analysis engines (events / snapshots / insights / stats) +// can read columns from EITHER an array of record objects (legacy / tests) OR a +// ColumnStore (the memory-bounded streaming path). Both expose the same handful +// of accessors the engines need: column(name), startMs(i), endMs(i), length. + +import { ColumnStore } from './column_store.js'; + +/** + * Wrap a records-array or a ColumnStore into a uniform column source. + * @param {Array|ColumnStore} source + * @param {object} spec parsed field_map.json + * @returns {{length:number, startMs:(i:number)=>number, endMs:(i:number)=>number, + * column:(name:string)=>(Float32Array|number[]), isStore:boolean}} + */ +export function asColumnSource(source, spec) { + if (source instanceof ColumnStore) { + return { + length: source.n, + isStore: true, + startMs: (i) => source.startMs[i], + endMs: (i) => source.endMs[i], + column: (name) => source.col(name), + }; + } + // records array + const records = source; + const fi = new Map(spec.fields.map((f) => [f.name, f.index])); + return { + length: records.length, + isStore: false, + startMs: (i) => records[i].startMs, + endMs: (i) => records[i].endMs, + column: (name) => { + const idx = fi.get(name); + if (idx === undefined) throw new Error(`spec is missing field ${name}`); + return Float32Array.from(records, (r) => r.floats[idx]); + }, + }; +} diff --git a/web/column_store.js b/web/column_store.js index a5de3c7..1997349 100644 --- a/web/column_store.js +++ b/web/column_store.js @@ -23,6 +23,7 @@ export const STORE_COLUMNS = Object.freeze([ 'freq_avg_Hz', // Power / apparent / reactive / power-factor totals 'P_total_avg_W', 'S_total_avg_VA', 'Q_total_avg_VAR', 'PF_total_avg', + 'DPF_total_avg', // Per-row energy (per-bucket kWh roll-ups) 'Wh_total', // THD per phase (IEEE 519) — V and I, avg only @@ -30,6 +31,10 @@ export const STORE_COLUMNS = Object.freeze([ 'I_THD_pct_a_avg', 'I_THD_pct_b_avg', 'I_THD_pct_c_avg', ]); +// Chart series the web UI renders straight from the store. If a chart needs a +// channel not in STORE_COLUMNS the renderer must fall back to read-through. +export const CHART_COLUMNS = STORE_COLUMNS; + /** * Resolve STORE_COLUMNS to spec float indices, once. * @param {object} spec parsed field_map.json diff --git a/web/events.js b/web/events.js index 92546da..2e2d6c8 100644 --- a/web/events.js +++ b/web/events.js @@ -1,6 +1,8 @@ // Event detection — JS port of python/src/fluke_3540/events.py. // Mirrors the same thresholds, mask logic, and Event shape. +import { asColumnSource } from './column_source.js'; + export const DEFAULT_RULES = Object.freeze({ outage_v_threshold: 50.0, dip_pct_of_nominal: 0.90, @@ -84,32 +86,26 @@ function inferNominalLnV(vAvgByPhase, outageThreshold) { } /** - * Detect events on an array of parsed Records. - * @param {Array<{index:number, startMs:number, endMs:number, floats:Float32Array}>} records + * Detect events on an array of parsed Records OR a ColumnStore. + * @param {Array<{index:number, startMs:number, endMs:number, floats:Float32Array}>|import('./column_store.js').ColumnStore} source * @param {object} spec - parsed field_map.json * @param {{nominalLnV?: number, rules?: object}} [opts] * @returns {Array} */ -export function detectEvents(records, spec, opts = {}) { +export function detectEvents(source, spec, opts = {}) { const rules = { ...DEFAULT_RULES, ...(opts.rules ?? {}) }; let { nominalLnV = null } = opts; - if (records.length === 0) return []; - - const VLNmin = PHASES.map((ph) => fieldIndex(spec, `V_LN_${ph}_min_V`)); - const VLNmax = PHASES.map((ph) => fieldIndex(spec, `V_LN_${ph}_max_V`)); - const VLNavg = PHASES.map((ph) => fieldIndex(spec, `V_LN_${ph}_avg_V`)); - const Imax = PHASES.map((ph) => fieldIndex(spec, `I_${ph}_max_A`)); - const freqIdx = fieldIndex(spec, 'freq_avg_Hz'); - const pIdx = fieldIndex(spec, 'P_total_avg_W'); + const src = asColumnSource(source, spec); + if (src.length === 0) return []; - const N = records.length; - // Extract columns once into typed arrays for speed. - const vMin = VLNmin.map((idx) => Float32Array.from(records, (r) => r.floats[idx])); - const vMax = VLNmax.map((idx) => Float32Array.from(records, (r) => r.floats[idx])); - const vAvg = VLNavg.map((idx) => Float32Array.from(records, (r) => r.floats[idx])); - const iMaxArr = Imax.map((idx) => Float32Array.from(records, (r) => r.floats[idx])); - const freqArr = Float32Array.from(records, (r) => r.floats[freqIdx]); - const pTotal = Float32Array.from(records, (r) => r.floats[pIdx]); + const N = src.length; + // Extract columns once into typed arrays for speed (no-copy for a store). + const vMin = PHASES.map((ph) => src.column(`V_LN_${ph}_min_V`)); + const vMax = PHASES.map((ph) => src.column(`V_LN_${ph}_max_V`)); + const vAvg = PHASES.map((ph) => src.column(`V_LN_${ph}_avg_V`)); + const iMaxArr = PHASES.map((ph) => src.column(`I_${ph}_max_A`)); + const freqArr = src.column('freq_avg_Hz'); + const pTotal = src.column('P_total_avg_W'); if (nominalLnV === null) { nominalLnV = inferNominalLnV(vAvg, rules.outage_v_threshold); @@ -127,8 +123,8 @@ export function detectEvents(records, spec, opts = {}) { const outageMask = notOutage.map((b) => !b); const events = []; - const startMs = (i) => records[i].startMs; - const endMs = (i) => records[i].endMs; + const startMs = (i) => src.startMs(i); + const endMs = (i) => src.endMs(i); const phaseChars = (phaseList) => phaseList.slice(); diff --git a/web/insights.js b/web/insights.js index 72d9610..60121c4 100644 --- a/web/insights.js +++ b/web/insights.js @@ -4,27 +4,26 @@ // insight_rules section. Same Finding shape, same rule logic, same // headline/detail templates so the web UI and CLI report look alike. +import { asColumnSource } from './column_source.js'; + function ruleVal(spec, key, fallback) { const v = spec?.insight_rules?.[key]; return v === undefined || v === null ? fallback : Number(v); } -function fieldIndex(spec, name) { - const f = spec.fields.find((f) => f.name === name); - if (!f) throw new Error(`spec missing field ${name}`); - return f.index; -} - -function col(records, spec, name) { - const idx = fieldIndex(spec, name); - return records.map((r) => r.floats[idx]); +// `src` is a column source (asColumnSource result). col() returns the channel +// as an Array for the .filter()/.map() patterns the rules use. +function col(src, name) { + return Array.from(src.column(name)); } -function nonOutageMask(records, spec) { - const a = col(records, spec, 'V_LN_a_avg_V'); - const b = col(records, spec, 'V_LN_b_avg_V'); - const c = col(records, spec, 'V_LN_c_avg_V'); - return records.map((_, i) => a[i] > 50 && b[i] > 50 && c[i] > 50); +function nonOutageMask(src) { + const a = src.column('V_LN_a_avg_V'); + const b = src.column('V_LN_b_avg_V'); + const c = src.column('V_LN_c_avg_V'); + const out = new Array(src.length); + for (let i = 0; i < src.length; i++) out[i] = a[i] > 50 && b[i] > 50 && c[i] > 50; + return out; } function mean(arr) { @@ -34,6 +33,14 @@ function mean(arr) { return s / arr.length; } +// Loop-based max — Math.max(...arr) overflows the call stack on a 7-day +// (~590 K element) session. +function arrayMax(arr) { + let m = -Infinity; + for (let i = 0; i < arr.length; i++) if (arr[i] > m) m = arr[i]; + return m; +} + // --- Individual rules ------------------------------------------------------- function ruleOutageSignatures(events, spec) { @@ -96,12 +103,12 @@ function ruleOutageSignatures(events, spec) { return findings; } -function rulePhaseAsymmetry(records, spec) { +function rulePhaseAsymmetry(src, spec) { const threshold = ruleVal(spec, 'phase_asymmetry_pct', 2.0); - const notOut = nonOutageMask(records, spec); - const a = col(records, spec, 'V_LN_a_avg_V'); - const b = col(records, spec, 'V_LN_b_avg_V'); - const c = col(records, spec, 'V_LN_c_avg_V'); + const notOut = nonOutageMask(src); + const a = col(src, 'V_LN_a_avg_V'); + const b = col(src, 'V_LN_b_avg_V'); + const c = col(src, 'V_LN_c_avg_V'); const aVals = a.filter((_, i) => notOut[i]); const bVals = b.filter((_, i) => notOut[i]); const cVals = c.filter((_, i) => notOut[i]); @@ -133,14 +140,14 @@ function rulePhaseAsymmetry(records, spec) { }]; } -function rulePfDrift(records, spec) { +function rulePfDrift(src, spec) { const threshold = ruleVal(spec, 'pf_drift_threshold', 0.85); const minFrac = ruleVal(spec, 'pf_drift_min_fraction', 0.10); - const notOut = nonOutageMask(records, spec); - const pf = col(records, spec, 'PF_total_avg'); - const q = col(records, spec, 'Q_total_avg_VAR'); - const s = col(records, spec, 'S_total_avg_VA'); - const lowMask = records.map((_, i) => + const notOut = nonOutageMask(src); + const pf = col(src, 'PF_total_avg'); + const q = col(src, 'Q_total_avg_VAR'); + const s = col(src, 'S_total_avg_VA'); + const lowMask = Array.from({ length: src.length }, (_, i) => notOut[i] && Math.abs(pf[i]) > 0 && Math.abs(pf[i]) < threshold ); const total = lowMask.length; @@ -171,15 +178,15 @@ function rulePfDrift(records, spec) { }]; } -function ruleImbalanceSustained(records, spec) { +function ruleImbalanceSustained(src, spec) { const pctThreshold = ruleVal(spec, 'imbalance_sustained_pct', 1.5); const secsThreshold = Math.round(ruleVal(spec, 'imbalance_sustained_secs', 60)); - const notOut = nonOutageMask(records, spec); - const a = col(records, spec, 'V_LN_a_avg_V'); - const b = col(records, spec, 'V_LN_b_avg_V'); - const c = col(records, spec, 'V_LN_c_avg_V'); - const imbal = new Array(records.length).fill(0); - for (let i = 0; i < records.length; i++) { + const notOut = nonOutageMask(src); + const a = col(src, 'V_LN_a_avg_V'); + const b = col(src, 'V_LN_b_avg_V'); + const c = col(src, 'V_LN_c_avg_V'); + const imbal = new Array(src.length).fill(0); + for (let i = 0; i < src.length; i++) { if (!notOut[i]) continue; const m = (a[i] + b[i] + c[i]) / 3; if (m <= 50) continue; @@ -220,11 +227,12 @@ function ruleImbalanceSustained(records, spec) { }]; } -function ruleFreqStiffness(records, events, spec) { +function ruleFreqStiffness(src, events, spec) { const thresholdHz = ruleVal(spec, 'freq_stiffness_hz', 0.05); const minCount = Math.round(ruleVal(spec, 'freq_stiffness_min_count', 3)); - const freq = col(records, spec, 'freq_avg_Hz'); - const indexByTime = new Map(records.map((r, i) => [r.startMs, i])); + const freq = col(src, 'freq_avg_Hz'); + const indexByTime = new Map(); + for (let i = 0; i < src.length; i++) indexByTime.set(src.startMs(i), i); const correlations = []; for (const ev of events) { if (ev.kind !== 'power_step') continue; @@ -256,11 +264,11 @@ function ruleFreqStiffness(records, events, spec) { }]; } -function ruleOutageFrequency(records, events, spec) { +function ruleOutageFrequency(src, events, spec) { const perDayThreshold = ruleVal(spec, 'outage_frequency_per_day', 1.0); const outages = events.filter((e) => e.kind === 'outage'); - if (outages.length === 0 || records.length === 0) return []; - const durSecs = (records[records.length - 1].endMs - records[0].startMs) / 1000; + if (outages.length === 0 || src.length === 0) return []; + const durSecs = (src.endMs(src.length - 1) - src.startMs(0)) / 1000; if (durSecs <= 0) return []; const days = durSecs / 86400; const rate = outages.length / days; @@ -281,17 +289,17 @@ function ruleOutageFrequency(records, events, spec) { }]; } -function ruleCurrentSpikeRatio(records, spec, opts = {}) { +function ruleCurrentSpikeRatio(src, spec, opts = {}) { const threshold = ruleVal(spec, 'current_to_mean_ratio_alert', 5.0); const breakerA = Number(opts.breakerRatingA) > 0 ? Number(opts.breakerRatingA) : null; const findings = []; - const notOut = nonOutageMask(records, spec); + const notOut = nonOutageMask(src); for (const phase of ['a', 'b', 'c']) { - const iMax = col(records, spec, `I_${phase}_max_A`); + const iMax = col(src, `I_${phase}_max_A`); const valid = iMax.filter((_, i) => notOut[i]); if (valid.length === 0) continue; const m = mean(valid); - const peak = Math.max(...valid); + const peak = arrayMax(valid); if (m <= 0) continue; const ratio = peak / m; if (ratio < threshold) continue; @@ -329,17 +337,17 @@ function ruleCurrentSpikeRatio(records, spec, opts = {}) { return findings; } -function ruleBreakerMargin(records, spec, opts = {}) { +function ruleBreakerMargin(src, spec, opts = {}) { const breakerA = Number(opts.breakerRatingA) > 0 ? Number(opts.breakerRatingA) : null; if (!breakerA) return []; - const notOut = nonOutageMask(records, spec); + const notOut = nonOutageMask(src); let worstPhase = null; let worstPeak = 0; for (const phase of ['a', 'b', 'c']) { - const iMax = col(records, spec, `I_${phase}_max_A`); + const iMax = col(src, `I_${phase}_max_A`); const valid = iMax.filter((_, i) => notOut[i]); if (!valid.length) continue; - const peak = Math.max(...valid); + const peak = arrayMax(valid); if (peak > worstPeak) { worstPeak = peak; worstPhase = phase; } } if (!worstPhase || worstPeak <= breakerA) return []; @@ -363,16 +371,17 @@ function ruleBreakerMargin(records, spec, opts = {}) { const SEV_RANK = { alert: 0, warn: 1, info: 2 }; -export function analyzeInsights(records, events, spec, _snapshots = [], _config = null, opts = {}) { +export function analyzeInsights(source, events, spec, _snapshots = [], _config = null, opts = {}) { + const src = asColumnSource(source, spec); const raw = [ ...ruleOutageSignatures(events, spec), - ...rulePhaseAsymmetry(records, spec), - ...rulePfDrift(records, spec), - ...ruleImbalanceSustained(records, spec), - ...ruleFreqStiffness(records, events, spec), - ...ruleOutageFrequency(records, events, spec), - ...ruleCurrentSpikeRatio(records, spec, opts), - ...ruleBreakerMargin(records, spec, opts), + ...rulePhaseAsymmetry(src, spec), + ...rulePfDrift(src, spec), + ...ruleImbalanceSustained(src, spec), + ...ruleFreqStiffness(src, events, spec), + ...ruleOutageFrequency(src, events, spec), + ...ruleCurrentSpikeRatio(src, spec, opts), + ...ruleBreakerMargin(src, spec, opts), ]; raw.sort((a, b) => (SEV_RANK[a.severity] - SEV_RANK[b.severity]) || a.kind.localeCompare(b.kind)); return raw.map((f, i) => ({ ...f, id: i })); diff --git a/web/plots.js b/web/plots.js index 8a735e1..e70d0d7 100644 --- a/web/plots.js +++ b/web/plots.js @@ -175,14 +175,21 @@ export function decimateSeries(data, targetPoints = 4000) { } /** - * Build uPlot data arrays from a set of records and a quantity definition. - * Returns [xs, ...ySeries] suitable for uPlot's data prop. + * Build uPlot data arrays from records OR a ColumnStore and a quantity + * definition. Returns [xs, ...ySeries] suitable for uPlot's data prop. + * + * A ColumnStore is detected by duck-typing (.cols + .startMs typed array); its + * channels are read column-wise with no per-record allocation, so the chart + * path is memory-bounded on the 7-day file. */ -function buildPlotData(records, spec, quantityDef, startMs, endMs) { +function buildPlotData(source, spec, quantityDef, startMs, endMs) { + if (source && source.cols && source.startMs instanceof Float64Array) { + return buildPlotDataFromStore(source, quantityDef, startMs, endMs); + } const fi = fieldIndexMap(spec); const xs = []; const ys = quantityDef.series.map(() => []); - for (const rec of records) { + for (const rec of source) { if (startMs !== null && rec.startMs < startMs) continue; if (endMs !== null && rec.startMs > endMs) continue; xs.push(rec.startMs / 1000); // uPlot expects unix seconds @@ -194,6 +201,29 @@ function buildPlotData(records, spec, quantityDef, startMs, endMs) { return [xs, ...ys]; } +function buildPlotDataFromStore(store, quantityDef, startMs, endMs) { + const cols = quantityDef.series.map((s) => { + const c = store.cols[s.name]; + if (!c) { + throw new Error( + `chart series ${s.name} not retained in the ColumnStore — ` + + 'add it to STORE_COLUMNS or use the records path' + ); + } + return [c, s.scale]; + }); + const xs = []; + const ys = quantityDef.series.map(() => []); + for (let i = 0; i < store.n; i++) { + const t = store.startMs[i]; + if (startMs !== null && t < startMs) continue; + if (endMs !== null && t > endMs) continue; + xs.push(t / 1000); + for (let k = 0; k < cols.length; k++) ys[k].push(cols[k][0][i] * cols[k][1]); + } + return [xs, ...ys]; +} + /** * Render one uPlot chart into a container. * @returns {{plot: uPlot, container: HTMLDivElement, data: any[], def: object}} diff --git a/web/range_select.js b/web/range_select.js index aa74fa3..1274c73 100644 --- a/web/range_select.js +++ b/web/range_select.js @@ -26,13 +26,28 @@ function fieldIndex(spec, name) { * @param {(range: {startMs: number, endMs: number} | null) => void} onRangeChange * @returns {{ destroy: () => void, setRange: (r) => void }} */ -export function renderRangeSelector(container, records, spec, onRangeChange) { +export function renderRangeSelector(container, source, spec, onRangeChange) { const uPlot = uplotOrThrow(); container.replaceChildren(); - if (records.length === 0) return { destroy() {}, setRange() {} }; - const pIdx = fieldIndex(spec, 'P_total_avg_W'); - const xs = records.map((r) => r.startMs / 1000); - const ys = records.map((r) => r.floats[pIdx] / 1000); + let xs; + let ys; + if (source && source.cols && source.startMs instanceof Float64Array) { + // ColumnStore path — read the P_total column directly (no per-record alloc). + if (source.n === 0) return { destroy() {}, setRange() {} }; + const p = source.cols.P_total_avg_W; + xs = new Array(source.n); + ys = new Array(source.n); + for (let i = 0; i < source.n; i++) { + xs[i] = source.startMs[i] / 1000; + ys[i] = p[i] / 1000; + } + } else { + const records = source; + if (records.length === 0) return { destroy() {}, setRange() {} }; + const pIdx = fieldIndex(spec, 'P_total_avg_W'); + xs = records.map((r) => r.startMs / 1000); + ys = records.map((r) => r.floats[pIdx] / 1000); + } const label = document.createElement('div'); label.className = 'range-label'; diff --git a/web/snapshots.js b/web/snapshots.js index 8b2993a..3fc66b4 100644 --- a/web/snapshots.js +++ b/web/snapshots.js @@ -1,11 +1,7 @@ // Snapshot picking — port of python/src/fluke_3540/snapshots.py. // Picks quiet, non-event windows by rolling stdev of P_total. -function fieldIndex(spec, name) { - const f = spec.fields.find((f) => f.name === name); - if (!f) throw new Error(`spec is missing field ${name}`); - return f.index; -} +import { asColumnSource } from './column_source.js'; function pstdev(values, start, end) { if (end - start + 1 < 2) return 0; @@ -26,18 +22,18 @@ function mean(values, start, end) { /** * Pick up to N snapshot windows that are quiet (low P_total stdev) and don't * overlap any detected event. - * @param {Array<{startMs:number, endMs:number, floats:Float32Array}>} records + * @param {Array<{startMs:number, endMs:number, floats:Float32Array}>|import('./column_store.js').ColumnStore} source * @param {Array<{tStartMs:number, tEndMs:number}>} events * @param {object} spec * @param {{n?: number, windowSecs?: number, minSeparationSecs?: number}} [opts] */ -export function pickSnapshots(records, events, spec, opts = {}) { +export function pickSnapshots(source, events, spec, opts = {}) { const { n = 3, windowSecs = 300, minSeparationSecs = 3600 } = opts; - if (records.length === 0) return []; + const src = asColumnSource(source, spec); + if (src.length === 0) return []; - const pIdx = fieldIndex(spec, 'P_total_avg_W'); - const p = Float32Array.from(records, (r) => r.floats[pIdx]); - const N = p.length; + const p = src.column('P_total_avg_W'); + const N = src.length; if (windowSecs <= 1 || windowSecs > N) return []; @@ -51,8 +47,8 @@ export function pickSnapshots(records, events, spec, opts = {}) { const eventIntervals = events.map((ev) => [ev.tStartMs, ev.tEndMs]); function overlapsEvent(i) { if (rolling[i] === null) return true; - const winStart = records[i - windowSecs + 1].startMs; - const winEnd = records[i].endMs; + const winStart = src.startMs(i - windowSecs + 1); + const winEnd = src.endMs(i); for (const [es, ee] of eventIntervals) { if (!(winEnd < es || winStart > ee)) return true; } @@ -70,8 +66,8 @@ export function pickSnapshots(records, events, spec, opts = {}) { const picked = []; const usedCenters = []; for (const [stdevVal, i] of candidates) { - const winStart = records[i - windowSecs + 1].startMs; - const winEnd = records[i].endMs; + const winStart = src.startMs(i - windowSecs + 1); + const winEnd = src.endMs(i); const center = (winStart + winEnd) / 2; let tooClose = false; for (const prev of usedCenters) { diff --git a/web/tariff.js b/web/tariff.js index 4b4e359..2e76fde 100644 --- a/web/tariff.js +++ b/web/tariff.js @@ -36,15 +36,20 @@ export function emptyCost(currency = 'USD') { }; } -export function computeCost(records, spec, tariff) { +export function computeCost(source, spec, tariff) { if (!tariff) return emptyCost(); const whIdx = fieldIndex(spec, 'Wh_total'); if (whIdx < 0) return emptyCost(tariff.currency); let pkImp = 0, pkExp = 0, opImp = 0, opExp = 0; - for (const r of records) { - const wh = r.floats[whIdx]; + // Accept a ColumnStore (Wh_total is a retained channel) or a records array. + const isStore = source && source.cols && source.startMs instanceof Float64Array; + const n = isStore ? source.n : source.length; + const whCol = isStore ? source.cols.Wh_total : null; + for (let i = 0; i < n; i++) { + const wh = isStore ? whCol[i] : source[i].floats[whIdx]; if (!wh) continue; - const hour = new Date(r.startMs).getUTCHours(); + const startMs = isStore ? source.startMs[i] : source[i].startMs; + const hour = new Date(startMs).getUTCHours(); const peak = isPeak(tariff, hour); if (wh > 0) { if (peak) pkImp += wh; else opImp += wh; diff --git a/web/tests/columnar.test.js b/web/tests/columnar.test.js index 3c68dbd..56be873 100644 --- a/web/tests/columnar.test.js +++ b/web/tests/columnar.test.js @@ -20,6 +20,9 @@ import { parseTrendColumnarStream, } from '../parser.js'; import { ColumnStore, STORE_COLUMNS } from '../column_store.js'; +import { detectEvents } from '../events.js'; +import { pickSnapshots } from '../snapshots.js'; +import { analyzeInsights } from '../insights.js'; const __dirname = dirname(fileURLToPath(import.meta.url)); const repoRoot = resolve(__dirname, '..', '..'); @@ -174,6 +177,48 @@ test('ColumnStore.fromRecords / toRecords round-trips retained channels', () => assert.equal(back[3].startMs, records[3].startMs); }); +test('analysis engines: ColumnStore path == records path', () => { + // Build records with planted events, then a store from those records, and + // assert detectEvents / pickSnapshots / analyzeInsights agree. + const nameToIdx = new Map(spec.fields.map((f) => [f.name, f.index])); + const baseMs = Date.UTC(2024, 0, 13, 22, 0, 0); + const N = 400; + const records = []; + for (let n = 0; n < N; n++) { + const floats = new Float32Array(spec.data_floats); + for (const ph of ['a', 'b', 'c']) { + for (const st of ['min', 'max', 'avg']) { + floats[nameToIdx.get(`V_LN_${ph}_${st}_V`)] = 277.0; + floats[nameToIdx.get(`I_${ph}_${st}_A`)] = 100.0; + } + } + floats[nameToIdx.get('freq_avg_Hz')] = 60.0; + floats[nameToIdx.get('P_total_avg_W')] = 50000.0; + floats[nameToIdx.get('PF_total_avg')] = 0.99; + // Plant an outage 100..119 and a dip 200..204. + if (n >= 100 && n <= 119) { + for (const ph of ['a', 'b', 'c']) { + for (const st of ['min', 'max', 'avg']) floats[nameToIdx.get(`V_LN_${ph}_${st}_V`)] = 0; + } + } + if (n >= 200 && n <= 204) floats[nameToIdx.get('V_LN_a_min_V')] = 200.0; + records.push({ index: n, startMs: baseMs + n * 1000, endMs: baseMs + (n + 1) * 1000, floats }); + } + const store = ColumnStore.fromRecords(records, spec); + + const evRec = detectEvents(records, spec, { nominalLnV: 277.0 }); + const evStore = detectEvents(store, spec, { nominalLnV: 277.0 }); + assert.deepEqual(evStore, evRec, 'events differ between store and records path'); + + const snRec = pickSnapshots(records, evRec, spec, { n: 2, windowSecs: 60, minSeparationSecs: 1 }); + const snStore = pickSnapshots(store, evStore, spec, { n: 2, windowSecs: 60, minSeparationSecs: 1 }); + assert.deepEqual(snStore, snRec, 'snapshots differ'); + + const inRec = analyzeInsights(records, evRec, spec); + const inStore = analyzeInsights(store, evStore, spec); + assert.deepEqual(inStore.map((f) => f.kind), inRec.map((f) => f.kind), 'insights differ'); +}); + test('ColumnStore.fromTransfer wraps a worker payload', () => { const ab = loadFixture(); const payload = parseTrendColumnar(ab, spec); From 0b04a79e79d797ca907ef10f593fc76c36d7f2ae Mon Sep 17 00:00:00 2001 From: Bill Bai Date: Tue, 2 Jun 2026 23:36:37 -0500 Subject: [PATCH 03/11] feat(web): JS stats port + Statistics panel + ToD chart (Feature B) Add web/analysis.js mirroring python analysis.py numerically: - RunningMoments (Welford), PercentileSketch (fixed-width histogram) - wholeSessionStats (per-channel count/min/p1/p5/median/mean/p95/p99/max/stdev + under-voltage / over-current second accounting) - classifyItic / eventItic (ITIC/CBEMA ride-through) - parsePeriod / parseTodWindow / timeOfDayProfile - correlateMarkers All accept a ColumnStore or records array via column_source. Surface in the web UI: a Statistics section (per-channel table + threshold note) and a time-of-day profile chart, computed off the resident store. Embed the whole-session stats table (and a narrative slot for Feature E) into the exported HTML report. Parity: python test_analysis_parity_golden.py emits a deterministic golden JSON; web/tests/analysis_parity.test.js recreates the identical session and asserts stats/ITIC/eventItic/ToD match within float tolerance (percentiles are exact since both use the same sketch). Web 88 -> 92 green; python 176 -> 177 green. Co-Authored-By: Claude Opus 4.8 (1M context) --- python/tests/fixtures/analysis_golden.json | 344 +++++++++++++++++++ python/tests/test_analysis_parity_golden.py | 92 ++++++ web/analysis.js | 345 ++++++++++++++++++++ web/app.js | 116 ++++++- web/html_report.js | 30 +- web/index.html | 8 + web/tests/analysis_parity.test.js | 139 ++++++++ 7 files changed, 1070 insertions(+), 4 deletions(-) create mode 100644 python/tests/fixtures/analysis_golden.json create mode 100644 python/tests/test_analysis_parity_golden.py create mode 100644 web/analysis.js create mode 100644 web/tests/analysis_parity.test.js diff --git a/python/tests/fixtures/analysis_golden.json b/python/tests/fixtures/analysis_golden.json new file mode 100644 index 0000000..02a62a9 --- /dev/null +++ b/python/tests/fixtures/analysis_golden.json @@ -0,0 +1,344 @@ +{ + "stats": { + "V_LN_a_avg_V": { + "unit": "V", + "count": 600, + "min": 240.0, + "p1": 240.05, + "p5": 274.05, + "median": 277.05, + "mean": 275.7550000000001, + "p95": 280.05, + "p99": 280.05, + "max": 280.0, + "stdev": 6.9239902994348 + }, + "V_LN_b_avg_V": { + "unit": "V", + "count": 600, + "min": 277.0, + "p1": 277.05, + "p5": 277.05, + "median": 277.05, + "mean": 277.0, + "p95": 277.05, + "p99": 277.05, + "max": 277.0, + "stdev": 0.0 + }, + "V_LN_c_avg_V": { + "unit": "V", + "count": 600, + "min": 277.0, + "p1": 277.05, + "p5": 277.05, + "median": 277.05, + "mean": 277.0, + "p95": 277.05, + "p99": 277.05, + "max": 277.0, + "stdev": 0.0 + }, + "I_a_avg_A": { + "unit": "A", + "count": 600, + "min": 100.0, + "p1": 100.125, + "p5": 100.125, + "median": 105.125, + "mean": 104.975, + "p95": 110.125, + "p99": 110.125, + "max": 110.0, + "stdev": 3.160860905934752 + }, + "I_b_avg_A": { + "unit": "A", + "count": 600, + "min": 100.0, + "p1": 100.125, + "p5": 100.125, + "median": 100.125, + "mean": 100.0, + "p95": 100.125, + "p99": 100.125, + "max": 100.0, + "stdev": 0.0 + }, + "I_c_avg_A": { + "unit": "A", + "count": 600, + "min": 100.0, + "p1": 100.125, + "p5": 100.125, + "median": 100.125, + "mean": 106.25000000000011, + "p95": 100.125, + "p99": 100.125, + "max": 850.0, + "stdev": 68.17945071647314 + }, + "freq_avg_Hz": { + "unit": "Hz", + "count": 600, + "min": 59.9900016784668, + "p1": 59.99125, + "p5": 59.99125, + "median": 60.00125, + "mean": 59.999999999999936, + "p95": 60.00875, + "p99": 60.00875, + "max": 60.0099983215332, + "stdev": 0.008163595346876485 + }, + "P_total_avg_W": { + "unit": "W", + "count": 600, + "min": 50000.0, + "p1": 50500.0, + "p5": 52500.0, + "median": 74500.0, + "mean": 74500.00000000013, + "p95": 97500.0, + "p99": 99500.0, + "max": 99000.0, + "stdev": 14430.86968966181 + }, + "S_total_avg_VA": { + "unit": "VA", + "count": 600, + "min": 0.0, + "p1": 500.0, + "p5": 500.0, + "median": 500.0, + "mean": 0.0, + "p95": 500.0, + "p99": 500.0, + "max": 0.0, + "stdev": 0.0 + }, + "Q_total_avg_VAR": { + "unit": "VAR", + "count": 600, + "min": 0.0, + "p1": 500.0, + "p5": 500.0, + "median": 500.0, + "mean": 0.0, + "p95": 500.0, + "p99": 500.0, + "max": 0.0, + "stdev": 0.0 + }, + "PF_total_avg": { + "unit": "", + "count": 600, + "min": 0.8999999761581421, + "p1": 0.9001125000000003, + "p5": 0.9001125000000003, + "median": 0.9200625000000002, + "mean": 0.9200000047683707, + "p95": 0.9400125000000001, + "p99": 0.9400125000000001, + "max": 0.9399999976158142, + "stdev": 0.01414213899548889 + }, + "_thresholds": { + "undervoltage_v": 250.0, + "sec_undervoltage": 20, + "pct_undervoltage": 3.3333333333333335, + "overcurrent_a": 800.0, + "sec_overcurrent": 5, + "pct_overcurrent": 0.8333333333333334, + "total_records": 600 + } + }, + "tod_rows": [ + { + "bin": "22:00", + "n": 60, + "n_days": 1, + "p_avg_kW": 71.16666666666667, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 276.9, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 104.75, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:01", + "n": 60, + "n_days": 1, + "p_avg_kW": 72.83333333333333, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 264.68333333333334, + "v_min_V": 240.0, + "v_max_V": 280.0, + "i_avg_A": 105.16666666666667, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:02", + "n": 60, + "n_days": 1, + "p_avg_kW": 74.5, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 276.96666666666664, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 104.85, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:03", + "n": 60, + "n_days": 1, + "p_avg_kW": 76.16666666666667, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 277.0, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 105.08333333333333, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:04", + "n": 60, + "n_days": 1, + "p_avg_kW": 77.83333333333333, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 277.03333333333336, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 104.95, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:05", + "n": 60, + "n_days": 1, + "p_avg_kW": 71.16666666666667, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 276.95, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 105.0, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:06", + "n": 60, + "n_days": 1, + "p_avg_kW": 72.83333333333333, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 277.1, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 105.05, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:07", + "n": 60, + "n_days": 1, + "p_avg_kW": 74.5, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 276.9, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 104.91666666666667, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:08", + "n": 60, + "n_days": 1, + "p_avg_kW": 76.16666666666667, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 277.05, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 105.15, + "i_min_A": 100.0, + "i_max_A": 110.0 + }, + { + "bin": "22:09", + "n": 60, + "n_days": 1, + "p_avg_kW": 77.83333333333333, + "p_min_kW": 50.0, + "p_max_kW": 99.0, + "v_avg_V": 276.96666666666664, + "v_min_V": 274.0, + "v_max_V": 280.0, + "i_avg_A": 104.83333333333333, + "i_min_A": 100.0, + "i_max_A": 110.0 + } + ], + "itic_points": [ + [ + 70.0, + 0.1 + ], + [ + 60.0, + 1.0 + ], + [ + 130.0, + 0.4 + ], + [ + 95.0, + 5.0 + ], + [ + 0.0, + 18.0 + ] + ], + "itic": [ + "no_interruption", + "no_damage", + "prohibited", + "no_interruption", + "no_damage" + ], + "event_itic": [ + { + "residual_pct": 72.0, + "duration_secs": 2.0, + "itic_class": "no_damage" + }, + { + "residual_pct": 0.0, + "duration_secs": 120.0, + "itic_class": "no_damage" + }, + { + "residual_pct": 113.99999999999999, + "duration_secs": 1.0, + "itic_class": "no_interruption" + } + ] +} \ No newline at end of file diff --git a/python/tests/test_analysis_parity_golden.py b/python/tests/test_analysis_parity_golden.py new file mode 100644 index 0000000..40261ed --- /dev/null +++ b/python/tests/test_analysis_parity_golden.py @@ -0,0 +1,92 @@ +"""Emit golden analysis outputs for the JS parity test (Feature B). + +This test builds a deterministic session, runs the Python analysis functions, +and writes the results to python/tests/fixtures/analysis_golden.json. The JS +side (web/tests/analysis_parity.test.js) loads the same JSON and asserts its +own port produces identical numbers within float tolerance. + +Keeping the generator in pytest (rather than a standalone script) means the +golden file is regenerated whenever the suite runs, so it can never drift from +the Python implementation. +""" +from __future__ import annotations + +import datetime as dt +import json +from pathlib import Path + +from fluke_3540.analysis import ( + classify_itic, event_itic, time_of_day_profile, whole_session_stats, +) +from fluke_3540.events import Event +from fluke_3540.store import ColumnStore + +from conftest import make_records, plant_window + + +GOLDEN_PATH = Path(__file__).parent / "fixtures" / "analysis_golden.json" + + +def _build_session() -> ColumnStore: + """A deterministic multi-shape session the JS test recreates exactly. + + - 600 records (10 minutes) starting 2024-01-13 22:00:00 UTC + - P_total ramps so mean/percentiles are non-trivial + - V_LN_a wobbles for a real stdev + - one undervoltage window and one overcurrent window for thresholds + - one dip window for the time-of-day / ITIC checks + """ + overrides: dict = {} + for i in range(600): + overrides.setdefault(i, {}) + overrides[i]["P_total_avg_W"] = 50_000.0 + (i % 50) * 1000.0 + overrides[i]["V_LN_a_avg_V"] = 277.0 + (i % 7) - 3.0 + overrides[i]["I_a_avg_A"] = 100.0 + (i % 11) + overrides[i]["PF_total_avg"] = 0.90 + (i % 5) * 0.01 + overrides[i]["freq_avg_Hz"] = 60.0 + ((i % 3) - 1) * 0.01 + plant_window(overrides, 100, 119, {"V_LN_a_avg_V": 240.0}) # undervoltage 20 s + plant_window(overrides, 200, 204, {"I_c_avg_A": 850.0}) # overcurrent 5 s + recs = make_records(600, overrides=overrides) + return ColumnStore.from_records(recs) + + +def test_emit_analysis_golden(): + store = _build_session() + stats = whole_session_stats(store) + tod = time_of_day_profile(store, window=(0, 1440), bin_minutes=1) + + # A handful of ITIC classifications spanning each region. + itic_points = [ + [70.0, 0.1], + [60.0, 1.0], + [130.0, 0.4], + [95.0, 5.0], + [0.0, 18.0], + ] + itic = [classify_itic(p, d) for p, d in itic_points] + + # event_itic on a representative dip + outage + swell. + base = dt.datetime(2024, 1, 13, 22, 0, 0, tzinfo=dt.timezone.utc) + sample_events = [ + Event(0, "dip", base, base + dt.timedelta(seconds=2), 0.72, ("a",)), + Event(1, "outage", base, base + dt.timedelta(seconds=120), 0.0, ("a", "b", "c")), + Event(2, "swell", base, base + dt.timedelta(seconds=1), 1.14, ("b",)), + ] + event_itic_out = [event_itic(e, 277.0) for e in sample_events] + + golden = { + "stats": stats, + "tod_rows": tod, + "itic_points": itic_points, + "itic": itic, + "event_itic": event_itic_out, + } + GOLDEN_PATH.parent.mkdir(parents=True, exist_ok=True) + GOLDEN_PATH.write_text(json.dumps(golden, indent=2), encoding="utf-8") + + # Sanity assertions so this test fails loudly if analysis regresses. + assert stats["_thresholds"]["sec_undervoltage"] == 20 + assert stats["_thresholds"]["sec_overcurrent"] == 5 + assert stats["P_total_avg_W"]["count"] == 600 + assert itic[0] == "no_interruption" + assert len(tod) > 0 diff --git a/web/analysis.js b/web/analysis.js new file mode 100644 index 0000000..0628207 --- /dev/null +++ b/web/analysis.js @@ -0,0 +1,345 @@ +// Round-2 analysis — JS port of python/src/fluke_3540/analysis.py. +// +// Mirrors the Python algorithms numerically so the web output matches the CLI: +// - RunningMoments (Welford running mean/variance) +// - PercentileSketch (fixed-width streaming histogram) +// - wholeSessionStats +// - classifyItic / eventItic (ITIC/CBEMA ride-through) +// - parsePeriod / bucketKey / bucketLabel / bucketSummaryRow +// - timeOfDayProfile +// - correlateMarkers +// +// All functions accept a ColumnStore OR a records array via asColumnSource. + +import { asColumnSource } from './column_source.js'; + +// --- Streaming percentile sketch ------------------------------------------- + +export class PercentileSketch { + constructor(lo, hi, nbins = 4000) { + if (hi <= lo) hi = lo + 1.0; + this.lo = lo; + this.hi = hi; + this.nbins = nbins; + this.bins = new Int32Array(nbins); + this.width = (hi - lo) / nbins; + this.n = 0; + this.minV = Infinity; + this.maxV = -Infinity; + } + + add(v) { + if (v !== v) return; // NaN + this.n += 1; + if (v < this.minV) this.minV = v; + if (v > this.maxV) this.maxV = v; + let idx = Math.floor((v - this.lo) / this.width); + if (idx < 0) idx = 0; + else if (idx >= this.nbins) idx = this.nbins - 1; + this.bins[idx] += 1; + } + + quantile(q) { + if (this.n === 0) return NaN; + const target = q * this.n; + let cum = 0; + for (let i = 0; i < this.nbins; i++) { + cum += this.bins[i]; + if (cum >= target) return this.lo + (i + 0.5) * this.width; + } + return this.hi; + } +} + +export class RunningMoments { + constructor() { this.n = 0; this.mean = 0.0; this.m2 = 0.0; } + add(v) { + if (v !== v) return; + this.n += 1; + const delta = v - this.mean; + this.mean += delta / this.n; + this.m2 += delta * (v - this.mean); + } + get variance() { return this.n > 0 ? this.m2 / this.n : 0.0; } + get stdev() { return Math.sqrt(this.variance); } +} + +// Channels reported in whole-session stats, with histogram ranges + units. +// Matches python analysis._STATS_CHANNELS exactly. +export const STATS_CHANNELS = [ + ['V_LN_a_avg_V', 0.0, 400.0, 'V'], + ['V_LN_b_avg_V', 0.0, 400.0, 'V'], + ['V_LN_c_avg_V', 0.0, 400.0, 'V'], + ['I_a_avg_A', 0.0, 1000.0, 'A'], + ['I_b_avg_A', 0.0, 1000.0, 'A'], + ['I_c_avg_A', 0.0, 1000.0, 'A'], + ['freq_avg_Hz', 55.0, 65.0, 'Hz'], + ['P_total_avg_W', -2000000.0, 2000000.0, 'W'], + ['S_total_avg_VA', -2000000.0, 2000000.0, 'VA'], + ['Q_total_avg_VAR', -2000000.0, 2000000.0, 'VAR'], + ['PF_total_avg', -1.05, 1.05, ''], +]; + +/** + * Per-channel streaming statistics + threshold time accounting. + * @param {import('./column_store.js').ColumnStore|Array} source + * @param {object} spec + * @param {{undervoltageV?:number, overcurrentA?:number}} [opts] + * @returns {object} keyed by channel name + "_thresholds" + */ +export function wholeSessionStats(source, spec, opts = {}) { + const undervoltageV = opts.undervoltageV ?? 250.0; + const overcurrentA = opts.overcurrentA ?? 800.0; + const src = asColumnSource(source, spec); + const nrec = src.length; + const out = {}; + const moments = {}; + const sketches = {}; + const cols = {}; + for (const [name, lo, hi] of STATS_CHANNELS) { + moments[name] = new RunningMoments(); + sketches[name] = new PercentileSketch(lo, hi); + cols[name] = src.column(name); + } + const va = src.column('V_LN_a_avg_V'); + const vb = src.column('V_LN_b_avg_V'); + const vc = src.column('V_LN_c_avg_V'); + const ia = src.column('I_a_avg_A'); + const ib = src.column('I_b_avg_A'); + const ic = src.column('I_c_avg_A'); + + let secUnder = 0; + let secOver = 0; + for (let i = 0; i < nrec; i++) { + for (const [name] of STATS_CHANNELS) { + const v = cols[name][i]; + moments[name].add(v); + sketches[name].add(v); + } + const notOutage = va[i] > 50.0 && vb[i] > 50.0 && vc[i] > 50.0; + if (notOutage && (va[i] < undervoltageV || vb[i] < undervoltageV || vc[i] < undervoltageV)) { + secUnder += 1; + } + if (ia[i] > overcurrentA || ib[i] > overcurrentA || ic[i] > overcurrentA) { + secOver += 1; + } + } + + for (const [name, , , unit] of STATS_CHANNELS) { + const m = moments[name]; + const sk = sketches[name]; + if (m.n === 0) continue; + out[name] = { + unit, + count: m.n, + min: sk.minV, + p1: sk.quantile(0.01), + p5: sk.quantile(0.05), + median: sk.quantile(0.50), + mean: m.mean, + p95: sk.quantile(0.95), + p99: sk.quantile(0.99), + max: sk.maxV, + stdev: m.stdev, + }; + } + out._thresholds = { + undervoltage_v: undervoltageV, + sec_undervoltage: secUnder, + pct_undervoltage: nrec ? (secUnder / nrec) * 100 : 0.0, + overcurrent_a: overcurrentA, + sec_overcurrent: secOver, + pct_overcurrent: nrec ? (secOver / nrec) * 100 : 0.0, + total_records: nrec, + }; + return out; +} + +// --- ITIC / CBEMA classification ------------------------------------------- + +const ITIC_LOWER = [ + [0.001, 0.0], + [0.003, 0.0], + [0.020, 70.0], + [0.500, 70.0], + [10.0, 80.0], + [1e9, 90.0], +]; +const ITIC_UPPER = [ + [0.001, 500.0], + [0.0001, 500.0], + [0.003, 200.0], + [0.5, 120.0], + [10.0, 120.0], + [1e9, 110.0], +]; + +function interpStep(table, duration) { + for (const [dmax, val] of table) { + if (duration <= dmax) return val; + } + return table[table.length - 1][1]; +} + +export function classifyItic(residualPct, durationSecs) { + if (durationSecs < 0) durationSecs = 0.0; + const lower = interpStep(ITIC_LOWER, durationSecs); + const upper = interpStep(ITIC_UPPER, durationSecs); + if (residualPct > upper) return 'prohibited'; + if (residualPct < lower) return 'no_damage'; + return 'no_interruption'; +} + +/** + * ITIC inputs + classification for a dip/outage/swell event. + * @param {{kind:string, tStartMs:number, tEndMs:number, severity:number}} ev + * @param {number} nominalLnV + */ +export function eventItic(ev, nominalLnV) { + const duration = (ev.tEndMs - ev.tStartMs) / 1000; + let residualPct; + if (ev.kind === 'dip') residualPct = ev.severity * 100.0; + else if (ev.kind === 'outage') residualPct = nominalLnV ? (ev.severity / nominalLnV) * 100.0 : 0.0; + else if (ev.kind === 'swell') residualPct = ev.severity * 100.0; + else return {}; + return { + residual_pct: residualPct, + duration_secs: duration, + itic_class: classifyItic(residualPct, duration), + }; +} + +// --- Time-bucket partitioning (--split-by) --------------------------------- + +export function parsePeriod(text) { + const t = String(text).trim().toLowerCase(); + if (t === 'hour' || t === 'hourly') return { kind: 'hour', seconds: 3600 }; + if (t === 'day' || t === 'daily') return { kind: 'day', seconds: 86400 }; + if (t === 'week' || t === 'weekly') return { kind: 'week', seconds: 7 * 86400 }; + const units = { s: 1, m: 60, h: 3600, d: 86400 }; + const unit = t.slice(-1); + const num = t.slice(0, -1); + if (t && units[unit] !== undefined && /^\d+$/.test(num)) { + const n = parseInt(num, 10); + if (n <= 0) throw new Error(`--split-by duration must be positive: ${text}`); + return { kind: 'duration', seconds: n * units[unit] }; + } + throw new Error( + `Unrecognized --split-by period ${text}. Use hour|day|week or a duration like 30m, 6h, 2d.` + ); +} + +// Time-of-day profile binning by minute-of-day (UTC), mirroring python. +export function parseTodWindow(text) { + const [a, b] = String(text).split('-'); + const toMin = (s) => { + const [hh, mm] = String(s).trim().split(':'); + return parseInt(hh, 10) * 60 + (mm ? parseInt(mm, 10) : 0); + }; + let start = toMin(a); + let end = toMin(b); + if (end === 0) end = 1440; + return [start, end]; +} + +/** + * Diurnal avg/min/max envelope per time-of-day bin (UTC clock). + * @returns {Array} rows with bin/n/n_days/p_avg_kW.../v.../i... + */ +export function timeOfDayProfile(source, spec, opts = {}) { + const [startMin, endMin] = opts.window ?? [0, 1440]; + const binMinutes = Math.max(1, opts.binMinutes ?? 1); + const src = asColumnSource(source, spec); + const nbins = Math.floor((1440 + binMinutes - 1) / binMinutes); + const p = src.column('P_total_avg_W'); + const va = src.column('V_LN_a_avg_V'); + const ia = src.column('I_a_avg_A'); + + const agg = new Array(nbins).fill(null); + const daysPerBin = Array.from({ length: nbins }, () => new Set()); + + for (let i = 0; i < src.length; i++) { + const d = new Date(src.startMs(i)); + const mod = d.getUTCHours() * 60 + d.getUTCMinutes(); + if (!(startMin <= mod && mod < endMin)) continue; + let b = Math.floor(mod / binMinutes); + if (b >= nbins) b = nbins - 1; + const pv = p[i] / 1000.0; + if (agg[b] === null) { + agg[b] = { + n: 0, + pSum: 0, pMin: Infinity, pMax: -Infinity, + vSum: 0, vMin: Infinity, vMax: -Infinity, + iSum: 0, iMin: Infinity, iMax: -Infinity, + }; + } + const a = agg[b]; + a.n += 1; + a.pSum += pv; a.pMin = Math.min(a.pMin, pv); a.pMax = Math.max(a.pMax, pv); + a.vSum += va[i]; a.vMin = Math.min(a.vMin, va[i]); a.vMax = Math.max(a.vMax, va[i]); + a.iSum += ia[i]; a.iMin = Math.min(a.iMin, ia[i]); a.iMax = Math.max(a.iMax, ia[i]); + // Day key = UTC date string. + daysPerBin[b].add(d.toISOString().slice(0, 10)); + } + + const rows = []; + for (let b = 0; b < nbins; b++) { + const a = agg[b]; + if (a === null || a.n === 0) continue; + const binStartMin = b * binMinutes; + const hh = String(Math.floor(binStartMin / 60)).padStart(2, '0'); + const mm = String(binStartMin % 60).padStart(2, '0'); + rows.push({ + bin: `${hh}:${mm}`, + n: a.n, + n_days: daysPerBin[b].size, + p_avg_kW: a.pSum / a.n, + p_min_kW: a.pMin, + p_max_kW: a.pMax, + v_avg_V: a.vSum / a.n, + v_min_V: a.vMin, + v_max_V: a.vMax, + i_avg_A: a.iSum / a.n, + i_min_A: a.iMin, + i_max_A: a.iMax, + }); + } + return rows; +} + +// --- Event markers / correlation ------------------------------------------- + +/** + * For each marker, find the nearest detected event + signed offset (s). + * @param {Array<{timeMs:number, label:string}>} markers + * @param {Array<{id:number, kind:string, tStartMs:number}>} events + */ +export function correlateMarkers(markers, events) { + const out = []; + for (const m of markers) { + let nearest = null; + let best = null; + for (const ev of events) { + const off = (m.timeMs - ev.tStartMs) / 1000; + if (best === null || Math.abs(off) < Math.abs(best)) { + best = off; + nearest = ev; + } + } + const entry = { + marker_time: new Date(m.timeMs).toISOString(), + label: m.label, + nearest_event: null, + }; + if (nearest !== null) { + entry.nearest_event = { + id: nearest.id, + kind: nearest.kind, + t_start: new Date(nearest.tStartMs).toISOString(), + offset_secs: best, + }; + } + out.push(entry); + } + return out; +} diff --git a/web/app.js b/web/app.js index 962648f..37b26f4 100644 --- a/web/app.js +++ b/web/app.js @@ -13,6 +13,7 @@ import { downloadPdfReport } from './pdf_export.js'; import { clearCache, getCached, hashBuffer, putCached } from './cache.js'; import { MultiSession } from './multi_session.js'; import { ColumnStore } from './column_store.js'; +import { wholeSessionStats, timeOfDayProfile } from './analysis.js'; import { computeCost, loadTariff, normalizeTariff, parsePeakHoursString, peakHoursToString, saveTariff, @@ -372,6 +373,7 @@ async function onParseDone(msg) { renderEventsTable(); renderSnapshotsList(); renderQuantityGrid(); + renderStatsPanel(spec); renderSessionsBar(); els.insightsSec.hidden = currentFindings.length === 0; els.sessionsSec.hidden = false; @@ -495,9 +497,114 @@ function recordsForExport(spec) { return []; } -// Statistics panel (Feature B) — populated from whole_session_stats; safe no-op -// until the stats UI is wired so the parse flow never throws on older markup. -function renderStatsPanel(_spec) { /* Feature B fills this in */ } +// Statistics panel (Feature B) — whole-session stats table + time-of-day chart, +// computed straight off the resident ColumnStore so the CLI and web agree. +let currentStats = null; // last computed whole_session_stats (for exports) +let currentTodRows = null; // last computed time-of-day profile (for exports) +let currentNarrative = null; // executive-summary narrative (Feature E) + +function renderStatsPanel(spec) { + const sec = document.getElementById('stats-section'); + const wrap = document.getElementById('stats-table-wrap'); + const status = document.getElementById('stats-status'); + if (!sec || !wrap) return; + const src = dataSource(); + if (!src) { sec.hidden = true; return; } + try { + currentStats = wholeSessionStats(src, spec); + currentTodRows = timeOfDayProfile(src, spec, { window: [0, 1440], binMinutes: 1 }); + } catch (e) { + console.warn('stats failed:', e); + sec.hidden = true; + return; + } + sec.hidden = false; + + // Stats table. + const table = document.createElement('table'); + table.className = 'stats-table'; + const thead = document.createElement('thead'); + const hr = document.createElement('tr'); + for (const h of ['Channel', 'Unit', 'Min', 'p1', 'p5', 'Median', 'Mean', 'p95', 'p99', 'Max', 'Stdev']) { + const th = document.createElement('th'); + th.textContent = h; + hr.appendChild(th); + } + thead.appendChild(hr); + table.appendChild(thead); + const tbody = document.createElement('tbody'); + const fmt = (v) => (Number.isFinite(v) ? (Math.abs(v) >= 1000 ? v.toFixed(0) : v.toFixed(2)) : '—'); + for (const [name, d] of Object.entries(currentStats)) { + if (name.startsWith('_')) continue; + const tr = document.createElement('tr'); + const cells = [name, d.unit, fmt(d.min), fmt(d.p1), fmt(d.p5), fmt(d.median), + fmt(d.mean), fmt(d.p95), fmt(d.p99), fmt(d.max), fmt(d.stdev)]; + for (const c of cells) { + const td = document.createElement('td'); + td.textContent = c; + tr.appendChild(td); + } + tbody.appendChild(tr); + } + table.appendChild(tbody); + + const th = currentStats._thresholds || {}; + const note = document.createElement('p'); + note.className = 'stats-thresholds'; + const us = document.createElement('small'); + us.textContent = + `Under-voltage (<${th.undervoltage_v} V, any phase, non-outage): ` + + `${th.sec_undervoltage} s (${(th.pct_undervoltage ?? 0).toFixed(2)}%). ` + + `Over-current (>${th.overcurrent_a} A, any phase): ` + + `${th.sec_overcurrent} s (${(th.pct_overcurrent ?? 0).toFixed(2)}%).`; + note.appendChild(us); + + wrap.replaceChildren(table, note); + if (status) status.firstChild + ? (status.firstChild.textContent = `${Object.keys(currentStats).length - 1} channels over ${(th.total_records || 0).toLocaleString()} records.`) + : (status.textContent = ''); + + renderTodChart(); +} + +function renderTodChart() { + const head = document.getElementById('tod-heading'); + const div = document.getElementById('tod-chart'); + const uPlot = window.uPlot; + if (!div) return; + div.replaceChildren(); + if (!uPlot || !currentTodRows || currentTodRows.length === 0) { + if (head) head.hidden = true; + return; + } + if (head) head.hidden = false; + // x = minute-of-day index; series = P avg (kW), V avg (V), I avg (A). + const xs = currentTodRows.map((_, i) => i); + const pAvg = currentTodRows.map((r) => r.p_avg_kW); + const vAvg = currentTodRows.map((r) => r.v_avg_V); + const iAvg = currentTodRows.map((r) => r.i_avg_A); + const plot = new uPlot({ + width: div.clientWidth || 900, + height: 240, + series: [ + { label: 'bin' }, + { label: 'P avg (kW)', stroke: '#cc0000', width: 1.4 }, + { label: 'V avg (V)', stroke: '#0066cc', width: 1, scale: 'v' }, + { label: 'I avg (A)', stroke: '#009933', width: 1, scale: 'i' }, + ], + scales: { x: { time: false } }, + axes: [ + { stroke: '#666', label: 'time-of-day bin' }, + { stroke: '#666', label: 'P (kW)' }, + { stroke: '#666', scale: 'v', side: 1, label: 'V' }, + ], + legend: { live: true }, + }, [xs, pAvg, vAvg, iAvg], div); + const resizeObs = new ResizeObserver(() => { + plot.setSize({ width: div.clientWidth, height: 240 }); + }); + resizeObs.observe(div); +} // --- Summary rendering ------------------------------------------------------ @@ -573,6 +680,7 @@ function resetUi() { els.progressSec.hidden = true; els.sessionsSec.hidden = true; els.insightsSec.hidden = true; + { const s = document.getElementById('stats-section'); if (s) s.hidden = true; } els.eventsSec.hidden = true; els.snapshotsSec.hidden = true; els.controlsSec.hidden = true; @@ -893,6 +1001,8 @@ async function exportHtmlReport() { title, config: currentConfig, records: scoped, spec, events: scopedEvents, snapshots: currentSnapshots, findings: scopedFindings, + wholeStats: currentRange ? null : currentStats, + narrative: currentRange ? null : currentNarrative, }); } diff --git a/web/html_report.js b/web/html_report.js index 0f4c51a..117b46d 100644 --- a/web/html_report.js +++ b/web/html_report.js @@ -195,7 +195,30 @@ function insightsHtml(findings) { return out.join('\n'); } -export function buildReportHtml({ title, config, records, spec, events, snapshots, findings = [] }) { +// Whole-session statistics table (Feature B) from a wholeSessionStats() dict. +function wholeStatsTableHtml(wholeStats) { + if (!wholeStats) return ''; + const cols = ['min', 'p1', 'p5', 'median', 'mean', 'p95', 'p99', 'max', 'stdev']; + const head = ['Channel', 'Unit', ...cols].map((h) => `${esc(h)}`).join(''); + const fmt = (v) => (Number.isFinite(v) + ? (Math.abs(v) >= 1000 ? v.toFixed(0) : v.toFixed(2)) : '—'); + const rows = []; + for (const [name, d] of Object.entries(wholeStats)) { + if (name.startsWith('_')) continue; + const cells = [esc(name), esc(d.unit), ...cols.map((c) => fmt(d[c]))] + .map((c) => `${c}`).join(''); + rows.push(`${cells}`); + } + const th = wholeStats._thresholds || {}; + const note = + `

Under-voltage (<${th.undervoltage_v} V): ${th.sec_undervoltage} s ` + + `(${(th.pct_undervoltage ?? 0).toFixed(2)}%). Over-current (>${th.overcurrent_a} A): ` + + `${th.sec_overcurrent} s (${(th.pct_overcurrent ?? 0).toFixed(2)}%).

`; + return `

Statistics

${head}` + + `${rows.join('')}
${note}`; +} + +export function buildReportHtml({ title, config, records, spec, events, snapshots, findings = [], wholeStats = null, narrative = null }) { const energy = summarizeRecords(records, spec); const stats = { 'Records (per-second)': records.length.toLocaleString(), @@ -207,10 +230,15 @@ export function buildReportHtml({ title, config, records, spec, events, snapshot 'Peak export (kW)': (energy.pNeg / 1000).toFixed(2), }; const charts = collectChartArtifacts(); + const narrativeHtml = narrative + ? `

Executive summary

${esc(narrative).replace(/\n/g, '
')}

` + : ''; const body = [ `

${esc(title)}

`, + narrativeHtml, '

Summary

', summaryDlHtml(stats, config), + wholeStatsTableHtml(wholeStats), insightsHtml(findings), '

Events

', eventsTableHtml(events), diff --git a/web/index.html b/web/index.html index c36ee7e..f3c04f1 100644 --- a/web/index.html +++ b/web/index.html @@ -121,6 +121,14 @@

Insights

+ +