Ebook Voice Player TTS Engine Deep Dive: Async Synthesis, Smart Caching & Cross-Platform Playback Watchdog

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_once dispatches 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_index unaffected
• Discard Pending: Queue entries > current_index purged to avoid speed mismatch
• Seamless Chain: Current segment finishes, auto-picks up new generator's audio
• Graceful Old Generator Exit: Added to _stale list, 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%

---

7. Troubleshooting Guide

Symptom	Likely Cause	Debug Steps
Stuck on first paragraph	Windows on_stop failure	Check watchdog logs [DBAUDIO] playing... / [DBAUDIO] play() called
403 on synthesis	System clock skew/old version	Sync system time, update app
Synthesis timeout/no response	Network/proxy/firewall	Packet capture to verify *.azure.com connectivity
Speed change not applied	Hot-swap not triggered	Verify set_speed() called with _playing=True
Cache dir bloating	Old version caches not cleaned	Manual clear_cache() or version migration

---

8. Extensibility: Local TTS Engine Interface Ready

Current architecture reserves extension points for future local TTS (Piper/Coqui):

# Future: swap TTSGenerator implementation
class LocalTTSGenerator(TTSGenerator):
    """Local offline synthesizer (reserved interface)"""
    def _run(self):
        # Load local model
        # Batch inference
        # Same event interface: paragraph/progress/error/finished
        pass

# TTSService unchanged, just swap generator instance
def switch_to_local_engine(self):
    self._replace_generator()
    self.generator = LocalTTSGenerator()
    self._connect_generator()

---

9. Conclusion

Ebook Voice Player's TTS engine demonstrates production-grade engineering for audio apps:

1. Async + Cache + Watchdog triad solves cloud synthesis latency, reliability, cross-platform playback
2. Clean State Machine enables hot-swap for "seamless" speed/voice/chapter switching
3. Error Classification + Localization turns technical exceptions into actionable user guidance
4. Resource Lifecycle Management prevents leaks, residue, supports manual cleanup

This engine architecture has been running stably in production, serving tens of thousands of daily users.

---

Try It: Microsoft Store - Ebook Voice Player

Source Code: tts_engine.py / tts_app/tts_engine.py