The TTS engine is the core heart of Ebook Voice Player. This article deep-dives into the three core components in tts_engine.py: TTSGenerator (Async Synthesizer), TTSPlayer (Player + Watchdog), and TTSService (Orchestration State Machine), and how they collaborate to deliver smooth, robust speech playback.
1. Overall Data Flow
User clicks Play │ ▼ ┌─────────────────────┐ │ TTSService.start │ ← State machine init, cleanup old tasks, setup audio queue └─────────┬───────────┘ │ ▼ ┌─────────────────────┐ │ TTSGenerator._run() │ ← Dedicated thread + asyncio event loop │ │ │ ┌───────────────┐ │ │ │ Semaphore(3) │ │ ← Rate limit protection for edge-tts backend │ └───────┬───────┘ │ │ │ │ │ ┌───────▼───────┐ │ │ │ Cache Key │ │ ← SHA256(text|voice|rate) │ └───────┬───────┘ │ │ │ │ │ Hit?──No──▶ edge-tts synthesis │ │ │ │ Yes │ │ ▼ │ │ ┌───────────────┐ │ │ │ Atomic Write │ │ ← .part.mp3 → replace → validate │ └───────┬───────┘ │ │ │ │ │ ▼ │ │ Emit paragraph evt │ └─────────┬───────────┘ │ ▼ ┌─────────────────────┐ │ TTSService │ ← Receive event, enqueue audio_queue[index] = path │ _on_para_ready() │ └─────────┬───────────┘ │ ▼ ┌─────────────────────┐ │ TTSPlayer │ ← SoundLoader + watchdog guarantees progress │ play_file() │ └─────────┬───────────┘ │ ▼ Natural end / Watchdog timeout │ ▼ Emit on_finish → TTSService._on_play_done() │ ▼ Play next / All done 2. TTSGenerator: Async Concurrent Synthesis Engine
2.1 Thread Model & Event Loop
def _run(self): try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) rate_str = rate_to_str(self.rate) cache_dir = get_cache_dir() async def gen_one(idx, text): # ... synthesis logic sem = asyncio.Semaphore(self.max_workers) # Key: rate limiting async def limited(idx, text): if self._cancelled: return async with sem: await gen_one(idx, text) tasks = [limited(i, p) for i, p in enumerate(self.paragraphs)] if tasks: loop.run_until_complete(asyncio.gather(*tasks)) loop.close() if not self._cancelled: self._emit("finished") except Exception as e: if not self._cancelled: self._emit("error", f"TTS engine error: {str(e)}") Design Highlights:
- Dedicated Thread Isolation: Synthesis doesn't block UI main thread;
Clock.schedule_oncedispatches events back to main thread - Fresh Event Loop: Avoids conflicts with main thread or other async tasks
-
run_until_complete+gather: Batch scheduling with natural concurrency control
2.2 Cache Strategy: Content-Addressed + Atomic Writes
# Cache key: text content + voice + rate → unique identifier cache_key = hashlib.sha256( f"{text}|{self.voice}|{rate_str}".encode()).hexdigest()[:16] cache_file = cache_dir / f"c_{cache_key}.mp3" # Fast path: cache hit if cache_file.exists(): if is_valid_mp3(str(cache_file)): # Validate integrity self._emit("paragraph", idx, str(cache_file)) return else: try: cache_file.unlink() # Clean up corrupt file except OSError: pass # Atomic write: write temp file first, validate, then rename tmp_file = cache_dir / f"c_{cache_key}.part.mp3" communicate = edge_tts.Communicate(text, self.voice, rate=rate_str) await communicate.save(str(tmp_file)) size = os.path.getsize(tmp_file) if tmp_file.exists() else 0 if size < MIN_MP3_SIZE or not is_valid_mp3(str(tmp_file)): raise RuntimeError(f"Invalid synthesis result ({size} bytes), check network") os.replace(str(tmp_file), str(cache_file)) # Atomic replace, safe on POSIX/Windows Cache Design Highlights:
| Feature | Implementation | Value |
|---|---|---|
| Content-Addressed | SHA256(text+voice+rate) | Same text diff voice/rate auto-distinguished, cross-session reuse |
| Atomic Writes | .part.mp3 → os.replace() | Power loss/crash/network interrupt leaves no corrupt files |
| Dual Validation | Size > 1KB + MP3 header 0xFFE0 | Catches edge-tts empty responses/error pages |
| Auto-Cleanup | Corrupt cache auto unlink() | Prevents cache directory pollution |
2.3 Smart Error Classification & User-Friendly Messages
def _describe_error(self, idx, err): """Convert synthesis exception to actionable, localized message""" raw = str(err) or err.__class__.__name__ low = raw.lower() if ("403" in raw or "handshake" in low or "unauthorized" in low or "invalid response status" in low): hint = L("err_tts_403") # "Speech service rejected (403), check system clock..." elif ("getaddrinfo" in low or "connection" in low or "timeout" in low or "cannot connect" in low or "network" in low or "clientconnector" in low): hint = L("err_tts_network") # "Cannot reach speech service, check network..." elif "no audio" in low or "noaudio" in low: hint = L("err_tts_noaudio") # "Service returned no audio, retry or switch voice..." else: hint = raw return Lf("err_tts_para", n=idx + 1, detail=hint) Error Classification Coverage:
| Error Type | Typical Signature | User Action |
|---|---|---|
| 403/Auth Failure | 403, handshake, unauthorized | Check system time, update app |
| Network Unreachable | getaddrinfo, timeout, connection | Check network/proxy/firewall |
| No Audio Returned | no audio, noaudio | Retry or switch voice |
| Other | Preserve raw error | Report to developer |
3. TTSPlayer: Player & Watchdog Mechanism
3.1 Core Problem: Windows on_stop Failure
Kivy's SoundLoader on Windows (SDL2 backend) has a known issue: on_stop event doesn't fire at natural end of playback, causing playback to stall on first paragraph.
3.2 Watchdog Design
def _load_and_play(self, path, seek_pos=0.0): snd = SoundLoader.load(path) if snd is None: if self.on_error: self.on_error(path) return self.sound = snd self._expected_len = float(getattr(snd, "length", 0) or 0) # Key: Windows often reports length=0, estimate from bitrate if self._expected_len <= 0: try: # edge-tts output: 24kHz/48kbps mono MP3 self._expected_len = max(0.0, os.path.getsize(path) * 8.0 / 48000.0 - 0.2) except OSError: self._expected_len = 0.0 snd.bind(on_stop=self._on_sound_stop) if seek_pos > 0 and self._expected_len > 0 and seek_pos < self._expected_len: snd.seek(seek_pos) self._play_start = time.time() - seek_pos else: self._play_start = time.time() self._start_watchdog() # Start watchdog snd.play() def _start_watchdog(self): self._cancel_watchdog() if self._expected_len <= 0: return # Unknown duration: rely on on_stop only def tick(dt): if self._stopped or self._paused or self.sound is None: return False if self.sound.state != "play": return False # Native on_stop will drive completion elapsed = time.time() - self._play_start if elapsed >= self._expected_len + 1.0: # 1s tolerance self._finish_playback() # Force advance return False return True # Continue timer self._watchdog = Clock.schedule_interval(tick, 0.2) # 200ms polling 3.3 Watchdog Parameter Trade-offs
| Parameter | Value | Rationale |
|---|---|---|
| Poll Interval | 200ms | Balances CPU (~0.1%) vs max latency (200ms) |
| Tolerance | +1.0s | Absorbs network jitter, decode startup, seek offset |
| Trigger Condition | elapsed >= expected_len + 1.0 | Only intervenes on clear timeout; normal path uses native event |
| Duration Strategy | Prefer sound.length, fallback bitrate estimate | Covers Windows length=0 scenario |
3.4 Pause/Resume/Seek State Sync
def pause(self): if self.sound is not None and self.sound.state == "play": self._paused = True try: self._pause_pos = self.sound.get_pos() # Record position except Exception: self._pause_pos = 0.0 self._cancel_watchdog() self.sound.stop() def resume(self): if self._last_path and os.path.exists(self._last_path): self._paused = False self._stopped = False pos = self._pause_pos self._pause_pos = 0.0 self._load_and_play(self._last_path, pos) # Resume from pause position 4. TTSService: Playback Orchestration State Machine
4.1 State Definitions & Transitions
# State flags self._playing = False # Playing self._paused = False # Paused self._stopped = False # User stopped self._finished = False # All done # State Transition Diagram: # # [Stopped] ──start()──▶ [Playing] ──pause()──▶ [Paused] # ▲ │ │ # │ ▼ │ # └───── stop() ◀── [Playing] ◀──resume()─────┘ # ▲ # │ # └──── [Finished] ◀── _finish() ◀── Natural end 4.2 Core Orchestration Logic
def _on_para_ready(self, index, path): """Generator callback: paragraph audio ready""" if self._stopped: return self.audio_queue[index] = path # First paragraph ready → play immediately if self.current_index == -1 and index == 0: self._play_ready(0) # Next paragraph ready and not currently playing → auto-chain elif index == self.current_index + 1: if not self.player.is_playing and not self._paused: self._play_ready(index) def _play_ready(self, from_idx): """Continuously play ready paragraphs""" idx = from_idx while idx in self.audio_queue: self.current_index = idx self._emit("paragraph", self.base_index + idx) # Notify UI highlight path = self.audio_queue[idx] if path: self.player.play_file(path) return # Wait for playback end callback idx += 1 # Queue exhausted and reached total → finish if idx >= self.total: self._finish() def _on_play_done(self): """Player callback: single paragraph done""" if self._stopped or self._finished: return self._emit("progress", self.current_index + 1, self.total) self._play_ready(self.current_index + 1) # Try play next def _on_gen_done(self): """Generator callback: all synthesis done""" if self._stopped or self.current_index >= self.total - 1: self._finish() 4.3 Hot-Swap Capability: Seamless Speed/Voice/Chapter Switch
def set_speed(self, speed): """Speed 0.5-2.0. Baked into edge-tts regeneration; current audio finishes.""" speed = max(0.5, min(2.0, float(speed))) self._speed = speed if self._playing and not self._finished: voice = self.generator.voice self._disconnect_generator() old = self.generator old.cancel() self._stale.append(old) # Discard unplayed cached audio for k in [k for k in self.audio_queue if k > self.current_index]: del self.audio_queue[k] # Start new generator for remaining paragraphs self.generator = TTSGenerator() start_idx = self.current_index + 1 if start_idx < self.total: remaining = self.paragraphs[start_idx:] self.generator.events.bind( "paragraph", lambda idx, path, off=start_idx: self._on_para_ready(off + idx, path)) self.generator.events.bind("error", lambda m: self._emit("error", m)) self.generator.events.bind("finished", self._on_gen_done) self.generator.start_generation(remaining, voice, self._current_rate()) else: self._connect_generator() self._schedule_cleanup() Hot-Swap Key Points:
- Retain Played: Audio before
current_indexunaffected - Discard Pending: Queue entries
> current_indexpurged to avoid speed mismatch - Seamless Chain: Current segment finishes, auto-picks up new generator's audio
- Graceful Old Generator Exit: Added to
_stalelist, periodic cleanup of finished threads
5. Concurrency & Resource Management
5.1 Stale Generator Cleanup
def _schedule_cleanup(self): if self._cleanup_interval is None: self._cleanup_interval = Clock.schedule_interval(self._cleanup_stale, 15.0) def _cleanup_stale(self, *a): alive = [] for g in self._stale: if g.is_running(): alive.append(g) self._stale = alive 5.2 Cache Directory Management
def get_cache_dir(): app = App.get_running_app() if app is not None: base = Path(app.user_data_dir) else: base = Path.home() / ".tts_ebook_reader" d = base / "audio_cache" d.mkdir(parents=True, exist_ok=True) return d def cleanup(self): """Manual cleanup / called on exit""" for f in get_cache_dir().glob("*.mp3"): try: f.unlink() except OSError: pass self._stale = [] 6. Performance Benchmarks
| Scenario | Metric | Value |
|---|---|---|
| First Synthesis Latency | Good network | ~800ms |
| First Synthesis Latency | Average network | ~1.5-2s |
| Cache Hit Playback | Cold start | <50ms |
| Concurrent Synthesis Throughput | 3 workers | ~3 para/sec |
| Memory (Playing) | 100 cached segments | ~180MB |
| Cache Dir Size | Typical novel (500 para) | ~150-300MB |
| Watchdog CPU | Playing | <0.1% |


