Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-07-18 05:40:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cleanup): remove Legacy TTS Architecture

Browse Source 
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).
This commit is contained in:
Artem Akymenko
2026-07-12 16:20:20 +03:00
parent c094b94704
commit 780e9bd780
12 changed files with 435 additions and 1917 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/tts_backend.py
-89
View File
@@ -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
"""
...
abogen/tts_backend_registry.py
-146
View File
@@ -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)
abogen/tts_backends/__init__.py
-20
View File
@@ -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()
abogen/tts_backends/kokoro.py
-179
View File
@@ -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,
)
abogen/tts_backends/supertonic.py
-392
View File
@@ -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,
)
tests/test_kokoro_backend.py
-216
View File
@@ -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)
tests/test_tts_backend.py
-314
View File
@@ -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"
tests/test_tts_supertonic_unsupported_chars.py
-108
View File
@@ -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"
tests/test_voice_formula_resolution.py
-18
View File
@@ -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