From 780e9bd78023db6db79a696c0edf4fabc835f0df Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Sun, 12 Jul 2026 10:29:59 +0000 Subject: [PATCH] refactor(cleanup): remove Legacy TTS Architecture Delete legacy backend infrastructure: - abogen/tts_backend.py (TTSBackend protocol, TTSBackendMetadata) - abogen/tts_backend_registry.py (TTSBackendRegistry, global singleton, register_backend) - abogen/tts_backends/ (kokoro.py, supertonic.py, __init__.py) Delete legacy tests: - tests/test_tts_backend.py - tests/test_kokoro_backend.py - tests/test_voice_formula_resolution.py - tests/test_tts_supertonic_unsupported_chars.py Production code now uses only Plugin Architecture via create_pipeline(). All contract, behavioral, and integration tests pass. 2 pre-existing failures in test_supertonic_plugin.py (mock engine mismatch). --- abogen/tts_backend.py | 89 ---- abogen/tts_backend_registry.py | 146 ------ abogen/tts_backends/__init__.py | 20 - abogen/tts_backends/kokoro.py | 179 ------- abogen/tts_backends/supertonic.py | 392 --------------- abogen/tts_plugin/types.py | 222 ++++----- abogen/tts_plugin/utils.py | 470 +++++++++--------- docs/architecture-amendment-001.md | 178 +++---- tests/test_kokoro_backend.py | 216 -------- tests/test_tts_backend.py | 314 ------------ .../test_tts_supertonic_unsupported_chars.py | 108 ---- tests/test_voice_formula_resolution.py | 18 - 12 files changed, 435 insertions(+), 1917 deletions(-) delete mode 100644 abogen/tts_backend.py delete mode 100644 abogen/tts_backend_registry.py delete mode 100644 abogen/tts_backends/__init__.py delete mode 100644 abogen/tts_backends/kokoro.py delete mode 100644 abogen/tts_backends/supertonic.py delete mode 100644 tests/test_kokoro_backend.py delete mode 100644 tests/test_tts_backend.py delete mode 100644 tests/test_tts_supertonic_unsupported_chars.py delete mode 100644 tests/test_voice_formula_resolution.py diff --git a/abogen/tts_backend.py b/abogen/tts_backend.py deleted file mode 100644 index 24caafd..0000000 --- a/abogen/tts_backend.py +++ /dev/null @@ -1,89 +0,0 @@ -""" -TTS Backend Interface - -This module defines the protocol for TTS backends and the -metadata model that describes a backend implementation. -""" - -from dataclasses import dataclass -from typing import Protocol, List, Dict, Any - - -@dataclass(frozen=True) -class TTSBackendMetadata: - """ - Immutable metadata describing a TTS backend implementation. - - Attributes: - id: Unique backend identifier (e.g. ``"kokoro"``, ``"supertonic"``). - name: Human-readable display name. - description: Short description of the backend. - voices: Tuple of supported voice identifiers. - """ - - id: str - name: str - description: str - voices: tuple[str, ...] = () - - -class TTSBackend(Protocol): - """ - Protocol for TTS backends. - - All TTS backends must implement this interface to be compatible - with the application. - """ - - @property - def metadata(self) -> TTSBackendMetadata: - ... - - def __init__(self, **kwargs) -> None: - """ - Initialize the TTS backend. - - Args: - **kwargs: Backend-specific configuration parameters - """ - ... - - def synthesize(self, text: str, **kwargs) -> bytes: - """ - Synthesize speech from text. - - Args: - text: Text to synthesize - **kwargs: Additional parameters for synthesis - - Returns: - Audio data as bytes - """ - ... - - def get_available_voices(self) -> List[str]: - """ - Get list of available voices. - - Returns: - List of voice identifiers - """ - ... - - def get_supported_formats(self) -> List[str]: - """ - Get list of supported audio formats. - - Returns: - List of supported audio formats - """ - ... - - def get_info(self) -> Dict[str, Any]: - """ - Get backend information. - - Returns: - Dictionary with backend information - """ - ... diff --git a/abogen/tts_backend_registry.py b/abogen/tts_backend_registry.py deleted file mode 100644 index 6280f1d..0000000 --- a/abogen/tts_backend_registry.py +++ /dev/null @@ -1,146 +0,0 @@ -""" -TTS Backend Registry - -Provides a global registry for TTS backend factories. -Backends register themselves with metadata and a factory callable. -The registry is universal and does not know about backend constructors. -""" - -from typing import Callable, Any - -from abogen.tts_backend import TTSBackend, TTSBackendMetadata - - -class TTSBackendRegistry: - """Registry of TTS backend factories. - - Stores metadata and factory callables for registered backends. - """ - - def __init__(self) -> None: - self._backends: dict[str, TTSBackendMetadata] = {} - self._factories: dict[str, Callable[..., TTSBackend]] = {} - - def register( - self, - metadata: TTSBackendMetadata, - factory: Callable[..., TTSBackend], - ) -> None: - """Register a backend with its metadata and factory callable.""" - self._backends[metadata.id] = metadata - self._factories[metadata.id] = factory - - def is_registered(self, backend_id: str) -> bool: - """Return True if a backend with the given id is registered.""" - return backend_id in self._backends - - def list_backends(self) -> list[TTSBackendMetadata]: - """Return metadata for all registered backends.""" - return list(self._backends.values()) - - def get_metadata(self, backend_id: str) -> TTSBackendMetadata: - """Get metadata for a specific backend. - - Raises: - KeyError: If backend with given id is not registered. - """ - if backend_id not in self._backends: - raise KeyError(f"Unknown backend: {backend_id}") - return self._backends[backend_id] - - def create_backend(self, backend_id: str, **kwargs: Any) -> TTSBackend: - """Create a backend instance by id. - - Raises: - KeyError: If backend with given id is not registered. - """ - if backend_id not in self._factories: - raise KeyError(f"Unknown backend: {backend_id}") - return self._factories[backend_id](**kwargs) - - def resolve_backend_for_voice( - self, - spec: str, - fallback: str = "kokoro", - ) -> str: - """Determine which backend owns the given voice specification. - - Resolution rules: - 1. Empty spec -> fallback - 2. Kokoro formula (contains '*' or '+') -> "kokoro" - 3. Exact voice ID match against registered backends -> backend id - 4. Unknown voice -> fallback - """ - raw = str(spec or "").strip() - if not raw: - return fallback - - if "*" in raw or "+" in raw: - return "kokoro" - - upper = raw.upper() - for metadata in self._backends.values(): - if upper in metadata.voices: - return metadata.id - - return fallback - - -_registry = TTSBackendRegistry() - - -def register_backend( - metadata: TTSBackendMetadata, - factory: Callable[..., TTSBackend], -) -> None: - """Register a TTS backend in the global registry.""" - _registry.register(metadata, factory) - - -def get_metadata(backend_id: str) -> TTSBackendMetadata: - """Get metadata for a specific backend by id. - - Ensures all backends are registered by importing the tts_backends - package on first access. - - Raises: - KeyError: If backend with given id is not registered. - """ - import abogen.tts_backends # noqa: F401 — triggers backend registration - return _registry.get_metadata(backend_id) - - -def get_default_voice(backend_id: str, fallback: str = "") -> str: - """Return the first voice of a backend, or *fallback* if none.""" - voices = get_metadata(backend_id).voices - return voices[0] if voices else fallback - - -def create_backend(backend_id: str, **kwargs: Any) -> TTSBackend: - """Create a TTS backend instance by provider id.""" - return _registry.create_backend(backend_id, **kwargs) - - -def is_registered_backend(backend_id: str) -> bool: - """Return True if *backend_id* is a registered TTS backend.""" - import abogen.tts_backends # noqa: F401 — triggers backend registration - return _registry.is_registered(backend_id) - - -def resolve_backend_for_voice( - spec: str, - fallback: str = "kokoro", -) -> str: - """Determine which backend owns the given voice specification. - - Ensures all backends are registered by importing the tts_backends - package on first access. - - Resolution rules: - 1. Empty spec -> fallback - 2. Kokoro formula (contains '*' or '+') -> "kokoro" - 3. Exact voice ID match against registered backends -> backend id - 4. Unknown voice -> fallback - """ - import abogen.tts_backends # noqa: F401 — triggers backend registration - return _registry.resolve_backend_for_voice(spec, fallback=fallback) diff --git a/abogen/tts_backends/__init__.py b/abogen/tts_backends/__init__.py deleted file mode 100644 index 71982f2..0000000 --- a/abogen/tts_backends/__init__.py +++ /dev/null @@ -1,20 +0,0 @@ -"""TTS backends package. - -Backend modules are auto-discovered and imported here. -Each backend module registers itself with the global registry -when imported. -""" - -import importlib -import pkgutil - - -def _discover_backends(): - """Import all modules in this package to trigger their registration.""" - package = __name__ - for _importer, modname, _ispkg in pkgutil.iter_modules(path=__path__): - importlib.import_module(f"{package}.{modname}") - - -_discover_backends() - diff --git a/abogen/tts_backends/kokoro.py b/abogen/tts_backends/kokoro.py deleted file mode 100644 index 9a4d7e9..0000000 --- a/abogen/tts_backends/kokoro.py +++ /dev/null @@ -1,179 +0,0 @@ -""" -Kokoro TTS Backend - -Encapsulates the Kokoro KPipeline as a TTSBackend implementation. -""" - -from __future__ import annotations - -from typing import Any, Dict, Iterator, List, Optional - -import numpy as np - -from abogen.tts_backend import TTSBackendMetadata - -# Internal voice list — source of truth for Kokoro voices. -# The rest of the project accesses voices via get_metadata("kokoro").voices. -_VOICES_INTERNAL = [ - "af_alloy", - "af_aoede", - "af_bella", - "af_heart", - "af_jessica", - "af_kore", - "af_nicole", - "af_nova", - "af_river", - "af_sarah", - "af_sky", - "am_adam", - "am_echo", - "am_eric", - "am_fenrir", - "am_liam", - "am_michael", - "am_onyx", - "am_puck", - "am_santa", - "bf_alice", - "bf_emma", - "bf_isabella", - "bf_lily", - "bm_daniel", - "bm_fable", - "bm_george", - "bm_lewis", - "ef_dora", - "em_alex", - "em_santa", - "ff_siwis", - "hf_alpha", - "hf_beta", - "hm_omega", - "hm_psi", - "if_sara", - "im_nicola", - "jf_alpha", - "jf_gongitsune", - "jf_nezumi", - "jf_tebukuro", - "jm_kumo", - "pf_dora", - "pm_alex", - "pm_santa", - "zf_xiaobei", - "zf_xiaoni", - "zf_xiaoxiao", - "zf_xiaoyi", - "zm_yunjian", - "zm_yunxi", - "zm_yunxia", - "zm_yunyang", -] - -_KOKORO_METADATA = TTSBackendMetadata( - id="kokoro", - name="Kokoro", - description="Kokoro TTS engine", - voices=tuple(_VOICES_INTERNAL), -) - - -def _load_kpipeline(): - """Lazy-load Kokoro dependencies.""" - from kokoro import KPipeline # type: ignore[import-not-found] - - return KPipeline - - -class KokoroBackend: - """TTSBackend implementation wrapping the Kokoro KPipeline. - - All interaction with KPipeline is encapsulated here. - The rest of the project depends only on this class. - """ - - def __init__(self, **kwargs: Any) -> None: - lang_code = kwargs["lang_code"] - repo_id = kwargs.get("repo_id", "hexgrad/Kokoro-82M") - device = kwargs.get("device", "cpu") - - KPipeline = _load_kpipeline() - self._pipeline = KPipeline( - lang_code=lang_code, - repo_id=repo_id, - device=device, - ) - self._lang_code = lang_code - - @property - def metadata(self) -> TTSBackendMetadata: - return _KOKORO_METADATA - - def __call__( - self, - text: str, - *, - voice: Any, - speed: float = 1.0, - split_pattern: Optional[str] = None, - ) -> Iterator[Any]: - """Delegate to KPipeline's __call__.""" - return self._pipeline( - text, - voice=voice, - speed=speed, - split_pattern=split_pattern, - ) - - def load_single_voice(self, voice_name: str) -> Any: - """Load a single voice tensor. Used by voice formula system.""" - return self._pipeline.load_single_voice(voice_name) - - def synthesize(self, text: str, **kwargs: Any) -> bytes: - """Synthesize speech from text. Returns raw audio bytes.""" - voice = kwargs.get("voice", "") - speed = kwargs.get("speed", 1.0) - split_pattern = kwargs.get("split_pattern", None) - - audio_parts: list[np.ndarray] = [] - for segment in self(text, voice=voice, speed=speed, split_pattern=split_pattern): - audio = segment.audio - if hasattr(audio, "numpy"): - audio = audio.numpy() - audio_parts.append(np.asarray(audio, dtype="float32")) - - if not audio_parts: - return b"" - - combined = np.concatenate(audio_parts).astype("float32", copy=False) - return combined.tobytes() - - def get_available_voices(self) -> List[str]: - """Return known Kokoro voice identifiers.""" - return list(self.metadata.voices) - - def get_supported_formats(self) -> List[str]: - """Kokoro outputs raw PCM float32 audio.""" - return ["pcm_float32"] - - def get_info(self) -> Dict[str, Any]: - return { - "id": "kokoro", - "name": "Kokoro", - "lang_code": self._lang_code, - } - - -def create_kokoro_backend(**kwargs: Any) -> KokoroBackend: - """Factory callable registered with TTSBackendRegistry.""" - return KokoroBackend(**kwargs) - - -# --- Registration --- -from abogen.tts_backend_registry import register_backend # noqa: E402 - -register_backend( - metadata=_KOKORO_METADATA, - factory=create_kokoro_backend, -) diff --git a/abogen/tts_backends/supertonic.py b/abogen/tts_backends/supertonic.py deleted file mode 100644 index eaccfc5..0000000 --- a/abogen/tts_backends/supertonic.py +++ /dev/null @@ -1,392 +0,0 @@ -from __future__ import annotations - -import ast -from dataclasses import dataclass -import logging -import math -import re -from typing import Any, Dict, Iterable, Iterator, List, Optional - -import numpy as np - - -logger = logging.getLogger(__name__) - - -DEFAULT_SUPERTONIC_VOICES = ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5") - -from abogen.tts_backend import TTSBackendMetadata - -_SUPERTONIC_METADATA = TTSBackendMetadata( - id="supertonic", - name="SuperTonic", - description="SuperTonic TTS engine", - voices=DEFAULT_SUPERTONIC_VOICES, -) - - -@dataclass -class SupertonicSegment: - graphemes: str - audio: np.ndarray - - -def _ensure_float32_mono(wav: Any) -> np.ndarray: - arr = np.asarray(wav, dtype="float32") - if arr.ndim == 2: - # (n, 1) or (1, n) or (n, channels) - if arr.shape[0] == 1 and arr.shape[1] > 1: - arr = arr.reshape(-1) - else: - arr = arr[:, 0] - return arr.reshape(-1) - - -def _resample_linear(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray: - if src_rate == dst_rate: - return audio - if audio.size == 0: - return audio - ratio = dst_rate / float(src_rate) - new_len = int(round(audio.size * ratio)) - if new_len <= 1: - return np.zeros(0, dtype="float32") - x_old = np.linspace(0.0, 1.0, num=audio.size, endpoint=False) - x_new = np.linspace(0.0, 1.0, num=new_len, endpoint=False) - return np.interp(x_new, x_old, audio).astype("float32", copy=False) - - -def _split_text( - text: str, *, split_pattern: Optional[str], max_chunk_length: int -) -> list[str]: - stripped = (text or "").strip() - if not stripped: - return [] - parts: list[str] - if split_pattern: - try: - parts = [p.strip() for p in re.split(split_pattern, stripped) if p.strip()] - except re.error: - parts = [stripped] - else: - parts = [stripped] - - # Enforce max length by hard-splitting long parts. - result: list[str] = [] - for part in parts: - if len(part) <= max_chunk_length: - result.append(part) - continue - start = 0 - while start < len(part): - end = min(len(part), start + max_chunk_length) - # Try to split at whitespace. - if end < len(part): - ws = part.rfind(" ", start, end) - if ws > start + 40: - end = ws - chunk = part[start:end].strip() - if chunk: - result.append(chunk) - start = end - return result - - -_UNSUPPORTED_CHARS_RE = re.compile( - r"unsupported character\(s\):\s*(\[[^\]]*\])", re.IGNORECASE -) - - -def _parse_unsupported_characters(error: BaseException) -> list[str]: - """Best-effort extraction of unsupported characters from SuperTonic errors.""" - - message = " ".join( - str(part) for part in getattr(error, "args", ()) if part is not None - ) or str(error) - match = _UNSUPPORTED_CHARS_RE.search(message) - if not match: - return [] - - raw = match.group(1) - try: - value = ast.literal_eval(raw) - except Exception: - return [] - - if isinstance(value, (list, tuple)): - out: list[str] = [] - for item in value: - if item is None: - continue - s = str(item) - if s: - out.append(s) - return out - - if isinstance(value, str) and value: - return [value] - - return [] - - -def _remove_unsupported_characters(text: str, unsupported: Iterable[str]) -> str: - result = text - for item in unsupported: - if not item: - continue - result = result.replace(item, "") - return result - - -def _configure_supertonic_gpu() -> None: - """Patch supertonic's config to enable GPU acceleration if available.""" - try: - import onnxruntime as ort - - available = ort.get_available_providers() - - # Use CUDA if available, skip TensorRT (requires extra libs not always present) - # TensorrtExecutionProvider may be listed as available but fail at runtime - # if TensorRT libraries (libnvinfer.so) are not installed - providers = [] - if "CUDAExecutionProvider" in available: - providers.append("CUDAExecutionProvider") - providers.append("CPUExecutionProvider") - - # Patch supertonic's config and loader before TTS import - # We must patch both because loader imports the value at module load time - import supertonic.config as supertonic_config - import supertonic.loader as supertonic_loader - - supertonic_config.DEFAULT_ONNX_PROVIDERS = providers - supertonic_loader.DEFAULT_ONNX_PROVIDERS = providers - logger.info("Supertonic ONNX providers configured: %s", providers) - except Exception as exc: - logger.warning("Could not configure supertonic GPU providers: %s", exc) - - -class SupertonicPipeline: - """Minimal adapter that mimics Kokoro's pipeline iteration interface.""" - - def __init__( - self, - *, - sample_rate: int, - auto_download: bool = True, - total_steps: int = 5, - max_chunk_length: int = 300, - ) -> None: - self.sample_rate = int(sample_rate) - self.total_steps = int(total_steps) - self.max_chunk_length = int(max_chunk_length) - - # Configure GPU providers before importing TTS - _configure_supertonic_gpu() - - try: - from supertonic import TTS # type: ignore[import-not-found] - except Exception as exc: # pragma: no cover - raise RuntimeError( - "Supertonic is not installed. Install it with `pip install supertonic`." - ) from exc - - self._tts = TTS(auto_download=auto_download) - - def __call__( - self, - text: str, - *, - voice: str, - speed: float, - split_pattern: Optional[str] = None, - total_steps: Optional[int] = None, - ) -> Iterator[SupertonicSegment]: - voice_name = (voice or "").strip() or "M1" - steps = int(total_steps) if total_steps is not None else self.total_steps - steps = max(2, min(15, steps)) - speed_value = float(speed) if speed is not None else 1.0 - speed_value = max(0.7, min(2.0, speed_value)) - - style = self._tts.get_voice_style(voice_name=voice_name) - chunks = _split_text( - text, split_pattern=split_pattern, max_chunk_length=self.max_chunk_length - ) - for chunk in chunks: - chunk_to_speak = chunk - removed: set[str] = set() - last_exc: Exception | None = None - - # SuperTonic can raise ValueError for unsupported characters; strip and retry. - for attempt in range(3): - try: - wav, duration = self._tts.synthesize( - text=chunk_to_speak, - voice_style=style, - total_steps=steps, - speed=speed_value, - max_chunk_length=self.max_chunk_length, - silence_duration=0.0, - verbose=False, - ) - break - except ValueError as exc: - last_exc = exc - unsupported = _parse_unsupported_characters(exc) - if not unsupported: - raise - - removed.update(unsupported) - sanitized = _remove_unsupported_characters( - chunk_to_speak, unsupported - ).strip() - - # If we didn't change anything, don't loop forever. - if sanitized == chunk_to_speak.strip(): - raise - - chunk_to_speak = sanitized - if not chunk_to_speak: - logger.warning( - "SuperTonic: dropped a chunk after removing unsupported characters: %s", - sorted(removed), - ) - break - - if attempt == 0: - logger.warning( - "SuperTonic: removed unsupported characters %s and retried.", - sorted(removed), - ) - else: - # Exhausted retries. - assert last_exc is not None - raise last_exc - - if not chunk_to_speak: - continue - - audio = _ensure_float32_mono(wav) - - # If duration is present, infer the source sample rate and resample if needed. - src_rate = self.sample_rate - try: - dur = float(duration) - if dur > 0 and audio.size > 0: - inferred = int(round(audio.size / dur)) - if 8000 <= inferred <= 96000: - src_rate = inferred - except Exception: - pass - - if src_rate != self.sample_rate: - audio = _resample_linear(audio, src_rate, self.sample_rate) - - yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio) - - -class SupertonicBackend: - """Supertonic TTS backend implementing the TTSBackend protocol. - - Encapsulates ``SupertonicPipeline`` as an internal implementation detail. - """ - - @property - def metadata(self) -> TTSBackendMetadata: - return _SUPERTONIC_METADATA - - def __init__(self, **kwargs: Any) -> None: - self._pipeline = SupertonicPipeline( - sample_rate=kwargs.get("sample_rate", 24000), - auto_download=kwargs.get("auto_download", True), - total_steps=kwargs.get("total_steps", 5), - ) - - def synthesize(self, text: str, **kwargs: Any) -> bytes: - """Synthesize speech and return raw audio bytes (WAV). - - Delegates to the internal :class:`SupertonicPipeline` and concatenates - all produced segments into a single byte buffer. - """ - import io - - import soundfile as sf - - voice = kwargs.get("voice", "M1") - speed = float(kwargs.get("speed", 1.0)) - split_pattern = kwargs.get("split_pattern") - total_steps = kwargs.get("total_steps") - - segments = self._pipeline( - text, - voice=voice, - speed=speed, - split_pattern=split_pattern, - total_steps=total_steps, - ) - - audio_parts: list[np.ndarray] = [] - for seg in segments: - audio_parts.append(seg.audio) - - if not audio_parts: - return b"" - - combined = np.concatenate(audio_parts) - buf = io.BytesIO() - sf.write(buf, combined, self._pipeline.sample_rate, format="WAV") - return buf.getvalue() - - def get_available_voices(self) -> List[str]: - """Return the list of built-in SuperTonic voice identifiers.""" - return list(self.metadata.voices) - - def get_supported_formats(self) -> List[str]: - return ["wav"] - - def get_info(self) -> Dict[str, Any]: - return { - "sample_rate": self._pipeline.sample_rate, - "total_steps": self._pipeline.total_steps, - "max_chunk_length": self._pipeline.max_chunk_length, - "voices": list(DEFAULT_SUPERTONIC_VOICES), - } - - def __call__( - self, - text: str, - *, - voice: str, - speed: float, - split_pattern: Optional[str] = None, - total_steps: Optional[int] = None, - ) -> Iterator[SupertonicSegment]: - """Backward-compatible call interface, delegates to the pipeline.""" - return self._pipeline( - text, - voice=voice, - speed=speed, - split_pattern=split_pattern, - total_steps=total_steps, - ) - - -def create_supertonic_backend(**kwargs: Any) -> SupertonicBackend: - """Create a SuperTonic TTS backend instance. - - Args: - sample_rate: Audio sample rate. Defaults to 24000. - auto_download: Auto-download models. Defaults to True. - total_steps: Inference steps. Defaults to 5. - - Returns: - SupertonicBackend instance. - """ - return SupertonicBackend(**kwargs) - - -from abogen.tts_backend_registry import register_backend # noqa: E402 - -register_backend( - metadata=_SUPERTONIC_METADATA, - factory=create_supertonic_backend, -) diff --git a/abogen/tts_plugin/types.py b/abogen/tts_plugin/types.py index 36e4275..5467c08 100644 --- a/abogen/tts_plugin/types.py +++ b/abogen/tts_plugin/types.py @@ -1,111 +1,111 @@ -"""Core domain types for the TTS Plugin Architecture. - -This module contains immutable value objects that form the core domain. -These types have zero dependencies and are used across the plugin system. -""" - -from __future__ import annotations - -from dataclasses import dataclass, field -from typing import Any, Mapping - - -@dataclass(frozen=True) -class AudioFormat: - """Immutable value object representing an audio format. - - Attributes: - mime: MIME type of the audio (e.g., "audio/wav", "audio/mpeg"). - extension: File extension (e.g., "wav", "mp3"). - """ - - mime: str - extension: str - - -@dataclass(frozen=True) -class Duration: - """Immutable value object representing a time duration. - - Attributes: - seconds: Duration in seconds. - """ - - seconds: float - - -@dataclass(frozen=True) -class VoiceSelection: - """Immutable value object for voice selection. Opaque to engine. - - Attributes: - source: Voice source identifier (e.g., "builtin", "clone"). - key: Voice key within the source. - payload: Optional payload for clone/blend sources. - """ - - source: str - key: str - payload: Any = None - - -@dataclass(frozen=True) -class ParameterValues: - """Immutable value object for synthesis parameters. Behaves like Mapping[str, Any]. - - Attributes: - values: Mapping of parameter names to their values. - """ - - values: Mapping[str, Any] = field(default_factory=dict) - - -@dataclass(frozen=True) -class SynthesisRequest: - """Immutable value object for a synthesis request. - - Attributes: - text: Text to synthesize. - voice: Voice selection. - parameters: Synthesis parameters. - format: Desired audio output format. - """ - - text: str - voice: VoiceSelection - parameters: ParameterValues - format: AudioFormat - - -@dataclass(frozen=True) -class SynthesizedAudio: - """Immutable value object for synthesized audio result. - - Attributes: - data: Raw audio bytes. - format: Audio format of the result. - duration: Duration of the audio. - """ - - data: bytes - format: AudioFormat - duration: Duration - - -@dataclass(frozen=True) -class EngineConfig: - """Immutable configuration of an Engine instance. - - Contains parameters that define how a particular Engine instance is - created and that remain constant throughout the lifetime of that Engine. - - Plugin implementations may ignore fields that are not applicable to them. - - Attributes: - device: Device to use (e.g., "cpu", "cuda:0"). - lang_code: Language code for the engine (e.g., "a" for Kokoro English). - Plugins that do not require a language code ignore this field. - """ - - device: str = "cpu" - lang_code: str = "a" +"""Core domain types for the TTS Plugin Architecture. + +This module contains immutable value objects that form the core domain. +These types have zero dependencies and are used across the plugin system. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Mapping + + +@dataclass(frozen=True) +class AudioFormat: + """Immutable value object representing an audio format. + + Attributes: + mime: MIME type of the audio (e.g., "audio/wav", "audio/mpeg"). + extension: File extension (e.g., "wav", "mp3"). + """ + + mime: str + extension: str + + +@dataclass(frozen=True) +class Duration: + """Immutable value object representing a time duration. + + Attributes: + seconds: Duration in seconds. + """ + + seconds: float + + +@dataclass(frozen=True) +class VoiceSelection: + """Immutable value object for voice selection. Opaque to engine. + + Attributes: + source: Voice source identifier (e.g., "builtin", "clone"). + key: Voice key within the source. + payload: Optional payload for clone/blend sources. + """ + + source: str + key: str + payload: Any = None + + +@dataclass(frozen=True) +class ParameterValues: + """Immutable value object for synthesis parameters. Behaves like Mapping[str, Any]. + + Attributes: + values: Mapping of parameter names to their values. + """ + + values: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class SynthesisRequest: + """Immutable value object for a synthesis request. + + Attributes: + text: Text to synthesize. + voice: Voice selection. + parameters: Synthesis parameters. + format: Desired audio output format. + """ + + text: str + voice: VoiceSelection + parameters: ParameterValues + format: AudioFormat + + +@dataclass(frozen=True) +class SynthesizedAudio: + """Immutable value object for synthesized audio result. + + Attributes: + data: Raw audio bytes. + format: Audio format of the result. + duration: Duration of the audio. + """ + + data: bytes + format: AudioFormat + duration: Duration + + +@dataclass(frozen=True) +class EngineConfig: + """Immutable configuration of an Engine instance. + + Contains parameters that define how a particular Engine instance is + created and that remain constant throughout the lifetime of that Engine. + + Plugin implementations may ignore fields that are not applicable to them. + + Attributes: + device: Device to use (e.g., "cpu", "cuda:0"). + lang_code: Language code for the engine (e.g., "a" for Kokoro English). + Plugins that do not require a language code ignore this field. + """ + + device: str = "cpu" + lang_code: str = "a" diff --git a/abogen/tts_plugin/utils.py b/abogen/tts_plugin/utils.py index 4a05fa9..acfb592 100644 --- a/abogen/tts_plugin/utils.py +++ b/abogen/tts_plugin/utils.py @@ -1,235 +1,235 @@ -"""TTS Plugin Architecture — direct utility functions. - -Provides helpers that replace the former compatibility adapter by -calling the Plugin Manager directly. -""" - -from __future__ import annotations - -from typing import Any, Iterator - -import numpy as np - -from abogen.tts_plugin.plugin_manager import get_plugin_manager - - -def get_voices(plugin_id: str) -> tuple[str, ...]: - """Return the voice-id tuple for *plugin_id*. - - Uses the official Plugin Architecture: PluginManager → Engine → VoiceLister. - First checks plugin manifest for static voice catalog. - """ - import logging - import tempfile - from pathlib import Path - - from abogen.tts_plugin.host_context import HostContext - from abogen.tts_plugin.types import EngineConfig - - manager = get_plugin_manager() - if not manager.has_plugin(plugin_id): - return () - - # Check manifest for static voice catalog - plugin_info = manager.get_plugin(plugin_id) - if plugin_info is not None: - manifest = plugin_info.get("manifest") - if manifest is not None and manifest.voices is not None: - return tuple(v.id for v in manifest.voices) - - ctx = HostContext( - config_dir=Path(tempfile.gettempdir()), - logger=logging.getLogger(f"abogen.utils.{plugin_id}"), - http_client=type("_StubHttpClient", (), { - "get": staticmethod(lambda url, **kw: None), - "post": staticmethod(lambda url, **kw: None), - })(), - ) - - try: - engine = manager.create_engine( - plugin_id, - context=ctx, - model_path=None, - config=EngineConfig(device="cpu"), - ) - except Exception: - return () - - try: - from abogen.tts_plugin.capabilities import VoiceLister - - if isinstance(engine, VoiceLister): - manifests = engine.listVoices("builtin") - return tuple(v.id for v in manifests) - return () - except Exception: - return () - finally: - engine.dispose() - - -def get_default_voice(plugin_id: str, fallback: str = "") -> str: - """Return the first voice of *plugin_id*, or *fallback*.""" - voices = get_voices(plugin_id) - return voices[0] if voices else fallback - - -def is_plugin_registered(plugin_id: str) -> bool: - """Check whether *plugin_id* is loaded by the Plugin Manager.""" - return get_plugin_manager().has_plugin(plugin_id) - - -def resolve_voice_to_plugin(spec: str, fallback: str = "kokoro") -> str: - """Determine which plugin owns the given voice specification. - - Resolution rules: - 1. Empty spec -> fallback - 2. Kokoro formula (contains '*' or '+') -> "kokoro" - 3. Exact voice-id match against loaded plugins -> plugin id - 4. Unknown voice -> fallback - """ - raw = str(spec or "").strip() - if not raw: - return fallback - - if "*" in raw or "+" in raw: - return "kokoro" - - upper = raw.upper() - manager = get_plugin_manager() - - for manifest in manager.list_plugins(): - for voice_source in manifest.engine.voiceSources: - if voice_source.type == "list" and isinstance(voice_source.config, dict): - try: - engine = manager.create_engine(manifest.id) - try: - if hasattr(engine, "listVoices"): - voice_manifests = engine.listVoices(voice_source.id) - voice_ids = [v.id.upper() for v in voice_manifests] - if upper in voice_ids: - return manifest.id - finally: - engine.dispose() - except Exception: - continue - - return fallback - - -class Pipeline: - """Callable wrapper around Engine / EngineSession. - - Presents the same interface that old callers expect:: - - pipeline = create_pipeline("kokoro", lang_code="a", device="cpu") - for segment in pipeline(text, voice="af_nova", speed=1.0): - audio = segment.audio - """ - - def __init__(self, engine: Any, **engine_kwargs: Any) -> None: - self._engine = engine - self._engine_kwargs = engine_kwargs - self._session: Any = None - - def _ensure_session(self) -> Any: - if self._session is None: - self._session = self._engine.createSession() - return self._session - - def __call__( - self, - text: str, - voice: str = "default", - speed: float = 1.0, - split_pattern: str | None = None, - **kwargs: Any, - ) -> Iterator[Any]: - from abogen.tts_plugin.types import ( - AudioFormat, - ParameterValues, - SynthesisRequest, - VoiceSelection, - ) - - session = self._ensure_session() - - params: dict[str, Any] = {"speed": speed} - if split_pattern is not None: - params["split_pattern"] = split_pattern - params.update(kwargs) - - request = SynthesisRequest( - text=text, - voice=VoiceSelection(source="builtin", key=voice), - parameters=ParameterValues(values=params), - format=AudioFormat(mime="audio/wav", extension="wav"), - ) - - result = session.synthesize(request) - audio_array = np.frombuffer(result.data, dtype=np.float32) - - from dataclasses import dataclass - - @dataclass - class Segment: - graphemes: str - audio: np.ndarray - - yield Segment(graphemes=text, audio=audio_array) - - def dispose(self) -> None: - if self._session is not None: - try: - self._session.dispose() - except Exception: - pass - self._session = None - - def __del__(self) -> None: - self.dispose() - - -def create_pipeline( - plugin_id: str, - *, - lang_code: str = "a", - device: str = "cpu", -) -> Pipeline: - """Create a callable TTS pipeline via the Plugin Architecture. - - Builds a proper HostContext and EngineConfig, then delegates to the - PluginManager to create the engine. Returns a :class:`Pipeline` whose - ``__call__`` interface matches the legacy ``TTSBackend`` callable protocol. - - Args: - plugin_id: Plugin identifier (e.g., "kokoro", "supertonic"). - lang_code: Language code for the engine. - device: Device to use (e.g., "cpu", "cuda:0"). - - Returns: - A callable Pipeline instance. - """ - import logging - import tempfile - from pathlib import Path - - from abogen.tts_plugin.host_context import HostContext - from abogen.tts_plugin.types import EngineConfig - - manager = get_plugin_manager() - - ctx = HostContext( - config_dir=Path(tempfile.gettempdir()), - logger=logging.getLogger(f"abogen.pipeline.{plugin_id}"), - http_client=type("_StubHttpClient", (), { - "get": staticmethod(lambda url, **kw: None), - "post": staticmethod(lambda url, **kw: None), - })(), - ) - - config = EngineConfig(device=device, lang_code=lang_code) - - engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config) - return Pipeline(engine) +"""TTS Plugin Architecture — direct utility functions. + +Provides helpers that replace the former compatibility adapter by +calling the Plugin Manager directly. +""" + +from __future__ import annotations + +from typing import Any, Iterator + +import numpy as np + +from abogen.tts_plugin.plugin_manager import get_plugin_manager + + +def get_voices(plugin_id: str) -> tuple[str, ...]: + """Return the voice-id tuple for *plugin_id*. + + Uses the official Plugin Architecture: PluginManager → Engine → VoiceLister. + First checks plugin manifest for static voice catalog. + """ + import logging + import tempfile + from pathlib import Path + + from abogen.tts_plugin.host_context import HostContext + from abogen.tts_plugin.types import EngineConfig + + manager = get_plugin_manager() + if not manager.has_plugin(plugin_id): + return () + + # Check manifest for static voice catalog + plugin_info = manager.get_plugin(plugin_id) + if plugin_info is not None: + manifest = plugin_info.get("manifest") + if manifest is not None and manifest.voices is not None: + return tuple(v.id for v in manifest.voices) + + ctx = HostContext( + config_dir=Path(tempfile.gettempdir()), + logger=logging.getLogger(f"abogen.utils.{plugin_id}"), + http_client=type("_StubHttpClient", (), { + "get": staticmethod(lambda url, **kw: None), + "post": staticmethod(lambda url, **kw: None), + })(), + ) + + try: + engine = manager.create_engine( + plugin_id, + context=ctx, + model_path=None, + config=EngineConfig(device="cpu"), + ) + except Exception: + return () + + try: + from abogen.tts_plugin.capabilities import VoiceLister + + if isinstance(engine, VoiceLister): + manifests = engine.listVoices("builtin") + return tuple(v.id for v in manifests) + return () + except Exception: + return () + finally: + engine.dispose() + + +def get_default_voice(plugin_id: str, fallback: str = "") -> str: + """Return the first voice of *plugin_id*, or *fallback*.""" + voices = get_voices(plugin_id) + return voices[0] if voices else fallback + + +def is_plugin_registered(plugin_id: str) -> bool: + """Check whether *plugin_id* is loaded by the Plugin Manager.""" + return get_plugin_manager().has_plugin(plugin_id) + + +def resolve_voice_to_plugin(spec: str, fallback: str = "kokoro") -> str: + """Determine which plugin owns the given voice specification. + + Resolution rules: + 1. Empty spec -> fallback + 2. Kokoro formula (contains '*' or '+') -> "kokoro" + 3. Exact voice-id match against loaded plugins -> plugin id + 4. Unknown voice -> fallback + """ + raw = str(spec or "").strip() + if not raw: + return fallback + + if "*" in raw or "+" in raw: + return "kokoro" + + upper = raw.upper() + manager = get_plugin_manager() + + for manifest in manager.list_plugins(): + for voice_source in manifest.engine.voiceSources: + if voice_source.type == "list" and isinstance(voice_source.config, dict): + try: + engine = manager.create_engine(manifest.id) + try: + if hasattr(engine, "listVoices"): + voice_manifests = engine.listVoices(voice_source.id) + voice_ids = [v.id.upper() for v in voice_manifests] + if upper in voice_ids: + return manifest.id + finally: + engine.dispose() + except Exception: + continue + + return fallback + + +class Pipeline: + """Callable wrapper around Engine / EngineSession. + + Presents the same interface that old callers expect:: + + pipeline = create_pipeline("kokoro", lang_code="a", device="cpu") + for segment in pipeline(text, voice="af_nova", speed=1.0): + audio = segment.audio + """ + + def __init__(self, engine: Any, **engine_kwargs: Any) -> None: + self._engine = engine + self._engine_kwargs = engine_kwargs + self._session: Any = None + + def _ensure_session(self) -> Any: + if self._session is None: + self._session = self._engine.createSession() + return self._session + + def __call__( + self, + text: str, + voice: str = "default", + speed: float = 1.0, + split_pattern: str | None = None, + **kwargs: Any, + ) -> Iterator[Any]: + from abogen.tts_plugin.types import ( + AudioFormat, + ParameterValues, + SynthesisRequest, + VoiceSelection, + ) + + session = self._ensure_session() + + params: dict[str, Any] = {"speed": speed} + if split_pattern is not None: + params["split_pattern"] = split_pattern + params.update(kwargs) + + request = SynthesisRequest( + text=text, + voice=VoiceSelection(source="builtin", key=voice), + parameters=ParameterValues(values=params), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + result = session.synthesize(request) + audio_array = np.frombuffer(result.data, dtype=np.float32) + + from dataclasses import dataclass + + @dataclass + class Segment: + graphemes: str + audio: np.ndarray + + yield Segment(graphemes=text, audio=audio_array) + + def dispose(self) -> None: + if self._session is not None: + try: + self._session.dispose() + except Exception: + pass + self._session = None + + def __del__(self) -> None: + self.dispose() + + +def create_pipeline( + plugin_id: str, + *, + lang_code: str = "a", + device: str = "cpu", +) -> Pipeline: + """Create a callable TTS pipeline via the Plugin Architecture. + + Builds a proper HostContext and EngineConfig, then delegates to the + PluginManager to create the engine. Returns a :class:`Pipeline` whose + ``__call__`` interface matches the legacy ``TTSBackend`` callable protocol. + + Args: + plugin_id: Plugin identifier (e.g., "kokoro", "supertonic"). + lang_code: Language code for the engine. + device: Device to use (e.g., "cpu", "cuda:0"). + + Returns: + A callable Pipeline instance. + """ + import logging + import tempfile + from pathlib import Path + + from abogen.tts_plugin.host_context import HostContext + from abogen.tts_plugin.types import EngineConfig + + manager = get_plugin_manager() + + ctx = HostContext( + config_dir=Path(tempfile.gettempdir()), + logger=logging.getLogger(f"abogen.pipeline.{plugin_id}"), + http_client=type("_StubHttpClient", (), { + "get": staticmethod(lambda url, **kw: None), + "post": staticmethod(lambda url, **kw: None), + })(), + ) + + config = EngineConfig(device=device, lang_code=lang_code) + + engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config) + return Pipeline(engine) diff --git a/docs/architecture-amendment-001.md b/docs/architecture-amendment-001.md index c8cb73e..51d60ac 100644 --- a/docs/architecture-amendment-001.md +++ b/docs/architecture-amendment-001.md @@ -1,89 +1,89 @@ -# Architecture Amendment #1: EngineConfig — `lang_code` field - -**Date:** 2026-07-12 -**Status:** Accepted -**PR:** #12 (Normalize Pipeline Public API) - -## Summary - -Add `lang_code: str = "a"` to `EngineConfig` and update its definition to clarify the architectural contract. - -## Background - -During migration from the old `KokoroBackend` to the Plugin Architecture, the `lang_code` parameter became a dead argument. The old backend read it from `**kwargs` and passed it to `KPipeline(lang_code=...)`. The new `KokoroPlugin.create_engine()` hardcodes `lang_code="a"`, ignoring the config entirely. Callers continued passing `lang_code` to `create_pipeline()`, unaware it had no effect. - -This is a functional regression relative to the pre-Plugin Architecture behavior. - -## Decision - -### 1. Updated EngineConfig definition - -**Before:** -``` -Immutable value object for engine initialization settings. -Contains only engine-specific settings, no resource references. -``` - -**After:** -``` -Immutable configuration of an Engine instance. -Contains parameters that define how a particular Engine instance is -created and that remain constant throughout the lifetime of that Engine. -Plugin implementations may ignore fields that are not applicable to them. -``` - -### 2. New field - -```python -@dataclass(frozen=True) -class EngineConfig: - device: str = "cpu" - lang_code: str = "a" -``` - -### 3. Architectural rules - -- **Fields in EngineConfig are optional unless explicitly required by a plugin.** -- **Plugins MUST ignore unsupported EngineConfig fields.** -- **All parameters that may vary between individual synthesis requests must remain in `SynthesisRequest.parameters`.** - -## Rationale - -Analysis of real TTS engines (Kokoro, Piper, XTTS, Coqui, StyleTTS2, Fish Speech) confirmed: - -| Parameter type | Where it belongs | Example | -|---------------|-----------------|---------| -| Engine instance config (immutable) | `EngineConfig` | `device`, `lang_code` | -| Synthesis parameters (per-request) | `SynthesisRequest.parameters` | `speed`, `split_pattern`, `total_steps` | - -`lang_code` determines the engine's behavior at creation time and cannot be changed during the engine's lifetime. It is not a synthesis parameter. - -## Impact on existing plugins - -| Plugin | `device` | `lang_code` | Notes | -|--------|----------|-------------|-------| -| Kokoro | Reads ✓ | Reads ✓ (was hardcoded, now from config) | Regression fixed | -| SuperTonic | Ignores | Ignores | No change — no language concept | -| Future plugins | May read | May ignore | Field-ignoring rule applies | - -## Contract tests added - -```python -class TestEngineConfigContract: - def test_default_lang_code(self) # EngineConfig().lang_code == "a" - def test_custom_lang_code(self) # EngineConfig(lang_code="j").lang_code == "j" - def test_immutability_lang_code(self) # frozen — cannot reassign - def test_plugins_may_ignore_irrelevant_fields(self) # field-ignoring rule - def test_engine_config_contains_engine_instance_configuration(self) # definition -``` - -## Files changed - -| File | Change | -|------|--------| -| `abogen/tts_plugin/types.py` | Updated docstring, added `lang_code: str = "a"` | -| `plugins/kokoro/__init__.py` | Reads `config.lang_code` instead of hardcoded `"a"` | -| `abogen/tts_plugin/utils.py` | `create_pipeline()` passes `lang_code` to `EngineConfig` | -| `tests/contracts/test_types_contract.py` | 5 new contract tests | -| `tests/contracts/test_plugin_manager_contract.py` | Updated assertion for `lang_code` | -| `tests/test_behavioral_regression.py` | Updated `test_engine_config_defaults` | +# Architecture Amendment #1: EngineConfig — `lang_code` field + +**Date:** 2026-07-12 +**Status:** Accepted +**PR:** #12 (Normalize Pipeline Public API) + +## Summary + +Add `lang_code: str = "a"` to `EngineConfig` and update its definition to clarify the architectural contract. + +## Background + +During migration from the old `KokoroBackend` to the Plugin Architecture, the `lang_code` parameter became a dead argument. The old backend read it from `**kwargs` and passed it to `KPipeline(lang_code=...)`. The new `KokoroPlugin.create_engine()` hardcodes `lang_code="a"`, ignoring the config entirely. Callers continued passing `lang_code` to `create_pipeline()`, unaware it had no effect. + +This is a functional regression relative to the pre-Plugin Architecture behavior. + +## Decision + +### 1. Updated EngineConfig definition + +**Before:** +``` +Immutable value object for engine initialization settings. +Contains only engine-specific settings, no resource references. +``` + +**After:** +``` +Immutable configuration of an Engine instance. +Contains parameters that define how a particular Engine instance is +created and that remain constant throughout the lifetime of that Engine. +Plugin implementations may ignore fields that are not applicable to them. +``` + +### 2. New field + +```python +@dataclass(frozen=True) +class EngineConfig: + device: str = "cpu" + lang_code: str = "a" +``` + +### 3. Architectural rules + +- **Fields in EngineConfig are optional unless explicitly required by a plugin.** +- **Plugins MUST ignore unsupported EngineConfig fields.** +- **All parameters that may vary between individual synthesis requests must remain in `SynthesisRequest.parameters`.** + +## Rationale + +Analysis of real TTS engines (Kokoro, Piper, XTTS, Coqui, StyleTTS2, Fish Speech) confirmed: + +| Parameter type | Where it belongs | Example | +|---------------|-----------------|---------| +| Engine instance config (immutable) | `EngineConfig` | `device`, `lang_code` | +| Synthesis parameters (per-request) | `SynthesisRequest.parameters` | `speed`, `split_pattern`, `total_steps` | + +`lang_code` determines the engine's behavior at creation time and cannot be changed during the engine's lifetime. It is not a synthesis parameter. + +## Impact on existing plugins + +| Plugin | `device` | `lang_code` | Notes | +|--------|----------|-------------|-------| +| Kokoro | Reads ✓ | Reads ✓ (was hardcoded, now from config) | Regression fixed | +| SuperTonic | Ignores | Ignores | No change — no language concept | +| Future plugins | May read | May ignore | Field-ignoring rule applies | + +## Contract tests added + +```python +class TestEngineConfigContract: + def test_default_lang_code(self) # EngineConfig().lang_code == "a" + def test_custom_lang_code(self) # EngineConfig(lang_code="j").lang_code == "j" + def test_immutability_lang_code(self) # frozen — cannot reassign + def test_plugins_may_ignore_irrelevant_fields(self) # field-ignoring rule + def test_engine_config_contains_engine_instance_configuration(self) # definition +``` + +## Files changed + +| File | Change | +|------|--------| +| `abogen/tts_plugin/types.py` | Updated docstring, added `lang_code: str = "a"` | +| `plugins/kokoro/__init__.py` | Reads `config.lang_code` instead of hardcoded `"a"` | +| `abogen/tts_plugin/utils.py` | `create_pipeline()` passes `lang_code` to `EngineConfig` | +| `tests/contracts/test_types_contract.py` | 5 new contract tests | +| `tests/contracts/test_plugin_manager_contract.py` | Updated assertion for `lang_code` | +| `tests/test_behavioral_regression.py` | Updated `test_engine_config_defaults` | diff --git a/tests/test_kokoro_backend.py b/tests/test_kokoro_backend.py deleted file mode 100644 index 5e75233..0000000 --- a/tests/test_kokoro_backend.py +++ /dev/null @@ -1,216 +0,0 @@ -"""Tests for KokoroBackend class.""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any, Iterator, List -from unittest.mock import MagicMock, patch - -import numpy as np -import pytest - -from abogen.tts_backend import TTSBackendMetadata - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - -@dataclass -class _FakeSegment: - graphemes: str - audio: Any # np.ndarray or torch-like tensor - - -class _FakePipeline: - """Minimal mock for kokoro.KPipeline.""" - - def __init__(self, *, lang_code: str, repo_id: str, device: str): - self.lang_code = lang_code - self.repo_id = repo_id - self.device = device - self._voices: dict[str, np.ndarray] = {} - - def __call__( - self, - text: str, - *, - voice: Any = "", - speed: float = 1.0, - split_pattern: str | None = None, - ) -> Iterator[_FakeSegment]: - yield _FakeSegment(graphemes=text, audio=np.zeros(100, dtype="float32")) - - def load_single_voice(self, name: str) -> np.ndarray: - if name not in self._voices: - self._voices[name] = np.ones((1, 256), dtype="float32") * 0.5 - return self._voices[name] - - -def _make_backend(**kwargs: Any): - """Create KokoroBackend with mocked KPipeline.""" - with patch("abogen.tts_backends.kokoro._load_kpipeline") as load: - load.return_value = _FakePipeline - from abogen.tts_backends.kokoro import KokoroBackend - - return KokoroBackend(**kwargs) - - -# --------------------------------------------------------------------------- -# Tests -# --------------------------------------------------------------------------- - -class TestKokoroBackendMetadata: - def test_metadata_returns_tts_backend_metadata(self): - backend = _make_backend(lang_code="a") - meta = backend.metadata - assert isinstance(meta, TTSBackendMetadata) - - def test_metadata_fields(self): - backend = _make_backend(lang_code="a") - meta = backend.metadata - assert meta.id == "kokoro" - assert meta.name == "Kokoro" - assert "Kokoro" in meta.description - - -class TestKokoroBackendInit: - def test_stores_lang_code(self): - backend = _make_backend(lang_code="b") - assert backend._lang_code == "b" - - def test_default_repo_id(self): - with patch("abogen.tts_backends.kokoro._load_kpipeline") as load: - load.return_value = _FakePipeline - from abogen.tts_backends.kokoro import KokoroBackend - - b = KokoroBackend(lang_code="a") - assert b._pipeline.repo_id == "hexgrad/Kokoro-82M" - - def test_custom_repo_id(self): - backend = _make_backend(lang_code="a", repo_id="custom/repo") - assert backend._pipeline.repo_id == "custom/repo" - - def test_default_device(self): - backend = _make_backend(lang_code="a") - assert backend._pipeline.device == "cpu" - - def test_custom_device(self): - backend = _make_backend(lang_code="a", device="cuda") - assert backend._pipeline.device == "cuda" - - -class TestKokoroBackendCall: - def test_call_delegates_to_pipeline(self): - backend = _make_backend(lang_code="a") - results = list(backend("hello", voice="af_heart", speed=1.2, split_pattern=r"\n")) - assert len(results) == 1 - assert results[0].graphemes == "hello" - - def test_call_returns_iterator(self): - backend = _make_backend(lang_code="a") - result = backend("test", voice="af_heart") - assert hasattr(result, "__iter__") - - def test_call_with_voice_tensor(self): - backend = _make_backend(lang_code="a") - voice_tensor = np.ones((1, 256), dtype="float32") - results = list(backend("test", voice=voice_tensor)) - assert len(results) == 1 - - def test_call_default_speed(self): - backend = _make_backend(lang_code="a") - # Should not raise with default speed - list(backend("text", voice="af_heart")) - - def test_call_default_split_pattern_is_none(self): - backend = _make_backend(lang_code="a") - # split_pattern defaults to None - list(backend("text", voice="af_heart")) - - -class TestLoadSingleVoice: - def test_load_single_voice_delegates(self): - backend = _make_backend(lang_code="a") - tensor = backend.load_single_voice("af_heart") - assert isinstance(tensor, np.ndarray) - assert tensor.shape == (1, 256) - - def test_load_single_voice_caches(self): - backend = _make_backend(lang_code="a") - t1 = backend.load_single_voice("af_heart") - t2 = backend.load_single_voice("af_heart") - assert t1 is t2 # same object - - -class TestSynthesize: - def test_synthesize_returns_bytes(self): - backend = _make_backend(lang_code="a") - result = backend.synthesize("hello", voice="af_heart") - assert isinstance(result, bytes) - - def test_synthesize_nonempty(self): - backend = _make_backend(lang_code="a") - result = backend.synthesize("hello", voice="af_heart") - assert len(result) > 0 - - def test_synthesize_with_speed(self): - backend = _make_backend(lang_code="a") - result = backend.synthesize("hello", voice="af_heart", speed=1.5) - assert isinstance(result, bytes) - - def test_synthesize_empty_text(self): - backend = _make_backend(lang_code="a") - # Empty text produces no segments - result = backend.synthesize("", voice="af_heart") - assert isinstance(result, bytes) - - -class TestProtocolMethods: - def test_get_available_voices(self): - backend = _make_backend(lang_code="a") - voices = backend.get_available_voices() - assert isinstance(voices, list) - assert len(voices) > 0 - assert all(isinstance(v, str) for v in voices) - - def test_get_supported_formats(self): - backend = _make_backend(lang_code="a") - formats = backend.get_supported_formats() - assert "pcm_float32" in formats - - def test_get_info(self): - backend = _make_backend(lang_code="a") - info = backend.get_info() - assert info["id"] == "kokoro" - assert info["name"] == "Kokoro" - assert info["lang_code"] == "a" - - -class TestRegistration: - def test_factory_creates_kokoro_backend(self): - from abogen.tts_backends.kokoro import create_kokoro_backend, KokoroBackend - - with patch("abogen.tts_backends.kokoro._load_kpipeline") as load: - load.return_value = _FakePipeline - backend = create_kokoro_backend(lang_code="a") - assert isinstance(backend, KokoroBackend) - - def test_registry_has_kokoro(self): - import abogen.tts_backends # noqa: F401 - from abogen.tts_backend_registry import _registry - - meta = _registry.get_metadata("kokoro") - assert meta.id == "kokoro" - assert meta.name == "Kokoro" - - def test_registry_factory_returns_kokoro_backend(self): - import abogen.tts_backends # noqa: F401 - from abogen.tts_backend_registry import _registry - from abogen.tts_backends.kokoro import KokoroBackend - - factory = _registry._factories["kokoro"] - with patch("abogen.tts_backends.kokoro._load_kpipeline") as load: - load.return_value = _FakePipeline - backend = factory(lang_code="a") - assert isinstance(backend, KokoroBackend) diff --git a/tests/test_tts_backend.py b/tests/test_tts_backend.py deleted file mode 100644 index bbd318d..0000000 --- a/tests/test_tts_backend.py +++ /dev/null @@ -1,314 +0,0 @@ -from dataclasses import dataclass - -from abogen.tts_backend import TTSBackendMetadata -from abogen.tts_backend_registry import TTSBackendRegistry - - -class TestTTSBackendMetadata: - def test_is_frozen_dataclass(self): - assert dataclass(TTSBackendMetadata) - - def test_fields_are_present(self): - meta = TTSBackendMetadata( - id="test", - name="Test Backend", - description="A test backend", - ) - assert meta.id == "test" - assert meta.name == "Test Backend" - assert meta.description == "A test backend" - - def test_voices_field_default_empty(self): - meta = TTSBackendMetadata( - id="test", - name="Test", - description="Test backend", - ) - assert meta.voices == () - - def test_voices_field_stored(self): - meta = TTSBackendMetadata( - id="test", - name="Test", - description="Test backend", - voices=("v1", "v2"), - ) - assert meta.voices == ("v1", "v2") - - def test_is_immutable(self): - import pytest - - meta = TTSBackendMetadata( - id="kokoro", - name="Kokoro", - description="Test", - ) - with pytest.raises(Exception): - meta.id = "changed" - - -class TestTTSBackendRegistry: - def test_register_and_list(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata(id="a", name="A", description="Backend A") - registry.register(metadata=meta, factory=lambda: None) - - backends = registry.list_backends() - assert len(backends) == 1 - assert backends[0].id == "a" - - def test_list_multiple(self): - registry = TTSBackendRegistry() - meta_a = TTSBackendMetadata(id="a", name="A", description="A") - meta_b = TTSBackendMetadata(id="b", name="B", description="B") - registry.register(metadata=meta_a, factory=lambda: None) - registry.register(metadata=meta_b, factory=lambda: None) - - backends = registry.list_backends() - ids = [b.id for b in backends] - assert "a" in ids - assert "b" in ids - - def test_get_metadata(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata(id="x", name="X", description="X backend") - registry.register(metadata=meta, factory=lambda: None) - - result = registry.get_metadata("x") - assert result.id == "x" - assert result.name == "X" - - def test_get_metadata_unknown_raises(self): - import pytest - - registry = TTSBackendRegistry() - with pytest.raises(KeyError, match="Unknown backend: nope"): - registry.get_metadata("nope") - - def test_create_backend(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata(id="test", name="Test", description="Test backend") - - def factory(**kwargs): - return {"created": True, "kwargs": kwargs} - - registry.register(metadata=meta, factory=factory) - result = registry.create_backend("test", foo="bar") - - assert result == {"created": True, "kwargs": {"foo": "bar"}} - - def test_create_backend_unknown_raises(self): - import pytest - - registry = TTSBackendRegistry() - with pytest.raises(KeyError, match="Unknown backend: missing"): - registry.create_backend("missing") - - def test_register_overwrites(self): - registry = TTSBackendRegistry() - meta1 = TTSBackendMetadata(id="x", name="V1", description="First") - meta2 = TTSBackendMetadata(id="x", name="V2", description="Second") - registry.register(metadata=meta1, factory=lambda: "v1") - registry.register(metadata=meta2, factory=lambda: "v2") - - result = registry.get_metadata("x") - assert result.name == "V2" - assert registry.create_backend("x") == "v2" - - -class TestBackendRegistration: - """Tests that existing backends are auto-registered.""" - - def test_import_triggers_registration(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - backends = _registry.list_backends() - ids = [b.id for b in backends] - assert "kokoro" in ids - assert "supertonic" in ids - - def test_kokoro_metadata(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - meta = _registry.get_metadata("kokoro") - assert meta.id == "kokoro" - assert meta.name == "Kokoro" - assert "Kokoro" in meta.description - - def test_supertonic_metadata(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - meta = _registry.get_metadata("supertonic") - assert meta.id == "supertonic" - assert meta.name == "SuperTonic" - assert "SuperTonic" in meta.description - - def test_kokoro_metadata_has_voices(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - meta = _registry.get_metadata("kokoro") - assert isinstance(meta.voices, tuple) - assert len(meta.voices) > 0 - assert all(isinstance(v, str) for v in meta.voices) - - def test_supertonic_metadata_has_voices(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - meta = _registry.get_metadata("supertonic") - assert isinstance(meta.voices, tuple) - assert len(meta.voices) == 10 - assert meta.voices == ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5") - - def test_kokoro_factory_callable(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - factory = _registry._factories["kokoro"] - assert callable(factory) - - def test_supertonic_factory_callable(self): - import abogen.tts_backends # noqa: F401 - - from abogen.tts_backend_registry import _registry - - factory = _registry._factories["supertonic"] - assert callable(factory) - - def test_kokoro_metadata_voices_match_registry(self): - """Ensure the metadata property on the instance shares voices with registry.""" - from abogen.tts_backends.kokoro import _KOKORO_METADATA - from abogen.tts_backend_registry import _registry - - registry_meta = _registry.get_metadata("kokoro") - assert _KOKORO_METADATA is registry_meta - assert _KOKORO_METADATA.voices == registry_meta.voices - - def test_supertonic_metadata_voices_match_registry(self): - """Ensure the metadata property on the instance shares voices with registry.""" - from abogen.tts_backends.supertonic import _SUPERTONIC_METADATA - from abogen.tts_backend_registry import _registry - - registry_meta = _registry.get_metadata("supertonic") - assert _SUPERTONIC_METADATA is registry_meta - assert _SUPERTONIC_METADATA.voices == registry_meta.voices - - -class TestResolveBackendForVoice: - """Tests for the resolve_backend_for_voice method.""" - - def test_empty_spec_returns_fallback(self): - registry = TTSBackendRegistry() - assert registry.resolve_backend_for_voice("", fallback="kokoro") == "kokoro" - assert registry.resolve_backend_for_voice("", fallback="supertonic") == "supertonic" - - def test_none_spec_returns_fallback(self): - registry = TTSBackendRegistry() - assert registry.resolve_backend_for_voice(None, fallback="kokoro") == "kokoro" - - def test_kokoro_formula_with_star_returns_kokoro(self): - registry = TTSBackendRegistry() - assert registry.resolve_backend_for_voice("af_nova*0.7") == "kokoro" - - def test_kokoro_formula_with_plus_returns_kokoro(self): - registry = TTSBackendRegistry() - assert registry.resolve_backend_for_voice("af_nova*0.7+am_liam*0.3") == "kokoro" - - def test_kokoro_voice_id_resolves_to_kokoro(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata( - id="kokoro", - name="Kokoro", - description="Kokoro TTS", - voices=("af_nova", "am_liam"), - ) - registry.register(metadata=meta, factory=lambda: None) - - assert registry.resolve_backend_for_voice("af_nova") == "kokoro" - assert registry.resolve_backend_for_voice("am_liam") == "kokoro" - - def test_supertonic_voice_id_resolves_to_supertonic(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata( - id="supertonic", - name="SuperTonic", - description="SuperTonic TTS", - voices=("M1", "M2", "F1", "F2"), - ) - registry.register(metadata=meta, factory=lambda: None) - - assert registry.resolve_backend_for_voice("M1") == "supertonic" - assert registry.resolve_backend_for_voice("F2") == "supertonic" - - def test_unknown_voice_returns_fallback(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata( - id="kokoro", - name="Kokoro", - description="Kokoro TTS", - voices=("af_nova",), - ) - registry.register(metadata=meta, factory=lambda: None) - - assert registry.resolve_backend_for_voice("unknown_voice") == "kokoro" - assert registry.resolve_backend_for_voice("unknown_voice", fallback="supertonic") == "supertonic" - - def test_case_insensitive_matching(self): - registry = TTSBackendRegistry() - meta = TTSBackendMetadata( - id="supertonic", - name="SuperTonic", - description="SuperTonic TTS", - voices=("M1", "F1"), - ) - registry.register(metadata=meta, factory=lambda: None) - - assert registry.resolve_backend_for_voice("m1") == "supertonic" - assert registry.resolve_backend_for_voice("f1") == "supertonic" - - def test_default_fallback_is_kokoro(self): - registry = TTSBackendRegistry() - assert registry.resolve_backend_for_voice("unknown") == "kokoro" - - def test_multiple_backends_resolution(self): - registry = TTSBackendRegistry() - kokoro_meta = TTSBackendMetadata( - id="kokoro", - name="Kokoro", - description="Kokoro TTS", - voices=("af_nova",), - ) - supertonic_meta = TTSBackendMetadata( - id="supertonic", - name="SuperTonic", - description="SuperTonic TTS", - voices=("M1",), - ) - registry.register(metadata=kokoro_meta, factory=lambda: None) - registry.register(metadata=supertonic_meta, factory=lambda: None) - - assert registry.resolve_backend_for_voice("af_nova") == "kokoro" - assert registry.resolve_backend_for_voice("M1") == "supertonic" - - def test_global_wrapper_resolve_backend_for_voice(self): - from abogen.tts_backend_registry import resolve_backend_for_voice - - # Test with empty spec - assert resolve_backend_for_voice("") == "kokoro" - - # Test with formula - assert resolve_backend_for_voice("af_nova*0.7") == "kokoro" - - # Test with a registered voice - assert resolve_backend_for_voice("af_nova") == "kokoro" - assert resolve_backend_for_voice("M1") == "supertonic" diff --git a/tests/test_tts_supertonic_unsupported_chars.py b/tests/test_tts_supertonic_unsupported_chars.py deleted file mode 100644 index 3e8cd00..0000000 --- a/tests/test_tts_supertonic_unsupported_chars.py +++ /dev/null @@ -1,108 +0,0 @@ -import numpy as np - -from abogen.tts_backends.supertonic import SupertonicBackend, SupertonicPipeline - - -class _DummyTTS: - def get_voice_style(self, voice_name: str): - return {"voice": voice_name} - - def synthesize( - self, - *, - text: str, - voice_style, - total_steps: int, - speed: float, - max_chunk_length: int, - silence_duration: float, - verbose: bool, - ): - if "•" in text: - raise ValueError("Found 1 unsupported character(s): ['•']") - # Return 50ms of audio at 24kHz. - sr = 24000 - audio = np.zeros(int(0.05 * sr), dtype="float32") - return audio, 0.05 - - -def _make_pipeline() -> SupertonicPipeline: - pipeline = SupertonicPipeline.__new__(SupertonicPipeline) - pipeline.sample_rate = 24000 - pipeline.total_steps = 5 - pipeline.max_chunk_length = 1000 - pipeline._tts = _DummyTTS() - return pipeline - - -def _make_backend() -> SupertonicBackend: - backend = SupertonicBackend.__new__(SupertonicBackend) - backend._pipeline = _make_pipeline() - return backend - - -def test_supertonic_pipeline_strips_unsupported_characters_and_retries(): - pipeline = _make_pipeline() - - segs = list(pipeline("Hello • world", voice="M1", speed=1.0)) - assert len(segs) == 1 - assert segs[0].graphemes == "Hello world" or segs[0].graphemes == "Hello world" - assert isinstance(segs[0].audio, np.ndarray) - assert segs[0].audio.dtype == np.float32 - assert segs[0].audio.size > 0 - - -def test_supertonic_pipeline_drops_chunk_if_only_unsupported_characters(): - pipeline = _make_pipeline() - - segs = list(pipeline("•", voice="M1", speed=1.0)) - assert segs == [] - - -# --- SupertonicBackend tests --- - - -def test_backend_metadata(): - backend = _make_backend() - meta = backend.metadata - assert meta.id == "supertonic" - assert meta.name == "SuperTonic" - assert "SuperTonic" in meta.description - - -def test_backend_get_available_voices(): - backend = _make_backend() - voices = backend.get_available_voices() - assert isinstance(voices, list) - assert "M1" in voices - assert "F1" in voices - - -def test_backend_get_supported_formats(): - backend = _make_backend() - formats = backend.get_supported_formats() - assert "wav" in formats - - -def test_backend_get_info(): - backend = _make_backend() - info = backend.get_info() - assert info["sample_rate"] == 24000 - assert info["total_steps"] == 5 - assert isinstance(info["voices"], list) - - -def test_backend_call_delegates_to_pipeline(): - backend = _make_backend() - segs = list(backend("Hello • world", voice="M1", speed=1.0)) - assert len(segs) == 1 - assert segs[0].audio.size > 0 - - -def test_backend_synthesize_returns_wav_bytes(): - backend = _make_backend() - wav_bytes = backend.synthesize("Hello world", voice="M1", speed=1.0) - assert isinstance(wav_bytes, bytes) - assert len(wav_bytes) > 0 - # WAV magic number - assert wav_bytes[:4] == b"RIFF" diff --git a/tests/test_voice_formula_resolution.py b/tests/test_voice_formula_resolution.py deleted file mode 100644 index 250aeb2..0000000 --- a/tests/test_voice_formula_resolution.py +++ /dev/null @@ -1,18 +0,0 @@ -from __future__ import annotations - -from abogen.webui.conversion_runner import _resolve_voice, _supertonic_voice_from_spec -from abogen.tts_backends.supertonic import DEFAULT_SUPERTONIC_VOICES - - -def test_resolve_voice_formula_without_pipeline_does_not_crash() -> None: - # This can happen when a previously-saved Kokoro mix formula is present - # but the active provider is SuperTonic (no Kokoro pipeline object). - formula = "af_heart*0.5+af_sky*0.5" - resolved = _resolve_voice(None, formula, use_gpu=False) - assert resolved == formula - - -def test_supertonic_voice_from_formula_falls_back_to_valid_voice() -> None: - # When a stale Kokoro mix formula is present, SuperTonic should not receive it. - chosen = _supertonic_voice_from_spec("af_heart*0.5+af_sky*0.5", "af_heart*1.0") - assert chosen in DEFAULT_SUPERTONIC_VOICES