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