tests/unit/streaming.test.js

import { describe, it, expect } from "vitest";
import { StreamingOnsetDetector, highpass } from "../../lib/detector.js";
import { textToSchedule, decodeOnsets } from "../../lib/wire.js";
import { makeTiming } from "../../lib/timing.js";

const SR = 48000;
const T = makeTiming(150);

// Same synthetic click renderer as detector.test.js, plus optional "rebound"
// echoes after each click to mimic the real antenna ringing we observed on
// hardware — the peakRatio + refractory gates should reject them.
function renderClicks(onsetsMs, { leadMs = 300, tailMs = 600, rebound = 0 } = {}) {
    const all = [];
    for (const t0 of onsetsMs) {
        all.push(t0);
        // quiet ringing echoes: one inside the refractory window, one later but
        // well below the peakRatio gate — the two gates should reject both.
        if (rebound) { all.push(t0 + 130); all.push(t0 + 250); }
    }
    const durMs = leadMs + (onsetsMs.length ? onsetsMs[onsetsMs.length - 1] : 0) + tailMs;
    const n = Math.ceil((durMs / 1000) * SR);
    const buf = new Float32Array(n);
    const clickLen = Math.round(0.01 * SR);
    const tau = 0.0015;
    for (const t0 of all) {
        const start = Math.round(((t0 + leadMs) / 1000) * SR);
        // rebounds are clearly quieter than the real strike (~15%)
        const amp = onsetsMs.includes(t0) ? 0.6 : 0.09;
        for (let j = 0; j < clickLen && start + j < n; j++) {
            const tt = j / SR;
            buf[start + j] += amp * Math.exp(-tt / tau) * Math.sin(2 * Math.PI * 4000 * tt);
        }
    }
    return buf;
}

function streamDecode(buf, opts = {}) {
    const filtered = highpass(buf, SR, 2000);
    const onsets = [];
    const det = new StreamingOnsetDetector({ sampleRate: SR, onOnset: (t) => onsets.push(t), ...opts });
    const BLOCK = 2048;
    for (let i = 0; i < filtered.length; i += BLOCK) {
        det.process(filtered.subarray(i, Math.min(i + BLOCK, filtered.length)), (i / SR) * 1000);
    }
    return { onsets, ...decodeOnsets(onsets, T) };
}

describe("StreamingOnsetDetector (live path)", () => {
    for (const phrase of ["SOS", "HELLO WORLD", "CQ"]) {
        it(`decodes "${phrase}" from clean clicks`, () => {
            const { onsets } = textToSchedule(phrase, T);
            const r = streamDecode(renderClicks(onsets));
            expect(r.onsets.length).toBe(onsets.length);
            expect(r.text).toBe(phrase);
        });

        it(`decodes "${phrase}" despite per-click ringing echoes`, () => {
            const { onsets } = textToSchedule(phrase, T);
            const r = streamDecode(renderClicks(onsets, { rebound: 1 }));
            expect(r.text).toBe(phrase);
        });
    }
});