import express from 'express'; import path from 'node:path'; import fs from 'node:fs'; import { execFile } from 'node:child_process'; import { promisify } from 'node:util'; import { fileURLToPath } from 'node:url'; import * as bt from './bt.js'; import * as rec from './recorder.js'; import * as db from './db.js'; import * as llm from './llm.js'; import * as diar from './diar.js'; import * as spk from './spk.js'; const pexec = promisify(execFile); // Glossary → { soundalike(lowercased): correctTerm }. Only entries WITH // soundalikes get a deterministic swap; term-only entries are left to the LLM hint. function glossaryMap(glossary) { const m = {}; for (const g of glossary || []) { if (!g.term || !g.soundalikes) continue; for (const s of String(g.soundalikes).split(',')) { const k = s.trim().toLowerCase(); if (k) m[k] = String(g.term).trim(); } } return m; } // Whole-word replace (Unicode-aware, case-insensitive match): swaps any token // equal to a known soundalike for its correct term, leaving punctuation intact. function applyGlossaryText(text, m) { if (!text || !m || !Object.keys(m).length) return text; return String(text).replace(/[\p{L}\p{N}]+/gu, (tok) => m[tok.toLowerCase()] || tok); } // Build the LLM input from stored words. If acoustic diarization tagged speakers, // produce a labelled dialogue ("Sprecher 1: …") so the LLM keeps real speakers // instead of guessing; otherwise the plain transcript. function buildBasis(rec, nameMap = {}) { const words = rec.words || []; const hasSpk = words.some(w => w.speaker); if (!hasSpk) return { text: rec.transcript || '', labeled: false }; const turns = diar.toTurns(words); const map = {}; let n = 0; const label = sp => nameMap[sp] || (map[sp] || (map[sp] = `Sprecher ${++n}`)); const text = turns.map(t => `${label(t.speaker)}: ${t.text}`).join('\n'); return { text, labeled: true }; } const REC_DIR2 = path.join(path.dirname(fileURLToPath(import.meta.url)), 'recordings'); // Match each diarization cluster against enrolled voiceprints → { speaker: name }. // Cuts the speaker's longest contiguous run from the kept wav, embeds it, cosine- // matches. Returns only confident matches (others stay "Sprecher N"). async function identifySpeakers(recData) { const words = recData.words || []; const speakers = [...new Set(words.filter(w => w.speaker).map(w => w.speaker))]; 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(); if (!vps.length) return {}; const best = {}; let cur = null; for (const w of words) { if (!cur || cur.speaker !== w.speaker) cur = { speaker: w.speaker, start: w.t_start_ms, end: w.t_end_ms }; else cur.end = w.t_end_ms; const len = cur.end - cur.start; if (!best[cur.speaker] || len > best[cur.speaker].len) best[cur.speaker] = { start: cur.start, end: cur.end, len }; } const map = {}; for (const sp of speakers) { const r = best[sp]; if (!r) continue; const seg = `/tmp/idseg_${recData.id}_${sp}.wav`; try { await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(r.start / 1000), '-to', String(r.end / 1000), '-ar', '16000', '-ac', '1', seg]); const emb = await spk.embed(seg); const m = spk.bestMatch(emb, vps); if (m) map[sp] = m.name; } catch (e) { console.error('identify', sp, e.message); } finally { try { fs.unlinkSync(seg); } catch {} } } return map; } // 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 // and a long Gemma generation hit 95-97°C uncapped on the fanless board (#70, // 16-min run) — same shutdown risk transcribeWav/rediarize already guard against. async function analyzeRecording(id) { // Free the resident rkllm (~5.9GB) BEFORE identifySpeakers: the TitaNet // spk-embed helper OOM-dies next to it ("spk beendet" per cluster, #71 got // 48 raw speakers). llm.analyze() below respawns it (~96s load, noise // against a 15-min analysis). llm.stopServer(); await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {}); try { const rec = await db.getRecording(id); if (!rec) return; const nameMap = await identifySpeakers(rec).catch(() => ({})); // Seed the per-recording speaker names from acoustic identification (fills // blanks only; manual edits stay). Manual names win for the LLM labelling. await db.mergeSpeakerNames(id, nameMap).catch(() => {}); const names = { ...nameMap, ...(rec.speaker_names || {}) }; const glossary = (await db.listGlossary().catch(() => [])).filter(g => g.enabled); // Deterministic soundalike replacement: the 3B model is unreliable at obeying // a "map Amy->Aime" instruction (it treats "Amy" as a valid name), so entries // WITH soundalikes are applied as a hard whole-word swap on the words FIRST — // then the LLM sees the correct term and the turn view shows it too. Entries // without soundalikes still rely on the (contextual) LLM glossary hint. const gmap = glossaryMap(glossary); if (Object.keys(gmap).length && Array.isArray(rec.words) && rec.words.length) { let changed = false; for (const w of rec.words) { const nt = applyGlossaryText(w.word, gmap); if (nt !== w.word) { w.word = nt; changed = true; } } if (changed) { await db.replaceWords(id, rec.words); console.log(`glossary: applied soundalike swaps to #${id} words`); } } const { text, labeled } = buildBasis(rec, names); const a = await llm.analyze(text, { labeled, glossary }); // Participants = the ACOUSTICALLY diarized speakers (named where known), // not the LLM's list — the model returned every name MENTIONED in the // conversation as a "Teilnehmer" (11 names on a 3-person call, #69). // LLM participants stay only as fallback when there is no diarization. const spkOrder = []; for (const w of (rec.words || [])) if (w.speaker && !spkOrder.includes(w.speaker)) spkOrder.push(w.speaker); if (spkOrder.length) a.participants = spkOrder.map((sp, i) => names[sp] || `Sprecher ${i + 1}`); // Belt-and-suspenders: also swap in the LLM outputs (cheap, catches any the // model reintroduced). for (const k of ['title', 'summary', 'transcriptClean']) if (a[k]) a[k] = applyGlossaryText(a[k], gmap); a.participants = (a.participants || []).map(p => applyGlossaryText(p, gmap)); a.todos = (a.todos || []).map(t => applyGlossaryText(t, gmap)); await db.updateAnalysis(id, { ...a, status: 'done' }); console.log(`analysis done for #${id}: ${a.title}`); } catch (e) { console.error(`analysis failed for #${id}:`, e.message); await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {}); } finally { await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {}); } } const __dirname = path.dirname(fileURLToPath(import.meta.url)); const app = express(); app.use(express.json()); const REC_DIR = path.join(__dirname, 'recordings'); app.use('/media', express.static(REC_DIR)); app.use(express.static(path.join(__dirname, 'web/dist'))); const wrap = fn => (req, res) => fn(req, res).catch(e => res.status(500).json({ error: String(e && e.message || e) })); app.get('/api/health', (req, res) => res.json({ ok: true, ...rec.status() })); // --- Bluetooth --- app.post('/api/bt/scan', wrap(async (req, res) => { const secs = Math.min(+(req.body && req.body.seconds) || 10, 30); res.json(await bt.scan(secs)); })); app.get('/api/bt/devices', wrap(async (req, res) => res.json(await bt.knownDevices()))); app.post('/api/bt/pair', wrap(async (req, res) => res.json(await bt.pair(req.body.mac)))); app.post('/api/bt/connect', wrap(async (req, res) => res.json(await bt.connect(req.body.mac)))); app.post('/api/bt/repair', wrap(async (req, res) => res.json(await bt.repair(req.body.mac)))); app.post('/api/bt/disconnect', wrap(async (req, res) => res.json(await bt.disconnect(req.body.mac)))); // --- Audio --- app.get('/api/audio/sources', wrap(async (req, res) => res.json(await rec.listSources()))); app.post('/api/audio/headset-mic', wrap(async (req, res) => res.json(await rec.enableHeadsetMic((req.body && req.body.codec) || 'auto')))); // --- Global NPU pipeline serializer --------------------------------------- // Transcription (whisper-turbo ENCODER on the NPU) and LLM analysis (rkllm on // the NPU) cannot run simultaneously: while the whisper encoder holds the NPU, // rkllm_init fails with "failed to allocate handle, ret:-1, errno:14, Bad // address", which surfaces as "rkllm READY timeout" and a failed/empty // analysis. This bit when two recordings finished within seconds of each other // (both fired transcribe+analyze fire-and-forget with no coordination). Every // heavy per-recording job (transcribe / analyze / re-diarize) now runs through // this single FIFO queue, so exactly one touches the NPU at a time; a second // recording simply waits its turn instead of colliding. let npuChain = Promise.resolve(); let npuDepth = 0; function withNpuLock(label, fn) { npuDepth++; const waiting = npuDepth - 1; const run = npuChain.then(async () => { if (waiting > 0) console.log(`npu-queue: ${label} start (${waiting} ahead done)`); try { return await fn(); } finally { npuDepth--; } }); npuChain = run.catch(() => {}); // keep the chain alive across a job's failure return run; } // --- Recording --- app.post('/api/record/start', wrap(async (req, res) => res.json(await rec.startRecording(req.body || {})))); // 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); await db.setTranscript(id, transcript, words); console.log(`transcription done for #${id}: ${words.length} segments`); } catch (e) { console.error(`transcription failed for #${id}:`, e.message); await db.setStatus(id, 'error').catch(() => {}); // also mark analysis 'error' so the UI stops polling and shows a retry await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {}); return; // no transcript -> skip LLM analysis } analyzeRecording(id); } app.post('/api/record/stop', wrap(async (req, res) => { // Persist wav + mp3 + DB row FIRST (crash-safe), respond immediately, then // transcribe/analyze in the background. The frontend already polls while // analysis_status === 'pending', so the transcript + analysis fill in live. const r = await rec.stopAndSave(); const id = await db.addRecording({ ...r, transcript: '', status: 'transcribing' }); res.json({ id, title: r.title, mp3_file: r.mp3_file, duration_ms: r.duration_ms, transcript: '', status: 'transcribing', analysis_status: 'pending' }); withNpuLock(`transcribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, r.numSpeakers)); })); // --- Recordings --- app.get('/api/recordings', wrap(async (req, res) => res.json(await db.listRecordings()))); app.get('/api/recordings/:id', wrap(async (req, res) => { const r = await db.getRecording(+req.params.id); r ? res.json(r) : res.status(404).json({ error: 'not found' }); })); app.delete('/api/recordings/:id', wrap(async (req, res) => { await db.deleteRecording(+req.params.id); res.json({ ok: true }); })); // --- 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 }))); })); 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 })))); app.post('/api/enroll/stop', wrap(async (req, res) => { const name = (req.body && req.body.name || '').trim(); if (!name) return res.status(400).json({ error: 'Name fehlt' }); const wav = await rec.stopRaw(); try { const emb = await spk.embed(wav); const id = await db.addVoiceprint(name, emb); res.json({ id, name, dim: emb.length }); } finally { try { fs.unlinkSync(wav); } catch {} } })); // --- Glossary (known proper nouns / terms; fed into LLM correction) --- app.get('/api/glossary', wrap(async (req, res) => res.json(await db.listGlossary()))); app.post('/api/glossary', wrap(async (req, res) => { const term = (req.body && req.body.term || '').trim(); const soundalikes = (req.body && req.body.soundalikes || '').trim(); if (!term) return res.status(400).json({ error: 'Begriff fehlt' }); res.json(await db.addGlossary(term, soundalikes)); })); app.delete('/api/glossary/:id', wrap(async (req, res) => { await db.deleteGlossary(+req.params.id); res.json({ ok: true }); })); app.put('/api/glossary/:id', wrap(async (req, res) => { const term = (req.body && req.body.term || '').trim(); const soundalikes = (req.body && req.body.soundalikes || '').trim(); if (!term) return res.status(400).json({ error: 'Begriff fehlt' }); const g = await db.updateGlossary(+req.params.id, term, soundalikes); if (!g) return res.status(404).json({ error: 'Eintrag nicht gefunden' }); res.json(g); })); // --- LLM --- app.get('/api/llm/status', (req, res) => res.json(llm.status())); app.get('/api/spk/status', (req, res) => res.json(spk.status())); // Re-diarize a recording's words with an (optional) fixed speaker count, then // persist the new speaker labels. Text/timestamps are untouched. Needs the wav. async function rediarize(id, numSpeakers) { const r = await db.getRecording(id); if (!r || !r.wav_file) throw new Error('wav für Re-Diarisierung fehlt'); const wav = path.join(REC_DIR, r.wav_file); if (!fs.existsSync(wav)) throw new Error('wav-Datei nicht gefunden: ' + r.wav_file); const words = r.words || []; // Free the resident rkllm first — diar's consolidation embeds every cluster // via TitaNet, which OOM-dies next to a resident 5.9GB LLM (see // analyzeRecording; on #71 that silently yielded 48 raw speakers). llm.stopServer(); // Diarization runs on the A76 cores; the board is fanless, so cap the clock // for the whole run (transcribeWav does this for the record path — rediarize // 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); diar.assignSpeakers(words, segs); await db.replaceWords(id, words); 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(() => {}); } } // Re-run analysis (e.g. the model wasn't downloaded yet at record time). Optional // body { numSpeakers } re-diarizes first (fixes over-segmentation on a known count). app.post('/api/recordings/:id/reanalyze', wrap(async (req, res) => { const id = +req.params.id; const r = await db.getRecording(id); if (!r) return res.status(404).json({ error: 'not found' }); const numSpeakers = +(req.body && req.body.numSpeakers) || 0; await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' }); await db.markAnalysisStart(id).catch(() => {}); res.json({ ok: true, status: 'pending', reDiarize: true }); // Always re-diarize: numSpeakers>0 pins the count, 0 = auto (now consolidated, // so auto no longer over-segments). This is the "fix my speakers" path. withNpuLock(`reanalyze #${id}`, async () => { try { await rediarize(id, numSpeakers); } catch (e) { console.error(`re-diarize failed for #${id}:`, e.message); } await analyzeRecording(id); }); })); // 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). app.post('/api/recordings/:id/reanalyze-llm', wrap(async (req, res) => { const id = +req.params.id; const r = await db.getRecording(id); if (!r) return res.status(404).json({ error: 'not found' }); await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' }); await db.markAnalysisStart(id).catch(() => {}); res.json({ ok: true, status: 'pending', reDiarize: false }); withNpuLock(`reanalyze-llm #${id}`, async () => { await analyzeRecording(id); }); })); // Re-run transcription from the saved wav (recovery: a crash/thermal-shutdown // during whisper leaves the row at status 'transcribing' with an empty // transcript, or 'error'; the wav is on disk). Optional { numSpeakers }. app.post('/api/recordings/:id/retranscribe', wrap(async (req, res) => { const id = +req.params.id; const r = await db.getRecording(id); if (!r) return res.status(404).json({ error: 'not found' }); if (!r.wav_file) return res.status(400).json({ error: 'kein wav für Re-Transkription' }); const numSpeakers = +(req.body && req.body.numSpeakers) || 0; await db.setStatus(id, 'transcribing'); await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' }); res.json({ ok: true, status: 'transcribing' }); withNpuLock(`retranscribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, numSpeakers)); })); // Longest contiguous run of one diarization cluster (best clean sample of a voice). function longestRun(words, speaker) { let best = null, cur = null; for (const w of words || []) { if (w.speaker !== speaker) { cur = null; continue; } if (!cur) cur = { start: w.t_start_ms, end: w.t_end_ms }; else cur.end = w.t_end_ms; if (!best || (cur.end - cur.start) > (best.end - best.start)) best = { start: cur.start, end: cur.end }; } return best; } // Name a diarized speaker in a recording and (default) LEARN their voice: fold // the speaker's longest clean segment into the named voiceprint centroid, so // future recordings auto-identify them. Empty name clears the label. app.post('/api/recordings/:id/speaker', wrap(async (req, res) => { const id = +req.params.id; const speaker = String((req.body && req.body.speaker) || '').trim(); const name = String((req.body && req.body.name) || '').trim(); const enroll = !(req.body && req.body.enroll === false); if (!speaker) return res.status(400).json({ error: 'speaker fehlt' }); const r = await db.getRecording(id); if (!r) return res.status(404).json({ error: 'not found' }); await db.setSpeakerName(id, speaker, name); let enrolled = false, learnMsg = null; if (enroll && name && r.wav_file) { const seg = longestRun(r.words, speaker); const wavPath = path.join(REC_DIR, r.wav_file); if (!seg || (seg.end - seg.start) < 1500) { learnMsg = 'Segment zu kurz zum Lernen (<1,5s)'; } else if (!fs.existsSync(wavPath)) { learnMsg = 'wav fehlt – nicht gelernt'; } else { const cut = `/tmp/enr_${id}_${speaker}.wav`; try { 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); enrolled = true; } catch (e) { console.error('enroll-from-recording failed:', e.message); learnMsg = 'Lernen fehlgeschlagen'; } finally { try { fs.unlinkSync(cut); } catch {} } } } const r2 = await db.getRecording(id); res.json({ ok: true, speaker_names: r2.speaker_names || {}, enrolled, learnMsg }); })); app.get(/^(?!\/api|\/media).*/, (req, res) => res.sendFile(path.join(__dirname, 'web/dist/index.html'))); const PORT = process.env.PORT || 3000; db.ensureSchema() .catch(e => console.error('ensureSchema:', e.message)) .finally(() => app.listen(PORT, '0.0.0.0', () => console.log(`callrec listening on :${PORT}`)));