diff --git a/cmd/transcriptor-ai/main.go b/cmd/transcriptor-ai/main.go index dd9b0aa..7dcd6e8 100644 --- a/cmd/transcriptor-ai/main.go +++ b/cmd/transcriptor-ai/main.go @@ -1,5 +1,5 @@ -// Command transcriptor-ai orchestrates local, real-time meeting transcription: -// it drives audiotee for audio capture, qwen-asr for speech recognition, and +// Command transcriptor-ai orchestrates local meeting transcription: it +// drives audiotee for audio capture, qwen-asr for speech recognition, and // serves the resulting transcript live over SSE. See CLAUDE.md for the full // architecture and the reasoning behind it. // @@ -11,6 +11,11 @@ // CLAUDE.md on why SSE, not WebSocket, and why one shared worker rather // than one per track). Post-processing (fuzzy glossary correction, LLM // reread) isn't wired in yet. +// +// Capture (and its raw PCM recording, see -recording-dir) always runs; +// -realtime=false skips VAD/ASR/SSE and only records, for later batch +// transcription via the "transcribe" subcommand (see transcribe.go) against +// the resulting .pcm files. package main import ( @@ -32,6 +37,13 @@ import ( ) func main() { + if len(os.Args) > 1 && os.Args[1] == "transcribe" { + if err := runTranscribe(os.Args[2:]); err != nil { + fmt.Fprintln(os.Stderr, "transcriptor-ai transcribe:", err) + os.Exit(1) + } + return + } if err := run(); err != nil { fmt.Fprintln(os.Stderr, "transcriptor-ai:", err) os.Exit(1) @@ -40,58 +52,79 @@ func main() { func run() error { audioteeBinary := flag.String("audiotee-binary", "", "path to the audiotee binary (optional; defaults to PATH then ~/bin/audiotee — see internal/capture)") - asrBinary := flag.String("asr-binary", "", "path to the qwen_asr binary (required)") - asrModelDir := flag.String("asr-model-dir", "", "path to the qwen_asr model directory (required)") + asrBinary := flag.String("asr-binary", "", "path to the qwen_asr binary (required if -realtime is enabled)") + asrModelDir := flag.String("asr-model-dir", "", "path to the qwen_asr model directory (required if -realtime is enabled)") prompt := flag.String("prompt", "", "glossary biasing prompt passed to qwen_asr") httpAddr := flag.String("http-addr", ":8420", "address to serve the live transcript SSE endpoint on") transcriptFile := flag.String("transcript-file", "", "path to append the live transcript to as plain text (optional; not written if empty)") captureSystem := flag.Bool("capture-system", true, "transcribe the system audio track (audiotee's system tap always runs regardless of this flag — disabling it only stops transcribing that track, see internal/capture.Config.CaptureMic doc)") captureMic := flag.Bool("capture-mic", true, "capture and transcribe the microphone track") + recordingDir := flag.String("recording-dir", "", "directory to save this session's raw PCM recording to (system/mic .pcm files, named by session start time); defaults to the OS temp dir if empty") + realtime := flag.Bool("realtime", true, "run live VAD+ASR+SSE transcription while capturing; disable to only capture and record audio, for later batch transcription via the 'transcribe' subcommand") flag.Parse() - if *asrBinary == "" || *asrModelDir == "" { - return fmt.Errorf("both -asr-binary and -asr-model-dir are required") - } if !*captureSystem && !*captureMic { return fmt.Errorf("at least one of -capture-system or -capture-mic must be enabled") } + if *realtime && (*asrBinary == "" || *asrModelDir == "") { + return fmt.Errorf("both -asr-binary and -asr-model-dir are required when -realtime is enabled") + } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - systemPath := filepath.Join(os.TempDir(), "transcriptor-ai-system.pcm") - micPath := filepath.Join(os.TempDir(), "transcriptor-ai-mic.pcm") + dir := *recordingDir + if dir == "" { + dir = os.TempDir() + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("create recording directory: %w", err) + } + sessionStamp := time.Now().Format("2006-01-02_15-04-05") + systemPath := filepath.Join(dir, sessionStamp+"-system.pcm") systemOut, err := os.Create(systemPath) if err != nil { return fmt.Errorf("create system output file: %w", err) } defer systemOut.Close() - micOut, err := os.Create(micPath) - if err != nil { - return fmt.Errorf("create mic output file: %w", err) + // Only opened when the mic track is actually captured — otherwise + // audiotee never emits TrackMic chunks and this would just be an empty, + // misleading file (see internal/capture.Config.CaptureMic doc). + var micOut *os.File + var micPath string + if *captureMic { + micPath = filepath.Join(dir, sessionStamp+"-mic.pcm") + micOut, err = os.Create(micPath) + if err != nil { + return fmt.Errorf("create mic output file: %w", err) + } + defer micOut.Close() } - defer micOut.Close() var fileWriter *transcript.FileWriter - if *transcriptFile != "" { - fileWriter, err = transcript.NewFileWriter(*transcriptFile) - if err != nil { - return err - } - defer fileWriter.Close() - fmt.Fprintln(os.Stderr, "transcript file ->", *transcriptFile) - } - - broadcaster := transcript.NewBroadcaster() - httpServer := &http.Server{Addr: *httpAddr, Handler: sseMux(broadcaster)} + var broadcaster *transcript.Broadcaster + var httpServer *http.Server httpErrs := make(chan error, 1) - go func() { - if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { - httpErrs <- err + if *realtime { + if *transcriptFile != "" { + fileWriter, err = transcript.NewFileWriter(*transcriptFile) + if err != nil { + return err + } + defer fileWriter.Close() + fmt.Fprintln(os.Stderr, "transcript file ->", *transcriptFile) } - }() + + broadcaster = transcript.NewBroadcaster() + httpServer = &http.Server{Addr: *httpAddr, Handler: sseMux(broadcaster)} + go func() { + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + httpErrs <- err + } + }() + } const sampleRate = 16000 c := capture.New(capture.Config{SampleRate: sampleRate, AudioteePath: *audioteeBinary, CaptureMic: *captureMic}) @@ -101,11 +134,19 @@ func run() error { fmt.Fprintln(os.Stderr, "capturing... Ctrl+C to stop") fmt.Fprintln(os.Stderr, "system track ->", systemPath) - fmt.Fprintln(os.Stderr, "mic track ->", micPath) - fmt.Fprintln(os.Stderr, "live transcript (SSE) -> curl -N http://"+httpAddrForDisplay(*httpAddr)+"/events") + if micOut != nil { + fmt.Fprintln(os.Stderr, "mic track ->", micPath) + } + if *realtime { + fmt.Fprintln(os.Stderr, "live transcript (SSE) -> curl -N http://"+httpAddrForDisplay(*httpAddr)+"/events") + } else { + fmt.Fprintln(os.Stderr, "realtime transcription disabled — capturing only, transcribe later with the 'transcribe' subcommand") + } - // Tee raw chunks to disk (for listening back later) while also feeding - // them to VAD. Buffered so the tee doesn't block segmentation. + // Tee raw chunks to disk (the durable recording — batch transcription + // input, see the 'transcribe' subcommand, and originally just "listening + // back later" before that existed) while also feeding them to VAD when + // realtime is on. Buffered so the tee doesn't block segmentation. tee := make(chan capture.Chunk, 64) go func() { defer close(tee) @@ -114,39 +155,52 @@ func run() error { case capture.TrackSystem: systemOut.Write(chunk.Data) case capture.TrackMic: - micOut.Write(chunk.Data) + if micOut != nil { + micOut.Write(chunk.Data) + } } tee <- chunk } }() - detector := vad.New(vad.DefaultConfig(sampleRate)) - segments := detector.Run(tee) - - // A single shared worker across both tracks, not one per track. - // qwen_asr has no quantized-weights option (BF16 only — see - // third_party/qwen-asr/README.md "Memory Requirements"); its static - // footprint alone is ~6.9GiB for the 1.7B model. Two per-track workers - // meant two concurrent qwen_asr subprocesses whenever both tracks had - // speech at once, doubling peak RSS to ~13GiB — over half the 24GB - // target machine's unified memory (see "Target hardware" in - // CLAUDE.md). Serializing onto one worker caps peak RSS to ~1 model - // instance, trading it for latency: if both tracks talk at the same - // time, one track's segment now waits for the other's to finish - // transcribing instead of running in parallel. Per-track order is - // still preserved, since `segments` (VAD's output) is itself already - // chronological and this worker drains it in that same order. - transcriber := asr.New(asr.Config{BinaryPath: *asrBinary, ModelDir: *asrModelDir, Prompt: *prompt}) - // transcribeWorker uses its own background context, not the - // shutdown-signal ctx above: segments still queued when Ctrl+C arrives - // (including the ones flushed by c.Stop() below) must still get - // transcribed during the graceful drain, not fail with "context - // canceled" because the shutdown signal already canceled ctx. - asrCtx := context.Background() + var segments <-chan vad.Segment asrSegs := make(chan vad.Segment, 16) var wg sync.WaitGroup - wg.Add(1) - go func() { defer wg.Done(); transcribeWorker(asrCtx, transcriber, asrSegs, broadcaster, fileWriter) }() + if *realtime { + detector := vad.New(vad.DefaultConfig(sampleRate)) + segments = detector.Run(tee) + + // A single shared worker across both tracks, not one per track. + // qwen_asr has no quantized-weights option (BF16 only — see + // third_party/qwen-asr/README.md "Memory Requirements"); its static + // footprint alone is ~6.9GiB for the 1.7B model. Two per-track workers + // meant two concurrent qwen_asr subprocesses whenever both tracks had + // speech at once, doubling peak RSS to ~13GiB — over half the 24GB + // target machine's unified memory (see "Target hardware" in + // CLAUDE.md). Serializing onto one worker caps peak RSS to ~1 model + // instance, trading it for latency: if both tracks talk at the same + // time, one track's segment now waits for the other's to finish + // transcribing instead of running in parallel. Per-track order is + // still preserved, since `segments` (VAD's output) is itself already + // chronological and this worker drains it in that same order. + transcriber := asr.New(asr.Config{BinaryPath: *asrBinary, ModelDir: *asrModelDir, Prompt: *prompt}) + // transcribeWorker uses its own background context, not the + // shutdown-signal ctx above: segments still queued when Ctrl+C arrives + // (including the ones flushed by c.Stop() below) must still get + // transcribed during the graceful drain, not fail with "context + // canceled" because the shutdown signal already canceled ctx. + asrCtx := context.Background() + wg.Add(1) + go func() { defer wg.Done(); transcribeWorker(asrCtx, transcriber, asrSegs, broadcaster, fileWriter) }() + } else { + // Nothing consumes VAD output in this mode — just drain the tee so + // the writer goroutine (and, upstream of it, capture itself) never + // blocks on a full channel. + go func() { + for range tee { + } + }() + } loop: for { @@ -181,19 +235,23 @@ loop: }() c.Stop() - // Drain any segments flushed by Stop() closing the chunk stream. - for seg := range segments { - logSegment(seg) - dispatch(seg, asrSegs, *captureSystem, *captureMic) + if *realtime { + // Drain any segments flushed by Stop() closing the chunk stream. + for seg := range segments { + logSegment(seg) + dispatch(seg, asrSegs, *captureSystem, *captureMic) + } } close(asrSegs) wg.Wait() // let in-flight/queued transcriptions finish before exiting, and // before their results are published — otherwise a connected SSE client // would miss the tail end of the transcript. - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - _ = httpServer.Shutdown(shutdownCtx) + if httpServer != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = httpServer.Shutdown(shutdownCtx) + } return nil } diff --git a/cmd/transcriptor-ai/transcribe.go b/cmd/transcriptor-ai/transcribe.go new file mode 100644 index 0000000..0d8dc9e --- /dev/null +++ b/cmd/transcriptor-ai/transcribe.go @@ -0,0 +1,138 @@ +package main + +import ( + "context" + "flag" + "fmt" + "io" + "os" + "sync" + "time" + + "github.com/stephanetailland/transcriptor-ai/internal/asr" + "github.com/stephanetailland/transcriptor-ai/internal/capture" + "github.com/stephanetailland/transcriptor-ai/internal/transcript" + "github.com/stephanetailland/transcriptor-ai/internal/vad" +) + +// runTranscribe implements the "transcribe" subcommand: batch, post-hoc +// transcription of a previously recorded PCM session (see -recording-dir on +// the default/live mode, which is what produces these files) — no capture, +// no live SSE, just VAD+ASR over the file(s) followed by a single transcript +// file write. Reuses transcribeWorker unchanged: reading segments off one +// channel synchronously here already serializes ASR calls the same way the +// live path's single shared worker does (see main.go's comment on why), +// just without the extra goroutine/channel indirection live mode needs to +// keep accepting new segments while a previous one is still transcribing. +func runTranscribe(args []string) error { + fs := flag.NewFlagSet("transcribe", flag.ExitOnError) + systemPCM := fs.String("system-pcm", "", "path to a previously recorded system-track .pcm file (raw s16le, 16kHz, mono)") + micPCM := fs.String("mic-pcm", "", "path to a previously recorded mic-track .pcm file (raw s16le, 16kHz, mono)") + asrBinary := fs.String("asr-binary", "", "path to the qwen_asr binary (required)") + asrModelDir := fs.String("asr-model-dir", "", "path to the qwen_asr model directory (required)") + prompt := fs.String("prompt", "", "glossary biasing prompt passed to qwen_asr") + transcriptFile := fs.String("transcript-file", "", "path to write the resulting transcript to as plain text (required)") + if err := fs.Parse(args); err != nil { + return err + } + + if *systemPCM == "" && *micPCM == "" { + return fmt.Errorf("at least one of -system-pcm or -mic-pcm is required") + } + if *asrBinary == "" || *asrModelDir == "" { + return fmt.Errorf("both -asr-binary and -asr-model-dir are required") + } + if *transcriptFile == "" { + return fmt.Errorf("-transcript-file is required") + } + + fileWriter, err := transcript.NewFileWriter(*transcriptFile) + if err != nil { + return err + } + defer fileWriter.Close() + + const sampleRate = 16000 + chunks := make(chan capture.Chunk, 64) + errs := make(chan error, 2) + var wg sync.WaitGroup + + // Both files started recording at the same instant (same live session), + // so a shared anchor lets VAD's duration-based thresholds (silence + // hangover, max segment duration) compare correctly across tracks — the + // same role Chunk.Timestamp plays for live capture, just derived from + // bytes-read instead of wall-clock reads, since a file is consumed as + // fast as disk I/O allows, not paced to real time. + base := time.Now() + if *systemPCM != "" { + wg.Add(1) + go readPCMFile(*systemPCM, capture.TrackSystem, sampleRate, base, chunks, &wg, errs) + } + if *micPCM != "" { + wg.Add(1) + go readPCMFile(*micPCM, capture.TrackMic, sampleRate, base, chunks, &wg, errs) + } + go func() { + wg.Wait() + close(chunks) + }() + + detector := vad.New(vad.DefaultConfig(sampleRate)) + segments := detector.Run(chunks) + transcriber := asr.New(asr.Config{BinaryPath: *asrBinary, ModelDir: *asrModelDir, Prompt: *prompt}) + // No HTTP server in batch mode — Publish to a Broadcaster with no + // subscribers is just a no-op, cheaper than special-casing a nil + // Broadcaster inside the shared transcribeWorker. + broadcaster := transcript.NewBroadcaster() + + fmt.Fprintln(os.Stderr, "transcribing...") + transcribeWorker(context.Background(), transcriber, segments, broadcaster, fileWriter) + + select { + case err := <-errs: + return fmt.Errorf("read pcm file: %w", err) + default: + } + + fmt.Fprintln(os.Stderr, "done ->", *transcriptFile) + return nil +} + +// readPCMFile replays a raw s16le/16kHz/mono PCM file as a stream of +// Chunks, the same shape live capture produces, so it can be fed through +// the same VAD detector unchanged. +func readPCMFile(path string, track capture.Track, sampleRate int, base time.Time, out chan<- capture.Chunk, wg *sync.WaitGroup, errs chan<- error) { + defer wg.Done() + f, err := os.Open(path) + if err != nil { + errs <- fmt.Errorf("open %s: %w", path, err) + return + } + defer f.Close() + + const chunkBytes = 3200 // matches internal/capture.Config's default (~100ms at 16kHz mono) + buf := make([]byte, chunkBytes) + var cumBytes int64 + for { + n, err := io.ReadFull(f, buf) + if n > 0 { + data := make([]byte, n) + copy(data, buf[:n]) + out <- capture.Chunk{Track: track, Data: data, Timestamp: base.Add(bytesToDuration(cumBytes, sampleRate))} + cumBytes += int64(n) + } + if err != nil { + if err != io.EOF && err != io.ErrUnexpectedEOF { + errs <- fmt.Errorf("read %s: %w", path, err) + } + return + } + } +} + +// bytesToDuration converts a count of 16-bit mono PCM sample bytes into the +// audio duration it represents at sampleRate. +func bytesToDuration(n int64, sampleRate int) time.Duration { + seconds := float64(n/2) / float64(sampleRate) + return time.Duration(seconds * float64(time.Second)) +}