"""Shared TTS client plumbing: cancellation, retries, chunk bookkeeping.""" import contextlib import logging import random import threading import time from pathlib import Path from typing import Optional from .. import config logger = logging.getLogger(__name__) class ConversionCancelled(Exception): """Raised when the run's cancel event is set (between requests).""" class NonRetryableTTSError(RuntimeError): """A deterministic server-side request/config error; retrying cannot help. Raised by clients for failures that will reproduce identically on every attempt (missing server-side reference transcripts, unknown model ids, missing model contracts, ...). process_chunk_with_retry skips its remaining attempts and back-off sleeps for these and re-raises, so the converter aborts with the actionable message instead of burning the retry budget on a request that can never succeed. Subclasses RuntimeError so except-handlers written for the plain HTTP-error case keep working. """ # How a run supplies its voice: a built-in CustomVoice speaker, by cloning # a reference audio clip (the faster and audiocpp backends always clone # server-side; only the Qwen client branches on this at request time), or # designed from an instruction (Qwen's VoiceDesign model). VOICE_MODE_CUSTOM = "custom_voice" VOICE_MODE_CLONE = "voice_clone" VOICE_MODE_DESIGN = "voice_design" VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE, VOICE_MODE_DESIGN) def resolve_request_seed() -> int: """Resolve the seed sent with every request. Returns config.SEED as-is, or (with CONSTANT_SEED and SEED < 0) one random value drawn per run, meant to be reused for every request so the voice stays consistent across chunk boundaries. Without CONSTANT_SEED, -1 is returned so the server re-samples the voice on every generation. """ seed = config.SEED if config.CONSTANT_SEED and seed < 0: seed = random.randrange(2 ** 31) return seed class BaseTTSClient: """Shared chunk retry logic, heartbeat, and chunk file bookkeeping. CHUNKS_DIR is the scratch folder the generated chunk files are written to — provided by the converter that owns the run's folders, never a module global, so concurrent runs (and tests) cannot step on each other. """ # Class-level defaults so a partially-constructed instance behaves like # a plain console run (tests build clients via __new__). cancel = None quiet = False def __init__(self, chunks_dir: Path, quiet: bool = False): self.chunks_dir = Path(chunks_dir) # Quiet silences console prints (the run view owns the screen). self.quiet = bool(quiet) # Set by the converter when the run is cancellable (the TUI run # view): a threading.Event that, once set, aborts the run between # requests (and interrupts retry back-off sleeps). self.cancel = None def _report(self, message: str) -> None: """Print a console line unless quiet (the run view owns the screen).""" if not self.quiet: print(message) def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: """Generate one audio chunk; returns its path in the chunks folder.""" raise NotImplementedError def _cancel_requested(self) -> bool: """True when the run's cancel event has been set (if any).""" return isinstance(self.cancel, threading.Event) \ and self.cancel.is_set() def _check_cancelled(self) -> None: """Raise ConversionCancelled when the cancel event is set.""" if self._cancel_requested(): raise ConversionCancelled("Cancelled by user") def _sleep(self, seconds: float) -> None: """Sleep SECONDS, cut short (raising) when the cancel event sets.""" if isinstance(self.cancel, threading.Event): if self.cancel.wait(seconds): raise ConversionCancelled("Cancelled by user") else: time.sleep(seconds) def _chunk_path(self, chunk_num: int, suffix: str) -> Path: """Resolve the target path for a chunk, removing stale files first. Any stale chunk file for this index is removed so a retry or extension change can never leave two files matching chunk_NNNN.*. """ for stale in self.chunks_dir.glob(f"chunk_{chunk_num:04d}.*"): try: stale.unlink() except OSError as exc: logger.debug("Could not remove stale chunk file %s: %s", stale, exc) return self.chunks_dir / f"chunk_{chunk_num:04d}{suffix}" def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]: """Process a chunk with retry logic. Returns the generated chunk file's path, or None when all attempts failed. Raises ConversionCancelled when the run was cancelled. NonRetryableTTSError is logged once and re-raised without consuming the remaining attempts (the identical request can never succeed). """ for attempt in range(config.MAX_RETRIES): self._check_cancelled() try: result = self.generate_chunk(text, chunk_num) if result and Path(result).exists(): return Path(result) logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) except ConversionCancelled: raise except NonRetryableTTSError as exc: logger.error("Chunk %d error (not retried): %s", chunk_num, exc) raise except Exception as exc: logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) if attempt < config.MAX_RETRIES - 1: sleep_time = 5 + (2 ** attempt) logger.info("Waiting %ds before retry...", sleep_time) self._sleep(sleep_time) logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) return None @contextlib.contextmanager def _chunk_heartbeat(self, chunk_num: int): """Log a periodic "still working" record while a request generates.""" stop = threading.Event() subject = f"Chunk {chunk_num}" def _beat(): start = time.time() while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): elapsed = time.time() - start if self.quiet: logger.info("%s still generating — %dm %ds elapsed", subject, int(elapsed // 60), int(elapsed % 60)) else: print(f"[...] {subject} still generating — " f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True) thread = threading.Thread(target=_beat, daemon=True) thread.start() try: yield finally: stop.set() thread.join()