Aime/app/spk.js

// Speaker voiceprints: talks to the persistent spk_embed process (sherpa-onnx
// NeMo TitaNet embedding on CPU). embed(wav) → 192-dim vector; cosine match a
// diarization cluster against enrolled voiceprints → a name. All offline.
import { spawn } from 'node:child_process';
import fs from 'node:fs';
import readline from 'node:readline';
import * as gate from './gate.js';

const BIN = '/opt/callrec/spk_embed';
const DIR = '/opt/sherpa/sherpa-onnx-v1.13.3-linux-aarch64-shared-cpu';
// 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();

function start() {
  if (proc) return ready;
  if (!fs.existsSync(EMB)) return Promise.reject(new Error('Embedding-Modell fehlt: ' + EMB));
  proc = spawn('taskset', ['-c', '4-7', BIN, EMB],
    { env: { ...process.env, LD_LIBRARY_PATH: `${DIR}/lib` }, stdio: ['pipe', 'pipe', 'pipe'] });
  proc.stderr.on('data', d => process.stderr.write('[spk] ' + d));
  ready = new Promise((res, rej) => {
    const to = setTimeout(() => rej(new Error('spk READY timeout')), 60000);
    rl = readline.createInterface({ input: proc.stdout });
    rl.on('line', line => {
      if (line.startsWith('READY')) { clearTimeout(to); res(); return; }
      if (!pending) return;
      const p = pending; pending = null;
      if (line.startsWith('OK ')) p.resolve(line.slice(3).split(',').map(Number));
      else p.reject(new Error(line.replace(/^ERR\s*/, '') || 'spk error'));
    });
  });
  const cleanup = () => { proc = null; ready = null; rl = null; if (pending) { pending.reject(new Error('spk beendet')); pending = null; } };
  proc.on('exit', cleanup); proc.on('error', cleanup);
  return ready;
}

// Extract the voiceprint for a 16 kHz mono wav file.
// Calls are serialized through `chain`, but a FAILED call must not poison it:
// .then() on a rejected promise skips the callback, so ONE crash used to make
// every later embed() fail instantly with the same "spk beendet" — that is why
// speaker auto-naming was dead for a whole recording after the first bad cut
// (root-caused 2026-07-27). The queue is continued from a swallowed copy.
// NOTE for callers: keep cuts short. onnxruntime ABORTS the helper process on
// very long inputs — reproduced at 180 s ("Non-zero status code returned while
// running Where"), fine at 20/60 s.
export function embed(wavPath) {
  const run = chain.then(async () => {
    await start();
    return await new Promise((res, rej) => { pending = { resolve: res, reject: rej }; proc.stdin.write(wavPath + '\n'); });
  });
  chain = run.then(() => {}, () => {});
  return run;
}

export function cosine(a, b) {
  let d = 0, na = 0, nb = 0;
  for (let i = 0; i < a.length; i++) { d += a[i] * b[i]; na += a[i] * a[i]; nb += b[i] * b[i]; }
  return d / (Math.sqrt(na) * Math.sqrt(nb) + 1e-9);
}

// Beste PERSON für ein Embedding, oder null unter der Schwelle.
//
// Bewertet wird pro Person mit ihrem BESTEN Abdruck (Maximum, kein Mittel): die
// Proben einer Person stammen aus verschiedenen Sessions und Headsets, und ein
// Mittel über sie verwischt genau das, was die Stimme ausmacht. `second` ist der
// beste Wert einer ANDEREN Person — nur damit ist ein Abstands-Kriterium
// überhaupt sinnvoll. Vorher lief der Abstand über einzelne ABDRÜCKE, und zwei
// Abdrücke derselben Person (Kosinus 0,928) haben ihn zuverlässig gerissen: die
// Zwangsbedingung in diar.js war dadurch faktisch abgeschaltet.
export function bestMatch(emb, voiceprints, threshold = THRESHOLD) {
  const perPerson = new Map();   // Identität → bester Kosinus
  const meta = new Map();
  for (const v of voiceprints) {
    const key = v.person_id != null ? `p${v.person_id}` : `n:${String(v.name || '').toLowerCase()}`;
    const s = cosine(emb, v.vec);
    if (!perPerson.has(key) || s > perPerson.get(key)) {
      perPerson.set(key, s);
      meta.set(key, { personId: v.person_id ?? null, name: v.name, printId: v.id });
    }
  }
  let bestKey = null, s1 = -Infinity, s2 = -Infinity;
  for (const [key, s] of perPerson) {
    if (s > s1) { s2 = s1; s1 = s; bestKey = key; }
    else if (s > s2) s2 = s;
  }
  if (bestKey === null || s1 < threshold) return null;
  const m = meta.get(bestKey);
  return { id: m.printId, personId: m.personId, key: bestKey, name: m.name, score: s1, second: s2 };
}

export function status() { return { model: EMB, modelId: modelId(), present: fs.existsSync(EMB), running: !!proc, threshold: THRESHOLD }; }

// Beendet den TitaNet-Helfer und wartet auf sein Ende (siehe gate.freeRam: hält
// in der Spitze 0,7 GB). Startet beim nächsten embed() faul neu. Wird NUR
// zwischen den Phasen gerufen, nie während der Diarisierung.
export function stop() {
  const p = proc;
  proc = null; ready = null; rl = null;
  return gate.killAndWait(p);
}