From 1187fd222c0cfeb73b4ba24c68e9bf57769fe489 Mon Sep 17 00:00:00 2001 From: StuBehan Date: Wed, 8 Jul 2026 00:28:57 +0100 Subject: [PATCH 1/2] feat: stream synthesis sentence-by-sentence for low-latency playback --- pyproject.toml | 6 +++- stackvox/engine.py | 82 +++++++++++++++++++++++++++++++++++++++++--- tests/test_engine.py | 75 +++++++++++++++++++++++++++++++++------- 3 files changed, 145 insertions(+), 18 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 18e0a58..887198e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -100,7 +100,11 @@ show_missing = true skip_covered = false [tool.mypy] -python_version = "3.10" +# Matches the interpreter CI runs mypy under, and must be >= 3.12: numpy's +# bundled stubs use PEP 695 `type` aliases, which mypy rejects as a syntax +# error when told to target an older Python. Runtime support for 3.10+ is +# still verified by the pytest matrix (py3.10–3.13). +python_version = "3.12" files = ["stackvox", "tests"] warn_unused_ignores = true warn_redundant_casts = true diff --git a/stackvox/engine.py b/stackvox/engine.py index c7082c8..71fd045 100644 --- a/stackvox/engine.py +++ b/stackvox/engine.py @@ -3,6 +3,8 @@ from __future__ import annotations import logging +import queue +import re import sys import threading import urllib.request @@ -69,6 +71,17 @@ def _ensure_models(cache_dir: Path) -> tuple[Path, Path]: return model_path, voices_path +# Split after ., !, or ? that is followed by whitespace, and on newlines. The +# whitespace requirement leaves decimals ("0.95") and mid-token dots +# ("file.ts") intact, which is good enough for speech chunking. +_SENTENCE_BOUNDARY = re.compile(r"(?<=[.!?])\s+|\n+") + + +def _split_sentences(text: str) -> list[str]: + """Break text into sentence-ish chunks for low-latency streaming playback.""" + return [part.strip() for part in _SENTENCE_BOUNDARY.split(text.strip()) if part.strip()] + + class Stackvox: """Reusable TTS engine. Load the model once, speak many times. @@ -89,6 +102,8 @@ def __init__( self.lang = lang model_path, voices_path = _ensure_models(cache_dir or _default_cache_dir()) self._kokoro = Kokoro(str(model_path), str(voices_path)) + self._stop_event = threading.Event() + self._play_thread: threading.Thread | None = None def synthesize( self, @@ -114,14 +129,73 @@ def speak( lang: str | None = None, blocking: bool = True, ) -> None: - """Synthesize and play through the system default output device.""" - samples, sample_rate = self.synthesize(text, voice=voice, speed=speed, lang=lang) - sd.play(samples, sample_rate) + """Synthesize and play through the system default output device. + + Streams sentence by sentence: the first sentence starts playing as soon + as it is synthesized (~0.2s) while the rest synthesize in the background, + rather than waiting for the whole text to synthesize first. + """ + self._stop_event.clear() if blocking: - sd.wait() + self._stream_play(text, voice=voice, speed=speed, lang=lang) + else: + self._play_thread = threading.Thread( + target=self._stream_play, + kwargs={"text": text, "voice": voice, "speed": speed, "lang": lang}, + daemon=True, + ) + self._play_thread.start() + + def _stream_play( + self, + text: str, + voice: str | None = None, + speed: float | None = None, + lang: str | None = None, + ) -> None: + """Synthesize sentence by sentence and play each as it is ready. + + Kokoro batches by phoneme count (~510), so a whole paragraph synthesizes + as one blob before any audio — the source of the lead-in lag. Splitting + into sentences ourselves means the first sentence (~0.2s to synthesize) + plays almost immediately. A producer thread synthesizes ahead into a + bounded queue while this thread plays, so later sentences are usually + ready by the time the previous one finishes. + """ + sentences = _split_sentences(text) + if not sentences: + return + + chunks: queue.Queue = queue.Queue(maxsize=8) + sentinel = object() + + def _produce() -> None: + try: + for sentence in sentences: + if self._stop_event.is_set(): + break + chunks.put(self.synthesize(sentence, voice=voice, speed=speed, lang=lang)) + except Exception: + logger.exception("synthesis error") + finally: + chunks.put(sentinel) + + producer = threading.Thread(target=_produce, daemon=True, name="stackvox-synth") + producer.start() + try: + while not self._stop_event.is_set(): + item = chunks.get() + if item is sentinel: + break + samples, sample_rate = item + sd.play(samples, sample_rate) + sd.wait() + finally: + producer.join(timeout=1.0) def stop(self) -> None: """Stop any in-progress playback started with blocking=False.""" + self._stop_event.set() sd.stop() def voices(self) -> list[str]: diff --git a/tests/test_engine.py b/tests/test_engine.py index e981717..7ca9770 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -99,25 +99,74 @@ def test_speed_zero_override_is_respected(self, fake_kokoro): class TestSpeak: - def test_blocking_calls_play_and_wait(self, fake_kokoro, fake_audio): - fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) - tts = engine.Stackvox() - tts.speak("hi") - engine.sd.play.assert_called_once() - engine.sd.wait.assert_called_once() + def test_blocking_synthesizes_and_plays_each_sentence(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10, dtype=np.float32), 24000) - def test_non_blocking_skips_wait(self, fake_kokoro, fake_audio): - fake_kokoro.return_value.create.return_value = (np.zeros(10), 24000) + engine.Stackvox().speak("One sentence. Two sentence. Three here.") + + # One synth + play cycle per sentence — that's what enables early audio. + assert fake_kokoro.return_value.create.call_count == 3 + assert engine.sd.play.call_count == 3 + assert engine.sd.wait.call_count == 3 + + def test_single_sentence_plays_once(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10, dtype=np.float32), 24000) + engine.Stackvox().speak("Just one sentence with no terminal punctuation") + assert engine.sd.play.call_count == 1 + + def test_non_blocking_plays_in_background_thread(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(10, dtype=np.float32), 24000) tts = engine.Stackvox() - tts.speak("hi", blocking=False) - engine.sd.play.assert_called_once() - engine.sd.wait.assert_not_called() - def test_stop_calls_sounddevice_stop(self, fake_kokoro, fake_audio): - engine.Stackvox().stop() + tts.speak("hi there", blocking=False) + + assert tts._play_thread is not None + tts._play_thread.join(timeout=5) + engine.sd.play.assert_called() + + def test_passes_engine_defaults_to_synthesis(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(4, dtype=np.float32), 24000) + engine.Stackvox(voice="af_sarah", speed=1.0, lang="en-us").speak("hello") + fake_kokoro.return_value.create.assert_called_once_with( + "hello", voice="af_sarah", speed=1.0, lang="en-us" + ) + + def test_per_call_overrides_take_priority(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(4, dtype=np.float32), 24000) + engine.Stackvox(voice="af_sarah", speed=1.0, lang="en-us").speak( + "hi", voice="bf_emma", speed=1.5, lang="en-gb" + ) + fake_kokoro.return_value.create.assert_called_once_with( + "hi", voice="bf_emma", speed=1.5, lang="en-gb" + ) + + def test_stop_sets_event_and_calls_sounddevice_stop(self, fake_kokoro, fake_audio): + tts = engine.Stackvox() + tts.stop() + assert tts._stop_event.is_set() engine.sd.stop.assert_called_once() +class TestSplitSentences: + def test_splits_on_sentence_punctuation(self): + assert engine._split_sentences("One. Two! Three?") == ["One.", "Two!", "Three?"] + + def test_leaves_decimals_and_mid_token_dots_intact(self): + assert engine._split_sentences("Set speed 0.95 in config.toml now.") == [ + "Set speed 0.95 in config.toml now." + ] + + def test_splits_on_newlines(self): + assert engine._split_sentences("Line one\nLine two\n\nLine three") == [ + "Line one", + "Line two", + "Line three", + ] + + def test_blank_text_yields_no_sentences(self): + assert engine._split_sentences(" \n ") == [] + + class TestVoices: def test_returns_sorted_voice_ids(self, fake_kokoro): fake_kokoro.return_value.get_voices.return_value = ["bf_emma", "af_sarah", "am_adam"] From b321a4e8ecc98a71c3067e57d13858e1401f3b76 Mon Sep 17 00:00:00 2001 From: StuBehan Date: Wed, 8 Jul 2026 15:55:49 +0100 Subject: [PATCH 2/2] fix(engine): propagate synthesis errors and stop stream thread leaks --- stackvox/engine.py | 68 +++++++++++++++++++++++++++++++++++--------- tests/test_engine.py | 33 +++++++++++++++++++++ 2 files changed, 87 insertions(+), 14 deletions(-) diff --git a/stackvox/engine.py b/stackvox/engine.py index 71fd045..b736970 100644 --- a/stackvox/engine.py +++ b/stackvox/engine.py @@ -121,6 +121,21 @@ def synthesize( ) return samples, sample_rate + def _cancel_active(self) -> None: + """Supersede any in-progress stream so a new ``speak`` starts clean. + + Signals the running stream's stop event and joins its thread, so we + never overlap two streams on the shared output device or orphan the + previous playback thread. + """ + thread = self._play_thread + self._play_thread = None + if thread is None or thread is threading.current_thread() or not thread.is_alive(): + return + self._stop_event.set() + sd.stop() + thread.join() + def speak( self, text: str, @@ -135,20 +150,28 @@ def speak( as it is synthesized (~0.2s) while the rest synthesize in the background, rather than waiting for the whole text to synthesize first. """ - self._stop_event.clear() + self._cancel_active() + # Each stream gets its own cancellation token, so a later speak() (or a + # stop() in between) can never clear an older stream's cancellation. + stop_event = threading.Event() + self._stop_event = stop_event if blocking: - self._stream_play(text, voice=voice, speed=speed, lang=lang) - else: - self._play_thread = threading.Thread( - target=self._stream_play, - kwargs={"text": text, "voice": voice, "speed": speed, "lang": lang}, - daemon=True, - ) - self._play_thread.start() + self._stream_play(text, stop_event, voice=voice, speed=speed, lang=lang) + return + + def _run() -> None: + try: + self._stream_play(text, stop_event, voice=voice, speed=speed, lang=lang) + except Exception: + logger.exception("stackvox playback failed") + + self._play_thread = threading.Thread(target=_run, daemon=True, name="stackvox-play") + self._play_thread.start() def _stream_play( self, text: str, + stop_event: threading.Event, voice: str | None = None, speed: float | None = None, lang: str | None = None, @@ -161,6 +184,10 @@ def _stream_play( plays almost immediately. A producer thread synthesizes ahead into a bounded queue while this thread plays, so later sentences are usually ready by the time the previous one finishes. + + A synthesis failure is re-raised rather than swallowed, so a blocking + caller sees the error instead of silent success; ``stop_event`` cancels + the stream between sentences. """ sentences = _split_sentences(text) if not sentences: @@ -172,26 +199,39 @@ def _stream_play( def _produce() -> None: try: for sentence in sentences: - if self._stop_event.is_set(): + if stop_event.is_set(): break chunks.put(self.synthesize(sentence, voice=voice, speed=speed, lang=lang)) - except Exception: - logger.exception("synthesis error") + except Exception as exc: # hand the failure to the consumer to re-raise + chunks.put(exc) finally: chunks.put(sentinel) producer = threading.Thread(target=_produce, daemon=True, name="stackvox-synth") producer.start() + error: Exception | None = None try: - while not self._stop_event.is_set(): + while not stop_event.is_set(): item = chunks.get() if item is sentinel: break + if isinstance(item, Exception): + error = item + break samples, sample_rate = item sd.play(samples, sample_rate) sd.wait() finally: - producer.join(timeout=1.0) + # Stop the producer and keep draining so it can never stay parked on + # a full queue — that's what guarantees the synth thread always exits. + stop_event.set() + while producer.is_alive(): + try: + chunks.get_nowait() + except queue.Empty: + producer.join(timeout=0.05) + if error is not None: + raise error def stop(self) -> None: """Stop any in-progress playback started with blocking=False.""" diff --git a/tests/test_engine.py b/tests/test_engine.py index 7ca9770..2d98929 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -8,6 +8,7 @@ from __future__ import annotations +import logging import threading import numpy as np @@ -146,6 +147,38 @@ def test_stop_sets_event_and_calls_sounddevice_stop(self, fake_kokoro, fake_audi assert tts._stop_event.is_set() engine.sd.stop.assert_called_once() + def test_blocking_surfaces_synthesis_errors(self, fake_kokoro, fake_audio): + # Regression: a Kokoro failure must propagate, not report silent success. + fake_kokoro.return_value.create.side_effect = RuntimeError("boom") + tts = engine.Stackvox() + with pytest.raises(RuntimeError, match="boom"): + tts.speak("hello there") + + def test_non_blocking_logs_synthesis_errors_without_crashing(self, fake_kokoro, fake_audio, caplog): + fake_kokoro.return_value.create.side_effect = RuntimeError("boom") + tts = engine.Stackvox() + with caplog.at_level(logging.ERROR): + tts.speak("hi there", blocking=False) + assert tts._play_thread is not None + tts._play_thread.join(timeout=5) + assert "playback failed" in caplog.text + + def test_each_speak_uses_a_fresh_stop_event(self, fake_kokoro, fake_audio): + # A per-stream event means a later speak() can't clear an older stream's + # cancellation (and stop() only affects the stream it was called on). + fake_kokoro.return_value.create.return_value = (np.zeros(4, dtype=np.float32), 24000) + tts = engine.Stackvox() + original = tts._stop_event + tts.speak("one sentence") + assert tts._stop_event is not original + + def test_speak_after_stop_still_plays(self, fake_kokoro, fake_audio): + fake_kokoro.return_value.create.return_value = (np.zeros(4, dtype=np.float32), 24000) + tts = engine.Stackvox() + tts.stop() + tts.speak("hello world") + engine.sd.play.assert_called() + class TestSplitSentences: def test_splits_on_sentence_punctuation(self):