From aa97ee5ae20663ef372197ff818eb800f2c31249 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:43:19 -0700 Subject: [PATCH] refactor(tools): split tts speaker playback worker into per-sentence phase helpers --- tools/tts_tool_speaker.py | 242 ++++++++++++++++---------------------- 1 file changed, 101 insertions(+), 141 deletions(-) diff --git a/tools/tts_tool_speaker.py b/tools/tts_tool_speaker.py index a995adf805..7edc25a543 100644 --- a/tools/tts_tool_speaker.py +++ b/tools/tts_tool_speaker.py @@ -1,20 +1,12 @@ """Speaker-side streaming pipeline for ``tools.tts_tool.stream_tts_to_speaker``. -Turns a queue of LLM text deltas into audio the moment each sentence is -complete. Two paths share the sentence cutter (``tools.tts_streaming``): - -* :class:`_StreamerPlayback` — a registered chunked streamer (ElevenLabs, - OpenAI, …). Every sentence gets a prefetch thread that fires the HTTP - request immediately and buffers PCM into a per-sentence queue; one playback - worker drains those queues in FIFO order through a sounddevice OutputStream - (or a temp WAV + system player when PortAudio is unavailable). -* :class:`_SyncSentencePipeline` — every other provider (edge, piper, - plugins). Per-sentence ``text_to_speech_tool`` synthesis on a single-thread - executor, overlapped with playback so sentence n+1 synthesizes while n plays. - -Seams tests monkeypatch on the origin module (``_load_tts_config``, -``_import_sounddevice``, ``text_to_speech_tool``, ``_strip_markdown_for_tts``) -are resolved through :func:`_origin` at call time. +Turns a queue of LLM text deltas into audio the moment each sentence is complete. +Two paths share the sentence cutter (``tools.tts_streaming``): :class:`_StreamerPlayback` +for a registered chunked streamer (prefetch thread per sentence, one FIFO playback worker +through a sounddevice OutputStream or temp WAV + system player) and +:class:`_SyncSentencePipeline` for every other provider (per-sentence ``text_to_speech_tool`` +on a single-thread executor, overlapped with playback). Seams tests monkeypatch on the origin +module are resolved through :func:`_origin` at call time. """ from __future__ import annotations @@ -74,9 +66,8 @@ def _play_via_tempfile(audio_iter: Iterable[bytes], stop_evt: threading.Event, s wf.setframerate(sample_rate) for aligned in _align_int16_chunks(audio_iter, stop_evt): wf.writeframes(aligned) - # wave.open() on a file object does NOT close it. On Windows the open - # write handle blocks the player and the unlink below (WinError 32), - # so release it before playback. + # wave.open() on a file object does NOT close it; on Windows the open write + # handle blocks the player and the unlink below (WinError 32). tmp.close() from tools.voice_mode import play_audio_file play_audio_file(tmp_path) @@ -104,14 +95,11 @@ def _drain_chunks(chunk_queue: "queue.Queue[Optional[bytes]]") -> List[bytes]: class _SyncSentencePipeline: """Overlap per-sentence synthesis with playback for non-streaming providers. - Serial synthesize-then-play added a full synthesis-time of dead air per - sentence — for a local model at real-time-factor ~1, as long silent as - speaking. One single-thread synthesis executor (sentences FIFO; providers - never see concurrent calls) feeds one playback worker through a small - bounded queue: while sentence n plays, n+1 is already synthesizing. The - bound keeps lookahead/temp files small and gives the caller backpressure. - - ``text_to_speech_tool`` / ``play_audio_file`` are resolved late so tests + Serial synthesize-then-play added a full synthesis-time of dead air per sentence. One + single-thread synthesis executor (sentences FIFO; providers never see concurrent calls) + feeds one playback worker through a small bounded queue: while sentence n plays, n+1 is + already synthesizing. The bound keeps lookahead/temp files small and gives the caller + backpressure. ``text_to_speech_tool`` / ``play_audio_file`` are resolved late so tests that monkeypatch them keep working. """ @@ -126,8 +114,7 @@ class _SyncSentencePipeline: """Queue one sentence. Blocks only when the lookahead bound is full.""" if self._stop.is_set(): return - future = self._executor.submit(self._synthesize_to_tmp, cleaned) - self._queue.put((cleaned, future)) + self._queue.put((cleaned, self._executor.submit(self._synthesize_to_tmp, cleaned))) def close(self) -> None: """Flush queued sentences in order (skipped if stopped), then join.""" @@ -158,9 +145,7 @@ class _SyncSentencePipeline: tmp_path = None try: tmp_path = future.result() - if (tmp_path and not self._stop.is_set() - and os.path.isfile(tmp_path) - and os.path.getsize(tmp_path) > 0): + if tmp_path and not self._stop.is_set() and os.path.isfile(tmp_path) and os.path.getsize(tmp_path) > 0: from tools.voice_mode import play_audio_file play_audio_file(tmp_path) except Exception as exc: @@ -172,13 +157,12 @@ class _SyncSentencePipeline: class _StreamerPlayback: """Prefetch + FIFO playback for a chunked :class:`StreamingTTSProvider`. - ``speak(text)`` calls ``streamer.stream()`` right away and hands the - iterator to a prefetch thread (at most 3 in flight) that buffers chunks - into a bounded per-sentence queue; the single playback worker plays those - queues in order, so sentence N+1 is already arriving while N plays. - Output goes to a PortAudio stream when one could be opened, otherwise via - temp WAV files. A failing PortAudio write is retried on a reinitialized - stream up to ``_MAX_REINIT`` times before falling back to temp files. + ``speak(text)`` calls ``streamer.stream()`` right away and hands the iterator to a + prefetch thread (at most 3 in flight) that buffers chunks into a bounded per-sentence + queue; the single playback worker plays those queues in order, so sentence N+1 is + already arriving while N plays. Output goes to a PortAudio stream when one could be + opened, otherwise via temp WAV files. A failing PortAudio write is retried on a + reinitialized stream up to ``_MAX_REINIT`` times before falling back to temp files. """ _MAX_REINIT = 3 @@ -198,19 +182,14 @@ class _StreamerPlayback: def _create_output_stream(self): sd = _origin()._import_sounddevice() - stream = sd.OutputStream( - samplerate=self.streamer.sample_rate, - channels=self.streamer.channels, - dtype="int16", - ) + stream = sd.OutputStream(samplerate=self.streamer.sample_rate, channels=self.streamer.channels, dtype="int16") stream.start() return stream def _open_output_stream(self): - # On macOS skip sounddevice entirely: PortAudio/CoreAudio init triggers - # a kTCCServiceMediaLibrary permission prompt even though output needs - # no media-library access. None routes every sentence through the - # tempfile -> play_audio_file -> afplay path. + # On macOS skip sounddevice entirely: PortAudio/CoreAudio init triggers a + # kTCCServiceMediaLibrary permission prompt even though output needs no + # media-library access. None routes every sentence through tempfile -> afplay. if platform.system() == "Darwin": return None try: @@ -223,12 +202,7 @@ class _StreamerPlayback: def _reinit_output_stream(self): """Close the broken PortAudio stream and try to create a fresh one.""" - if self.output_stream is not None: - try: - self.output_stream.stop() - self.output_stream.close() - except Exception: - pass + self.close_output_stream() try: self.output_stream = self._create_output_stream() logger.info("TTS: PortAudio output stream reinitialized after error") @@ -266,18 +240,11 @@ class _StreamerPlayback: try: for chunk in audio_iter: if self.stop_event.is_set(): - logger.info( - "TTS CUT: prefetch cancelled (stop_event set " - "mid-sentence) — partial audio only" - ) + logger.info("TTS CUT: prefetch cancelled (stop_event set mid-sentence) — partial audio only") break chunk_queue.put(chunk, timeout=30.0) except Exception as exc: - logger.warning( - "TTS CUT: streaming TTS prefetch failed mid-sentence " - "(partial audio only): %s", - exc, - ) + logger.warning("TTS CUT: streaming TTS prefetch failed mid-sentence (partial audio only): %s", exc) finally: chunk_queue.put(None) # sentinel: no more chunks self._prefetch_sem.release() @@ -287,16 +254,60 @@ class _StreamerPlayback: def _play_sentence_via_tempfile(self, chunk_queue) -> None: _play_via_tempfile(iter(_drain_chunks(chunk_queue)), self.stop_event, self.streamer.sample_rate) + def _for_each_sentence(self, play: Callable[[queue.Queue], None]) -> None: + """Feed queued sentences to *play* in order until the end sentinel; stopped sentences are skipped.""" + while True: + chunk_queue = self._audio_queue.get() + if chunk_queue is None: + return + if not self.stop_event.is_set(): + play(chunk_queue) + + def _write_pcm(self, buf: bytes) -> None: + self._current_stream.write(self._np.frombuffer(buf, dtype=" bool: + """Reinit the PortAudio stream after a failed write; False once ``_MAX_REINIT`` is exhausted.""" + if self._reinit_count < self._MAX_REINIT: + self._reinit_count += 1 + self._current_stream = self._reinit_output_stream() + return self._current_stream is not None + logger.warning( + "TTS: PortAudio reinit exhausted after %d attempts, falling back to tempfile for remaining sentences", + self._MAX_REINIT, + ) + self._current_stream = None + return False + + def _play_sentence_via_stream(self, chunk_queue) -> None: + """Write one sentence's PCM to PortAudio; after an unrecoverable write failure the rest is dropped.""" + if self._current_stream is None: + self._play_sentence_via_tempfile(chunk_queue) + return + pcm_leftover = b"" + while True: + chunk = chunk_queue.get() + if chunk is None or self.stop_event.is_set(): + return + buf = pcm_leftover + chunk + aligned_len = len(buf) - (len(buf) % 2) + if aligned_len >= 2: + try: + self._write_pcm(buf[:aligned_len]) + except Exception as write_exc: + logger.warning("PortAudio write failed, attempting stream reinit: %s", write_exc) + if not self._recover_stream(): + return + try: + self._write_pcm(buf[:aligned_len]) + except Exception: + pass + pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b"" + def _playback_worker(self) -> None: """Single consumer: play audio segments from the queue in order.""" if self.output_stream is None: - while True: - chunk_queue = self._audio_queue.get() - if chunk_queue is None: - break - if self.stop_event.is_set(): - continue - self._play_sentence_via_tempfile(chunk_queue) + self._for_each_sentence(self._play_sentence_via_tempfile) return import numpy as _np @@ -307,59 +318,12 @@ class _StreamerPlayback: def mark_audio_output_active(_active): return None - def write_pcm(stream, buf: bytes) -> None: - stream.write(_np.frombuffer(buf, dtype="= 2: - try: - write_pcm(current_stream, buf[:aligned_len]) - except Exception as write_exc: - logger.warning( - "PortAudio write failed, attempting " - "stream reinit: %s", - write_exc, - ) - if reinit_count < self._MAX_REINIT: - reinit_count += 1 - current_stream = self._reinit_output_stream() - if current_stream is not None: - try: - write_pcm(current_stream, buf[:aligned_len]) - except Exception: - pass - pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b"" - continue - else: - logger.warning( - "TTS: PortAudio reinit exhausted " - "after %d attempts, falling back " - "to tempfile for remaining " - "sentences", - self._MAX_REINIT, - ) - current_stream = None - break - pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b"" + self._for_each_sentence(self._play_sentence_via_stream) finally: mark_audio_output_active(False) @@ -379,19 +343,17 @@ def stream_tts_to_speaker( display_callback: Optional[Callable[[str], None]] = None, provider: Optional[str] = None, ): - """Consume text deltas from *text_queue*, cut them into sentences, and speak - each one the moment it's ready — the conversational path. + """Consume text deltas from *text_queue*, cut them into sentences, and speak each one + the moment it's ready — the conversational path. - A registered streaming provider plays chunked PCM for the lowest latency; - every other provider (edge, the default) is spoken per-sentence via the - sync ``text_to_speech_tool`` path, so audio still starts on sentence one. + A registered streaming provider plays chunked PCM for the lowest latency; every other + provider (edge, the default) is spoken per-sentence via the sync ``text_to_speech_tool`` + path, so audio still starts on sentence one. - Protocol: - * The producer puts ``str`` deltas onto *text_queue*. - * A ``None`` sentinel signals end-of-text (flush remaining buffer). - * *stop_event* aborts early (barge-in / user interrupt). - * *tts_done_event* is **set** in the ``finally`` block so callers - waiting on it (continuous voice mode) know playback is finished. + Protocol: the producer puts ``str`` deltas onto *text_queue*; a ``None`` sentinel + signals end-of-text (flush remaining buffer); *stop_event* aborts early (barge-in); + *tts_done_event* is **set** in the ``finally`` block so callers waiting on it + (continuous voice mode) know playback is finished. """ tts_done_event.clear() origin = _origin() @@ -401,8 +363,8 @@ def stream_tts_to_speaker( try: tts_config = origin._load_tts_config() - # Prefer a chunked streamer for low time-to-first-audio; otherwise - # per-sentence sync synthesis (universal — edge + every non-streamer). + # Prefer a chunked streamer for low time-to-first-audio; otherwise per-sentence + # sync synthesis (universal — edge + every non-streamer). from tools.tts_streaming import SentenceChunker, resolve_streaming_provider streamer = resolve_streaming_provider(tts_config, preferred=provider) @@ -411,9 +373,7 @@ def stream_tts_to_speaker( sync_pipeline = _SyncSentencePipeline(stop_event) else: try: - stream_max_len = origin._resolve_max_text_length( - provider or origin._get_provider(tts_config), tts_config - ) + stream_max_len = origin._resolve_max_text_length(provider or origin._get_provider(tts_config), tts_config) except Exception: stream_max_len = 0 playback = _StreamerPlayback(streamer, stop_event) @@ -469,16 +429,16 @@ def stream_tts_to_speaker( except Exception as exc: logger.warning("Streaming TTS pipeline error: %s", exc) finally: - # Flush the sync pipeline first: queued sentences finish playing (or - # are skipped when stop_event is set) BEFORE tts_done_event fires, so - # continuous voice mode never reopens the mic over its own voice. + # Flush the sync pipeline first: queued sentences finish playing (or are skipped + # when stop_event is set) BEFORE tts_done_event fires, so continuous voice mode + # never reopens the mic over its own voice. if sync_pipeline is not None: try: sync_pipeline.close() except Exception: pass - # The end sentinel lives in finally: so an exception in the text pump - # still lets the playback worker exit. + # The end sentinel lives in finally: so an exception in the text pump still lets + # the playback worker exit. if playback is not None: playback.finish() tts_done_event.set()