From f7cdb2984a90e0c4bb9390a11ca610d4e8cbac8c Mon Sep 17 00:00:00 2001 From: will wade Date: Sun, 16 Aug 2026 14:56:15 +0000 Subject: [PATCH] feat(sherpaonnx): stream audio per sentence batch via the generate callback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The crate's README (and engine docs) claimed sherpa-onnx could not stream — it can: the generate progress callback receives each batch's NEWLY generated samples (verified in sherpa-onnx csrc: the vits/kokoro/ matcha/supertonic impls call the callback after every sentence batch; the Rust crate's 'samples generated so far' doc comment was wrong). With max_num_sentences=1 already set, batches are sentence-sized. - speak() now emits each batch through on_audio immediately (volume/ pitch applied per batch — batches are sentence-aligned, so per-batch resampling never seams mid-speech), giving sentence-level streaming for multi-sentence utterances - estimated word boundaries fire progressively via the shared EstimateFirer (1/speed time-scale anchors estimates to delivered audio; reported times stay on the rate-1.0 baseline for existing rate-compensating callers) - the generate callback must be 'static: on_audio/on_boundary are stashed as lifetime-erased pointers in thread-locals for the synchronous call (same-thread callback per the C++ docs; serialised by the tts_instance mutex), and the firer is shared via Arc so the outer scope flushes the remainder afterwards - EstimateFirer/EstimatePlan moved to a new always-compiled boundaries module (shared by cloud + sherpa); EstimateFirer now owns its plan so it can move into 'static callbacks Live-verified with piper-nl-rdh-low: a boundary event provably fires before the final audio chunk and first audio precedes the last (sherpa_streams_audio_per_sentence_batch). js-tts-wrapper parity note: its synthToBytestream enqueues one whole-clip buffer — this goes further. --- README.md | 2 +- src/boundaries.rs | 238 +++++++++++++++++++++++++++++++++++++++ src/cloud_engine.rs | 197 ++++++++++---------------------- src/lib.rs | 1 + src/sherpaonnx_engine.rs | 209 +++++++++++++++++++++++++++------- tests/sherpaonnx_live.rs | 69 ++++++++++++ 6 files changed, 538 insertions(+), 178 deletions(-) create mode 100644 src/boundaries.rs diff --git a/README.md b/README.md index a2dd915..8586b6c 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ Cross-platform TTS (Text-to-Speech) wrapper with C ABI. Mirrors [js-tts-wrapper] | xAI | Cloud | API Key | Chunked | — | Estimated | Platform-aware | | ModelsLab | Cloud | API Key | Chunked | — | Estimated | Platform-aware | -- **Streaming**: Audio is delivered through the `on_audio` callback in chunks. REST engines and Edge stream as bytes arrive over the network (MP3 is decoded to PCM16 mono incrementally, on a background reader thread); Azure's WebSocket delivers PCM frames per message. Engines whose APIs return a single JSON document with base64 audio (Google, ElevenLabs `with-timestamps`) deliver only once the response completes — an API limitation. Sherpa-ONNX synthesises the whole clip first, then slices the rendered PCM into 8 KB chunks (matching the cloud delivery shape and the js-tts-wrapper / swift-tts-wrapper siblings). Estimated word boundaries (engines without API timing data) fire progressively during streaming, anchored to delivered audio, rather than all at once when the response completes. +- **Streaming**: Audio is delivered through the `on_audio` callback in chunks. REST engines and Edge stream as bytes arrive over the network (MP3 is decoded to PCM16 mono incrementally, on a background reader thread); Azure's WebSocket delivers PCM frames per message. Engines whose APIs return a single JSON document with base64 audio (Google, ElevenLabs `with-timestamps`) deliver only once the response completes — an API limitation. Sherpa-ONNX delivers each sentence batch as it is synthesised (sentence-level streaming via the generate progress callback; single-sentence utterances still complete before delivery). Estimated word boundaries (engines without API timing data) fire progressively during streaming, anchored to delivered audio, rather than all at once when the response completes. Estimated word boundaries (engines without API timing data) fire progressively during streaming, anchored to delivered audio, rather than all at once when the response completes. - **Native engine varies by platform**: the table shows `system` (Linux speech-dispatcher); macOS uses `avsynth` (AVSpeechSynthesizer) and Windows uses `sapi`. "22 total" counts one native engine + Sherpa-ONNX + the 20 cloud engines, per platform. ## Formatting & Testing diff --git a/src/boundaries.rs b/src/boundaries.rs new file mode 100644 index 0000000..a02dd4c --- /dev/null +++ b/src/boundaries.rs @@ -0,0 +1,238 @@ +//! Progressive word-boundary firing, shared by the cloud and sherpa-onnx +//! engines. +//! +//! Engines without real API timing data get 150-wpm estimates. Firing +//! them in one batch after synthesis leaves callers that interleave marks +//! with playback (e.g. the VoiceGarden-SPD speech-dispatcher module) +//! unable to highlight in sync on long utterances. [`EstimateFirer`] +//! anchors the estimates onto the delivered-audio clock instead: estimate +//! *i* fires once ≥ its start-time worth of PCM has actually been emitted. + +use crate::engine::estimate_word_boundaries; +use crate::types::WordBoundary; + +/// One estimated boundary event with source-text position resolved. +#[derive(Debug, Clone)] +pub struct EstimateEvent { + /// The spoken word. + pub word: String, + /// Estimate start time in seconds (rate-1.0 baseline). + pub start_s: f32, + /// Estimate end time in seconds (rate-1.0 baseline). + pub end_s: f32, + /// Byte offset into the spoken plain text (-1 when unresolvable). + pub char_offset: i32, + /// Character length of the word. + pub char_len: i32, +} + +/// Pre-resolved estimated boundaries for an utterance, in firing order. +pub struct EstimatePlan { + events: Vec, +} + +impl EstimatePlan { + /// Build from the crate's 150-wpm estimator, resolving char offsets in + /// the spoken text. SSML input is stripped first so offsets and word + /// lists match what is actually spoken. + #[must_use] + pub fn build(text: &str) -> Self { + let plain = if text.trim_start().to_ascii_lowercase().starts_with(" Self { + let mut events = Vec::with_capacity(estimated.len()); + let mut search_from = 0usize; + for b in estimated { + #[allow(clippy::cast_possible_truncation)] + let char_offset = plain[search_from..] + .find(&b.text) + .map_or(-1, |pos| (search_from + pos) as i32); + if char_offset >= 0 { + search_from = char_offset as usize + b.text.len(); + } + #[allow(clippy::cast_precision_loss)] + let start = b.offset as f32 / 1000.0; + #[allow(clippy::cast_precision_loss)] + let end = (b.offset + b.duration) as f32 / 1000.0; + let char_len = b.text.chars().count() as i32; + events.push(EstimateEvent { + word: b.text.clone(), + start_s: start, + end_s: end, + char_offset, + char_len, + }); + } + Self { events } + } + + /// Number of events in the plan. + #[must_use] + pub fn len(&self) -> usize { + self.events.len() + } + + /// The event at `idx` (firing order). + #[must_use] + pub fn event(&self, idx: usize) -> Option<&EstimateEvent> { + self.events.get(idx) + } + + /// True when the plan has no events. + #[must_use] + pub fn is_empty(&self) -> bool { + self.events.is_empty() + } +} + +/// Fires [`EstimatePlan`] events as cumulative delivered audio crosses +/// each estimate's start time (scaled by `time_scale` to account for the +/// engine's actual speech rate). Owns the plan so it can be moved into +/// 'static callbacks. +pub struct EstimateFirer { + plan: Box, + next: usize, + samples: u64, + rate: Option, + time_scale: f32, +} + +impl EstimateFirer { + /// Create a firer for `plan`. `time_scale` converts estimate seconds + /// into delivered-audio seconds (e.g. `1/speed` for engines whose + /// rate parameter compresses duration; 1.0 when estimates already + /// match delivery). + #[must_use] + pub fn new(plan: EstimatePlan, time_scale: f32) -> Self { + Self { + plan: Box::new(plan), + next: 0, + samples: 0, + rate: None, + time_scale, + } + } + + /// Record `samples` newly-emitted PCM16-mono samples and fire every + /// estimate whose (scaled) start time has been reached. + pub fn on_samples( + &mut self, + samples: u64, + rate_now: Option, + fire: &mut dyn FnMut(&EstimateEvent), + ) { + self.samples += samples; + if let Some(r) = rate_now { + self.rate = Some(r); + } + let Some(rate) = self.rate else { return }; + #[allow(clippy::cast_precision_loss)] + let rate_f = rate as f32; + while self.next < self.plan.events.len() { + let e = &self.plan.events[self.next]; + #[allow(clippy::cast_precision_loss)] + let threshold = (e.start_s * self.time_scale * rate_f) as u64; + if self.samples >= threshold { + fire(e); + self.next += 1; + } else { + break; + } + } + } + + /// Fire every remaining estimate (stream ended before their times). + pub fn flush(&mut self, fire: &mut dyn FnMut(&EstimateEvent)) { + while self.next < self.plan.events.len() { + fire(&self.plan.events[self.next]); + self.next += 1; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn estimate_event(word: &str, start_s: f32, end_s: f32) -> EstimateEvent { + EstimateEvent { + word: word.into(), + start_s, + end_s, + char_offset: -1, + char_len: word.len() as i32, + } + } + + #[test] + fn firer_fires_at_scaled_thresholds() { + let events = vec![ + estimate_event("one", 0.0, 0.5), + estimate_event("two", 0.5, 1.0), + ]; + let plan = EstimatePlan { events }; + let mut firer = EstimateFirer::new(plan, 1.0); + let mut seen: Vec = Vec::new(); + firer.on_samples(6000, Some(24_000), &mut |e| seen.push(e.word.clone())); + assert_eq!(seen, vec!["one"], "0.25s of audio → only first word"); + firer.on_samples(6000, None, &mut |e| seen.push(e.word.clone())); + assert_eq!(seen, vec!["one", "two"], "0.5s total → second word"); + } + + #[test] + fn firer_applies_time_scale() { + // Speed 2× → audio half as long → estimates scale by 1/2. + let events = vec![ + estimate_event("one", 0.0, 0.5), + estimate_event("two", 0.5, 1.0), + ]; + let plan = EstimatePlan { events }; + let mut firer = EstimateFirer::new(plan, 0.5); + let mut count = 0usize; + firer.on_samples(6000, Some(24_000), &mut |_| count += 1); + assert_eq!(count, 2, "0.25s audio at 2× covers both estimates"); + } + + #[test] + fn firer_flush_fires_remainder() { + let events = vec![estimate_event("one", 10.0, 10.5)]; + let plan = EstimatePlan { events }; + let mut firer = EstimateFirer::new(plan, 1.0); + let mut count = 0usize; + firer.on_samples(1000, Some(8000), &mut |_| count += 1); + assert_eq!(count, 0); + firer.flush(&mut |_| count += 1); + assert_eq!(count, 1); + } + + #[test] + fn plan_from_estimates_resolves_offsets() { + use crate::types::WordBoundary; + let est = vec![ + WordBoundary { + text: "hello".into(), + offset: 0, + duration: 400, + }, + WordBoundary { + text: "world".into(), + offset: 400, + duration: 400, + }, + ]; + let plan = EstimatePlan::from_estimates(&est, "hello world"); + assert_eq!(plan.len(), 2); + assert_eq!(plan.events[0].char_offset, 0); + assert_eq!(plan.events[1].char_offset, 6); + } +} diff --git a/src/cloud_engine.rs b/src/cloud_engine.rs index 9794de1..6baf89c 100644 --- a/src/cloud_engine.rs +++ b/src/cloud_engine.rs @@ -19,6 +19,7 @@ pub fn set_viseme_callback(cb: Option>) { VISEME_CB.with(|cell| *cell.borrow_mut() = cb); } +use crate::boundaries::{EstimateFirer, EstimatePlan}; use crate::engine::{estimate_word_boundaries, preprocess_speech_markdown, TtsEngine}; use crate::types::{ normalize_gender, Gender, LanguageCode, TtsError, TtsResult, Voice, WordBoundary, @@ -462,14 +463,15 @@ enum StreamEvt<'x> { /// callers interleaving marks with playback (e.g. the VoiceGarden-SPD /// speech-dispatcher module) can report them in sync. #[cfg(feature = "cloud")] +#[allow(clippy::too_many_lines)] fn stream_body_to_on_audio( mut body: impl std::io::Read + Send + 'static, is_pcm: bool, pcm_rate: u32, - plan: Option<&EstimatePlan>, + plan: Option, on_event: &mut dyn FnMut(StreamEvt<'_>), ) -> Result { - let mut firer = plan.map(EstimateFirer::new); + let mut firer = plan.map(|p| EstimateFirer::new(p, 1.0)); if is_pcm { let mut buf = [0u8; STREAMING_CHUNK_SIZE]; @@ -485,13 +487,27 @@ fn stream_body_to_on_audio( total += n; if let Some(f) = firer.as_mut() { // PCM16 mono: 2 bytes per sample. - f.on_samples((n / 2) as u64, Some(pcm_rate), &mut |w, s, e, o, l| { - on_event(StreamEvt::Boundary(w, s, e, o, l)); + f.on_samples((n / 2) as u64, Some(pcm_rate), &mut |ev| { + on_event(StreamEvt::Boundary( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + )); }); } } if let Some(f) = firer.as_mut() { - f.flush(&mut |w, s, e, o, l| on_event(StreamEvt::Boundary(w, s, e, o, l))); + f.flush(&mut |ev| { + on_event(StreamEvt::Boundary( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + )); + }); } return Ok(total); } @@ -528,11 +544,15 @@ fn stream_body_to_on_audio( total += chunk.len(); if let Some(f) = firer.as_mut() { // Decoded PCM16 mono: one i16 per sample. - f.on_samples( - chunk.len() as u64, - dec.sample_rate(), - &mut |w, s, e, o, l| on_event(StreamEvt::Boundary(w, s, e, o, l)), - ); + f.on_samples(chunk.len() as u64, dec.sample_rate(), &mut |ev| { + on_event(StreamEvt::Boundary( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + )); + }); } } Ok(None) => break, @@ -546,7 +566,15 @@ fn stream_body_to_on_audio( } eprintln!("rust-tts-wrapper: streaming decode error after {total} bytes: {e}"); if let Some(f) = firer.as_mut() { - f.flush(&mut |w, s, e, o, l| on_event(StreamEvt::Boundary(w, s, e, o, l))); + f.flush(&mut |ev| { + on_event(StreamEvt::Boundary( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + )); + }); } return Ok(total); } @@ -554,129 +582,19 @@ fn stream_body_to_on_audio( } let _ = reader.join(); if let Some(f) = firer.as_mut() { - f.flush(&mut |w, s, e, o, l| on_event(StreamEvt::Boundary(w, s, e, o, l))); + f.flush(&mut |ev| { + on_event(StreamEvt::Boundary( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + )); + }); } Ok(total) } -// ============================================================================ -// Progressive estimated word boundaries -// ============================================================================ - -/// One estimated boundary event with source-text position resolved. -#[cfg(feature = "cloud")] -struct EstimateEvent { - word: String, - start_s: f32, - end_s: f32, - char_offset: i32, - char_len: i32, -} - -/// Pre-resolved estimated boundaries for an utterance, in firing order. -#[cfg(feature = "cloud")] -struct EstimatePlan { - events: Vec, -} - -impl EstimatePlan { - /// Build from the crate's 150-wpm estimator, resolving char offsets in - /// the spoken text. SSML input is stripped first so offsets and word - /// lists match what is actually spoken. - #[must_use] - fn build(text: &str) -> Self { - let plain = if text.trim_start().to_ascii_lowercase().starts_with("= 0 { - search_from = char_offset as usize + b.text.len(); - } - #[allow(clippy::cast_precision_loss)] - let start = b.offset as f32 / 1000.0; - #[allow(clippy::cast_precision_loss)] - let end = (b.offset + b.duration) as f32 / 1000.0; - let char_len = b.text.chars().count() as i32; - events.push(EstimateEvent { - word: b.text.clone(), - start_s: start, - end_s: end, - char_offset, - char_len, - }); - } - Self { events } - } -} - -/// Fires [`EstimatePlan`] events as cumulative delivered audio crosses -/// each estimate's start time. Anchors the text-based estimates onto the -/// real audio clock: if the voice speaks slower than the 150-wpm -/// baseline, marks still fire in sync with what the caller has actually -/// emitted (late words clamp to the final flush). -#[cfg(feature = "cloud")] -struct EstimateFirer<'a> { - plan: &'a EstimatePlan, - next: usize, - samples: u64, - rate: Option, -} - -#[cfg(feature = "cloud")] -impl<'a> EstimateFirer<'a> { - fn new(plan: &'a EstimatePlan) -> Self { - Self { - plan, - next: 0, - samples: 0, - rate: None, - } - } - - /// Record `samples` newly-emitted PCM16-mono samples and fire every - /// estimate whose start time has been reached. - fn on_samples( - &mut self, - samples: u64, - rate_now: Option, - fire: &mut dyn FnMut(&str, f32, f32, i32, i32), - ) { - self.samples += samples; - if let Some(r) = rate_now { - self.rate = Some(r); - } - let Some(rate) = self.rate else { return }; - while self.next < self.plan.events.len() { - let e = &self.plan.events[self.next]; - #[allow(clippy::cast_precision_loss)] - let threshold = (e.start_s * rate as f32) as u64; - if self.samples >= threshold { - fire(&e.word, e.start_s, e.end_s, e.char_offset, e.char_len); - self.next += 1; - } else { - break; - } - } - } - - /// Fire every remaining estimate (stream ended before their times). - fn flush(&mut self, fire: &mut dyn FnMut(&str, f32, f32, i32, i32)) { - while self.next < self.plan.events.len() { - let e = &self.plan.events[self.next]; - fire(&e.word, e.start_s, e.end_s, e.char_offset, e.char_len); - self.next += 1; - } - } -} /// Sniff the first few bytes for an MP3 sync word or ID3 tag. Kept as a /// diagnostic helper but not used for delivery routing — raw PCM16 audio /// frequently contains 0xFF 0xE0+ byte pairs that false-positive, so format @@ -2384,7 +2302,7 @@ impl TtsEngine for CloudEngine { // Raw-PCM providers here (Azure, Cartesia) are pinned to // 24 kHz in their CloudConfigs. 24_000, - plan.as_ref(), + plan, &mut on_event, ) .map_err(TtsError)?; @@ -3263,7 +3181,8 @@ mod tests { // the flush path must cover audio shorter than the estimates. let mp3 = make_silent_mp3(80); // ≈ 2.1 s of audio let plan = EstimatePlan::build("one two three four five six seven"); - assert!(!plan.events.is_empty()); + let expected = plan.len(); + assert!(expected > 0); let reader = DribbleReader { data: mp3, @@ -3272,14 +3191,14 @@ mod tests { }; let mut audio_chunks = 0usize; let mut boundaries: Vec = Vec::new(); - stream_body_to_on_audio(reader, false, 24_000, Some(&plan), &mut |ev| match ev { + stream_body_to_on_audio(reader, false, 24_000, Some(plan), &mut |ev| match ev { StreamEvt::Audio(_) => audio_chunks += 1, StreamEvt::Boundary(word, ..) => boundaries.push(word.to_string()), }) .expect("stream"); assert!(!boundaries.is_empty(), "no boundaries fired"); // Every estimate eventually fired (flush covers short audio). - assert_eq!(boundaries.len(), plan.events.len()); + assert_eq!(boundaries.len(), expected); assert!(audio_chunks > 1); } @@ -3307,7 +3226,7 @@ mod tests { StreamEvt::Audio(..) => seq.push(AudioEvt), StreamEvt::Boundary(..) => seq.push(BoundaryEvt), }; - stream_body_to_on_audio(reader, false, 24_000, Some(&plan), &mut record).expect("stream"); + stream_body_to_on_audio(reader, false, 24_000, Some(plan), &mut record).expect("stream"); // Find a Boundary that is followed by at least one more Audio → // it fired during streaming, not at the end flush. let interleaved = seq @@ -3324,12 +3243,14 @@ mod tests { fn estimate_plan_strips_ssml_before_estimating() { let plain = EstimatePlan::build("hello world"); let ssml = EstimatePlan::build("hello world"); - assert_eq!(plain.events.len(), ssml.events.len()); - let words: Vec<&str> = ssml.events.iter().map(|e| e.word.as_str()).collect(); + assert_eq!(plain.len(), ssml.len()); + let words: Vec = (0..ssml.len()) + .map(|i| ssml.event(i).expect("in range").word.clone()) + .collect(); assert_eq!(words, vec!["hello", "world"]); // Offsets resolved into the stripped text: "world" found at a // valid position (not -1). - assert!(ssml.events[1].char_offset > 0); + assert!(ssml.event(1).expect("in range").char_offset > 0); } #[test] diff --git a/src/lib.rs b/src/lib.rs index 167d720..4b5e22a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -32,6 +32,7 @@ #[cfg(all(feature = "avsynth", target_os = "macos"))] mod avsynth_engine; +pub mod boundaries; #[cfg(feature = "cloud")] mod cloud_engine; pub mod engine; diff --git a/src/sherpaonnx_engine.rs b/src/sherpaonnx_engine.rs index d3b0e82..9e9b024 100644 --- a/src/sherpaonnx_engine.rs +++ b/src/sherpaonnx_engine.rs @@ -1,6 +1,7 @@ //! Sherpa-ONNX offline TTS engine with model registry. -use crate::engine::{estimate_word_boundaries, TtsEngine}; +use crate::boundaries::{EstimateFirer, EstimatePlan}; +use crate::engine::TtsEngine; use crate::types::{ Gender, LanguageCode, SherpaLanguage, SherpaModelInfo, TtsError, TtsResult, Voice, }; @@ -8,7 +9,7 @@ use std::collections::HashMap; use std::fmt; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; /// Embedded model registry compiled from `models.json`. static MODELS_JSON: &str = include_str!("models.json"); @@ -16,6 +17,22 @@ static MODELS_JSON: &str = include_str!("models.json"); /// Shared cancellation flag — set by `stop()`, read by the progress callback. static CANCEL_REQUESTED: AtomicBool = AtomicBool::new(false); +// The sherpa-onnx generate callback must be 'static, but speak() holds the +// caller's on_audio/on_boundary borrows only for the method body. The C++ +// runtime invokes the callback synchronously on the same thread inside the +// generate call, so we stash the callback pointers here for the duration +// (same technique as VISEME_CB in the cloud engine) and clear them after. +// Synthesis per engine instance is serialised by the tts_instance mutex. +type AudioCbPtr = *mut dyn FnMut(&[u8]); +type BoundaryCbPtr = *mut dyn FnMut(&str, f32, f32, i32, i32); + +thread_local! { + static STREAM_AUDIO_CB: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; + static STREAM_BOUNDARY_CB: std::cell::RefCell> = + const { std::cell::RefCell::new(None) }; +} + /// PCM delivery chunk size. Sherpa-ONNX synthesises the whole clip up-front, /// so we slice the rendered PCM into 8 KB chunks before pushing them through /// `on_audio` — matching the cloud engines' streamed-chunk shape so callers @@ -454,53 +471,156 @@ impl TtsEngine for SherpaOnnxEngine { // Reset cancellation flag before synthesis. CANCEL_REQUESTED.store(false, Ordering::SeqCst); - // Sherpa-ONNX synthesises the whole clip before its `generate` call - // returns (the progress callback reports cumulative samples, not a - // streamable pipe), so we generate once, apply volume/pitch, then - // deliver the PCM through `on_audio` in 8 KB chunks. This matches the - // delivery shape the cloud engines and the js-tts-wrapper / swift- - // tts-wrapper siblings use: callers receive multiple `on_audio` - // callbacks rather than one monolithic buffer. - let audio = tts - .generate_with_config( + // Streaming delivery: sherpa-onnx's generate callback fires after + // each sentence batch (max_num_sentences=1 above) with the batch's + // NEWLY generated samples — not cumulative — so audio is delivered + // through `on_audio` while later sentences are still synthesising. + // Volume/pitch are applied per batch (batches are sentence-aligned, + // so per-batch resampling for pitch doesn't seam mid-speech). + // + // Word-boundary estimates (150-wpm baseline) fire progressively, + // anchored to delivered samples and scaled by 1/speed so they track + // the audio actually emitted; the reported times stay on the + // rate-1.0 baseline (existing callers, e.g. VoiceGarden-SPD and the + // SAPI adapter, compensate for rate themselves). + // + // The generate callback must be `'static`, so it cannot capture the + // method-lifetime `on_audio`/`on_boundary` borrows directly. The C++ + // runtime invokes the callback synchronously on this same thread + // while the borrows are live, so the pointers are stashed in + // thread-locals for the duration of the call (the same technique as + // VISEME_CB above) and cleared afterwards. + #[allow(clippy::cast_precision_loss)] + let time_scale = 1.0 / rate.max(0.1); + let sample_rate_out = model_info.sample_rate.max(1); + + if wants_callback { + let plan = on_boundary.is_some().then(|| EstimatePlan::build(text)); + + // SAFETY (stash): the raw pointers are only dereferenced inside + // the generate callback, which sherpa-onnx calls synchronously + // on this thread between the stash and the clear below, while + // `on_audio`/`on_boundary` are still borrowed. Synthesis on a + // given engine instance is serialised by the `tts_instance` + // mutex held for the whole call. + // ptr→ptr transmute + explicit borrow-to-pointer are the + // standard lifetime-erasure idioms for synchronous callback + // stashing; the safety argument is documented above. + #[allow(clippy::transmute_ptr_to_ptr)] + let audio_ptr: Option = on_audio.as_mut().map(|cb| { + // SAFETY: the fat pointer's layout is identical; only the + // lifetime is erased. See the thread-local docs for why + // use is confined to this call. + unsafe { + std::mem::transmute::<*mut (dyn FnMut(&[u8]) + '_), AudioCbPtr>( + std::ptr::from_mut(&mut **cb), + ) + } + }); + #[allow(clippy::transmute_ptr_to_ptr)] + let boundary_ptr: Option = on_boundary.as_mut().map(|cb| { + // SAFETY: as above. + unsafe { + std::mem::transmute::< + *mut (dyn FnMut(&str, f32, f32, i32, i32) + '_), + BoundaryCbPtr, + >(std::ptr::from_mut(&mut **cb)) + } + }); + STREAM_AUDIO_CB.with(|c| *c.borrow_mut() = audio_ptr); + STREAM_BOUNDARY_CB.with(|c| *c.borrow_mut() = boundary_ptr); + + // The firer owns the plan and is shared with the 'static + // callback via Arc>, so the outer scope can flush + // the remainder after generation ends. + let firer = plan.map(|p| Arc::new(Mutex::new(EstimateFirer::new(p, time_scale)))); + let firer_for_cb = firer.clone(); + + // A `None` result means the engine rejected the config outright; + // cancellation mid-stream still returns Some(audio-so-far), + // which we ignore — every batch was already streamed. + let result = tts.generate_with_config( text, &gen_config, - Some(|_s: &[f32], _p: f32| -> bool { !CANCEL_REQUESTED.load(Ordering::SeqCst) }), - ) - .ok_or_else(|| TtsError("SherpaOnnx synthesis returned no audio".into()))?; - let sample_rate = audio.sample_rate(); - let processed = apply_volume_and_pitch(audio.samples(), volume_factor, pitch_factor); + Some(move |batch: &[f32], _progress: f32| -> bool { + if CANCEL_REQUESTED.load(Ordering::SeqCst) { + return false; + } + let processed = apply_volume_and_pitch(batch, volume_factor, pitch_factor); + let pcm = samples_to_le_bytes(&processed); + STREAM_AUDIO_CB.with(|c| { + if let Some(ptr) = *c.borrow() { + // SAFETY (stash): see the thread-local docs. + unsafe { (*ptr)(&pcm) }; + } + }); + if let Some(f) = firer_for_cb.as_ref() { + if let Ok(mut guard) = f.lock() { + guard.on_samples( + batch.len() as u64, + Some(sample_rate_out), + &mut |ev| { + STREAM_BOUNDARY_CB.with(|c| { + if let Some(ptr) = *c.borrow() { + // SAFETY (stash): see above. + unsafe { + (*ptr)( + &ev.word, + ev.start_s, + ev.end_s, + ev.char_offset, + ev.char_len, + ); + } + } + }); + }, + ); + } + } + !CANCEL_REQUESTED.load(Ordering::SeqCst) + }), + ); - if wants_callback { - if let Some(cb) = on_audio.as_mut() { - // Volume + pitch already baked into `processed` above. - deliver_pcm(*cb, &processed, 1.0); + STREAM_AUDIO_CB.with(|c| *c.borrow_mut() = None); + STREAM_BOUNDARY_CB.with(|c| *c.borrow_mut() = None); + + if result.is_none() { + return Err(TtsError("SherpaOnnx synthesis returned no audio".into())); + } + + // Fire any estimates whose audio never arrived (short audio, + // cancellation): the boundary set is now closed. + if let (Some(f), Some(cb)) = (firer.as_ref(), on_boundary.as_mut()) { + if let Ok(mut f) = f.lock() { + f.flush(&mut |ev| { + cb(&ev.word, ev.start_s, ev.end_s, ev.char_offset, ev.char_len); + }); + } } } else { + let audio = tts + .generate_with_config( + text, + &gen_config, + Some(|_s: &[f32], _p: f32| -> bool { + !CANCEL_REQUESTED.load(Ordering::SeqCst) + }), + ) + .ok_or_else(|| TtsError("SherpaOnnx synthesis returned no audio".into()))?; + let processed = apply_volume_and_pitch(audio.samples(), volume_factor, pitch_factor); let filename = std::env::temp_dir().join("rust-tts-wrapper-sherpa.wav"); - if write_wav(&filename, &processed, sample_rate) { + if write_wav(&filename, &processed, audio.sample_rate()) { play_wav_file(&filename); } - } - if let Some(cb) = on_boundary.as_mut() { - let estimated = estimate_word_boundaries(text); - let mut search_from = 0usize; - for b in &estimated { - #[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)] - let char_offset = text[search_from..] - .find(&b.text) - .map_or(-1, |pos| (search_from + pos) as i32); - - if char_offset >= 0 { - search_from = char_offset as usize + b.text.len(); + // No streaming took place; estimates fire as before. + if let Some(cb) = on_boundary.as_mut() { + let plan = EstimatePlan::build(text); + for i in 0..plan.len() { + let ev = plan.event(i).expect("in range"); + cb(&ev.word, ev.start_s, ev.end_s, ev.char_offset, ev.char_len); } - #[allow(clippy::cast_precision_loss)] - let start = b.offset as f32 / 1000.0; - #[allow(clippy::cast_precision_loss)] - let end = (b.offset + b.duration) as f32 / 1000.0; - let char_len = b.text.chars().count() as i32; - cb(&b.text, start, end, char_offset, char_len); } } @@ -611,6 +731,17 @@ fn apply_volume_and_pitch(samples: &[f32], volume: f32, pitch: f32) -> Vec } } +/// Convert f32 samples to little-endian PCM16 bytes (one allocation). +#[allow(clippy::cast_possible_truncation)] +fn samples_to_le_bytes(samples: &[f32]) -> Vec { + let mut pcm = Vec::with_capacity(samples.len() * 2); + for &s in samples { + let s16 = (s.clamp(-1.0, 1.0) * 32767.0) as i16; + pcm.extend_from_slice(&s16.to_le_bytes()); + } + pcm +} + /// Scale `samples` by `volume_factor`, convert to little-endian PCM16 bytes, /// and push them through `cb` in `STREAMING_CHUNK_SIZE`-byte chunks. Volume /// and pitch are applied to the full buffer by `apply_volume_and_pitch` diff --git a/tests/sherpaonnx_live.rs b/tests/sherpaonnx_live.rs index 3cd2b11..4e27392 100644 --- a/tests/sherpaonnx_live.rs +++ b/tests/sherpaonnx_live.rs @@ -233,6 +233,75 @@ fn vits_piper_volume_changes_amplitude() { ); } +#[test] +#[ignore] +fn sherpa_streams_audio_per_sentence_batch() { + // The generate progress callback delivers each sentence batch as it is + // synthesised (max_num_sentences=1). With on_audio + on_boundary set, + // boundary events must interleave with audio chunks — i.e. at least one + // boundary fires BEFORE the final audio chunk, proving delivery is + // incremental rather than one end-batched buffer. Additionally, the + // first audio chunk must arrive before synthesis completes. + let id = model_id("SHERPA_VITS_MODEL", "piper-nl-rdh-low"); + let engine = engine_for(&id); + let text = "First sentence arrives early. Second sentence synthesises later. Third one closes the stream."; + + #[derive(PartialEq, Debug, Clone, Copy)] + enum Ev { + Audio, + Boundary, + } + use Ev::{Audio as A, Boundary as B}; + + let seq = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let t_start = std::time::Instant::now(); + let first_audio_at = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let last_audio_at = std::sync::Arc::new(std::sync::Mutex::new(t_start.elapsed())); + + let seq_a = seq.clone(); + let first_a = first_audio_at.clone(); + let last_a = last_audio_at.clone(); + let seq_b = seq.clone(); + engine + .speak( + text, + None, + 1.0, + 1.0, + 1.0, + Some(&mut move |_chunk: &[u8]| { + let mut f = first_a.lock().unwrap(); + if f.is_none() { + *f = Some(t_start.elapsed()); + } + *last_a.lock().unwrap() = t_start.elapsed(); + seq_a.lock().unwrap().push(A); + }), + Some(&mut move |_w, _s, _e, _o, _l| seq_b.lock().unwrap().push(B)), + ) + .expect("speak"); + + let seq = seq.lock().unwrap().clone(); + let total = t_start.elapsed(); + assert!(seq.contains(&A), "no audio delivered"); + assert!(seq.contains(&B), "no boundaries delivered"); + let interleaved = seq + .iter() + .enumerate() + .any(|(i, e)| *e == B && seq[i + 1..].contains(&A)); + assert!( + interleaved, + "no boundary fired before the final audio chunk; seq = {seq:?}" + ); + let first = first_audio_at.lock().unwrap().expect("audio delivered"); + let last = *last_audio_at.lock().unwrap(); + assert!( + first < last, + "all audio arrived in one batch (first {first:?} == last {last:?})" + ); + let _ = total; +} + #[test] #[ignore] fn vits_piper_word_boundaries_fire_per_word() {