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

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 

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)}") 

# 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 

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) 

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 

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 

# 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 

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() 

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() 

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 

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 = []