add -realtime toggle and transcribe subcommand for batch transcription
-realtime=false skips VAD/ASR/SSE and only captures+records PCM (-recording-dir, timestamped per session, replacing the old fixed temp-dir filenames that a second session would silently overwrite). The new "transcribe" subcommand replays a previously recorded PCM file through the same VAD/ASR pipeline for post-hoc transcription, without live capture or a network dependency. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
+123
-65
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
Reference in New Issue
Block a user