diff options
Diffstat (limited to 'vendor/patches')
| -rw-r--r-- | vendor/patches/canary-rs.patch | 1767 | ||||
| -rw-r--r-- | vendor/patches/llama.cpp.patch | 13 | ||||
| -rw-r--r-- | vendor/patches/qwentts.cpp.patch | 41 |
3 files changed, 1821 insertions, 0 deletions
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<dyn std::error::Error>> { +- // 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::<f32>::with_capacity(chunk_samples * 2))); +- let (tx, rx) = mpsc::channel::<Vec<f32>>(); +- +- 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::<Job>(); ++ let (update_tx, update_rx) = mpsc::channel::<Update>(); ++ ++ let mut workers: Vec<JoinHandle<()>> = Vec::new(); ++ let final_tx = match final_model { ++ Some(final_model) => { ++ let (final_tx, final_rx) = mpsc::channel::<Job>(); ++ 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::<Capture>(); ++ 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<f32> = 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{:<width$}", "", width = last_len); +- } +- print!("\r{}", final_text); +- println!(); +- let _ = std::io::stdout().flush(); +- last_len = 0; +- } +- } else if last_len > 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{:<width$}", final_line, width = last_len); +- let _ = std::io::stdout().flush(); +- } ++impl Options { ++ fn from_env() -> 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<f32> { ++ std::env::var(key).ok()?.trim().parse().ok() ++} ++ ++fn env_usize(key: &str) -> Option<usize> { ++ std::env::var(key).ok()?.trim().parse().ok() ++} ++ ++fn env_bool(key: &str) -> Option<bool> { ++ 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<f32>, ++ 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<f32>, ++ at: Instant, ++ }, ++ /// A finished utterance, to be decoded in full. ++ Final { seq: u64, audio: Vec<f32> }, ++ /// 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<Capture>) -> Result<Self, Box<dyn std::error::Error>> { ++ 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<SupportedStreamConfig, Box<dyn std::error::Error>> { ++ 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<T: Sample>( +- input: &[T], +- chunk_samples: usize, +- buffer: &Arc<Mutex<Vec<f32>>>, +- sender: &mpsc::Sender<Vec<f32>>, +-) where +- f32: cpal::FromSample<T>, +-{ +- 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<Capture>, ++ job_tx: Sender<Job>, ++ final_tx: Option<Sender<Job>>, ++ 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<f32> = Vec::with_capacity(frame_samples * 2); ++ let mut preroll: std::collections::VecDeque<f32> = std::collections::VecDeque::new(); ++ let mut utterance: Vec<f32> = 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<f32> = frame.drain(..frame_samples).collect(); ++ let level = rms(&chunk); + +- while buffer.len() >= chunk_samples { +- let chunk: Vec<f32> = 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<f32>, format: InputFormat) -> Vec<f32> { ++ let mono = if format.channels > 1 { ++ samples ++ .chunks(format.channels) ++ .map(|frame| frame.iter().sum::<f32>() / 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<Job>, tx: Sender<Update>, 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<f32> = 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<Job>, tx: Sender<Update>, 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; ++ } ++ }</ |