From 48fa6b4f9e3e963863a26962b1562de791ae69e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 27 Jul 2026 17:00:45 +0200 Subject: [PATCH] Diarization v2: Fenster-Clustering mit gelernter Schwelle + Live-Namensmarker #74 (22 Teilnehmer) ergab automatisch 44 Sprecher, dieselbe Stimme mehrfach. Ursache gemessen: ein Zentroid pro sherpa-Cluster aus max. 3 Schnitten, fester Kosinus auf rohen Embeddings (Kreuz-Turn-Kosinus ~0,4 = keine Trennschaerfe). v2: sherpa liefert nur Turn-Grenzen, Identitaet aus 3-s-Fenster-Embeddings, Mittelwert-Zentrierung, Average-Linkage-AHC, Weak-Fold, Schwelle pro Aufnahme per 2-Gauss-EM gelernt. Validiert: #74 44->20, #70 ->4, #68 ->2. Live-Namensmarker: Namen waehrend der Aufnahme tippen; Zeitstempel benennt spaeter den Cluster und dient als must-/cannot-link-Zwangsbedingung. --- app/db.js | 67 ++++- app/diar.js | 474 ++++++++++++++++++++++++++++++++++- app/lab/cluster-lab.mjs | 156 ++++++++++++ app/lab/lab-dump-segs.sh | 23 ++ app/lab/lab-embed.mjs | 117 +++++++++ app/lab/lab-eval.mjs | 23 ++ app/lab/lab-run-embed.sh | 14 ++ app/lab/lab-run2.sh | 19 ++ app/lab/lab-thr.mjs | 52 ++++ app/recorder.js | 35 ++- app/server.js | 122 ++++++++- app/spk.js | 16 +- app/web/src/main.js | 111 +++++++- app/web/src/style.css | 23 ++ docs/03-aufnahme-pipeline.md | 7 +- docs/04-diarization.md | 123 ++++++--- docs/08-changelog.md | 42 ++++ 17 files changed, 1357 insertions(+), 67 deletions(-) create mode 100644 app/lab/cluster-lab.mjs create mode 100644 app/lab/lab-dump-segs.sh create mode 100644 app/lab/lab-embed.mjs create mode 100644 app/lab/lab-eval.mjs create mode 100644 app/lab/lab-run-embed.sh create mode 100644 app/lab/lab-run2.sh create mode 100644 app/lab/lab-thr.mjs diff --git a/app/db.js b/app/db.js index 08a892c..de9ba5b 100644 --- a/app/db.js +++ b/app/db.js @@ -27,6 +27,11 @@ export async function ensureSchema() { created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)`); // n = number of samples averaged into this voiceprint (running centroid). await pool.query(`ALTER TABLE voiceprints ADD COLUMN IF NOT EXISTS n INT NOT NULL DEFAULT 1`); + // Embedding model the vector came from: prints of different models are NOT + // comparable (cosine across models is noise), so identification only ever + // uses the rows of the model currently loaded. NULL = pre-2026-07-27 TitaNet. + await pool.query(`ALTER TABLE voiceprints ADD COLUMN IF NOT EXISTS model VARCHAR(64) NULL`); + await pool.query(`UPDATE voiceprints SET model='nemo_titanet_small' WHERE model IS NULL`); // Domain glossary: known proper nouns / terms the ASR mis-hears (e.g. "Aime" // heard as "Amy"). Fed into the LLM correction step so soundalikes get mapped // to the correct spelling. `soundalikes` = optional comma-list of mis-hearings. @@ -36,6 +41,41 @@ export async function ensureSchema() { soundalikes TEXT NULL, enabled TINYINT NOT NULL DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)`); + // Live speaker markers: names typed DURING the recording ("who is talking + // right now"). t_ms = offset from recording start. After diarization each + // marker is resolved to the acoustic cluster that was speaking around t_ms + // (`speaker`), which names that cluster — ground truth instead of guessing. + await pool.query(`CREATE TABLE IF NOT EXISTS markers ( + id INT AUTO_INCREMENT PRIMARY KEY, + recording_id INT NOT NULL, + t_ms INT NOT NULL, + name VARCHAR(128) NOT NULL, + speaker VARCHAR(32) NULL, + conf VARCHAR(8) NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + INDEX rec_idx (recording_id))`); +} + +// --- live speaker markers --------------------------------------------------- +export async function addMarkers(recId, markers) { + const list = (markers || []).filter(m => m && m.name); + if (!list.length) return; + await pool.query('INSERT INTO markers (recording_id, t_ms, name) VALUES ?', + [list.map(m => [recId, Math.max(0, Math.round(m.t_ms || 0)), String(m.name).slice(0, 128)])]); +} +export async function listMarkers(recId) { + const [rows] = await pool.query( + 'SELECT id, t_ms, name, speaker, conf FROM markers WHERE recording_id=? ORDER BY t_ms', [recId]); + return rows; +} +export async function setMarkerSpeaker(id, speaker, conf) { + await pool.execute('UPDATE markers SET speaker=?, conf=? WHERE id=?', [speaker || null, conf || null, id]); +} +export async function deleteMarker(id) { await pool.execute('DELETE FROM markers WHERE id=?', [id]); } +export async function addMarker(recId, tMs, name) { + const [res] = await pool.execute('INSERT INTO markers (recording_id, t_ms, name) VALUES (?,?,?)', + [recId, Math.max(0, Math.round(tMs || 0)), String(name).slice(0, 128)]); + return res.insertId; } export async function listGlossary() { @@ -58,9 +98,10 @@ export async function updateGlossary(id, term, soundalikes) { return rows[0] ? { ...rows[0], enabled: !!rows[0].enabled } : null; } -export async function addVoiceprint(name, vec) { +export async function addVoiceprint(name, vec, model) { const [res] = await pool.execute( - 'INSERT INTO voiceprints (name, dim, vec) VALUES (?,?,?)', [name, vec.length, JSON.stringify(vec)]); + 'INSERT INTO voiceprints (name, dim, vec, model) VALUES (?,?,?,?)', + [name, vec.length, JSON.stringify(vec), model || null]); return res.insertId; } @@ -68,11 +109,13 @@ export async function addVoiceprint(name, vec) { // centroid (mean of all samples). This is how the app "learns who is who": every // time a speaker is named in a recording, their voice is folded into the centroid, // so future recordings match them more reliably. New name -> fresh row. -export async function upsertVoiceprint(name, vec) { - const [rows] = await pool.query('SELECT id, vec, n FROM voiceprints WHERE name=? ORDER BY id LIMIT 1', [name]); +export async function upsertVoiceprint(name, vec, model) { + const [rows] = await pool.query( + 'SELECT id, vec, n FROM voiceprints WHERE name=? AND (model=? OR (model IS NULL AND ? IS NULL)) ORDER BY id LIMIT 1', + [name, model || null, model || null]); if (!rows.length) { - const [res] = await pool.execute('INSERT INTO voiceprints (name, dim, vec, n) VALUES (?,?,?,1)', - [name, vec.length, JSON.stringify(vec)]); + const [res] = await pool.execute('INSERT INTO voiceprints (name, dim, vec, n, model) VALUES (?,?,?,1,?)', + [name, vec.length, JSON.stringify(vec), model || null]); return res.insertId; } const old = rows[0]; @@ -83,8 +126,12 @@ export async function upsertVoiceprint(name, vec) { [JSON.stringify(mean), on + 1, mean.length, old.id]); return old.id; } -export async function listVoiceprints() { - const [rows] = await pool.query('SELECT id,name,dim,vec,created_at FROM voiceprints ORDER BY name'); +// model given → only the prints of that embedding model (what identification +// must use); omitted → all rows (GUI listing). +export async function listVoiceprints(model) { + const [rows] = model + ? await pool.query('SELECT id,name,dim,vec,model,created_at FROM voiceprints WHERE model=? ORDER BY name', [model]) + : await pool.query('SELECT id,name,dim,vec,model,created_at FROM voiceprints ORDER BY name'); return rows.map(r => ({ ...r, vec: JSON.parse(r.vec) })); } export async function deleteVoiceprint(id) { await pool.query('DELETE FROM voiceprints WHERE id=?', [id]); } @@ -185,9 +232,11 @@ export async function getRecording(id) { 'SELECT word,t_start_ms,t_end_ms,idx,speaker FROM words WHERE recording_id=? ORDER BY idx', [id]); let speaker_names = {}; try { speaker_names = JSON.parse(rec.speaker_names || '{}') || {}; } catch {} - return { ...parseTodos(rec), speaker_names, words }; + const markers = await listMarkers(id).catch(() => []); + return { ...parseTodos(rec), speaker_names, words, markers }; } export async function deleteRecording(id) { + await pool.query('DELETE FROM markers WHERE recording_id=?', [id]).catch(() => {}); await pool.query('DELETE FROM recordings WHERE id=?', [id]); } diff --git a/app/diar.js b/app/diar.js index 5b603c9..bb08b10 100644 --- a/app/diar.js +++ b/app/diar.js @@ -53,12 +53,57 @@ const THRESHOLD = process.env.DIAR_THRESHOLD || '0.7'; function env() { return { ...process.env, LD_LIBRARY_PATH: `${DIR}/lib` }; } -// Returns [{ start, end, speaker }] (seconds, speaker = "speaker_00"…). -// numSpeakers > 0 fixes the count (--clustering.num-clusters) when the user knows -// it; otherwise auto via the cluster threshold. -export async function diarize(wavPath, numSpeakers) { - const n = parseInt(numSpeakers, 10); - const clusterArg = n > 0 ? `--clustering.num-clusters=${n}` : `--clustering.cluster-threshold=${THRESHOLD}`; +// --- v2 pipeline (default) -------------------------------------------------- +// The old path let sherpa cluster its own window embeddings and then tried to +// repair the result by merging whole clusters. On a 23-person weekly (#74) that +// produced 44 speakers with the same voice spread over several of them: one +// centroid per (already wrong) cluster from at most 3 cuts is too coarse to +// separate 20+ voices, and a single fixed cosine threshold on RAW embeddings +// can't work at all — untouched TitaNet vectors of one channel share a large +// common component (mean cross-turn cosine ≈ 0.4). +// +// v2 therefore ignores sherpa's labels and only keeps its (pyannote) turn +// BOUNDARIES, then does the identity part itself: +// 1. slice every turn into 3 s windows (1.5 s hop), +// 2. embed each window, +// 3. subtract the per-recording mean and L2-normalise (channel/room +// compensation — this alone moves cross-turn cosine to ≈ 0 and lifts the +// same/different separation to d' ≈ 3.5), +// 4. average-linkage agglomerative clustering (all pairs decide a merge, not +// one greedy first-match link as before) at a FINE threshold, +// 5. fold clusters with almost no talk time into the nearest real speaker, +// 6. keep merging the remaining (strong) clusters up to the FINAL threshold, +// or until a user-pinned speaker count is reached, +// 7. duration-weighted majority label per turn. +// Live name markers (see resolveMarkers) additionally constrain step 4/6: +// windows around markers with the SAME name must land in one cluster, windows +// around markers with DIFFERENT names must never be merged. +// DIAR_MODE=legacy restores the old behaviour. +const MODE = process.env.DIAR_MODE || 'v2'; +const WIN_SEC = +(process.env.DIAR_WIN_SEC || 3.0); +const HOP_SEC = +(process.env.DIAR_HOP_SEC || 1.5); +const MIN_WIN_SEC = +(process.env.DIAR_MIN_WIN_SEC || 0.6); +const MAX_WINDOWS = +(process.env.DIAR_MAX_WINDOWS || 4000); +// Merge thresholds on MEAN-CENTRED cosine (see cluster-lab measurements). +// Stage A over-clusters at a fixed high threshold (0.45 measured stable on #74). +const COS_FINE = +(process.env.DIAR_COS_FINE || 0.45); +// Stage B's threshold is derived PER RECORDING (see mergeThreshold): a fixed one +// cannot serve both a clean 23-person conference and a noisy 4-person call — +// 0.30 gave 21 speakers on #74 (~right) but 9 on #70 (truly 4), while 0.15 fixed +// #70 and collapsed #74 to 14. DIAR_COS_FINAL pins it for debugging. +const COS_FINAL_ENV = process.env.DIAR_COS_FINAL ? +process.env.DIAR_COS_FINAL : null; +const COS_FINAL_MIN = +(process.env.DIAR_COS_FINAL_MIN || 0.05); +const COS_FINAL_MAX = +(process.env.DIAR_COS_FINAL_MAX || 0.35); +// sherpa cluster threshold for the v2 pass: its labels are discarded, we only +// want fine turn boundaries (and min-duration-off=0 so adjacent turns of the +// same label are NOT glued together). +const FINE_THRESHOLD = process.env.DIAR_FINE_THRESHOLD || '0.2'; + +// Raw sherpa turns. mode 'fine' = boundaries only (v2), else the legacy call +// with its own clustering (threshold or pinned count). +async function sherpaTurns(wavPath, numSpeakers, fine) { + const clusterArg = fine ? `--clustering.cluster-threshold=${FINE_THRESHOLD}` + : (numSpeakers > 0 ? `--clustering.num-clusters=${numSpeakers}` : `--clustering.cluster-threshold=${THRESHOLD}`); const args = [ 'taskset', '-c', '4-7', BIN, `--segmentation.pyannote-model=${SEG}`, @@ -66,6 +111,7 @@ export async function diarize(wavPath, numSpeakers) { `--embedding.model=${EMB}`, `--embedding.num-threads=4`, clusterArg, + ...(fine ? ['--min-duration-on=0.2', '--min-duration-off=0.0'] : []), wavPath, ]; const { stdout } = await pexec(args[0], args.slice(1), { env: env(), maxBuffer: 1 << 22, timeout: diarTimeoutMs(wavPath) }); @@ -74,14 +120,384 @@ export async function diarize(wavPath, numSpeakers) { const m = line.match(/([\d.]+)\s*--\s*([\d.]+)\s*(speaker_\d+)/); if (m) segs.push({ start: +m[1], end: +m[2], speaker: m[3] }); } + segs.sort((a, b) => a.start - b.start); + return segs; +} + +// Returns [{ start, end, speaker }] (seconds, speaker = "speaker_00"…). +// numSpeakers > 0 fixes the count when the user knows it, otherwise the count +// is found automatically (people can join/leave a conference call). +export async function diarize(wavPath, numSpeakers, opts = {}) { + const n = parseInt(numSpeakers, 10) > 0 ? parseInt(numSpeakers, 10) : 0; + if (MODE === 'v2') { + const turns = await sherpaTurns(wavPath, n, true); + if (turns.length) { + try { return await reclusterTurns(wavPath, turns, n, opts.markers || []); } + catch (e) { console.log('diar: v2 re-clustering failed, falling back to legacy:', e.message); } + } + } + const segs = await sherpaTurns(wavPath, n, false); // Only auto mode over-segments; a user-pinned count is already trusted. - if (!(n > 0) && segs.length) { + if (!n && segs.length) { try { return await consolidateSpeakers(wavPath, segs); } catch (e) { console.log('diar: consolidation failed, using raw clusters:', e.message); } } return segs; } +// --- wav slicing (16 kHz mono s16, no ffmpeg per window) -------------------- +function loadWav(p) { + const b = fs.readFileSync(p); + if (b.toString('ascii', 0, 4) !== 'RIFF' || b.toString('ascii', 8, 12) !== 'WAVE') throw new Error('kein RIFF/WAVE: ' + p); + let off = 12, fmt = null, data = null; + while (off + 8 <= b.length) { + const id = b.toString('ascii', off, off + 4), size = b.readUInt32LE(off + 4); + if (id === 'fmt ') fmt = { ch: b.readUInt16LE(off + 10), rate: b.readUInt32LE(off + 12), bits: b.readUInt16LE(off + 22) }; + if (id === 'data') { data = { start: off + 8, size: Math.min(size, b.length - off - 8) }; break; } + off += 8 + size + (size & 1); + } + if (!fmt || !data) throw new Error('wav-Chunks fehlen'); + if (fmt.bits !== 16) throw new Error('wav ist nicht s16 (' + fmt.bits + ')'); + return { buf: b, fmt, data }; +} +function writeSlice(w, startSec, endSec, outPath) { + const bpf = 2 * w.fmt.ch; + const s = Math.max(0, Math.floor(startSec * w.fmt.rate) * bpf); + const e = Math.min(w.data.size, Math.ceil(endSec * w.fmt.rate) * bpf); + const n = Math.max(0, e - s); + const h = Buffer.alloc(44); + h.write('RIFF', 0); h.writeUInt32LE(36 + n, 4); h.write('WAVE', 8); + h.write('fmt ', 12); h.writeUInt32LE(16, 16); h.writeUInt16LE(1, 20); + h.writeUInt16LE(w.fmt.ch, 22); h.writeUInt32LE(w.fmt.rate, 24); + h.writeUInt32LE(w.fmt.rate * bpf, 28); h.writeUInt16LE(bpf, 32); h.writeUInt16LE(16, 34); + h.write('data', 36); h.writeUInt32LE(n, 40); + fs.writeFileSync(outPath, Buffer.concat([h, w.buf.subarray(w.data.start + s, w.data.start + e)])); + return n / (w.fmt.rate * bpf); +} + +// --- vector helpers --------------------------------------------------------- +function l2(v) { let s = 0; for (const x of v) s += x * x; s = Math.sqrt(s) + 1e-9; return v.map(x => x / s); } +function meanVec(vs) { + const d = vs[0].length, m = new Array(d).fill(0); + for (const v of vs) for (let i = 0; i < d; i++) m[i] += v[i] / vs.length; + return m; +} +function dot(a, b) { let s = 0; for (let i = 0; i < a.length; i++) s += a[i] * b[i]; return s; } + +// Per-recording merge threshold. The pairwise window-similarity histogram of one +// recording is bimodal: a big "different speaker" mode around 0 and a "same +// speaker" mode further right. How far right depends entirely on the audio — on +// a clean conference recording the same-speaker mode sits at ~0.30, on a noisy +// BT call at ~0.21 with much more spread. A 2-Gaussian mixture (EM) is fitted to +// a deterministic sample of the pairs and the point where the two weighted +// densities cross becomes the merge threshold: measured 0.20 for #74 (→ 20 +// speakers for 23 participants) and 0.07 for #70 (→ exactly the 4 known ones), +// where no single fixed value could do both. +function mergeThreshold(X) { + const n = X.length; + const vals = []; + for (let k = 1; k < 400000 && vals.length < 60000; k++) { + const i = (k * 7919) % n, j = (k * 104729 + 13) % n; + if (i !== j) vals.push(dot(X[i], X[j])); + } + if (vals.length < 50) return (COS_FINAL_MIN + COS_FINAL_MAX) / 2; + const N = (x, m, s) => Math.exp(-((x - m) ** 2) / (2 * s * s)) / (s * Math.sqrt(2 * Math.PI)); + let m1 = -0.1, m2 = 0.5, s1 = 0.15, s2 = 0.15, w1 = 0.8; + for (let it = 0; it < 100; it++) { + let sw1 = 0, sw2 = 0, sm1 = 0, sm2 = 0, sv1 = 0, sv2 = 0; + for (const x of vals) { + const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2); + const r = a / (a + b + 1e-300); + sw1 += r; sw2 += 1 - r; sm1 += r * x; sm2 += (1 - r) * x; + } + const nm1 = sm1 / sw1, nm2 = sm2 / sw2; + for (const x of vals) { + const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2); + const r = a / (a + b + 1e-300); + sv1 += r * (x - nm1) ** 2; sv2 += (1 - r) * (x - nm2) ** 2; + } + m1 = nm1; m2 = nm2; + s1 = Math.max(0.02, Math.sqrt(sv1 / sw1)); + s2 = Math.max(0.02, Math.sqrt(sv2 / sw2)); + w1 = sw1 / vals.length; + } + if (!(m2 > m1)) return (COS_FINAL_MIN + COS_FINAL_MAX) / 2; + let lo = m1, hi = m2, t = (lo + hi) / 2; + for (let i = 0; i < 60; i++) { + t = (lo + hi) / 2; + if (w1 * N(t, m1, s1) - (1 - w1) * N(t, m2, s2) > 0) lo = t; else hi = t; + } + return Math.min(COS_FINAL_MAX, Math.max(COS_FINAL_MIN, t)); +} + +// Average-linkage agglomerative clustering over unit vectors X. +// Stops when the best average-linkage cosine drops below `threshold`, or when +// `kStop` clusters remain (user-pinned speaker count). Best-match caching keeps +// this ~O(n^2) for the few-thousand windows of a long recording. +// initGroups: continue from an existing clustering (arrays of X indices) +// names: per-init-group name tag; two clusters that both carry a (then +// necessarily different) name are never merged — the live markers' +// cannot-link constraint. +// Returns the surviving groups (arrays of X indices) and their name tags. +function ahcAverage(X, threshold, kStop = 0, initGroups = null, names = null) { + const groups = (initGroups || X.map((_, i) => [i])).map(g => [...g]); + const tag = groups.map((_, i) => (names && names[i]) || null); + const m = groups.length; + const A = new Float32Array(m * m); + for (let i = 0; i < m; i++) { + A[i * m + i] = 1; + for (let j = i + 1; j < m; j++) { + let s = 0; + for (const a of groups[i]) for (const b of groups[j]) s += dot(X[a], X[b]); + s /= groups[i].length * groups[j].length; + A[i * m + j] = s; A[j * m + i] = s; + } + } + const cnt = new Int32Array(m); groups.forEach((g, i) => cnt[i] = g.length); + const alive = new Uint8Array(m).fill(1); + const bestJ = new Int32Array(m).fill(-1); + const bestS = new Float32Array(m).fill(-Infinity); + const blocked = (i, j) => tag[i] && tag[j] && tag[i] !== tag[j]; + const rescan = i => { + let bj = -1, bs = -Infinity; + for (let j = 0; j < m; j++) { + if (j === i || !alive[j] || blocked(i, j)) continue; + const s = A[i * m + j]; if (s > bs) { bs = s; bj = j; } + } + bestJ[i] = bj; bestS[i] = bs; + }; + for (let i = 0; i < m; i++) rescan(i); + let k = m; + while (k > 1) { + let a = -1, bs = -Infinity; + for (let i = 0; i < m; i++) if (alive[i] && bestJ[i] >= 0 && bestS[i] > bs) { bs = bestS[i]; a = i; } + if (a < 0) break; + const b = bestJ[a]; + if (kStop > 0 ? k <= kStop : bs < threshold) break; + const ca = cnt[a], cb = cnt[b]; + for (let c = 0; c < m; c++) { + if (!alive[c] || c === a || c === b) continue; + const v = (A[a * m + c] * ca + A[b * m + c] * cb) / (ca + cb); + A[a * m + c] = v; A[c * m + a] = v; + } + cnt[a] = ca + cb; alive[b] = 0; k--; + groups[a] = groups[a].concat(groups[b]); groups[b] = []; + const gained = !tag[a] && !!tag[b]; + if (gained) tag[a] = tag[b]; + rescan(a); + // A cluster that just gained a name blocks pairs that were allowed before, + // so every cached best-match has to be re-checked; otherwise only those + // pointing at the two merged clusters. + for (let i = 0; i < m; i++) + if (alive[i] && (gained || bestJ[i] === b || bestJ[i] === a)) rescan(i); + } + const out = [], outTags = []; + for (let i = 0; i < m; i++) if (alive[i]) { out.push(groups[i]); outTags.push(tag[i]); } + return { groups: out, tags: outTags }; +} + +// v2 entry: embed the windows, then cluster them. +async function reclusterTurns(wavPath, turns, numSpeakers = 0, markers = []) { + const ok = await embedWindows(wavPath, turns); + return clusterWindows(turns, ok, numSpeakers, markers); +} + +// Map each live marker to the turn the named person was speaking in: the turn +// covering the marker, else the last one ending shortly before it (the name is +// typed a few seconds AFTER the voice was heard), else the next one starting +// right after. Returns { name -> [turnIndex…] }. +const ANCHOR_BACK_SEC = +(process.env.MARK_PRE_MS || 25000) / 1000; +const ANCHOR_FWD_SEC = +(process.env.MARK_POST_MS || 8000) / 1000; +const ANCHOR_MIN_SEC = 1.0; // too short to identify a voice from +function markerAnchors(turns, markers) { + const byName = new Map(); + const usable = turns.map((t, i) => ({ i, ...t })).filter(t => t.end - t.start >= ANCHOR_MIN_SEC); + for (const mk of markers || []) { + const t = (mk.t_ms || 0) / 1000; + let pick = -1, bd = Infinity; + for (const c of usable) { + const back = t - c.end; // >0: turn ended before the marker + const fwd = c.start - t; // >0: turn starts after the marker + let d = Infinity; + if (back <= 0 && fwd <= 0) d = 0; // covers it + else if (back > 0 && back <= ANCHOR_BACK_SEC) d = back; // typing lag + else if (fwd > 0 && fwd <= ANCHOR_FWD_SEC) d = fwd * (ANCHOR_BACK_SEC / Math.max(1, ANCHOR_FWD_SEC)); + if (d < bd) { bd = d; pick = c.i; } + } + if (pick < 0) continue; + const key = String(mk.name).trim().toLowerCase(); + (byName.get(key) || byName.set(key, []).get(key)).push(pick); + } + return byName; +} + +// Slice + embed every window of every turn (the only part that touches audio). +export async function embedWindows(wavPath, turns) { + const wav = loadWav(wavPath); + // Adaptive hop: keep the window count (and the O(n^2) clustering) bounded on + // very long recordings. + const winCount = (hop) => turns.reduce((a, t) => { + const d = t.end - t.start; + return a + (d < MIN_WIN_SEC ? 0 : d <= WIN_SEC ? 1 : Math.max(1, Math.ceil((d - MIN_WIN_SEC) / hop))); + }, 0); + let hop = HOP_SEC; + while (winCount(hop) > MAX_WINDOWS && hop < 60) hop *= 1.5; + + const wins = []; + turns.forEach((t, ti) => { + const d = t.end - t.start; + if (d < MIN_WIN_SEC) return; + if (d <= WIN_SEC) { wins.push({ ti, start: t.start, end: t.end }); return; } + for (let s = t.start; s + MIN_WIN_SEC <= t.end; s += hop) wins.push({ ti, start: s, end: Math.min(s + WIN_SEC, t.end) }); + }); + if (!wins.length) throw new Error('keine Fenster für Sprecher-Embeddings'); + + const tmp = `/tmp/diarwin_${process.pid}.wav`; + const t0 = Date.now(); + for (const w of wins) { + writeSlice(wav, w.start, w.end, tmp); + try { w.vec = await spk.embed(tmp); } catch (e) { w.err = e.message; } + } + try { fs.unlinkSync(tmp); } catch {} + const ok = wins.filter(w => Array.isArray(w.vec) && w.vec.length); + if (ok.length < 2) throw new Error('Embedding fehlgeschlagen (' + (wins[0] && wins[0].err || 'unbekannt') + ')'); + console.log(`diar: ${ok.length}/${wins.length} Fenster-Embeddings (hop ${hop.toFixed(1)}s) in ${((Date.now() - t0) / 1000).toFixed(0)}s`); + return ok; +} + +// The heart of v2: window embeddings → clusters → per-turn speaker labels. +// Split from the audio/embedding part so the clustering can be replayed offline +// on cached embeddings (lab/lab-eval.mjs) — the tuning ran against THIS code. +export function clusterWindows(turns, ok, numSpeakers = 0, markers = []) { + // Channel compensation: L2 → subtract the recording mean → L2 again. + let X = ok.map(w => l2(w.vec)); + const mu = meanVec(X); + X = X.map(v => l2(v.map((x, i) => x - mu[i]))); + + // Windows of a turn, and the turn label from its windows (duration-weighted). + const winsOfTurn = new Map(); + ok.forEach((w, i) => { (winsOfTurn.get(w.ti) || winsOfTurn.set(w.ti, []).get(w.ti)).push(i); }); + const labelTurns = (groups) => { + const of = new Int32Array(ok.length).fill(-1); + groups.forEach((g, gi) => g.forEach(i => of[i] = gi)); + const lab = new Array(turns.length).fill(-1); + for (const [ti, idxs] of winsOfTurn) { + const m = new Map(); + for (const i of idxs) m.set(of[i], (m.get(of[i]) || 0) + (ok[i].end - ok[i].start)); + let bl = -1, bd = -1; + for (const [l, d] of m) if (d > bd) { bd = d; bl = l; } + lab[ti] = bl; + } + const talk = new Map(); + turns.forEach((t, ti) => { if (lab[ti] >= 0) talk.set(lab[ti], (talk.get(lab[ti]) || 0) + (t.end - t.start)); }); + return { lab, talk }; + }; + const centroidOf = g => l2(meanVec(g.map(i => X[i]))); + + const cosFinal = COS_FINAL_ENV !== null ? COS_FINAL_ENV : mergeThreshold(X); + + // Stage A: deliberate over-clustering — every real speaker ends up in one or + // more clusters, but (almost) no cluster mixes two speakers. + let { groups } = ahcAverage(X, COS_FINE, 0); + const fineCount = groups.length; + + // Fold clusters with almost no talk time (overlap, crosstalk, door noise) into + // the acoustically nearest real speaker; they must not survive as "speakers" + // and must not be merge partners in stage B. The floor scales with the length + // of the recording — a flat 20 s would swallow a real second speaker on a + // 2-minute call, while 3 s would let noise through on a 33-minute conference. + const speechSec = turns.reduce((a, t) => a + (t.end - t.start), 0); + const minTotal = Math.min(MIN_TOTAL_SEC, Math.max(3, speechSec * 0.02)); + const { talk } = labelTurns(groups); + const strongIdx = groups.map((_, i) => i).filter(i => (talk.get(i) || 0) >= minTotal); + let folded = 0; + if (strongIdx.length && strongIdx.length < groups.length) { + const cents = new Map(strongIdx.map(i => [i, centroidOf(groups[i])])); + const keep = new Map(strongIdx.map(i => [i, [...groups[i]]])); + groups.forEach((g, i) => { + if (keep.has(i)) return; + const c = centroidOf(g); + let best = strongIdx[0], bs = -Infinity; + for (const s of strongIdx) { const v = dot(c, cents.get(s)); if (v > bs) { bs = v; best = s; } } + keep.get(best).push(...g); + folded++; + }); + groups = [...keep.values()]; + } + + // Live markers → must-link (same name in one group) + cannot-link (different + // names never merge). Applied to the folded stage-A clusters: the cluster a + // marker's turn landed in carries that name. + const anchors = markerAnchors(turns, markers); + let tags = null; + if (anchors.size) { + const { lab } = labelTurns(groups); + const nameOfGroup = new Map(); // groupIdx -> name (first/most-voted wins) + const votes = new Map(); + for (const [name, tis] of anchors) for (const ti of tis) { + const g = lab[ti]; + if (g < 0) continue; + const key = g + '\0' + name; + votes.set(key, (votes.get(key) || 0) + 1); + } + const bestPerGroup = new Map(); + for (const [key, v] of votes) { + const [g, name] = key.split('\0'); + const cur = bestPerGroup.get(g); + if (!cur || v > cur.v) bestPerGroup.set(g, { name, v }); + } + for (const [g, { name }] of bestPerGroup) nameOfGroup.set(+g, name); + // must-link: all groups tagged with the same name become ONE group + const byName = new Map(); + for (const [g, name] of nameOfGroup) (byName.get(name) || byName.set(name, []).get(name)).push(g); + const used = new Set(); + const merged = [], mtags = []; + for (const [name, gs] of byName) { + merged.push(gs.flatMap(g => groups[g])); + mtags.push(name); + gs.forEach(g => used.add(g)); + } + groups.forEach((g, i) => { if (!used.has(i)) { merged.push(g); mtags.push(null); } }); + groups = merged; tags = mtags; + console.log(`diar v2: ${anchors.size} Marker-Name(n) als Zwangsbedingung (${[...byName.values()].reduce((a, g) => a + g.length, 0)} Cluster verankert)`); + } + + // Stage B: merge the remaining strong clusters up to the final threshold — or + // down to the user-pinned speaker count. Marker constraints are respected. + const stageB = ahcAverage(X, cosFinal, numSpeakers, groups, tags); + groups = stageB.groups; + + const { lab: turnLab } = labelTurns(groups); + // Turns too short to embed inherit the temporally nearest labelled turn. + for (let i = 0; i < turns.length; i++) { + if (turnLab[i] >= 0) continue; + let best = -1, bd = Infinity; + for (let j = 0; j < turns.length; j++) { + if (turnLab[j] < 0) continue; + const d = Math.min(Math.abs(turns[j].end - turns[i].start), Math.abs(turns[i].end - turns[j].start)); + if (d < bd) { bd = d; best = turnLab[j]; } + } + turnLab[i] = best >= 0 ? best : 0; + } + + // Renumber by first appearance and glue adjacent turns of the same speaker + // (gap <= 0.5 s — what sherpa's min-duration-off would have done). + const rename = new Map(); let nSp = 0; + const out = []; + turns.forEach((t, ti) => { + const l = turnLab[ti]; + if (!rename.has(l)) rename.set(l, `speaker_${String(nSp++).padStart(2, '0')}`); + const sp = rename.get(l); + const last = out[out.length - 1]; + if (last && last.speaker === sp && t.start - last.end <= 0.5) last.end = Math.max(last.end, t.end); + else out.push({ start: t.start, end: t.end, speaker: sp }); + }); + console.log(`diar v2: ${turns.length} Turns, ${ok.length} Fenster → ${fineCount} Cluster (fein ${COS_FINE})` + + ` → ${folded} schwache <${minTotal.toFixed(0)}s eingefaltet → ${nSp} Sprecher` + + (numSpeakers ? ` [fix ${numSpeakers}]` : ` (final ${cosFinal.toFixed(3)}${COS_FINAL_ENV === null ? ' automatisch' : ''})`)); + return out; +} + // Average voiceprint of a cluster: embed its longest few segments and mean them // (short cuts give noisy embeddings, so prefer the longest and average out noise). async function clusterCentroid(wavPath, clusterSegs) { @@ -196,6 +612,50 @@ export function assignSpeakers(words, segs) { return words; } +// --- live name markers ------------------------------------------------------ +// Contiguous runs of one speaker over the word list (ms). +export function speakerRuns(words) { + const runs = []; + for (const w of words || []) { + if (!w.speaker) continue; + const last = runs[runs.length - 1]; + if (last && last.speaker === w.speaker) last.end = w.t_end_ms; + else runs.push({ speaker: w.speaker, start: w.t_start_ms, end: w.t_end_ms }); + } + return runs; +} + +// Resolve names the user typed DURING the recording to diarization clusters. +// A marker is stamped when the name is submitted, i.e. typically a few seconds +// AFTER that person started talking (hear voice → type → Enter), occasionally +// just before they start. So we score every speaker run in an asymmetric window +// around the marker by overlap × proximity, and prefer the run that covers the +// marker itself. Returns [{ ...marker, speaker, conf }] (conf: high | low | null). +const MARK_PRE = +(process.env.MARK_PRE_MS || 25000); // look back (typing lag) +const MARK_POST = +(process.env.MARK_POST_MS || 8000); // look ahead (typed early) +const MARK_DECAY = +(process.env.MARK_DECAY_MS || 10000); +export function resolveMarkers(words, markers) { + const runs = speakerRuns(words); + return (markers || []).map(m => { + const t = m.t_ms || 0, lo = t - MARK_PRE, hi = t + MARK_POST; + const score = {}, covering = {}; + for (const r of runs) { + const ov = Math.min(hi, r.end) - Math.max(lo, r.start); + if (ov <= 0) continue; + const dist = r.start > t ? r.start - t : (r.end < t ? t - r.end : 0); + score[r.speaker] = (score[r.speaker] || 0) + ov * Math.exp(-dist / MARK_DECAY); + if (r.start <= t && r.end >= t) covering[r.speaker] = Math.max(covering[r.speaker] || 0, r.end - r.start); + } + const ranked = Object.entries(score).sort((a, b) => b[1] - a[1]); + if (!ranked.length) return { ...m, speaker: null, conf: null }; + const [sp, s1] = ranked[0]; + const s2 = ranked[1] ? ranked[1][1] : 0; + const conf = (covering[sp] >= 1500 && s1 >= 1.5 * s2) ? 'high' + : (s1 >= 2 * s2 || ranked.length === 1) ? 'mid' : 'low'; + return { ...m, speaker: sp, conf }; + }); +} + // Group consecutive same-speaker words into turns [{ speaker, text }]. export function toTurns(words) { const turns = []; diff --git a/app/lab/cluster-lab.mjs b/app/lab/cluster-lab.mjs new file mode 100644 index 0000000..92a36cd --- /dev/null +++ b/app/lab/cluster-lab.mjs @@ -0,0 +1,156 @@ +// Lab: offline clustering experiments on window embeddings from lab-embed.mjs. +// Pipeline under test (mirrors the planned diar.js v2): +// L2 → subtract recording mean → L2 +// stage A: average-linkage AHC at a FINE threshold (deliberate over-clustering) +// fold: clusters with < minTotal s of turn time → nearest strong centroid +// stage B: continue average-linkage merging of the strong clusters until +// thrFinal (or until k clusters remain) +// Usage: node cluster-lab.mjs [--expect N] [--fine 0.35] [--final a,b,c] +// [--k N] [--minTotal 15] [--center 0|1] [--quiet] +import fs from 'node:fs'; + +const args = process.argv.slice(2); +const file = args[0]; +const opt = (k, d) => { const i = args.indexOf('--' + k); return i >= 0 ? args[i + 1] : d; }; +const has = k => args.includes('--' + k); +const EXPECT = +(opt('expect', 0)); +const CENTER = opt('center', '1') === '1'; +const FINE = +(opt('fine', 0.35)); +const MIN_TOTAL = +(opt('minTotal', 15)); +const KFIX = +(opt('k', 0)); + +const data = JSON.parse(fs.readFileSync(file, 'utf8')); +const turns = data.turns; +const wins = data.wins.filter(w => w.v); +const f = x => (x === undefined || x === null || Number.isNaN(x) ? 'NaN' : x.toFixed(3)); + +const norm = v => { let s = 0; for (const x of v) s += x * x; s = Math.sqrt(s) + 1e-9; return v.map(x => x / s); }; +let X = wins.map(w => norm(w.v)); +if (CENTER) { + const dim = X[0].length, mu = new Array(dim).fill(0); + for (const v of X) for (let i = 0; i < dim; i++) mu[i] += v[i] / X.length; + X = X.map(v => norm(v.map((x, i) => x - mu[i]))); +} +const dot = (a, b) => { let s = 0; for (let i = 0; i < a.length; i++) s += a[i] * b[i]; return s; }; +const n = X.length; +const S = new Float32Array(n * n); +for (let i = 0; i < n; i++) { S[i * n + i] = 1; for (let j = i + 1; j < n; j++) { const s = dot(X[i], X[j]); S[i * n + j] = s; S[j * n + i] = s; } } + +console.log(`# ${data.model.split('/').pop()} dim=${data.dim} windows=${n}/${data.wins.length} turns=${turns.length} center=${CENTER}`); + +// ---- separability diagnostics --------------------------------------------- +const byTurn = new Map(); +wins.forEach((w, i) => { (byTurn.get(w.ti) || byTurn.set(w.ti, []).get(w.ti)).push(i); }); +const pos = []; +for (const idxs of byTurn.values()) + for (let a = 0; a < idxs.length; a++) for (let b = a + 1; b < idxs.length; b++) + if (Math.abs(wins[idxs[a]].s - wins[idxs[b]].s) >= 3) pos.push(S[idxs[a] * n + idxs[b]]); +const rnd = []; +for (let k = 0; k < 40000; k++) { + const i = (k * 7919) % n, j = (k * 104729 + 13) % n; + if (i !== j && wins[i].ti !== wins[j].ti) rnd.push(S[i * n + j]); +} +const stat = a => { + const m = a.reduce((x, y) => x + y, 0) / a.length; + const sd = Math.sqrt(a.reduce((x, y) => x + (y - m) ** 2, 0) / a.length); + const s = [...a].sort((x, y) => x - y); + return { n: a.length, mean: m, sd, p05: s[(s.length * 0.05) | 0], p50: s[(s.length * 0.5) | 0], p95: s[(s.length * 0.95) | 0] }; +}; +const sp = stat(pos), sr = stat(rnd); +console.log(`# same-turn mean=${f(sp.mean)} sd=${f(sp.sd)} p05=${f(sp.p05)} cross-turn mean=${f(sr.mean)} sd=${f(sr.sd)} p95=${f(sr.p95)} d'=${f((sp.mean - sr.mean) / Math.sqrt((sp.sd ** 2 + sr.sd ** 2) / 2))}`); + +// ---- average-linkage AHC, optionally continuing from given groups ----------- +function ahc(threshold, kStop = 0, initGroups = null) { + const groups = initGroups ? initGroups.map(g => [...g]) : X.map((_, i) => [i]); + const m = groups.length; + const A = new Float32Array(m * m); + for (let i = 0; i < m; i++) { + A[i * m + i] = 1; + for (let j = i + 1; j < m; j++) { + let s = 0; + for (const a of groups[i]) for (const b of groups[j]) s += S[a * n + b]; + s /= groups[i].length * groups[j].length; + A[i * m + j] = s; A[j * m + i] = s; + } + } + const cnt = new Int32Array(m); groups.forEach((g, i) => cnt[i] = g.length); + const alive = new Uint8Array(m).fill(1); + const bestJ = new Int32Array(m).fill(-1), bestS = new Float32Array(m).fill(-Infinity); + const rescan = i => { + let bj = -1, bs = -Infinity; + for (let j = 0; j < m; j++) { if (j === i || !alive[j]) continue; const s = A[i * m + j]; if (s > bs) { bs = s; bj = j; } } + bestJ[i] = bj; bestS[i] = bs; + }; + for (let i = 0; i < m; i++) rescan(i); + let k = m; + while (k > 1) { + let a = -1, bs = -Infinity; + for (let i = 0; i < m; i++) if (alive[i] && bestS[i] > bs) { bs = bestS[i]; a = i; } + if (a < 0) break; + const b = bestJ[a]; if (b < 0) break; + if (kStop > 0 ? k <= kStop : bs < threshold) break; + const ca = cnt[a], cb = cnt[b]; + for (let c = 0; c < m; c++) { + if (!alive[c] || c === a || c === b) continue; + const v = (A[a * m + c] * ca + A[b * m + c] * cb) / (ca + cb); + A[a * m + c] = v; A[c * m + a] = v; + } + cnt[a] = ca + cb; alive[b] = 0; k--; + groups[a] = groups[a].concat(groups[b]); groups[b] = []; + rescan(a); + for (let i = 0; i < m; i++) if (alive[i] && (bestJ[i] === b || bestJ[i] === a)) rescan(i); + } + return groups.filter(g => g.length); +} + +// ---- turn labels + per-cluster talk time ----------------------------------- +function turnLabels(groups) { + const of = new Int32Array(n).fill(-1); + groups.forEach((g, gi) => g.forEach(i => of[i] = gi)); + const acc = new Map(); + wins.forEach((w, i) => { + const mm = acc.get(w.ti) || acc.set(w.ti, new Map()).get(w.ti); + mm.set(of[i], (mm.get(of[i]) || 0) + (w.e - w.s)); + }); + const lab = new Array(turns.length).fill(-1); + for (const [ti, mm] of acc) { let bl = -1, bd = -1; for (const [l, d] of mm) if (d > bd) { bd = d; bl = l; } lab[ti] = bl; } + const talk = new Map(); + turns.forEach((t, ti) => { if (lab[ti] >= 0) talk.set(lab[ti], (talk.get(lab[ti]) || 0) + (t.end - t.start)); }); + return { lab, talk }; +} +const centroid = g => norm(g.reduce((acc, i) => (X[i].forEach((x, d) => acc[d] += x / g.length), acc), new Array(X[0].length).fill(0))); + +// ---- the full pipeline ----------------------------------------------------- +function run(tag, { fine = FINE, final = null, k = 0, minTotal = MIN_TOTAL }) { + let groups = ahc(fine); + const { talk } = turnLabels(groups); + // fold weak clusters into the nearest strong centroid + const strongIdx = groups.map((g, i) => i).filter(i => (talk.get(i) || 0) >= minTotal); + let folded = 0; + if (strongIdx.length && strongIdx.length < groups.length) { + const cents = new Map(strongIdx.map(i => [i, centroid(groups[i])])); + const merged = new Map(strongIdx.map(i => [i, [...groups[i]]])); + groups.forEach((g, i) => { + if (merged.has(i)) return; + const c = centroid(g); + let best = strongIdx[0], bs = -Infinity; + for (const s of strongIdx) { const v = dot(c, cents.get(s)); if (v > bs) { bs = v; best = s; } } + merged.get(best).push(...g); folded++; + }); + groups = [...merged.values()]; + } + const afterFold = groups.length; + if (final !== null || k > 0) groups = ahc(final === null ? 0 : final, k, groups); + const { lab, talk: t2 } = turnLabels(groups); + const order = [...t2.entries()].sort((a, b) => b[1] - a[1]); + const switches = lab.filter((l, i) => i > 0 && l >= 0 && lab[i - 1] >= 0 && l !== lab[i - 1]).length; + const first = new Map(), last = new Map(); + turns.forEach((t, ti) => { const l = lab[ti]; if (l < 0) return; if (!first.has(l)) first.set(l, t.start); last.set(l, t.end); }); + console.log(`${tag}: fine=${f(fine)}${final !== null ? ` final=${f(final)}` : ''}${k ? ` k=${k}` : ''} → afterFine=${afterFold + folded} folded=${folded} speakers=${groups.length} switches=${switches}${EXPECT ? ` expect=${EXPECT}` : ''}`); + if (!has('quiet')) console.log(' ' + order.map(([l, d]) => `${d.toFixed(0)}s[${(first.get(l) / 60).toFixed(1)}-${(last.get(l) / 60).toFixed(1)}]`).join(' ')); + return groups.length; +} + +if (KFIX) run('pinned', { k: KFIX }); +for (const t of (opt('final', '') ? opt('final').split(',').map(Number) : [0.10, 0.15, 0.20, 0.25, 0.30])) + run('auto', { final: t }); diff --git a/app/lab/lab-dump-segs.sh b/app/lab/lab-dump-segs.sh new file mode 100644 index 0000000..a35cfa3 --- /dev/null +++ b/app/lab/lab-dump-segs.sh @@ -0,0 +1,23 @@ +#!/bin/bash +# Lab: dump RAW sherpa diarization turns for a recording, at maximum temporal +# granularity (min-duration-off=0 -> no adjacent-same-label merging, small +# cluster threshold -> fine labels). We only reuse the TURN BOUNDARIES; the +# labels get thrown away and re-clustered offline by cluster-lab.js. +# Usage: lab-dump-segs.sh [cluster-threshold] +DIR=/opt/sherpa/sherpa-onnx-v1.13.3-linux-aarch64-shared-cpu +BIN=$DIR/bin/sherpa-onnx-offline-speaker-diarization +SEG=/opt/sherpa/sherpa-onnx-pyannote-segmentation-3-0/model.onnx +EMB=${EMB:-/opt/sherpa/nemo_titanet_small.onnx} +WAV="$1"; OUT="$2"; THR="${3:-0.2}" +/opt/callrec/aime-cpucap.sh 1608000 +export LD_LIBRARY_PATH=$DIR/lib +echo "start $(date -Is)" > "$OUT.log" +taskset -c 4-7 "$BIN" \ + --segmentation.pyannote-model=$SEG --segmentation.num-threads=4 \ + --embedding.model=$EMB --embedding.num-threads=4 \ + --clustering.cluster-threshold=$THR \ + --min-duration-on=0.2 --min-duration-off=0.0 \ + "$WAV" > "$OUT" 2>> "$OUT.log" +echo "end $(date -Is) rc=$?" >> "$OUT.log" +/opt/callrec/aime-cpucap.sh 2256000 +grep -c 'speaker_' "$OUT" >> "$OUT.log" diff --git a/app/lab/lab-embed.mjs b/app/lab/lab-embed.mjs new file mode 100644 index 0000000..1521fb8 --- /dev/null +++ b/app/lab/lab-embed.mjs @@ -0,0 +1,117 @@ +// Lab: window-level speaker embeddings for a recording. +// Reads sherpa's raw turn boundaries, slices the 16k mono wav into fixed windows +// (no ffmpeg: direct PCM slicing), embeds each window with a given sherpa +// embedding model via the spk_embed helper, and dumps JSON for offline +// clustering experiments (cluster-lab.mjs). +// +// Usage: node lab-embed.mjs [winSec] [hopSec] +import fs from 'node:fs'; +import { spawn } from 'node:child_process'; +import readline from 'node:readline'; +import os from 'node:os'; + +const [segsPath, wavPath, model, outPath, winSecArg, hopSecArg] = process.argv.slice(2); +const WIN = +(winSecArg || 3.0), HOP = +(hopSecArg || 1.5); +const MIN_WIN = 0.6; // shortest slice we bother embedding +const DIR = '/opt/sherpa/sherpa-onnx-v1.13.3-linux-aarch64-shared-cpu'; +const TMP = fs.mkdtempSync(`${os.tmpdir()}/labemb_`); + +// ---- wav (PCM s16 mono) ---------------------------------------------------- +function loadWav(p) { + const b = fs.readFileSync(p); + if (b.toString('ascii', 0, 4) !== 'RIFF' || b.toString('ascii', 8, 12) !== 'WAVE') throw new Error('not RIFF/WAVE'); + let off = 12, fmt = null, data = null; + while (off + 8 <= b.length) { + const id = b.toString('ascii', off, off + 4), size = b.readUInt32LE(off + 4); + if (id === 'fmt ') fmt = { ch: b.readUInt16LE(off + 10), rate: b.readUInt32LE(off + 12), bits: b.readUInt16LE(off + 22) }; + if (id === 'data') { data = { start: off + 8, size: Math.min(size, b.length - off - 8) }; break; } + off += 8 + size + (size & 1); + } + if (!fmt || !data) throw new Error('wav chunks missing'); + if (fmt.bits !== 16) throw new Error('need s16, got ' + fmt.bits); + return { buf: b, fmt, data }; +} +function writeSlice(w, startSec, endSec, path) { + const bytesPerFrame = 2 * w.fmt.ch; + const s = Math.max(0, Math.floor(startSec * w.fmt.rate) * bytesPerFrame); + const e = Math.min(w.data.size, Math.ceil(endSec * w.fmt.rate) * bytesPerFrame); + const n = Math.max(0, e - s); + const h = Buffer.alloc(44); + h.write('RIFF', 0); h.writeUInt32LE(36 + n, 4); h.write('WAVE', 8); + h.write('fmt ', 12); h.writeUInt32LE(16, 16); h.writeUInt16LE(1, 20); + h.writeUInt16LE(w.fmt.ch, 22); h.writeUInt32LE(w.fmt.rate, 24); + h.writeUInt32LE(w.fmt.rate * bytesPerFrame, 28); h.writeUInt16LE(bytesPerFrame, 32); h.writeUInt16LE(16, 34); + h.write('data', 36); h.writeUInt32LE(n, 40); + fs.writeFileSync(path, Buffer.concat([h, w.buf.subarray(w.data.start + s, w.data.start + e)])); + return n / (w.fmt.rate * bytesPerFrame); +} + +// ---- spk_embed pipe -------------------------------------------------------- +function embedder(modelPath) { + const proc = spawn('taskset', ['-c', '4-7', '/opt/callrec/spk_embed', modelPath], + { env: { ...process.env, LD_LIBRARY_PATH: `${DIR}/lib` }, stdio: ['pipe', 'pipe', 'pipe'] }); + let pending = null; + const rl = readline.createInterface({ input: proc.stdout }); + const ready = new Promise((res, rej) => { + const to = setTimeout(() => rej(new Error('READY timeout')), 180000); + rl.on('line', line => { + if (line.startsWith('READY')) { clearTimeout(to); return res(); } + if (!pending) return; + const p = pending; pending = null; + if (line.startsWith('OK ')) p.res(line.slice(3).split(',').map(Number)); + else p.rej(new Error(line)); + }); + }); + proc.stderr.on('data', d => process.stderr.write('[spk] ' + d)); + let chain = ready; + return { + embed(path) { + chain = chain.then(() => new Promise((res, rej) => { pending = { res, rej }; proc.stdin.write(path + '\n'); })); + return chain; + }, + close() { try { proc.kill(); } catch {} }, + }; +} + +// ---- main ------------------------------------------------------------------ +const turns = []; +for (const line of fs.readFileSync(segsPath, 'utf8').split('\n')) { + const m = line.match(/([\d.]+)\s*--\s*([\d.]+)\s*(speaker_\d+)/); + if (m) turns.push({ start: +m[1], end: +m[2], raw: m[3] }); +} +turns.sort((a, b) => a.start - b.start); +console.log(`turns=${turns.length} rawLabels=${new Set(turns.map(t => t.raw)).size} speech=${turns.reduce((a, t) => a + t.end - t.start, 0).toFixed(0)}s`); + +const wav = loadWav(wavPath); +console.log('wav', wav.fmt, (wav.data.size / (wav.fmt.rate * 2 * wav.fmt.ch)).toFixed(1) + 's'); + +const wins = []; +turns.forEach((t, ti) => { + const dur = t.end - t.start; + if (dur < MIN_WIN) return; + if (dur <= WIN) { wins.push({ ti, start: t.start, end: t.end }); return; } + for (let s = t.start; s + MIN_WIN <= t.end; s += HOP) wins.push({ ti, start: s, end: Math.min(s + WIN, t.end) }); +}); +console.log(`windows=${wins.length}`); + +const emb = embedder(model); +const t0 = Date.now(); +let done = 0; +for (const w of wins) { + const p = `${TMP}/w.wav`; + writeSlice(wav, w.start, w.end, p); + try { w.vec = await emb.embed(p); } catch (e) { w.err = e.message; } + if (++done % 100 === 0) { + const rate = done / ((Date.now() - t0) / 1000); + console.log(` ${done}/${wins.length} ${rate.toFixed(1)}/s eta ${((wins.length - done) / rate).toFixed(0)}s`); + } +} +emb.close(); +const ok = wins.filter(w => w.vec); +fs.writeFileSync(outPath, JSON.stringify({ + model, wav: wavPath, win: WIN, hop: HOP, + dim: ok.length ? ok[0].vec.length : 0, + turns, wins: wins.map(w => ({ ti: w.ti, s: +w.start.toFixed(3), e: +w.end.toFixed(3), v: w.vec ? w.vec.map(x => +x.toFixed(5)) : null })), +})); +console.log(`done ok=${ok.length}/${wins.length} dim=${ok.length ? ok[0].vec.length : 0} in ${((Date.now() - t0) / 1000).toFixed(0)}s -> ${outPath}`); +try { fs.rmSync(TMP, { recursive: true, force: true }); } catch {} diff --git a/app/lab/lab-eval.mjs b/app/lab/lab-eval.mjs new file mode 100644 index 0000000..08ecf91 --- /dev/null +++ b/app/lab/lab-eval.mjs @@ -0,0 +1,23 @@ +// Replay the REAL production clustering (diar.js clusterWindows) on cached +// window embeddings, so thresholds can be tuned without re-running sherpa or the +// embedding pass. Usage: +// [DIAR_COS_FINE=..] [DIAR_COS_FINAL=..] node lab-eval.mjs [expect] [numSpeakers] [markers.json] +import fs from 'node:fs'; +import { clusterWindows } from './diar-v2.js'; + +const [file, expect, numSp, markerFile] = process.argv.slice(2); +const d = JSON.parse(fs.readFileSync(file, 'utf8')); +const ok = d.wins.filter(w => w.v).map(w => ({ ti: w.ti, start: w.s, end: w.e, vec: w.v })); +const markers = markerFile ? JSON.parse(fs.readFileSync(markerFile, 'utf8')) : []; +const segs = clusterWindows(d.turns, ok, +(numSp || 0), markers); + +const talk = new Map(), first = new Map(), last = new Map(); +for (const s of segs) { + talk.set(s.speaker, (talk.get(s.speaker) || 0) + (s.end - s.start)); + if (!first.has(s.speaker)) first.set(s.speaker, s.start); + last.set(s.speaker, s.end); +} +const order = [...talk.entries()].sort((a, b) => b[1] - a[1]); +console.log(`RESULT ${d.model.split('/').pop()} fine=${process.env.DIAR_COS_FINE || 0.35} final=${process.env.DIAR_COS_FINAL || 0.30}` + + ` → speakers=${order.length}${expect ? ` (expect ${expect})` : ''} segs=${segs.length}`); +console.log(' ' + order.map(([sp, t]) => `${sp.replace('speaker_', '')}:${t.toFixed(0)}s[${(first.get(sp) / 60).toFixed(1)}-${(last.get(sp) / 60).toFixed(1)}]`).join(' ')); diff --git a/app/lab/lab-run-embed.sh b/app/lab/lab-run-embed.sh new file mode 100644 index 0000000..a3ba7da --- /dev/null +++ b/app/lab/lab-run-embed.sh @@ -0,0 +1,14 @@ +#!/bin/bash +# Embed the same window set with every candidate speaker-embedding model, so the +# clustering lab can compare their speaker separability on OUR audio. +cd /opt/callrec/lab +/opt/callrec/aime-cpucap.sh 1800000 +for m in nemo_titanet_small wespeaker_en_voxceleb_resnet34_LM 3dspeaker_speech_campplus_sv_zh_en_16k-common_advanced nemo_en_titanet_large 3dspeaker_speech_eres2netv2_sv_zh-cn_16k-common; do + out=emb74.$m.json + [ -f "$out" ] && { echo "skip $m"; continue; } + echo "=== $m $(date -Is)" + taskset -c 4-7 node lab-embed.mjs segs74.txt /opt/callrec/recordings/rec_1784280581035.wav \ + /opt/sherpa/$m.onnx "$out" 3.0 1.5 2>&1 | tail -12 +done +/opt/callrec/aime-cpucap.sh 2256000 +echo "ALLDONE $(date -Is)" diff --git a/app/lab/lab-run2.sh b/app/lab/lab-run2.sh new file mode 100644 index 0000000..07a1148 --- /dev/null +++ b/app/lab/lab-run2.sh @@ -0,0 +1,19 @@ +#!/bin/bash +# Regression set: the two recordings whose true speaker count we know +# (#70 = 4 speakers, #71 = 2). Same pipeline as #74 so the chosen thresholds can +# be checked against small calls, not just the 23-person weekly. +cd /opt/callrec/lab +/opt/callrec/aime-cpucap.sh 1416000 +run() { # $1 = tag, $2 = wav + [ -f segs$1.txt ] || ./lab-dump-segs.sh "$2" segs$1.txt 0.2 + for m in nemo_titanet_small 3dspeaker_speech_campplus_sv_zh_en_16k-common_advanced nemo_en_titanet_large; do + out=emb$1.$m.json + [ -f "$out" ] && continue + echo "=== $1 $m $(date -Is)" + taskset -c 4-7 node lab-embed.mjs segs$1.txt "$2" /opt/sherpa/$m.onnx "$out" 3.0 1.5 2>&1 | tail -3 + done +} +run 70 /opt/callrec/recordings/rec_1784192426082.wav +run 71 /opt/callrec/recordings/rec_1784205293914.wav +/opt/callrec/aime-cpucap.sh 2256000 +echo "ALLDONE2 $(date -Is)" diff --git a/app/lab/lab-thr.mjs b/app/lab/lab-thr.mjs new file mode 100644 index 0000000..dea9a4d --- /dev/null +++ b/app/lab/lab-thr.mjs @@ -0,0 +1,52 @@ +// Which merge threshold does a recording itself suggest? Fits a 2-component +// Gaussian mixture (EM) to the pairwise window-similarity histogram — the +// "different speaker" and "same speaker" modes — and takes the point where the +// two weighted densities cross. Compares that with the fixed thresholds. +import fs from 'node:fs'; +const f = x => x.toFixed(3); +const norm = v => { let s = 0; for (const x of v) s += x * x; s = Math.sqrt(s) + 1e-9; return v.map(x => x / s); }; +const dot = (a, b) => { let s = 0; for (let i = 0; i < a.length; i++) s += a[i] * b[i]; return s; }; + +for (const file of process.argv.slice(2)) { + const d = JSON.parse(fs.readFileSync(file, 'utf8')); + const wins = d.wins.filter(w => w.v); + let X = wins.map(w => norm(w.v)); + const dim = X[0].length, mu = new Array(dim).fill(0); + for (const v of X) for (let i = 0; i < dim; i++) mu[i] += v[i] / X.length; + X = X.map(v => norm(v.map((x, i) => x - mu[i]))); + const n = X.length; + // sample pairs (deterministic stride, no RNG) + const vals = []; + for (let k = 1; k < 400000 && vals.length < 60000; k++) { + const i = (k * 7919) % n, j = (k * 104729 + 13) % n; + if (i !== j) vals.push(dot(X[i], X[j])); + } + // EM for 2 Gaussians + let m1 = -0.1, m2 = 0.5, s1 = 0.15, s2 = 0.15, w1 = 0.8; + const N = (x, m, s) => Math.exp(-((x - m) ** 2) / (2 * s * s)) / (s * Math.sqrt(2 * Math.PI)); + for (let it = 0; it < 100; it++) { + let sw1 = 0, sm1 = 0, sm2 = 0, sv1 = 0, sv2 = 0, sw2 = 0; + for (const x of vals) { + const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2); + const r = a / (a + b + 1e-300); + sw1 += r; sw2 += 1 - r; sm1 += r * x; sm2 += (1 - r) * x; + } + const nm1 = sm1 / sw1, nm2 = sm2 / sw2; + for (const x of vals) { + const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2); + const r = a / (a + b + 1e-300); + sv1 += r * (x - nm1) ** 2; sv2 += (1 - r) * (x - nm2) ** 2; + } + m1 = nm1; m2 = nm2; + s1 = Math.max(0.02, Math.sqrt(sv1 / sw1)); s2 = Math.max(0.02, Math.sqrt(sv2 / sw2)); + w1 = sw1 / vals.length; + } + // crossing point between the means: w1*N1(x) = w2*N2(x) + let lo = m1, hi = m2, t = (lo + hi) / 2; + for (let i = 0; i < 60; i++) { + t = (lo + hi) / 2; + const g = w1 * N(t, m1, s1) - (1 - w1) * N(t, m2, s2); + if (g > 0) lo = t; else hi = t; + } + console.log(`${file.split('.').slice(0, 2).join('.')}: diff N(${f(m1)},${f(s1)}) w=${f(w1)} | same N(${f(m2)},${f(s2)}) w=${f(1 - w1)} → crossing=${f(t)}`); +} diff --git a/app/recorder.js b/app/recorder.js index 9ec10a1..c96fced 100644 --- a/app/recorder.js +++ b/app/recorder.js @@ -158,13 +158,35 @@ function wavDurationSec(wavPath) { try { return Math.max(0, (fs.statSync(wavPath).size - 44) / 32000); } catch { return 0; } } -let current = null; // { proc, wavPath, startTs, source, title } +let current = null; // { proc, wavPath, startTs, source, title, markers } export function status() { - return current ? { recording: true, startedAt: current.startTs, source: current.source } : { recording: false }; + return current ? { recording: true, startedAt: current.startTs, source: current.source, + markers: current.markers } : { recording: false }; } export function isRecording() { return !!current; } +// --- live speaker markers --------------------------------------------------- +// While recording, the user types the name of whoever is talking. We stamp the +// offset from recording start; after diarization the name is attached to the +// acoustic cluster that was speaking around that offset (see diar.resolveMarkers). +// This is the only reliable way to name 20+ participants of a conference call. +export function addMarker(name) { + if (!current) throw new Error('not recording'); + const n = String(name || '').trim(); + if (!n) throw new Error('Name fehlt'); + const m = { t_ms: Date.now() - current.startTs, name: n.slice(0, 128) }; + current.markers.push(m); + return { ...m, index: current.markers.length - 1, markers: current.markers }; +} +export function listMarkers() { return current ? current.markers : []; } +export function removeMarker(index) { + if (!current) throw new Error('not recording'); + const i = +index; + if (i >= 0 && i < current.markers.length) current.markers.splice(i, 1); + return current.markers; +} + // List real capture devices (one entry per recording device). A "source" in // PulseAudio/PipeWire includes monitor sources (the loopback of every OUTPUT) // and both A2DP/HFP variants of a BT card — confusing for the user. We only @@ -263,7 +285,7 @@ export async function startRecording({ source, title, codec, numSpeakers } = {}) // in the list (mSBC 16 kHz vs CVSD 8 kHz). const codecTag = hfp && hfp.profile ? (/cvsd/i.test(hfp.profile) ? ' [CVSD 8k]' : ' [mSBC 16k]') : ''; current = { - proc, wavPath, startTs: ts, source: source || 'default', numSpeakers, + proc, wavPath, startTs: ts, source: source || 'default', numSpeakers, markers: [], title: (title || `Aufnahme ${new Date(ts).toLocaleString('de-DE')}`) + codecTag, err: () => err }; return { startedAt: ts }; @@ -328,13 +350,14 @@ export async function stopAndSave() { catch { fs.copyFileSync(c.wavPath, path.join(REC_DIR, wavFile)); try { fs.unlinkSync(c.wavPath); } catch {} } return { mp3_file: `${base}.mp3`, wav_file: wavFile, duration_ms: dur, - language: 'de', title: c.title, device: c.source, numSpeakers: c.numSpeakers }; + language: 'de', title: c.title, device: c.source, numSpeakers: c.numSpeakers, + markers: c.markers }; } // Transcribe a saved wav (in REC_DIR) with whisper (thermal-guarded) + diarize. // Returns { transcript, words }. Idempotent — safe to re-run for recovery of a // recording whose transcription was interrupted (crash) or failed. -export async function transcribeWav(wavFile, numSpeakers) { +export async function transcribeWav(wavFile, numSpeakers, opts = {}) { const wavPath = path.isAbsolute(wavFile) ? wavFile : path.join(REC_DIR, wavFile); if (!fs.existsSync(wavPath)) throw new Error('wav-Datei fehlt: ' + wavFile); const outBase = `/tmp/${path.basename(wavPath, '.wav')}`; @@ -389,7 +412,7 @@ export async function transcribeWav(wavFile, numSpeakers) { // Acoustic diarization under the A76 clock cap (both paths). Non-fatal. await setA76Max(A76_CAP); try { - const segs = await diar.diarize(wavPath, numSpeakers); + const segs = await diar.diarize(wavPath, numSpeakers, { markers: opts.markers || [] }); diar.assignSpeakers(words, segs); } catch (e) { console.error('diarize failed:', e.message); } try { fs.unlinkSync(`${outBase}.json`); } catch {} diff --git a/app/server.js b/app/server.js index c5c6ff2..763517c 100644 --- a/app/server.js +++ b/app/server.js @@ -56,7 +56,7 @@ async function identifySpeakers(recData) { if (!speakers.length || !recData.wav_file) return {}; const wavPath = path.join(REC_DIR2, recData.wav_file); if (!fs.existsSync(wavPath)) return {}; - const vps = await db.listVoiceprints(); + const vps = await db.listVoiceprints(spk.modelId()); if (!vps.length) return {}; const best = {}; let cur = null; for (const w of words) { @@ -80,6 +80,73 @@ async function identifySpeakers(recData) { return map; } +// Fold a speaker's longest clean run into the named voiceprint centroid, so the +// next recording can auto-identify them. Shared by the manual rename popup and +// the live-marker resolution. Returns true if a sample was learned. +async function enrollSpeakerVoice(recData, speaker, name) { + if (!name || !recData.wav_file) return false; + const seg = longestRun(recData.words, speaker); + if (!seg || (seg.end - seg.start) < 1500) return false; + const wavPath = path.join(REC_DIR2, recData.wav_file); + if (!fs.existsSync(wavPath)) return false; + const cut = `/tmp/enr_${recData.id}_${speaker}.wav`; + try { + await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(seg.start / 1000), + '-to', String(Math.min(seg.end, seg.start + 20000) / 1000), '-ar', '16000', '-ac', '1', cut]); + const emb = await spk.embed(cut); + await db.upsertVoiceprint(name, emb, spk.modelId()); + return true; + } catch (e) { console.error('enroll failed:', e.message); return false; } + finally { try { fs.unlinkSync(cut); } catch {} } +} + +// Apply the names the user typed DURING the recording to the diarized clusters. +// Runs after every (re-)diarization, before the LLM analysis, so the summary +// already talks about real people. Two effects: +// 1. each resolved cluster gets its typed name (authoritative, manual-level), +// 2. if ONE name confidently lands on SEVERAL clusters, those clusters are the +// same person split by the diarizer → they are merged (user-supervised fix +// for the residual over-segmentation). +async function applyMarkers(id) { + const rec = await db.getRecording(id); + if (!rec || !(rec.markers || []).length || !(rec.words || []).some(w => w.speaker)) return null; + const resolved = diar.resolveMarkers(rec.words, rec.markers); + for (const m of resolved) await db.setMarkerSpeaker(m.id, m.speaker, m.conf).catch(() => {}); + + // name -> clusters (only confident hits drive the merge) + const byName = {}; + for (const m of resolved) { + if (!m.speaker || m.conf === 'low' || m.conf === null) continue; + (byName[m.name] = byName[m.name] || new Set()).add(m.speaker); + } + const talk = {}; + for (const w of rec.words) if (w.speaker) talk[w.speaker] = (talk[w.speaker] || 0) + (w.t_end_ms - w.t_start_ms); + + const fold = {}; // extra cluster -> primary cluster + const nameOf = {}; // cluster -> typed name + for (const [name, set] of Object.entries(byName)) { + const cls = [...set].sort((a, b) => (talk[b] || 0) - (talk[a] || 0)); + const primary = cls[0]; + nameOf[primary] = name; + for (const c of cls.slice(1)) fold[c] = primary; + } + let merged = 0; + if (Object.keys(fold).length) { + for (const w of rec.words) if (fold[w.speaker]) w.speaker = fold[w.speaker]; + await db.replaceWords(id, rec.words); + merged = Object.keys(fold).length; + console.log(`markers: merged ${merged} cluster(s) into marker-named speakers on #${id}`); + } + for (const [sp, name] of Object.entries(nameOf)) await db.setSpeakerName(id, sp, name).catch(() => {}); + // Learn the voices so the NEXT recording identifies them automatically. + const rec2 = await db.getRecording(id); + let learned = 0; + for (const [sp, name] of Object.entries(nameOf)) if (await enrollSpeakerVoice(rec2, sp, name)) learned++; + const named = Object.keys(nameOf).length; + console.log(`markers on #${id}: ${resolved.length} marker(s) → ${named} named speaker(s), ${merged} merged, ${learned} voice(s) learned`); + return { resolved, named, merged, learned }; +} + // Run LLM analysis for a recording and persist title/summary/todos/participants/ // clean dialogue. Background (NPU ~30-90s) so record/stop stays fast. // A76 clock cap for the whole run: rkllm burns ~1.7 CPU cores alongside the NPU @@ -189,15 +256,25 @@ function withNpuLock(label, fn) { // --- Recording --- app.post('/api/record/start', wrap(async (req, res) => res.json(await rec.startRecording(req.body || {})))); +// Live speaker markers: type the name of whoever is talking WHILE recording. +app.post('/api/record/marker', wrap(async (req, res) => res.json(rec.addMarker((req.body && req.body.name) || '')))); +app.get('/api/record/markers', (req, res) => res.json(rec.listMarkers())); +app.delete('/api/record/marker/:index', wrap(async (req, res) => res.json({ markers: rec.removeMarker(req.params.index) }))); // Transcribe (thermal-guarded whisper + diarize) then analyze, in the background. // The wav + DB row already exist, so a crash here loses nothing — the recording // stays with status 'transcribing' and can be recovered via /retranscribe. async function transcribeAndAnalyze(id, wavFile, numSpeakers) { await db.markAnalysisStart(id).catch(() => {}); try { - const { transcript, words } = await rec.transcribeWav(wavFile, numSpeakers); + // Live markers constrain the diarization itself (same name = one speaker, + // different names = never merged), so they are loaded before transcribing. + const markers = await db.listMarkers(id).catch(() => []); + const { transcript, words } = await rec.transcribeWav(wavFile, numSpeakers, { markers }); await db.setTranscript(id, transcript, words); console.log(`transcription done for #${id}: ${words.length} segments`); + // Names typed live → cluster names (+ user-supervised cluster merge) BEFORE + // the LLM sees the dialogue. + await applyMarkers(id).catch(e => console.error('applyMarkers:', e.message)); } catch (e) { console.error(`transcription failed for #${id}:`, e.message); await db.setStatus(id, 'error').catch(() => {}); @@ -214,8 +291,10 @@ app.post('/api/record/stop', wrap(async (req, res) => { // analysis_status === 'pending', so the transcript + analysis fill in live. const r = await rec.stopAndSave(); const id = await db.addRecording({ ...r, transcript: '', status: 'transcribing' }); + await db.addMarkers(id, r.markers).catch(e => console.error('addMarkers:', e.message)); res.json({ id, title: r.title, mp3_file: r.mp3_file, duration_ms: r.duration_ms, - transcript: '', status: 'transcribing', analysis_status: 'pending' }); + transcript: '', status: 'transcribing', analysis_status: 'pending', + markers: (r.markers || []).length }); withNpuLock(`transcribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, r.numSpeakers)); })); @@ -230,7 +309,9 @@ app.delete('/api/recordings/:id', wrap(async (req, res) => { await db.deleteReco // --- Voiceprints / enrollment --- app.get('/api/voiceprints', wrap(async (req, res) => { const vps = await db.listVoiceprints(); - res.json(vps.map(v => ({ id: v.id, name: v.name, dim: v.dim, created_at: v.created_at }))); + const cur = spk.modelId(); + res.json(vps.map(v => ({ id: v.id, name: v.name, dim: v.dim, model: v.model, + stale: v.model !== cur, created_at: v.created_at }))); })); app.delete('/api/voiceprints/:id', wrap(async (req, res) => { await db.deleteVoiceprint(+req.params.id); res.json({ ok: true }); })); app.post('/api/enroll/start', wrap(async (req, res) => res.json(await rec.startRecording({ source: req.body && req.body.source })))); @@ -240,7 +321,7 @@ app.post('/api/enroll/stop', wrap(async (req, res) => { const wav = await rec.stopRaw(); try { const emb = await spk.embed(wav); - const id = await db.addVoiceprint(name, emb); + const id = await db.addVoiceprint(name, emb, spk.modelId()); res.json({ id, name, dim: emb.length }); } finally { try { fs.unlinkSync(wav); } catch {} } })); @@ -283,9 +364,13 @@ async function rediarize(id, numSpeakers) { // is the other entry point and must do it too, or a long re-diar overheats). await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {}); try { - const segs = await diar.diarize(wav, numSpeakers); + const segs = await diar.diarize(wav, numSpeakers, { markers: r.markers || [] }); diar.assignSpeakers(words, segs); await db.replaceWords(id, words); + // Re-diarization RENUMBERS the clusters, so every previously stored name + // ("speaker_02" = Fabio) would now point at a different voice. Drop them; the + // live markers and the voiceprint identification re-fill them right after. + await db.pool.query('UPDATE recordings SET speaker_names=NULL WHERE id=?', [id]).catch(() => {}); console.log(`re-diarized #${id} (numSpeakers=${numSpeakers || 'auto'}) → ${new Set(segs.map(s => s.speaker)).size} speakers`); } finally { await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {}); @@ -307,10 +392,33 @@ app.post('/api/recordings/:id/reanalyze', wrap(async (req, res) => { withNpuLock(`reanalyze #${id}`, async () => { try { await rediarize(id, numSpeakers); } catch (e) { console.error(`re-diarize failed for #${id}:`, e.message); } + await applyMarkers(id).catch(e => console.error('applyMarkers:', e.message)); await analyzeRecording(id); }); })); +// --- markers of a stored recording ----------------------------------------- +app.get('/api/recordings/:id/markers', wrap(async (req, res) => res.json(await db.listMarkers(+req.params.id)))); +// Add a marker after the fact (e.g. while listening back) — same resolution path. +app.post('/api/recordings/:id/markers', wrap(async (req, res) => { + const name = String((req.body && req.body.name) || '').trim(); + const tMs = +(req.body && req.body.t_ms); + if (!name || !Number.isFinite(tMs)) return res.status(400).json({ error: 'name/t_ms fehlt' }); + await db.addMarker(+req.params.id, tMs, name); + res.json({ ok: true, markers: await db.listMarkers(+req.params.id) }); +})); +app.delete('/api/recordings/:id/markers/:mid', wrap(async (req, res) => { + await db.deleteMarker(+req.params.mid); + res.json({ ok: true, markers: await db.listMarkers(+req.params.id) }); +})); +// Re-resolve the markers against the current diarization (no LLM, no sherpa). +app.post('/api/recordings/:id/apply-markers', wrap(async (req, res) => { + const out = await applyMarkers(+req.params.id); + const r = await db.getRecording(+req.params.id); + res.json({ ok: true, ...(out || { named: 0, merged: 0, learned: 0 }), + speaker_names: (r && r.speaker_names) || {}, markers: (r && r.markers) || [] }); +})); + // LLM-only re-analysis: re-run title/summary/todos on the EXISTING words + // speaker labels, WITHOUT re-diarizing (fast — skips the ~40min sherpa pass when // the speakers are already correct and only the LLM/prompts changed). @@ -378,7 +486,7 @@ app.post('/api/recordings/:id/speaker', wrap(async (req, res) => { await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(seg.start / 1000), '-to', String(seg.end / 1000), '-ar', '16000', '-ac', '1', cut]); const emb = await spk.embed(cut); - await db.upsertVoiceprint(name, emb); + await db.upsertVoiceprint(name, emb, spk.modelId()); enrolled = true; } catch (e) { console.error('enroll-from-recording failed:', e.message); learnMsg = 'Lernen fehlgeschlagen'; } finally { try { fs.unlinkSync(cut); } catch {} } diff --git a/app/spk.js b/app/spk.js index 0b4cf63..bb4c0d0 100644 --- a/app/spk.js +++ b/app/spk.js @@ -7,8 +7,20 @@ import readline from 'node:readline'; const BIN = '/opt/callrec/spk_embed'; const DIR = '/opt/sherpa/sherpa-onnx-v1.13.3-linux-aarch64-shared-cpu'; -const EMB = process.env.DIAR_EMB || '/opt/sherpa/nemo_titanet_small.onnx'; +// Identity/clustering model. Benchmarked 2026-07-27 on our own conference audio +// (#74, 1181 mean-centred windows, same-turn vs cross-turn cosine): +// TitaNet-large d' 3.75 / 197 s CAM++ (3D-Speaker) d' 3.72 / 82 s +// TitaNet-small d' 3.54 / 57 s wespeaker VoxCeleb resnet34 d' 2.43 / 185 s +// Fed through the full v2 pipeline the three good models ended up within one +// speaker of each other (21/22/21 on a 23-person weekly), so TitaNet-small stays +// the default: it is the fastest AND keeps every already-learned voiceprint +// valid. SPK_MODEL switches models — prints are then stored per model (see +// db.upsertVoiceprint), because cosine across models is meaningless. +const EMB = process.env.SPK_MODEL || '/opt/sherpa/nemo_titanet_small.onnx'; export const THRESHOLD = +(process.env.SPK_THRESHOLD || 0.5); +// Voiceprints are only comparable within the SAME embedding model — stored with +// every print so a model swap can't silently match against stale vectors. +export function modelId() { return EMB.split('/').pop().replace(/\.onnx$/, ''); } let proc = null, ready = null, rl = null, pending = null, chain = Promise.resolve(); @@ -56,4 +68,4 @@ export function bestMatch(emb, voiceprints, threshold = THRESHOLD) { return (best && bs >= threshold) ? { id: best.id, name: best.name, score: bs } : null; } -export function status() { return { model: EMB, present: fs.existsSync(EMB), running: !!proc, threshold: THRESHOLD }; } +export function status() { return { model: EMB, modelId: modelId(), present: fs.existsSync(EMB), running: !!proc, threshold: THRESHOLD }; } diff --git a/app/web/src/main.js b/app/web/src/main.js index cdc0b23..f94d1d0 100644 --- a/app/web/src/main.js +++ b/app/web/src/main.js @@ -8,7 +8,7 @@ const api = { }; const app = document.querySelector('#app'); -const state = { tab: 'recordings', recording: false, recStart: 0, timer: null, selectedId: null, collapsedDates: new Set(), lastListSig: null }; +const state = { tab: 'recordings', recording: false, recStart: 0, timer: null, selectedId: null, collapsedDates: new Set(), lastListSig: null, markers: [], mkDraft: '' }; function fmt(ms) { const s = Math.floor(ms / 1000); return `${String(Math.floor(s / 60)).padStart(2, '0')}:${String(s % 60).padStart(2, '0')}`; } function toast(msg, err) { const t = document.createElement('div'); t.className = 'toast' + (err ? ' err' : ''); t.textContent = msg; document.body.appendChild(t); setTimeout(() => t.remove(), 4000); } @@ -34,6 +34,10 @@ function render() {
${state.recording ? fmt(Date.now() - state.recStart) : '00:00'} + ${state.recording ? ` + + 🏷 ${state.markers.length}` : ''}