1import {spawn, type ChildProcess} from 'node:child_process'
2import * as fs from 'node:fs'
3import * as path from 'node:path'
4import {mergeTurns, renderJson, renderMarkdown, type Item} from './render.js'
5import type {WavWriter} from './wav.js'
7// Live transcription during recording (the record command's --transcribe).
8//
9// A single persistent python process loads faster-whisper ONCE and then
10// serves transcription requests over a line-delimited JSON stdin/stdout
11// protocol; each request names a WAV file and a frame range, which the helper
12// reads raw from disk (only flushed bytes are ever requested, so reading a
13// file that is still being appended to is safe — the header is bypassed).
14//
15// The recorder feeds every decoded audio frame through onAudio, which tracks
16// silence so chunks can be cut at natural pauses (>= 300 ms below the RMS
17// threshold) — never mid-word. Every tick, each segment with >= MIN_CHUNK of
18// speech up to a silence cut is dispatched (or force-cut at MAX_CHUNK of
19// unbroken speech). transcript.md / transcript.json are re-rendered after
20// every result, so the transcript grows while the meeting is happening.
21//
22// If the helper dies, recording is NEVER affected: live transcription
23// disables itself with a log line, and `transcribe` can be run afterwards.
25// Tuned for ~8 s from end-of-utterance to text on screen: pause detect
26// (~0.3 s) + WAV flush (~0.5 s avg) + tick (~1.5 s avg) + ASR (1-4 s with
27// `small` on CPU) + render debounce (0.5 s). Smaller chunks trade a little
28// per-chunk context/efficiency for latency; turn merging glues the text back.
29const QUIET_RMS = 300 // ~ -41 dBFS: below this a 10 ms frame counts as quiet
30const QUIET_CUT_MS = 300 // this much consecutive quiet = a safe cut point
31const TICK_MS = 3000
32const MIN_CHUNK_SEC = 2 // don't bother the model with less than this
33const MAX_CHUNK_SEC = 30 // force a cut after this much unbroken speech
34const FINISH_TIMEOUT_MS = 180000
36interface AsrSegment {
37 start: number
38 end: number
39 text: string
40}
42const LIVE_PY = `
43import sys, json
44import numpy as np
45from faster_whisper import WhisperModel
46model_name = sys.argv[1]
47language = None if sys.argv[2] == "-" else sys.argv[2]
48model = WhisperModel(model_name, device="auto", compute_type="auto")
49print(json.dumps({"ready": True}), flush=True)
50while True:
51 # readline, not iteration: iterating sys.stdin read-ahead-buffers and can
52 # sit on a complete line without yielding it
53 line = sys.stdin.readline()
54 if not line:
55 break
56 req = json.loads(line)
57 ch = req["channels"]
58 with open(req["wav"], "rb") as f:
59 f.seek(44 + req["startFrame"] * ch * 2)
60 raw = f.read((req["endFrame"] - req["startFrame"]) * ch * 2)
61 a = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0
62 if ch > 1:
63 a = a.reshape(-1, ch).mean(axis=1)
64 rate = req["rate"]
65 if rate != 16000:
66 if rate % 16000 == 0:
67 k = rate // 16000
68 n = (len(a) // k) * k
69 a = a[:n].reshape(-1, k).mean(axis=1)
70 else:
71 xi = np.arange(0, len(a), rate / 16000.0)
72 a = np.interp(xi, np.arange(len(a)), a)
73 segs, info = model.transcribe(a.astype(np.float32), language=language, vad_filter=True)
74 out = [{"start": s.start, "end": s.end, "text": s.text} for s in segs]
75 print(json.dumps({"id": req["id"], "segments": out}), flush=True)
76`
78interface AsrRequest {
79 wav: string
80 rate: number
81 channels: number
82 startFrame: number
83 endFrame: number
84}
86/** The persistent faster-whisper helper process. */
87class PythonAsr {
88 private proc: ChildProcess
89 private pending = new Map<
90 number,
91 {resolve: (segs: AsrSegment[]) => void; reject: (err: Error) => void}
92 >()
93 private nextId = 1
94 private buf = ''
95 private readyResolve!: () => void
96 private readyReject!: (err: Error) => void
97 readonly ready: Promise<void>
98 dead = false
100 constructor(model: string, language: string | null, onLog: (l: string) => void) {
101 this.ready = new Promise((resolve, reject) => {
102 this.readyResolve = resolve
103 this.readyReject = reject
104 })
105 // NB: `python3 -c <script>` — NOT `python3 -` + script on stdin, which
106 // would read stdin to EOF before executing and never see our requests.
107 // With -c, sys.argv is ['-c', model, language].
108 this.proc = spawn('python3', ['-c', LIVE_PY, model, language ?? '-'], {
109 stdio: ['pipe', 'pipe', 'pipe']
110 })
111 this.proc.stdout!.on('data', d => {
112 this.buf += d
113 const lines = this.buf.split('\n')
114 this.buf = lines.pop() ?? ''
115 for (const line of lines) {
116 if (!line.trim()) continue
117 let msg: {ready?: boolean; id?: number; segments?: AsrSegment[]}
118 try {
119 msg = JSON.parse(line)
120 } catch {
121 continue
122 }
123 if (msg.ready) {
124 this.readyResolve()
125 continue
126 }
127 if (typeof msg.id === 'number') {
128 const p = this.pending.get(msg.id)
129 this.pending.delete(msg.id)
130 p?.resolve(msg.segments ?? [])
131 }
132 }
133 })
134 let errTail = ''
135 this.proc.stderr!.on('data', d => {
136 const line = String(d).trim()
137 if (line) errTail = line.slice(0, 300)
138 })
139 this.proc.on('error', err => this.die(err.message))
140 this.proc.on('exit', code => {
141 if (!this.dead && code !== 0) {
142 this.die(`helper exited with code ${code}${errTail ? ` (${errTail})` : ''}`)
143 onLog(`live transcription disabled: ${errTail || `helper exited (${code})`}`)
144 }
145 })
146 }
148 private die(reason: string) {
149 if (this.dead) return
150 this.dead = true
151 this.readyReject(new Error(reason))
152 // ready may already be resolved; a handled rejection after that is fine
153 this.ready.catch(() => undefined)
154 for (const p of this.pending.values()) p.reject(new Error(reason))
155 this.pending.clear()
156 }
158 request(req: AsrRequest): Promise<AsrSegment[]> {
159 if (this.dead) return Promise.reject(new Error('helper is dead'))
160 const id = this.nextId++
161 return new Promise((resolve, reject) => {
162 this.pending.set(id, {resolve, reject})
163 this.proc.stdin!.write(JSON.stringify({id, ...req}) + '\n')
164 })
165 }
167 close() {
168 this.dead = true
169 try {
170 this.proc.stdin!.end()
171 } catch {
172 /* ignore */
173 }
174 setTimeout(() => {
175 try {
176 this.proc.kill('SIGKILL')
177 } catch {
178 /* ignore */
179 }
180 }, 2000).unref()
181 }
182}
184interface LiveSeg {
185 file: string // relative, as in the manifest
186 absPath: string
187 name: string
188 rate: number
189 channels: number
190 startMs: number
191 writer: WavWriter | null // null once the segment has ended
192 /** Frames handed to ASR (or skipped as silence) so far. */
193 transcribedUpTo: number
194 /** Latest safe cut point (end of a >= 300 ms quiet stretch). */
195 lastCutFrame: number
196 lastEndFrame: number
197 quietMs: number
198 /** Any non-quiet audio since transcribedUpTo? */
199 hasSpeech: boolean
200 utterances: AsrSegment[] // in-file seconds
201 ended: boolean
202 endFrames: number
203 pendingOps: number
204 cacheWritten: boolean
205}
207export interface LiveTranscriberOptions {
208 outDir: string
209 room: string
210 model: string | null
211 language: string | null
212 startedAtMs: number
213 onLog: (line: string) => void
214}
216export class LiveTranscriber {
217 private asr: PythonAsr
218 private segs = new Map<string, LiveSeg>()
219 private eventItems: Item[] = []
220 private inFlight = new Set<Promise<unknown>>()
221 private timer: ReturnType<typeof setInterval> | null = null
222 private renderTimer: ReturnType<typeof setTimeout> | null = null
223 private model: string
225 constructor(private opts: LiveTranscriberOptions) {
226 this.model = opts.model ?? 'small'
227 this.asr = new PythonAsr(this.model, opts.language, opts.onLog)
228 opts.onLog(`live transcription: loading model ${this.model}...`)
229 void this.asr.ready.then(
230 () => opts.onLog('live transcription ready'),
231 () => undefined // logged by PythonAsr
232 )
233 this.timer = setInterval(() => this.tick(), TICK_MS)
234 }
236 onSegmentStart(
237 file: string,
238 absPath: string,
239 name: string,
240 rate: number,
241 channels: number,
242 startMs: number,
243 writer: WavWriter
244 ) {
245 this.segs.set(file, {
246 file,
247 absPath,
248 name,
249 rate,
250 channels,
251 startMs,
252 writer,
253 transcribedUpTo: 0,
254 lastCutFrame: 0,
255 lastEndFrame: 0,
256 quietMs: 0,
257 hasSpeech: false,
258 utterances: [],
259 ended: false,
260 endFrames: 0,
261 pendingOps: 0,
262 cacheWritten: false
263 })
264 }
266 /** Called for every decoded frame batch; endFrame = writer.framesWritten
267 * AFTER appending (so padded gaps show up as position jumps). */
268 onAudio(file: string, endFrame: number, samples: Int16Array) {
269 const seg = this.segs.get(file)
270 if (!seg || seg.ended) return
271 const frames = samples.length / seg.channels
272 const startFrame = endFrame - frames
273 if (startFrame > seg.lastEndFrame) {
274 // A silence-padded gap was inserted before this batch: it is all quiet,
275 // so the start of the current batch is a safe cut point.
276 seg.quietMs = 0
277 seg.lastCutFrame = startFrame
278 }
279 seg.lastEndFrame = endFrame
280 let sumSq = 0
281 for (let i = 0; i < samples.length; i++) sumSq += samples[i]! * samples[i]!
282 const rms = Math.sqrt(sumSq / samples.length)
283 if (rms < QUIET_RMS) {
284 seg.quietMs += (frames / seg.rate) * 1000
285 if (seg.quietMs >= QUIET_CUT_MS) seg.lastCutFrame = endFrame
286 } else {
287 seg.quietMs = 0
288 seg.hasSpeech = true
289 }
290 }
292 onSegmentEnd(file: string) {
293 const seg = this.segs.get(file)
294 if (!seg || seg.ended) return
295 seg.ended = true
296 seg.endFrames = seg.writer?.framesWritten ?? seg.lastEndFrame
297 seg.writer = null
298 this.dispatch(seg, seg.endFrames)
299 this.maybeWriteCache(seg)
300 }
302 /** Chat / join / left items from the recorder, for the rendered timeline. */
303 onEvent(item: Item) {
304 this.eventItems.push(item)
305 this.scheduleRender()
306 }
308 private tick() {
309 for (const seg of this.segs.values()) {
310 if (seg.ended || !seg.writer) continue
311 const flushed = seg.writer.flushedFrames
312 const cut = Math.min(seg.lastCutFrame, flushed)
313 if (cut - seg.transcribedUpTo >= MIN_CHUNK_SEC * seg.rate) {
314 this.dispatch(seg, cut)
315 } else if (flushed - seg.transcribedUpTo >= MAX_CHUNK_SEC * seg.rate) {
316 this.dispatch(seg, flushed) // unbroken speech: cut anyway
317 }
318 }
319 }
321 private dispatch(seg: LiveSeg, to: number) {
322 const from = seg.transcribedUpTo
323 if (to <= from) return
324 seg.transcribedUpTo = to
325 const hadSpeech = seg.hasSpeech
326 seg.hasSpeech = false
327 if (!hadSpeech || this.asr.dead) return
328 seg.pendingOps++
329 const op = this.asr
330 .request({
331 wav: seg.absPath,
332 rate: seg.rate,
333 channels: seg.channels,
334 startFrame: from,
335 endFrame: to
336 })
337 .then(segments => {
338 for (const s of segments) {
339 const text = s.text.trim()
340 if (!text) continue
341 seg.utterances.push({
342 start: s.start + from / seg.rate,
343 end: s.end + from / seg.rate,
344 text
345 })
346 }
347 if (segments.length > 0) this.scheduleRender()
348 })
349 .catch(() => undefined) // helper death is logged once by PythonAsr
350 .finally(() => {
351 seg.pendingOps--
352 this.inFlight.delete(op)
353 this.maybeWriteCache(seg)
354 })
355 this.inFlight.add(op)
356 }
358 /** Once a segment has ended and drained, persist its ASR results in the
359 * same asr/*.json format the offline transcribe subcommand uses/caches. */
360 private maybeWriteCache(seg: LiveSeg) {
361 if (!seg.ended || seg.pendingOps > 0 || seg.cacheWritten) return
362 seg.cacheWritten = true
363 if (this.asr.dead && seg.utterances.length === 0) return // let offline redo it
364 const asrDir = path.join(this.opts.outDir, 'asr')
365 try {
366 fs.mkdirSync(asrDir, {recursive: true})
367 fs.writeFileSync(
368 path.join(asrDir, path.basename(seg.file).replace(/\.wav$/i, '') + '.json'),
369 JSON.stringify(
370 {
371 engine: 'faster-whisper',
372 model: this.model,
373 segments: [...seg.utterances].sort((a, b) => a.start - b.start)
374 },
375 null,
376 2
377 ) + '\n'
378 )
379 } catch {
380 /* ignore */
381 }
382 }
384 private scheduleRender() {
385 if (this.renderTimer !== null) return
386 this.renderTimer = setTimeout(() => {
387 this.renderTimer = null
388 this.render(null)
389 }, 500)
390 }
392 private render(endedAtMs: number | null) {
393 const items: Item[] = [...this.eventItems]
394 for (const seg of this.segs.values()) {
395 for (const u of seg.utterances) {
396 items.push({
397 timeMs: seg.startMs + u.start * 1000,
398 endMs: seg.startMs + u.end * 1000,
399 type: 'speech',
400 speaker: seg.name,
401 text: u.text
402 })
403 }
404 }
405 const merged = mergeTurns(items)
406 const meta = {
407 room: this.opts.room,
408 startedAtMs: this.opts.startedAtMs,
409 endedAtMs,
410 speakers: [...new Set([...this.segs.values()].map(s => s.name))],
411 engine: 'faster-whisper (live)',
412 model: this.model
413 }
414 try {
415 fs.writeFileSync(
416 path.join(this.opts.outDir, 'transcript.md'),
417 renderMarkdown(meta, merged)
418 )
419 fs.writeFileSync(
420 path.join(this.opts.outDir, 'transcript.json'),
421 renderJson(meta, merged)
422 )
423 } catch {
424 /* ignore */
425 }
426 }
428 /** All segments have been ended by the recorder; drain and finalize. */
429 async finish(endedAtMs: number): Promise<void> {
430 if (this.timer !== null) clearInterval(this.timer)
431 this.timer = null
432 if (this.renderTimer !== null) clearTimeout(this.renderTimer)
433 this.renderTimer = null
434 if (this.inFlight.size > 0) {
435 this.opts.onLog(
436 `waiting for ${this.inFlight.size} transcription chunk(s) to finish...`
437 )
438 await Promise.race([
439 Promise.allSettled([...this.inFlight]),
440 new Promise(r => setTimeout(r, FINISH_TIMEOUT_MS))
441 ])
442 }
443 this.asr.close()
444 for (const seg of this.segs.values()) this.maybeWriteCache(seg)
445 this.render(endedAtMs)
446 this.opts.onLog(`wrote ${path.join(this.opts.outDir, 'transcript.md')}`)
447 }
448}