feat: stream synthesis sentence-by-sentence for low-latency playback

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

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,
@@ -106,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,
@@ -114,14 +144,98 @@ 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._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:
-            sd.wait()
+            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,
+    ) -> 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.
+
+        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:
+            return
+
+        chunks: queue.Queue = queue.Queue(maxsize=8)
+        sentinel = object()
+
+        def _produce() -> None:
+            try:
+                for sentence in sentences:
+                    if stop_event.is_set():
+                        break
+                    chunks.put(self.synthesize(sentence, voice=voice, speed=speed, lang=lang))
+            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 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:
+            # 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."""
+        self._stop_event.set()
         sd.stop()
 
     def voices(self) -> list[str]:

tests/test_engine.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import logging
 import threading
 
 import numpy as np
@@ -99,24 +100,105 @@ 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()
 
+    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):
+        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):