From f8f98e83481e7376a235fb305a095552433f79ab Mon Sep 17 00:00:00 2001 From: elvis Date: Sun, 6 Sep 2026 19:21:16 -0300 Subject: Local voice assistant on top of Canary, llama.cpp and qwentts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pipeline en Rust de hilos y canales que une los tres motores: el micrófono alimenta un segmentador con VAD, las intervenciones cerradas van al reconocedor, la transcripción al modelo y cada frase que este cierra sale hacia el sintetizador sin esperar al resto de la respuesta. Dos reglas sostienen el diseño: ninguna etapa bloquea a la anterior —quien va sobrado descarta trabajo en lugar de acumular retraso— y todo lo que viaja por los canales lleva el turno al que pertenece, así que interrumpir es subir el contador y levantar dos banderas de cancelación. Seis crates: core (configuración, eventos, HTTP, telemetría, herramientas), audio (cpal, VAD, anillo de reproducción), asr, llm, tts y app (supervisor de procesos y orquestador). Los motores van como submódulos fijados a un commit, con los cambios locales en vendor/patches. Midiendo el pipeline aparecieron tres cuellos de botella de configuración que valieron más que cualquier cambio de código, todos documentados en docs/RENDIMIENTO.md: - tts-server decodificaba el audio en bloques de 24 s, de modo que el modo «streaming» llegaba de una pieza: 4948 ms -> 585 ms hasta el primer audio. - La plantilla de chat del modelo abre y no lo cierra nunca, sin variable que lo apague: 8630 ms -> 413 ms hasta el primer token, con una copia de la plantilla que deja el bloque cerrado de entrada. - Cualquier indicación de estilo junto a la guía de herramientas hace que este modelo de 2B deje de llamarlas y se invente el dato (8/8 aciertos con la guía sola, 0/8 con la persona de asistente de voz). El turno alterna ahora entre dos instrucciones de sistema. La ejecución de órdenes del sistema queda implementada y apagada, tras cuatro barreras: lista blanca sobre el ejecutable, rutas rechazadas, sin shell que interprete metacaracteres y plazo máximo. 68 pruebas unitarias sin modelos, más seis de integración que se saltan solas si no hay servidores y se turnan la GPU: en paralelo, los dos servidores no caben en 4 GB y miden contención en vez de latencia. Claude-Session: https://claude.ai/code/session_01FNxz5cSdQSscJH9H7b8uGU --- vendor/canary-rs | 1 + vendor/extra/LEEME.md | 10 + vendor/extra/canary-rs/examples/bench_live.rs | 91 ++ vendor/extra/qwentts.cpp/examples/all.sh | 99 ++ vendor/extra/qwentts.cpp/examples/server-test.sh | 51 + vendor/llama.cpp | 1 + vendor/patches/canary-rs.patch | 1767 ++++++++++++++++++++++ vendor/patches/llama.cpp.patch | 13 + vendor/patches/qwentts.cpp.patch | 41 + vendor/qwentts.cpp | 1 + 10 files changed, 2075 insertions(+) create mode 160000 vendor/canary-rs create mode 100644 vendor/extra/LEEME.md create mode 100644 vendor/extra/canary-rs/examples/bench_live.rs create mode 100755 vendor/extra/qwentts.cpp/examples/all.sh create mode 100755 vendor/extra/qwentts.cpp/examples/server-test.sh create mode 160000 vendor/llama.cpp create mode 100644 vendor/patches/canary-rs.patch create mode 100644 vendor/patches/llama.cpp.patch create mode 100644 vendor/patches/qwentts.cpp.patch create mode 160000 vendor/qwentts.cpp (limited to 'vendor') diff --git a/vendor/canary-rs b/vendor/canary-rs new file mode 160000 index 0000000..c7e68d2 --- /dev/null +++ b/vendor/canary-rs @@ -0,0 +1 @@ +Subproject commit c7e68d2d3b08525f2ea81a34fd2541806732f486 diff --git a/vendor/extra/LEEME.md b/vendor/extra/LEEME.md new file mode 100644 index 0000000..658bde9 --- /dev/null +++ b/vendor/extra/LEEME.md @@ -0,0 +1,10 @@ +# Ficheros sueltos de los submódulos + +Los parches de `vendor/patches/` sólo llevan cambios a ficheros que sus +repositorios ya versionaban. Estos otros existían únicamente en el árbol de +trabajo local, así que se guardan aquí enteros y `scripts/bootstrap.sh` los +copia al submódulo correspondiente tras aplicar el parche. + +Hace falta al menos uno de ellos para que el submódulo compile: +`canary-rs/Cargo.toml` declara el ejemplo `bench_live`, y cargo se niega a +resolver el paquete si el fichero no está. diff --git a/vendor/extra/canary-rs/examples/bench_live.rs b/vendor/extra/canary-rs/examples/bench_live.rs new file mode 100644 index 0000000..71c8420 --- /dev/null +++ b/vendor/extra/canary-rs/examples/bench_live.rs @@ -0,0 +1,91 @@ +#[path = "shared/utils.rs"] +mod utils; + +use canary_rs::{Canary, StreamConfig}; +use std::time::Instant; + +/// Benchmarks the pieces that drive live-streaming latency: +/// +/// * encoder cost as a function of window length, +/// * decoder cost per generated token, +/// * end-to-end cost of one streaming window vs. one full-utterance decode. +/// +/// Usage: `bench_live [lang]`, honouring the same `CANARY_*` environment +/// variables as the other examples. +fn main() -> Result<(), Box> { + let wav = match std::env::args().nth(1) { + Some(path) => path, + None => return Err("usage: bench_live [lang]".into()), + }; + let lang = std::env::args().nth(2).unwrap_or_else(|| "en".to_string()); + + let config = utils::execution_config_from_env(); + let model_dir = + std::env::var("CANARY_MODEL_DIR").unwrap_or_else(|_| "canary-180m-flash".to_string()); + + let t0 = Instant::now(); + let model = Canary::from_pretrained(&model_dir, Some(config))?; + println!("model load: {:.2?}", t0.elapsed()); + + let mut reader = hound::WavReader::open(&wav)?; + let spec = reader.spec(); + let samples: Vec = reader + .samples::() + .map(|s| s.map(|v| v as f32 / 32768.0)) + .collect::>()?; + let sr = spec.sample_rate as usize; + let duration = samples.len() as f32 / sr as f32; + println!("audio: {:.2}s @ {} Hz, {} ch", duration, sr, spec.channels); + + let mut session = model.session(); + let warm = &samples[..(sr).min(samples.len())]; + let _ = session.transcribe_samples(warm, sr, 1, &lang, &lang)?; + + // A live window always ends mid-utterance, so mirror that here: take the + // *last* `window` seconds and append the same tail silence the stream uses. + let tail = StreamConfig::default().tail_silence_duration; + println!("\nrolling windows (tail silence {:.2}s):", tail); + for window_s in [2.0f32, 4.0, 6.0, 8.0] { + if window_s > duration { + continue; + } + let start = samples.len() - (sr as f32 * window_s) as usize; + let mut win = samples[start..].to_vec(); + win.resize(win.len() + (sr as f32 * tail) as usize, 0.0); + + let mut best = f64::MAX; + let mut tokens = 0; + for _ in 0..3 { + let t = Instant::now(); + let r = session.transcribe_samples(&win, sr, 1, &lang, &lang)?; + best = best.min(t.elapsed().as_secs_f64()); + tokens = r.tokens.len(); + } + // No tokens means the model saw nothing worth transcribing, and the time is encoder-only. + let per_token = match tokens { + 0 => "encoder only".to_string(), + n => format!("{:.1} ms/token", best * 1000.0 / n as f64), + }; + println!( + " {:>4.1}s window -> {:>6.0} ms ({:>3} tokens, {})", + window_s, + best * 1000.0, + tokens, + per_token + ); + } + + let t = Instant::now(); + let r = session.transcribe_samples(&samples, sr, 1, &lang, &lang)?; + let full = t.elapsed().as_secs_f64(); + println!( + "\nfull utterance ({:.1}s) -> {:.0} ms ({} tokens, {:.1}x realtime)", + duration, + full * 1000.0, + r.tokens.len(), + duration as f64 / full + ); + println!("text: {}", r.text); + + Ok(()) +} diff --git a/vendor/extra/qwentts.cpp/examples/all.sh b/vendor/extra/qwentts.cpp/examples/all.sh new file mode 100755 index 0000000..f0b701c --- /dev/null +++ b/vendor/extra/qwentts.cpp/examples/all.sh @@ -0,0 +1,99 @@ +#!/bin/bash +# Matriz completa de pruebas de qwentts.cpp con los modelos disponibles. +# Genera un WAV por cada posibilidad en examples/out/ y un resumen al final. +# Uso: ./all.sh +set -u +cd "$(dirname "$0")" + +TTS=../build/qwen-tts +CODEC=../build/qwen-codec +COD=../models/qwen-tokenizer-12hz-Q8_0.gguf +T06B=../models/qwen-talker-0.6b-base-Q4_K_M.gguf +T06C=../models/qwen-talker-0.6b-customvoice-Q4_K_M.gguf +T17B=../models/qwen-talker-1.7b-base-Q8_0.gguf +T17C=../models/qwen-talker-1.7b-customvoice-Q8_0.gguf +T17V=../models/qwen-talker-1.7b-voicedesign-Q8_0.gguf +OUT=out +mkdir -p "$OUT" + +EN="qwentts runs entirely offline on your own machine." +ES="Hola, esto es una prueba de sintesis de voz totalmente local." +ZH="这是一个完全本地运行的语音合成测试。" + +n=0; ok=0 +# run [args extra de qwen-tts...] +run() { + local desc="$1" out="$2" text="$3"; shift 3 + n=$((n+1)) + printf '%2d) %-46s ' "$n" "$desc" + local t0 t1 + t0=$(date +%s.%N) + if printf '%s' "$text" | "$TTS" --codec "$COD" --seed 42 "$@" \ + -o "$OUT/$out" >"$OUT/$out.log" 2>&1; then + t1=$(date +%s.%N) + printf 'OK %5s %4.0fs %s\n' "$(du -h "$OUT/$out" | cut -f1)" \ + "$(echo "$t1 - $t0" | bc)" "$out" + ok=$((ok+1)) + else + printf 'FALLO (ver %s)\n' "$OUT/$out.log" + fi +} + +echo "########## A. BASE (voz por defecto) ##########" +run "Base 0.6B english" base_0.6b_en.wav "$EN" --model "$T06B" --lang english +run "Base 1.7B english" base_1.7b_en.wav "$EN" --model "$T17B" --lang english +run "Base 1.7B auto (autodetec)" base_1.7b_auto.wav "$EN" --model "$T17B" +run "Base 1.7B spanish" base_1.7b_es.wav "$ES" --model "$T17B" --lang spanish +run "Base 1.7B chinese" base_1.7b_zh.wav "$ZH" --model "$T17B" --lang chinese + +echo "########## B. CLONACION (referencia freeman, Base) ##########" +run "Clone ref-wav (encode interno)" clone_refwav_1.7b.wav "$EN" --model "$T17B" --lang english \ + --ref-wav freeman.wav --ref-text freeman.txt +run "Clone x-vector only (solo .spk)" clone_xvec_1.7b.wav "$EN" --model "$T17B" --lang english \ + --ref-spk freeman.spk +run "Clone ICL (.spk+.rvq+.txt)" clone_icl_1.7b.wav "$EN" --model "$T17B" --lang english \ + --ref-spk freeman.spk --ref-rvq freeman.rvq --ref-text freeman.txt +run "Clone ICL 0.6B" clone_icl_0.6b.wav "$EN" --model "$T06B" --lang english \ + --ref-spk freeman.spk --ref-rvq freeman.rvq --ref-text freeman.txt + +echo "########## C. CUSTOMVOICE (speakers con nombre) ##########" +for spk in serena vivian uncle_fu ryan aiden ono_anna sohee; do + run "CustomVoice 1.7B $spk" "cv_1.7b_${spk}.wav" "$EN" --model "$T17C" --lang english --speaker "$spk" +done +run "CustomVoice 1.7B eric (sichuan, ZH)" cv_1.7b_eric_zh.wav "$ZH" --model "$T17C" --lang chinese --speaker eric +run "CustomVoice 1.7B dylan (beijing, ZH)" cv_1.7b_dylan_zh.wav "$ZH" --model "$T17C" --lang chinese --speaker dylan +run "CustomVoice 0.6B vivian" cv_0.6b_vivian.wav "$EN" --model "$T06C" --lang english --speaker vivian +run "CustomVoice 1.7B vivian + instruct" cv_1.7b_vivian_instr.wav "$EN" --model "$T17C" --lang english \ + --speaker vivian --instruct "happy, energetic" + +echo "########## D. VOICEDESIGN (instruccion de atributos, 1.7B) ##########" +run "VoiceDesign male young" vd_male_young.wav "$EN" --model "$T17V" --lang english \ + --instruct "male, young adult, moderate pitch" +run "VoiceDesign female elderly" vd_female_elderly.wav "$EN" --model "$T17V" --lang english \ + --instruct "female, elderly, low pitch, calm" +run "VoiceDesign teen excited" vd_teen_excited.wav "$EN" --model "$T17V" --lang english \ + --instruct "female, teenager, high pitch, excited and fast" + +echo "########## E. MUESTREO / FORMATO (Base 1.7B) ##########" +run "Greedy (determinista)" fmt_greedy.wav "$EN" --model "$T17B" --lang english --greedy +run "Sampled temp=1.2 top-p=0.9" fmt_temp.wav "$EN" --model "$T17B" --lang english --temp 1.2 --top-p 0.9 +run "Formato wav24" fmt_wav24.wav "$EN" --model "$T17B" --lang english --format wav24 +run "Formato wav32" fmt_wav32.wav "$EN" --model "$T17B" --lang english --format wav32 + +echo "########## F. QWEN-CODEC (encode/decode round-trip) ##########" +cp -f freeman.wav "$OUT/rt.wav" +echo -n "F1) codec encode (wav -> .rvq + .spk) " +if "$CODEC" --model "$COD" --talker "$T17B" -i "$OUT/rt.wav" >"$OUT/codec_encode.log" 2>&1; then + echo "OK ($(du -h "$OUT/rt.rvq" 2>/dev/null | cut -f1) rvq, $(du -h "$OUT/rt.spk" 2>/dev/null | cut -f1) spk)" +else echo "FALLO (ver $OUT/codec_encode.log)"; fi +cp -f "$OUT/rt.rvq" "$OUT/roundtrip.rvq" +echo -n "F2) codec decode (.rvq -> wav round-trip) " +if "$CODEC" --model "$COD" -i "$OUT/roundtrip.rvq" >"$OUT/codec_decode.log" 2>&1; then + echo "OK ($(du -h "$OUT/roundtrip.wav" 2>/dev/null | cut -f1))" +else echo "FALLO (ver $OUT/codec_decode.log)"; fi + +echo +echo "########## RESUMEN ##########" +echo "Sintesis OK: $ok / $n" +echo "WAVs en $(pwd)/$OUT/:" +/bin/ls -1 "$OUT"/*.wav 2>/dev/null | sed 's|^| |' diff --git a/vendor/extra/qwentts.cpp/examples/server-test.sh b/vendor/extra/qwentts.cpp/examples/server-test.sh new file mode 100755 index 0000000..7242afe --- /dev/null +++ b/vendor/extra/qwentts.cpp/examples/server-test.sh @@ -0,0 +1,51 @@ +#!/bin/bash +# Prueba de punta a punta del tts-server (API OpenAI-compatible). +set -u +cd "$(dirname "$0")" +OUT=out; mkdir -p "$OUT" +PORT=8080; H=127.0.0.1; B="http://$H:$PORT" + +echo "== arrancando tts-server (base 1.7B) ==" +../build/tts-server \ + --model ../models/qwen-talker-1.7b-base-Q8_0.gguf \ + --codec ../models/qwen-tokenizer-12hz-Q8_0.gguf \ + --alias qwen3-tts-base --host $H --port $PORT --lang auto \ + >"$OUT/server.log" 2>&1 & +SRV=$! +trap 'kill $SRV 2>/dev/null' EXIT + +echo -n "esperando readiness" +for i in $(seq 1 60); do + curl -sf "$B/v1/models" >/dev/null 2>&1 && { echo " listo"; break; } + kill -0 $SRV 2>/dev/null || { echo " el server murio"; tail -20 "$OUT/server.log"; exit 1; } + printf '.'; sleep 1 +done + +echo "== 1) GET /v1/models ==" +curl -s "$B/v1/models"; echo + +echo "== 2) voz por defecto -> wav ==" +curl -s -X POST "$B/v1/audio/speech" -H "Content-Type: application/json" \ + -d '{"input":"This is the default server voice.","response_format":"wav","seed":42}' \ + -o "$OUT/srv_default.wav" +echo " $(du -h "$OUT/srv_default.wav" | cut -f1) $OUT/srv_default.wav" + +echo "== 3) registrar voz clonada 'freeman' (spk+rvq b64) ==" +curl -s -X POST "$B/v1/audio/voices" -H "Content-Type: application/json" \ + -d "{\"name\":\"freeman\",\"ref_text\":\"$(cat freeman.txt)\", + \"spk_b64\":\"$(base64 -w0 freeman.spk)\",\"rvq_b64\":\"$(base64 -w0 freeman.rvq)\"}" +echo + +echo "== 4) sintesis con la voz clonada -> wav ==" +curl -s -X POST "$B/v1/audio/speech" -H "Content-Type: application/json" \ + -d '{"input":"Now I am speaking with the cloned voice.","voice":"freeman","response_format":"wav","seed":42,"temperature":0.8}' \ + -o "$OUT/srv_freeman.wav" +echo " $(du -h "$OUT/srv_freeman.wav" | cut -f1) $OUT/srv_freeman.wav" + +echo "== 5) streaming PCM (s16le) -> archivo raw ==" +curl -s -X POST "$B/v1/audio/speech" -H "Content-Type: application/json" \ + -d '{"input":"Streaming pcm chunks.","response_format":"pcm","seed":42}' \ + -o "$OUT/srv_stream.pcm" +echo " $(du -h "$OUT/srv_stream.pcm" | cut -f1) $OUT/srv_stream.pcm (s16le mono 24k)" + +echo "== ok, parando server ==" diff --git a/vendor/llama.cpp b/vendor/llama.cpp new file mode 160000 index 0000000..4fc4ec5 --- /dev/null +++ b/vendor/llama.cpp @@ -0,0 +1 @@ +Subproject commit 4fc4ec5541b243957ae5099edb67372f8f3b550e diff --git a/vendor/patches/canary-rs.patch b/vendor/patches/canary-rs.patch new file mode 100644 index 0000000..c579ee0 --- /dev/null +++ b/vendor/patches/canary-rs.patch @@ -0,0 +1,1767 @@ +diff --git a/Cargo.toml b/Cargo.toml +index 575a037..65469d9 100644 +--- a/Cargo.toml ++++ b/Cargo.toml +@@ -41,6 +41,10 @@ path = "examples/inspect_model.rs" + name = "decoder_smoke" + path = "examples/decoder_smoke.rs" + ++[[example]] ++name = "bench_live" ++path = "examples/bench_live.rs" ++ + [features] + default = ["cpu", "ort-defaults"] + cpu = [] +diff --git a/README.md b/README.md +index ea711f1..34ecfc5 100644 +--- a/README.md ++++ b/README.md +@@ -41,6 +41,21 @@ let mut stream = model.stream("en", "en", stream_cfg)?; + // stream.push_samples(&audio_chunk, sample_rate, channels)?; + ``` + ++### Streaming ++ ++`CanaryStream` decodes a rolling window of audio, and its defaults are tuned for live input where ++decoding is slower than real time: ++ ++- `tail_silence_duration` (0.4 s) pads the end of every window. Canary was trained on complete ++ utterances and very often emits `<|endoftext|>` immediately — returning *no text at all* — for ++ audio that stops mid-word, which is what a live window always looks like. ++- `max_windows_per_push` (1) decodes only the newest window and drops the backlog, so latency ++ stays bounded by one decode instead of growing without limit. Set it to `0` when replaying ++ recorded audio, where every window matters and wall-clock does not. ++- `min_commit_duration` (2.0 s) keeps the text of very short windows out of `delta_text`. The ++ first fractions of a second of an utterance decode poorly, and a committed transcript cannot be ++ taken back. Their text is still available in `StreamChunk::result` for display. ++ + **Note**: When using `canary-180m-flash` don't enable the `use_itn` option in `SessionConfig`, as this model doesn't seem to be trained with inverse text normalization and enabling it causes empty output. + + ## Features +@@ -55,13 +70,54 @@ See the `examples` directory for more usage examples, including live microphone + + ```bash + # Defaults to CPU as execution provider +-cargo run --example live ++cargo run --release --example live -- en en + ``` + ++The `live` example runs a threaded pipeline — audio callback, capture/VAD, decode worker(s) and ++rendering each on their own thread, connected by channels — so a slow decode never blocks capture ++and stale windows are dropped rather than queued. It shows the stabilized transcript plus a dimmed, ++still-changing tail, and re-decodes each finished utterance for a clean final line. Tuning knobs: ++ ++| Variable | Default | Meaning | ++| --- | --- | --- | ++| `CANARY_LIVE_WINDOW` | `6.0` | Rolling window in seconds; longer means more context, fewer updates | ++| `CANARY_LIVE_STEP` | `0.4` | How far the window advances between decodes | ++| `CANARY_LIVE_STABILITY` | `2` | Windows that must agree before a word is committed | ++| `CANARY_LIVE_TAIL_SILENCE` | `0.4` | Silence appended to each decoded window | ++| `CANARY_LIVE_SILENCE_HOLD` | `0.8` | Silence that ends an utterance | ++| `CANARY_LIVE_MIN_UTTERANCE` | `0.3` | Shortest utterance worth a final decode | ++| `CANARY_LIVE_MAX_UTTERANCE` | `20.0` | Longest utterance before one is forced | ++| `CANARY_LIVE_FINAL_MODEL` | `0` | Load a second model instance so final decodes run in parallel | ++| `CANARY_LIVE_STATS` | `0` | Print decode time, lag and dropped windows to stderr | ++| `CANARY_LIVE_WIDTH` | `100` | Characters of the live line to keep on screen | ++ ++`CANARY_INTER_THREADS` / `CANARY_INTRA_THREADS` set the ONNX Runtime thread pools for every ++example. ++ + You can provide an environment variable called `CANARY_EXECUTION_PROVIDER` with the matching feature flag to select an execution provider. For example, to use CUDA: + + ```bash +-CANARY_EXECUTION_PROVIDER=cuda CANARY_CUDA_DEVICE_ID=1 cargo run --example transcribe --features cuda ++CANARY_EXECUTION_PROVIDER=cuda CANARY_CUDA_DEVICE_ID=0 cargo run --example transcribe --features cuda ++``` ++ ++By default, a requested execution provider that cannot be loaded is a hard error rather than a ++silent fall back to CPU — ONNX Runtime's own behaviour, which otherwise looks like a working GPU ++session running an order of magnitude slower than expected. The CUDA provider needs cuDNN 9 ++alongside the CUDA runtime; without it you will see ++ ++``` ++Failed to load library libonnxruntime_providers_cuda.so with error: libcudnn.so.9: cannot open shared object file ++``` ++ ++Set `CANARY_REQUIRE_EP=0` (or `ExecutionConfig::with_require_execution_provider(false)`) to opt ++back into the silent fallback. ++ ++Note that the published Canary ONNX exports are int8-quantized, and the CUDA provider has limited ++coverage of int8 operators — on a modest GPU it can be no faster than a multi-core CPU. Measure ++with the `bench_live` example before assuming the GPU wins: ++ ++```bash ++cargo run --release --example bench_live -- audio.wav en + ``` + + Or to use CoreML with Neural Engine + low precision: +diff --git a/examples/live.rs b/examples/live.rs +index 46d3ad8..a60cfdf 100644 +--- a/examples/live.rs ++++ b/examples/live.rs +@@ -1,248 +1,505 @@ + #[path = "shared/utils.rs"] + mod utils; + +-use canary_rs::{Canary, StreamConfig}; +-use cpal::Sample; ++use canary_rs::{Canary, CanarySession, StreamConfig}; + use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; ++use cpal::{Sample, SampleFormat, SupportedStreamConfig}; + use std::io::Write; + use std::sync::atomic::{AtomicBool, Ordering}; +-use std::sync::{Arc, Mutex, mpsc}; +-use std::time::Duration; +- +-// Live streaming example: +-// - Captures microphone audio via cpal. +-// - Runs a rolling window decode for low-latency updates. +-// - Tracks a simple noise floor to avoid decoding silence. +-// - On detected utterance end, re-decodes the full utterance for a final result. ++use std::sync::mpsc::{self, Receiver, Sender}; ++use std::sync::Arc; ++use std::thread::{self, JoinHandle}; ++use std::time::{Duration, Instant}; ++ ++// Live streaming transcription. ++// ++// Decoding a window is slower than recording it — on a laptop CPU a 6 s window costs roughly ++// 800 ms while it only advances 0.4 s of audio — so anything that decodes every window in order ++// falls behind the speaker without bound. The pipeline below is built around that fact: each ++// stage runs on its own thread, connected by channels, and the decode stage is allowed to throw ++// work away. ++// ++// cpal callback ─samples─> capture thread ─jobs─> decode worker(s) ─updates─> render (main) ++// realtime resample + VAD Canary sessions terminal ++// ++// * The capture thread never blocks the audio callback: it owns resampling, the noise-floor VAD ++// and the utterance buffer. ++// * A decode worker drains its whole queue before running the model, keeps only the newest ++// rolling window and drops the backlog, so latency is bounded by one decode instead of growing. ++// * The end of an utterance re-decodes the whole thing for a clean final line. That decode is ++// long, so it can optionally run on its own thread against a second model instance ++// (`CANARY_LIVE_FINAL_MODEL=1`) rather than stalling the partials. ++ ++/// Sample rate the model works at; the input stream is opened here directly when the device ++/// allows it, which removes resampling from the hot path entirely. ++const MODEL_SAMPLE_RATE: u32 = 16_000; ++ ++/// How much audio the VAD looks at per decision. ++const VAD_FRAME_SECONDS: f32 = 0.1; ++/// Audio kept before speech is detected, so an utterance does not start clipped. ++const PREROLL_SECONDS: f32 = 0.3; ++ ++const MIN_SILENCE_THRESHOLD: f32 = 0.0008; ++const MAX_SILENCE_THRESHOLD: f32 = 0.02; ++ + fn main() -> Result<(), Box> { +- // Use en -> en or the first and second command line arguments as source and target language codes +- let source_lang = match std::env::args().nth(1) { +- Some(lang) => lang, +- None => "en".to_string(), +- }; +- let target_lang = match std::env::args().nth(2) { +- Some(lang) => lang, +- None => source_lang.clone(), +- }; ++ let opts = Options::from_env(); + println!( + "Source language: {}, Target language: {}", +- source_lang, target_lang ++ opts.source_lang, opts.target_lang + ); + + println!("Loading Canary model..."); +- + let config = utils::execution_config_from_env(); + let model_dir = + std::env::var("CANARY_MODEL_DIR").unwrap_or_else(|_| "canary-180m-flash".to_string()); +- let model = Canary::from_pretrained(&model_dir, Some(config))?; +- let stream_cfg = StreamConfig::new() +- .with_window_duration(8.0) +- .with_step_duration(0.5) +- .with_emit_partial(true) +- .with_pad_partial(false) +- .with_stability_window(3); +- let mut stream_state = +- model.stream(source_lang.clone(), target_lang.clone(), stream_cfg.clone())?; +- let mut full_session = model.session(); +- ++ let model = Canary::from_pretrained(&model_dir, Some(config.clone()))?; ++ ++ // A second instance means finals and partials decode in parallel: sessions cloned from one ++ // `Canary` share its ONNX sessions behind a mutex, so they would serialize instead. ++ let final_model = if opts.dedicated_final_model { ++ println!("Loading a second model instance for final decodes..."); ++ Some(Canary::from_pretrained(&model_dir, Some(config))?) ++ } else { ++ None ++ }; + println!("Model loaded successfully!"); +- +- let host = cpal::default_host(); +- let device = host +- .default_input_device() +- .ok_or("No input device available")?; +- let supported_config = device.default_input_config()?; +- let sample_rate = supported_config.sample_rate() as usize; +- let channels = supported_config.channels() as usize; +- let cpal_config: cpal::StreamConfig = supported_config.clone().into(); +- +- // Feed chunks at roughly the streaming step size to avoid multi-second output bursts. +- let chunk_seconds = stream_cfg.step_duration.max(0.05); +- let chunk_samples = ((sample_rate as f32 * chunk_seconds).round() as usize).max(1) * channels; +- let min_silence_threshold = 0.0008_f32; +- let max_silence_threshold = 0.02_f32; +- let silence_hold_seconds = 0.8_f32; +- let silence_hold_samples = +- ((sample_rate as f32 * silence_hold_seconds).round() as usize).max(1) * channels; +- let min_utterance_seconds = 0.3_f32; +- let min_utterance_samples = +- ((sample_rate as f32 * min_utterance_seconds).round() as usize).max(1) * channels; +- +- let buffer = Arc::new(Mutex::new(Vec::::with_capacity(chunk_samples * 2))); +- let (tx, rx) = mpsc::channel::>(); +- +- let err_fn = |err| eprintln!("stream error: {}", err); +- +- let buffer_clone = Arc::clone(&buffer); +- let tx_clone = tx.clone(); +- let stream = match supported_config.sample_format() { +- cpal::SampleFormat::F32 => device.build_input_stream( +- &cpal_config, +- move |data: &[f32], _| { +- push_input_data(data, chunk_samples, &buffer_clone, &tx_clone); +- }, +- err_fn, +- None, +- )?, +- cpal::SampleFormat::I16 => device.build_input_stream( +- &cpal_config, +- move |data: &[i16], _| { +- push_input_data(data, chunk_samples, &buffer_clone, &tx_clone); +- }, +- err_fn, +- None, +- )?, +- cpal::SampleFormat::U16 => device.build_input_stream( +- &cpal_config, +- move |data: &[u16], _| { +- push_input_data(data, chunk_samples, &buffer_clone, &tx_clone); +- }, +- err_fn, +- None, +- )?, +- sample_format => { +- return Err(format!("Unsupported sample format: {:?}", sample_format).into()); ++ let (job_tx, job_rx) = mpsc::channel::(); ++ let (update_tx, update_rx) = mpsc::channel::(); ++ ++ let mut workers: Vec> = Vec::new(); ++ let final_tx = match final_model { ++ Some(final_model) => { ++ let (final_tx, final_rx) = mpsc::channel::(); ++ let opts = opts.clone(); ++ let updates = update_tx.clone(); ++ workers.push(thread::spawn(move || { ++ run_final_worker(final_model, final_rx, updates, &opts) ++ })); ++ Some(final_tx) + } ++ None => None, ++ }; ++ { ++ let opts = opts.clone(); ++ let updates = update_tx.clone(); ++ workers.push(thread::spawn(move || { ++ run_partial_worker(model, job_rx, updates, &opts) ++ })); ++ } ++ drop(update_tx); ++ ++ let (capture_tx, capture_rx) = mpsc::channel::(); ++ let input = InputStream::open(capture_tx)?; ++ eprintln!( ++ "Input: {} Hz, {} ch ({:?}){}", ++ input.config.sample_rate(), ++ input.config.channels(), ++ input.config.sample_format(), ++ if input.needs_conversion() { ++ " — converting to 16 kHz mono" ++ } else { ++ "" ++ } ++ ); ++ ++ let capture = { ++ let opts = opts.clone(); ++ let format = input.format(); ++ thread::spawn(move || run_capture(capture_rx, job_tx, final_tx, format, &opts)) + }; + + println!("Listening... press Enter to stop."); +- stream.play()?; ++ input.stream.play()?; + + let stop = Arc::new(AtomicBool::new(false)); +- let stop_clone = Arc::clone(&stop); +- std::thread::spawn(move || { +- let mut line = String::new(); +- let _ = std::io::stdin().read_line(&mut line); +- stop_clone.store(true, Ordering::SeqCst); +- }); +- +- let mut last_len = 0usize; +- let mut final_line = String::new(); +- let mut silence_samples = 0usize; +- let mut noise_floor = 0.0_f32; +- let mut utterance_audio: Vec = Vec::new(); +- let mut in_utterance = false; +- while !stop.load(Ordering::SeqCst) { +- match rx.recv_timeout(Duration::from_millis(100)) { +- Ok(chunk) => { +- let chunk_rms = rms(&chunk); +- if noise_floor == 0.0 { +- noise_floor = chunk_rms; +- } else if chunk_rms < noise_floor * 1.5 { +- noise_floor = noise_floor * 0.95 + chunk_rms * 0.05; +- } +- let silence_threshold = +- (noise_floor * 3.0).clamp(min_silence_threshold, max_silence_threshold); +- +- if chunk_rms < silence_threshold { +- if in_utterance { +- silence_samples = silence_samples.saturating_add(chunk.len()); +- utterance_audio.extend_from_slice(&chunk); +- if silence_samples >= silence_hold_samples { +- if utterance_audio.len() >= min_utterance_samples { +- let final_result = full_session.transcribe_samples( +- &utterance_audio, +- sample_rate, +- channels, +- &source_lang, +- &target_lang, +- )?; +- let final_text = +- normalize_punctuation_spacing(final_result.text.trim()); +- if !final_text.is_empty() { +- if last_len > 0 { +- print!("\r{: 0 { +- println!(); +- last_len = 0; +- } +- +- stream_state.reset(); +- final_line.clear(); +- utterance_audio.clear(); +- silence_samples = 0; +- in_utterance = false; +- } +- } +- continue; +- } +- silence_samples = 0; +- in_utterance = true; +- utterance_audio.extend_from_slice(&chunk); +- +- let results = stream_state.push_samples(&chunk, sample_rate, channels)?; +- for result in results { +- let delta = result.delta_text.trim(); +- if delta.is_empty() { +- continue; +- } ++ { ++ let stop = Arc::clone(&stop); ++ thread::spawn(move || { ++ let mut line = String::new(); ++ let _ = std::io::stdin().read_line(&mut line); ++ stop.store(true, Ordering::SeqCst); ++ }); ++ } + +- let delta = normalize_punctuation_spacing(delta); +- let delta = strip_duplicate_leading_punct(&final_line, &delta); +- if delta.is_empty() { +- continue; +- } ++ let mut render = Renderer::new(&opts); ++ let mut stream = Some(input.stream); ++ loop { ++ match update_rx.recv_timeout(Duration::from_millis(100)) { ++ Ok(update) => render.apply(update), ++ Err(mpsc::RecvTimeoutError::Timeout) => {} ++ Err(mpsc::RecvTimeoutError::Disconnected) => break, ++ } ++ // Dropping the stream ends the capture thread, which ends the workers, which closes ++ // `update_rx` — but only after the in-flight decodes have been rendered. ++ if stop.load(Ordering::SeqCst) { ++ stream = None; ++ } ++ } ++ drop(stream); ++ render.finish(); + +- append_with_space(&mut final_line, &delta); ++ let _ = capture.join(); ++ for worker in workers { ++ let _ = worker.join(); ++ } ++ Ok(()) ++} + +- let mut line_break = false; +- if ends_with_sentence_punct(&final_line) { +- line_break = true; +- } ++/// Runtime knobs, all overridable from the environment so the pipeline can be tuned without a ++/// rebuild. ++#[derive(Clone)] ++struct Options { ++ source_lang: String, ++ target_lang: String, ++ /// Rolling window length; longer means more context but fewer updates per second. ++ window: f32, ++ /// How far the window advances between decodes. ++ step: f32, ++ /// Windows that must agree before a word is committed to the stable transcript. ++ stability: usize, ++ /// Silence appended to each decoded window; see `StreamConfig::tail_silence_duration`. ++ tail_silence: f32, ++ /// Silence that ends an utterance. ++ silence_hold: f32, ++ /// Shortest utterance worth a final decode. ++ min_utterance: f32, ++ /// Longest utterance before one is forced, bounding final-decode cost and memory. ++ max_utterance: f32, ++ dedicated_final_model: bool, ++ stats: bool, ++ width: usize, ++} + +- if !final_line.is_empty() && has_alnum(&final_line) { +- last_len = last_len.max(final_line.len()); +- print!("\r{: Self { ++ let source_lang = std::env::args().nth(1).unwrap_or_else(|| "en".to_string()); ++ let target_lang = std::env::args().nth(2).unwrap_or_else(|| source_lang.clone()); ++ Self { ++ source_lang, ++ target_lang, ++ window: env_f32("CANARY_LIVE_WINDOW").unwrap_or(6.0), ++ step: env_f32("CANARY_LIVE_STEP").unwrap_or(0.4), ++ stability: env_usize("CANARY_LIVE_STABILITY").unwrap_or(2), ++ tail_silence: env_f32("CANARY_LIVE_TAIL_SILENCE") ++ .unwrap_or(StreamConfig::default().tail_silence_duration), ++ silence_hold: env_f32("CANARY_LIVE_SILENCE_HOLD").unwrap_or(0.8), ++ min_utterance: env_f32("CANARY_LIVE_MIN_UTTERANCE").unwrap_or(0.3), ++ max_utterance: env_f32("CANARY_LIVE_MAX_UTTERANCE").unwrap_or(20.0), ++ dedicated_final_model: env_bool("CANARY_LIVE_FINAL_MODEL").unwrap_or(false), ++ stats: env_bool("CANARY_LIVE_STATS").unwrap_or(false), ++ width: env_usize("CANARY_LIVE_WIDTH") ++ .or_else(|| env_usize("COLUMNS")) ++ .unwrap_or(100), ++ } ++ } + +- if line_break { +- println!(); +- last_len = 0; +- final_line.clear(); +- } +- } +- } +- Err(mpsc::RecvTimeoutError::Timeout) => {} +- Err(mpsc::RecvTimeoutError::Disconnected) => break, ++ fn stream_config(&self) -> StreamConfig { ++ StreamConfig::new() ++ .with_window_duration(self.window) ++ .with_step_duration(self.step) ++ .with_emit_partial(true) ++ .with_pad_partial(false) ++ .with_stability_window(self.stability) ++ .with_tail_silence_duration(self.tail_silence) ++ // The whole point of the pipeline: never work through a backlog of stale windows. ++ .with_max_windows_per_push(1) ++ } ++} ++ ++fn env_f32(key: &str) -> Option { ++ std::env::var(key).ok()?.trim().parse().ok() ++} ++ ++fn env_usize(key: &str) -> Option { ++ std::env::var(key).ok()?.trim().parse().ok() ++} ++ ++fn env_bool(key: &str) -> Option { ++ match std::env::var(key).ok()?.trim().to_lowercase().as_str() { ++ "1" | "true" | "yes" => Some(true), ++ "0" | "false" | "no" => Some(false), ++ _ => None, ++ } ++} ++ ++/// One buffer straight from the audio callback, stamped so the pipeline can report how far ++/// behind the speaker it is running. ++struct Capture { ++ samples: Vec, ++ at: Instant, ++} ++ ++/// Work handed to a decode thread. `seq` identifies the utterance so that a final arriving late ++/// is not mistaken for the transcript of the utterance already in progress. ++enum Job { ++ /// Newly captured audio for the rolling partial window, at 16 kHz mono. ++ Audio { ++ seq: u64, ++ samples: Vec, ++ at: Instant, ++ }, ++ /// A finished utterance, to be decoded in full. ++ Final { seq: u64, audio: Vec }, ++ /// The utterance ended with nothing worth decoding. ++ Reset { seq: u64 }, ++} ++ ++enum Update { ++ /// Stable transcript so far plus the still-changing tail of the newest window. ++ Partial { ++ seq: u64, ++ committed: String, ++ volatile: String, ++ stats: Stats, ++ }, ++ /// Whole-utterance transcript; replaces everything shown for `seq`. ++ Final { ++ seq: u64, ++ text: String, ++ stats: Stats, ++ }, ++ /// The utterance produced nothing. ++ Discard { seq: u64 }, ++ Error(String), ++} ++ ++#[derive(Clone, Copy)] ++struct Stats { ++ decode: Duration, ++ /// Age of the newest sample in the decoded audio when the result was ready. ++ lag: Duration, ++ /// Steps of audio thrown away as stale before this decode. ++ dropped: usize, ++} ++ ++// --------------------------------------------------------------------------------------------- ++// Capture ++// --------------------------------------------------------------------------------------------- ++ ++/// The input stream plus the format it actually opened with. ++struct InputStream { ++ stream: cpal::Stream, ++ config: SupportedStreamConfig, ++} ++ ++impl InputStream { ++ /// Opens the default input device, preferring 16 kHz mono so no conversion is needed. ++ fn open(tx: Sender) -> Result> { ++ let host = cpal::default_host(); ++ let device = host ++ .default_input_device() ++ .ok_or("No input device available")?; ++ ++ let config = Self::preferred_config(&device)?; ++ let stream_config: cpal::StreamConfig = config.clone().into(); ++ let err_fn = |err| eprintln!("stream error: {}", err); ++ ++ // The callback runs on a realtime thread: convert to f32, hand it over, return. ++ macro_rules! build { ++ ($sample:ty) => { ++ device.build_input_stream( ++ &stream_config, ++ move |data: &[$sample], _: &_| { ++ let samples = data.iter().map(|s| f32::from_sample(*s)).collect(); ++ let _ = tx.send(Capture { ++ samples, ++ at: Instant::now(), ++ }); ++ }, ++ err_fn, ++ None, ++ )? ++ }; + } ++ ++ let stream = match config.sample_format() { ++ SampleFormat::F32 => build!(f32), ++ SampleFormat::I16 => build!(i16), ++ SampleFormat::U16 => build!(u16), ++ other => return Err(format!("Unsupported sample format: {:?}", other).into()), ++ }; ++ ++ Ok(Self { stream, config }) + } + +- Ok(()) ++ fn preferred_config( ++ device: &cpal::Device, ++ ) -> Result> { ++ let native = device ++ .supported_input_configs()? ++ .filter(|range| range.channels() == 1) ++ .filter(|range| { ++ range.min_sample_rate() <= MODEL_SAMPLE_RATE ++ && MODEL_SAMPLE_RATE <= range.max_sample_rate() ++ }) ++ .find(|range| range.sample_format() == SampleFormat::F32) ++ .map(|range| range.with_sample_rate(MODEL_SAMPLE_RATE)); ++ ++ match native { ++ Some(config) => Ok(config), ++ None => Ok(device.default_input_config()?), ++ } ++ } ++ ++ fn needs_conversion(&self) -> bool { ++ self.config.sample_rate() != MODEL_SAMPLE_RATE || self.config.channels() != 1 ++ } ++ ++ fn format(&self) -> InputFormat { ++ InputFormat { ++ sample_rate: self.config.sample_rate() as usize, ++ channels: self.config.channels() as usize, ++ } ++ } ++} ++ ++#[derive(Clone, Copy)] ++struct InputFormat { ++ sample_rate: usize, ++ channels: usize, + } + +-fn push_input_data( +- input: &[T], +- chunk_samples: usize, +- buffer: &Arc>>, +- sender: &mpsc::Sender>, +-) where +- f32: cpal::FromSample, +-{ +- let mut buffer = match buffer.lock() { +- Ok(guard) => guard, +- Err(poisoned) => poisoned.into_inner(), ++/// Turns captured audio into decode jobs: converts to 16 kHz mono, tracks the noise floor, and ++/// cuts the stream into utterances. ++fn run_capture( ++ rx: Receiver, ++ job_tx: Sender, ++ final_tx: Option>, ++ format: InputFormat, ++ opts: &Options, ++) { ++ let rate = MODEL_SAMPLE_RATE as f32; ++ let frame_samples = (rate * VAD_FRAME_SECONDS).max(1.0) as usize; ++ let preroll_samples = (rate * PREROLL_SECONDS) as usize; ++ let silence_hold_samples = (rate * opts.silence_hold) as usize; ++ let min_utterance_samples = (rate * opts.min_utterance) as usize; ++ let max_utterance_samples = (rate * opts.max_utterance) as usize; ++ ++ let mut frame: Vec = Vec::with_capacity(frame_samples * 2); ++ let mut preroll: std::collections::VecDeque = std::collections::VecDeque::new(); ++ let mut utterance: Vec = Vec::new(); ++ let mut noise_floor = 0.0f32; ++ let mut silence_run = 0usize; ++ let mut speaking = false; ++ let mut seq = 0u64; ++ ++ // Sending a boundary is the same work whether or not a dedicated final worker exists. ++ let send_boundary = |job: Job| -> bool { ++ match (&final_tx, &job) { ++ (Some(final_tx), Job::Final { seq, .. }) => { ++ job_tx.send(Job::Reset { seq: *seq }).is_ok() && final_tx.send(job).is_ok() ++ } ++ _ => job_tx.send(job).is_ok(), ++ } + }; + +- buffer.extend(input.iter().map(|sample| f32::from_sample(*sample))); ++ while let Ok(capture) = rx.recv() { ++ let mono = to_model_audio(capture.samples, format); ++ frame.extend_from_slice(&mono); ++ ++ while frame.len() >= frame_samples { ++ let chunk: Vec = frame.drain(..frame_samples).collect(); ++ let level = rms(&chunk); + +- while buffer.len() >= chunk_samples { +- let chunk: Vec = buffer.drain(..chunk_samples).collect(); +- let _ = sender.send(chunk); ++ // Track the quietest recent level as the noise floor, and only adapt downwards so a ++ // steady voice cannot drag the threshold up over itself. ++ if noise_floor == 0.0 { ++ noise_floor = level; ++ } else if level < noise_floor * 1.5 { ++ noise_floor = noise_floor * 0.95 + level * 0.05; ++ } ++ let threshold = (noise_floor * 3.0).clamp(MIN_SILENCE_THRESHOLD, MAX_SILENCE_THRESHOLD); ++ ++ if level < threshold && !speaking { ++ preroll.extend(chunk.iter().copied()); ++ while preroll.len() > preroll_samples { ++ preroll.pop_front(); ++ } ++ continue; ++ } ++ ++ if !speaking { ++ speaking = true; ++ seq += 1; ++ utterance.clear(); ++ utterance.extend(preroll.drain(..)); ++ } ++ ++ if level < threshold { ++ silence_run += chunk.len(); ++ } else { ++ silence_run = 0; ++ } ++ utterance.extend_from_slice(&chunk); ++ ++ let ended = silence_run >= silence_hold_samples; ++ let too_long = utterance.len() >= max_utterance_samples; ++ if !ended && !too_long { ++ let job = Job::Audio { ++ seq, ++ samples: chunk, ++ at: capture.at, ++ }; ++ if job_tx.send(job).is_err() { ++ return; ++ } ++ continue; ++ } ++ ++ let long_enough = utterance.len() >= min_utterance_samples; ++ let boundary = if long_enough { ++ Job::Final { ++ seq, ++ audio: std::mem::take(&mut utterance), ++ } ++ } else { ++ Job::Reset { seq } ++ }; ++ if !send_boundary(boundary) { ++ return; ++ } ++ ++ utterance.clear(); ++ preroll.clear(); ++ silence_run = 0; ++ // A forced cut lands mid-sentence, so keep listening as if speech never stopped. ++ speaking = too_long && !ended; ++ if speaking { ++ seq += 1; ++ } ++ } + } + } + +-fn ends_with_sentence_punct(text: &str) -> bool { +- text.chars() +- .rev() +- .find(|ch| !ch.is_whitespace()) +- .map_or(false, |ch| matches!(ch, '.' | '!' | '?')) ++/// Converts a captured buffer to the 16 kHz mono the model expects. ++fn to_model_audio(samples: Vec, format: InputFormat) -> Vec { ++ let mono = if format.channels > 1 { ++ samples ++ .chunks(format.channels) ++ .map(|frame| frame.iter().sum::() / format.channels as f32) ++ .collect() ++ } else { ++ samples ++ }; ++ ++ if format.sample_rate == MODEL_SAMPLE_RATE as usize { ++ return mono; ++ } ++ // Cheap linear resampling is enough here: the input is already band-limited by the device ++ // and the alternative costs more than the decode it feeds. ++ let ratio = MODEL_SAMPLE_RATE as f64 / format.sample_rate as f64; ++ let out_len = (mono.len() as f64 * ratio) as usize; ++ (0..out_len) ++ .map(|i| { ++ let pos = i as f64 / ratio; ++ let idx = pos as usize; ++ let frac = (pos - idx as f64) as f32; ++ let a = mono.get(idx).copied().unwrap_or(0.0); ++ let b = mono.get(idx + 1).copied().unwrap_or(a); ++ a + (b - a) * frac ++ }) ++ .collect() + } + + fn rms(samples: &[f32]) -> f32 { +@@ -253,19 +510,398 @@ fn rms(samples: &[f32]) -> f32 { + (sum / samples.len() as f32).sqrt() + } + +-fn has_alnum(text: &str) -> bool { +- text.chars().any(|ch| ch.is_alphanumeric()) ++// --------------------------------------------------------------------------------------------- ++// Decoding ++// --------------------------------------------------------------------------------------------- ++ ++/// Runs the rolling-window decode, and the final decode too unless a dedicated worker took it. ++fn run_partial_worker(model: Canary, rx: Receiver, tx: Sender, opts: &Options) { ++ let mut stream = match model.stream( ++ opts.source_lang.clone(), ++ opts.target_lang.clone(), ++ opts.stream_config(), ++ ) { ++ Ok(stream) => stream, ++ Err(err) => { ++ let _ = tx.send(Update::Error(format!("stream setup failed: {}", err))); ++ return; ++ } ++ }; ++ let mut session = model.session(); ++ let mut committed = String::new(); ++ ++ for batch in Batches::new(rx) { ++ // Audio queued behind an utterance boundary belongs to a transcript that is about to be ++ // decoded in full, so decoding a window of it would only spend time to be overwritten. ++ let mut pending: Vec = Vec::new(); ++ let mut pending_seq = 0u64; ++ let mut newest = Instant::now(); ++ let mut dropped = 0usize; ++ ++ for job in batch { ++ match job { ++ Job::Audio { seq, samples, at } => { ++ pending_seq = seq; ++ pending.extend_from_slice(&samples); ++ newest = at; ++ } ++ Job::Reset { seq } => { ++ dropped += steps_in(&pending, opts); ++ pending.clear(); ++ stream.reset(); ++ committed.clear(); ++ let _ = tx.send(Update::Discard { seq }); ++ } ++ Job::Final { seq, audio } => { ++ dropped += steps_in(&pending, opts); ++ pending.clear(); ++ stream.reset(); ++ committed.clear(); ++ if decode_final(&mut session, &audio, seq, &tx, opts).is_err() { ++ return; ++ } ++ } ++ } ++ } ++ ++ if pending.is_empty() { ++ continue; ++ } ++ dropped += steps_in(&pending, opts).saturating_sub(1); ++ ++ let started = Instant::now(); ++ let chunks = match stream.push_samples(&pending, MODEL_SAMPLE_RATE as usize, 1) { ++ Ok(chunks) => chunks, ++ Err(err) => { ++ let _ = tx.send(Update::Error(format!("decode failed: {}", err))); ++ continue; ++ } ++ }; ++ let Some(chunk) = chunks.last() else { continue }; ++ if is_degenerate(&chunk.result.text) { ++ continue; ++ } ++ ++ append_delta(&mut committed, chunk.delta_text.trim()); ++ let volatile = volatile_tail(&committed, chunk.result.text.trim()); ++ let update = Update::Partial { ++ seq: pending_seq, ++ committed: committed.clone(), ++ volatile, ++ stats: Stats { ++ decode: started.elapsed(), ++ lag: newest.elapsed(), ++ dropped, ++ }, ++ }; ++ if tx.send(update).is_err() { ++ return; ++ } ++ } ++} ++ ++/// Decodes whole utterances only; used when a second model instance keeps finals off the ++/// partial worker's thread. ++fn run_final_worker(model: Canary, rx: Receiver, tx: Sender, opts: &Options) { ++ let mut session = model.session(); ++ for batch in Batches::new(rx) { ++ // Only the newest utterance is worth the wait if several piled up. ++ let last = batch ++ .into_iter() ++ .filter_map(|job| match job { ++ Job::Final { seq, audio } => Some((seq, audio)), ++ _ => None, ++ }) ++ .next_back(); ++ let Some((seq, audio)) = last else { continue }; ++ if decode_final(&mut session, &audio, seq, &tx, opts).is_err() { ++ return; ++ } ++ } ++} ++ ++fn decode_final( ++ session: &mut CanarySession, ++ audio: &[f32], ++ seq: u64, ++ tx: &Sender, ++ opts: &Options, ++) -> Result<(), ()> { ++ let started = Instant::now(); ++ let result = session.transcribe_samples( ++ audio, ++ MODEL_SAMPLE_RATE as usize, ++ 1, ++ &opts.source_lang, ++ &opts.target_lang, ++ ); ++ let update = match result { ++ Ok(result) => { ++ let text = normalize_punctuation_spacing(result.text.trim()); ++ if text.is_empty() { ++ Update::Discard { seq } ++ } else { ++ Update::Final { ++ seq, ++ text, ++ stats: Stats { ++ decode: started.elapsed(), ++ lag: started.elapsed(), ++ dropped: 0, ++ }, ++ } ++ } ++ } ++ Err(err) => Update::Error(format!("final decode failed: {}", err)), ++ }; ++ tx.send(update).map_err(|_| ()) ++} ++ ++fn steps_in(pending: &[f32], opts: &Options) -> usize { ++ let step = (opts.step * MODEL_SAMPLE_RATE as f32).max(1.0) as usize; ++ pending.len() / step ++} ++ ++/// Yields everything queued on a channel as one batch, blocking until at least one job arrives. ++struct Batches { ++ rx: Receiver, ++} ++ ++impl Batches { ++ fn new(rx: Receiver) -> Self { ++ Self { rx } ++ } ++} ++ ++impl Iterator for Batches { ++ type Item = Vec; ++ ++ fn next(&mut self) -> Option> { ++ let mut batch = vec![self.rx.recv().ok()?]; ++ while let Ok(job) = self.rx.try_recv() { ++ batch.push(job); ++ } ++ Some(batch) ++ } ++} ++ ++// --------------------------------------------------------------------------------------------- ++// Rendering ++// --------------------------------------------------------------------------------------------- ++ ++/// Draws a single live line: the stable transcript, then the still-changing tail dimmed. ++struct Renderer { ++ width: usize, ++ stats: bool, ++ seq: u64, ++ committed: String, ++ volatile: String, ++ line_open: bool, ++} ++ ++impl Renderer { ++ fn new(opts: &Options) -> Self { ++ Self { ++ width: opts.width.max(20), ++ stats: opts.stats, ++ seq: 0, ++ committed: String::new(), ++ volatile: String::new(), ++ line_open: false, ++ } ++ } ++ ++ fn apply(&mut self, update: Update) { ++ match update { ++ Update::Partial { ++ seq, ++ committed, ++ volatile, ++ stats, ++ } => { ++ if seq != self.seq { ++ self.seq = seq; ++ } ++ self.committed = committed; ++ self.volatile = volatile; ++ self.draw(); ++ self.report(stats, "partial"); ++ } ++ Update::Final { seq, text, stats } => { ++ // A final can land after the next utterance has already started drawing; print ++ // it above the live line and restore what was there. ++ self.commit_line(&text); ++ if seq == self.seq { ++ self.committed.clear(); ++ self.volatile.clear(); ++ } else { ++ self.draw(); ++ } ++ self.report(stats, "final"); ++ } ++ Update::Discard { seq } => { ++ if seq == self.seq { ++ self.clear_line(); ++ self.committed.clear(); ++ self.volatile.clear(); ++ } ++ } ++ Update::Error(message) => { ++ self.clear_line(); ++ eprintln!("{}", message); ++ self.draw(); ++ } ++ } ++ } ++ ++ fn draw(&mut self) { ++ let text = visible_tail(&self.committed, &self.volatile, self.width); ++ if text.is_empty() { ++ self.clear_line(); ++ return; ++ } ++ print!("\r\x1b[2K{}", text); ++ let _ = std::io::stdout().flush(); ++ self.line_open = true; ++ } ++ ++ fn commit_line(&mut self, text: &str) { ++ print!("\r\x1b[2K{}\n", text); ++ let _ = std::io::stdout().flush(); ++ self.line_open = false; ++ } ++ ++ fn clear_line(&mut self) { ++ if self.line_open { ++ print!("\r\x1b[2K"); ++ let _ = std::io::stdout().flush(); ++ self.line_open = false; ++ } ++ } ++ ++ fn report(&self, stats: Stats, kind: &str) { ++ if !self.stats { ++ return; ++ } ++ eprint!( ++ "\r\x1b[2K[{} {:>5} ms, lag {:>5} ms, {} dropped]\n", ++ kind, ++ stats.decode.as_millis(), ++ stats.lag.as_millis(), ++ stats.dropped ++ ); ++ } ++ ++ fn finish(&mut self) { ++ if self.line_open { ++ println!(); ++ self.line_open = false; ++ } ++ } + } + +-fn append_with_space(line: &mut String, chunk: &str) { +- if chunk.is_empty() { ++/// Builds the display line, keeping the most recent `width` characters — the tail is where the ++/// speaker is. The unstable part is dimmed, and the stable part is cut back to whole words. ++fn visible_tail(committed: &str, volatile: &str, width: usize) -> String { ++ const DIM: &str = "\x1b[2m"; ++ const RESET: &str = "\x1b[0m"; ++ /// Below this, a truncated prefix is more distracting than helpful. ++ const MIN_COMMITTED: usize = 16; ++ ++ let volatile_len = volatile.chars().count(); ++ if volatile_len + MIN_COMMITTED >= width { ++ return format!("{}{}{}", DIM, tail_words(volatile, width), RESET); ++ } ++ ++ let separator = if committed.is_empty() || volatile.is_empty() { ++ "" ++ } else { ++ " " ++ }; ++ let committed = tail_words(committed, width - volatile_len - separator.len()); ++ if volatile.is_empty() { ++ return committed; ++ } ++ format!("{}{}{}{}{}", committed, separator, DIM, volatile, RESET) ++} ++ ++/// Last `max` characters of `text`, cut at a word boundary and marked with an ellipsis. ++fn tail_words(text: &str, max: usize) -> String { ++ if text.chars().count() <= max { ++ return text.to_string(); ++ } ++ let keep = max.saturating_sub(2); ++ let tail: String = text.chars().skip(text.chars().count() - keep).collect(); ++ let tail = match tail.find(char::is_whitespace) { ++ Some(space) => &tail[space + 1..], ++ None => tail.as_str(), ++ }; ++ format!("…{}", tail) ++} ++ ++// --------------------------------------------------------------------------------------------- ++// Text assembly ++// --------------------------------------------------------------------------------------------- ++ ++/// Appends newly stabilized words, dropping punctuation the previous delta already ended with. ++fn append_delta(line: &mut String, delta: &str) { ++ let delta = normalize_punctuation_spacing(delta); ++ let delta = strip_duplicate_leading_punct(line, &delta); ++ if delta.is_empty() { + return; + } +- let needs_space = !line.is_empty() && !starts_with_punct(chunk); +- if needs_space { ++ if !line.is_empty() && !starts_with_punct(&delta) { + line.push(' '); + } +- line.push_str(chunk); ++ line.push_str(&delta); ++} ++ ++/// The part of the newest window's text that has not been committed yet. ++fn volatile_tail(committed: &str, window_text: &str) -> String { ++ let window_words: Vec<&str> = window_text.split_whitespace().collect(); ++ if window_words.is_empty() { ++ return String::new(); ++ } ++ let committed_words: Vec<&str> = committed.split_whitespace().collect(); ++ ++ // Compare loosely: the same word routinely comes back from a later window with different ++ // capitalisation or a comma attached, and an exact match would then show it twice. ++ let max_len = committed_words.len().min(window_words.len()); ++ let overlap = (1..=max_len) ++ .rev() ++ .find(|&len| { ++ committed_words[committed_words.len() - len..] ++ .iter() ++ .zip(&window_words[..len]) ++ .all(|(a, b)| same_word(a, b)) ++ }) ++ .unwrap_or(0); ++ normalize_punctuation_spacing(&window_words[overlap..].join(" ")) ++} ++ ++/// Whether a window collapsed into a repetition loop. ++/// ++/// Near-silent or clipped audio can send the decoder into emitting one token over and over until ++/// it hits the length cap; the result is noise on screen and is never worth showing. ++fn is_degenerate(text: &str) -> bool { ++ let words: Vec<&str> = text.split_whitespace().collect(); ++ if words.len() < 8 { ++ return false; ++ } ++ let mut counts: std::collections::HashMap = std::collections::HashMap::new(); ++ for word in &words { ++ *counts.entry(word.to_lowercase()).or_default() += 1; ++ } ++ counts.values().copied().max().unwrap_or(0) * 2 > words.len() ++} ++ ++/// Whether two words are the same once case and attached punctuation are ignored. ++fn same_word(a: &str, b: &str) -> bool { ++ let trim = |word: &str| { ++ word.trim_matches(|ch: char| !ch.is_alphanumeric()) ++ .to_lowercase() ++ }; ++ trim(a) == trim(b) + } + + fn starts_with_punct(text: &str) -> bool { +@@ -314,14 +950,9 @@ fn strip_duplicate_leading_punct(line: &str, chunk: &str) -> String { + + let mut chars = chunk.chars().peekable(); + if let Some(last_punct) = last_punct { +- while let Some(&ch) = chars.peek() { +- if ch == last_punct { +- chars.next(); +- continue; +- } +- break; ++ while chars.peek() == Some(&last_punct) { ++ chars.next(); + } + } +- + chars.collect() + } +diff --git a/examples/shared/utils.rs b/examples/shared/utils.rs +index 81eaf57..945f22e 100644 +--- a/examples/shared/utils.rs ++++ b/examples/shared/utils.rs +@@ -7,6 +7,9 @@ use canary_rs::{CoreMLComputeUnits, ExecutionConfig, ExecutionProvider, SessionC + /// - `CANARY_COREML_COMPUTE_UNITS`: `all`, `cpu_and_ne` (default), `cpu_and_gpu`, `cpu_only` + /// - `CANARY_COREML_LOW_PRECISION`: `1` or `true` + /// - `CANARY_CUDA_DEVICE_ID`: integer device index (default: `0`) ++/// - `CANARY_INTER_THREADS` / `CANARY_INTRA_THREADS`: ORT thread pool sizes ++/// - `CANARY_REQUIRE_EP`: `0`/`false` to allow a silent CPU fallback when the requested ++/// execution provider cannot be loaded (default: fail loudly) + /// - `CANARY_ITN`: `1`/`true` = force `<|itn|>`, `0`/`false` = force `<|noitn|>`, unset = omit token + pub fn execution_config_from_env() -> ExecutionConfig { + let provider = s