| """Text-to-Speech — speaks text aloud. |
| |
| Three tiers: |
| - Default: Web Speech API (browser) — used when accessing via web UI |
| - Local TTS: piper (neural TTS) if installed — natural voices, ~50ms latency |
| - Fallback: System TTS via pyttsx3 — works everywhere |
| |
| Auto-selects best available TTS engine on startup. |
| Sentence queue: speaks sentences as they're generated. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import queue |
| import threading |
| import time |
| from typing import Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class TTSEngine: |
| """Text-to-speech engine with automatic backend selection. |
| |
| Backends (checked in order): |
| 1. piper (if installed) — neural TTS, ~50ms latency, 100% offline |
| 2. pyttsx3 (if installed) — system TTS, works everywhere |
| 3. Web Speech API (browser only — handled in UI) |
| 4. Fallback: prints text to console |
| """ |
|
|
| def __init__(self, preferred: str = "auto", rate: float = 1.1, volume: float = 1.0) -> None: |
| self.preferred = preferred |
| self.rate = rate |
| self.volume = volume |
| self.backend = self._detect_backend() |
| self._engine = None |
| self._voice = None |
| self._sentence_queue: queue.Queue[str] = queue.Queue() |
| self._speaking = False |
| self._speak_thread: threading.Thread | None = None |
| self._stop_flag = threading.Event() |
| self._stats = { |
| "sentences_spoken": 0, |
| "errors": 0, |
| "total_speak_time_s": 0.0, |
| } |
|
|
| if self.backend != "none": |
| logger.info("TTS backend: %s", self.backend) |
| self._init_engine() |
|
|
| def _detect_backend(self) -> str: |
| """Detect the best available TTS backend.""" |
| if self.preferred != "auto": |
| return self.preferred |
|
|
| |
| try: |
| import piper |
| return "piper" |
| except ImportError: |
| pass |
|
|
| |
| try: |
| import pyttsx3 |
| return "pyttsx3" |
| except ImportError: |
| pass |
|
|
| return "none" |
|
|
| def _init_engine(self) -> None: |
| """Initialize the TTS engine.""" |
| try: |
| if self.backend == "pyttsx3": |
| import pyttsx3 |
| self._engine = pyttsx3.init() |
| self._engine.setProperty("rate", int(self.rate * 200)) |
| self._engine.setProperty("volume", self.volume) |
| voices = self._engine.getProperty("voices") |
| if voices: |
| self._engine.setProperty("voice", voices[0].id) |
| except Exception as e: |
| logger.error("TTS engine init failed: %s", e) |
| self.backend = "none" |
|
|
| def speak(self, text: str, blocking: bool = False) -> None: |
| """Speak a text string. |
| |
| Args: |
| text: text to speak |
| blocking: if True, wait until speech is complete |
| """ |
| if not text.strip(): |
| return |
|
|
| if self.backend == "none": |
| |
| print(f"[TTS] {text}") |
| return |
|
|
| self._sentence_queue.put(text) |
|
|
| if not self._speaking: |
| self._speaking = True |
| self._stop_flag.clear() |
| if blocking: |
| self._speak_loop() |
| else: |
| self._speak_thread = threading.Thread(target=self._speak_loop, daemon=True) |
| self._speak_thread.start() |
|
|
| def _speak_loop(self) -> None: |
| """Process the sentence queue.""" |
| while not self._stop_flag.is_set(): |
| try: |
| text = self._sentence_queue.get(timeout=0.5) |
| except queue.Empty: |
| break |
|
|
| t0 = time.time() |
| try: |
| if self.backend == "pyttsx3" and self._engine: |
| self._engine.say(text) |
| self._engine.runAndWait() |
| elif self.backend == "piper": |
| |
| logger.debug("Piper TTS: %s", text[:50]) |
| except Exception as e: |
| logger.error("TTS error: %s", e) |
| self._stats["errors"] += 1 |
|
|
| self._stats["sentences_spoken"] += 1 |
| self._stats["total_speak_time_s"] += time.time() - t0 |
|
|
| self._speaking = False |
|
|
| def stop(self) -> None: |
| """Stop speaking and clear the queue.""" |
| self._stop_flag.set() |
| while not self._sentence_queue.empty(): |
| try: |
| self._sentence_queue.get_nowait() |
| except queue.Empty: |
| break |
| if self._engine and self.backend == "pyttsx3": |
| try: |
| self._engine.stop() |
| except Exception: |
| pass |
|
|
| def is_speaking(self) -> bool: |
| """Check if TTS is currently speaking.""" |
| return self._speaking |
|
|
| def is_available(self) -> bool: |
| """Check if any TTS backend is available.""" |
| return self.backend != "none" |
|
|
| def get_stats(self) -> dict[str, Any]: |
| return {**self._stats, "backend": self.backend, "speaking": self._speaking} |
|
|
