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: remove compatibility layer, use Plugin Architecture directly

Browse Source 
- Delete abogen/tts_plugin/compat.py (CompatBackend, create_backend, get_metadata, etc.)
- Add abogen/tts_plugin/utils.py with direct Plugin Manager functions:
  get_voices, get_default_voice, is_plugin_registered, resolve_voice_to_plugin, create_pipeline
- Update all 16 consumer files to import from utils instead of compat
- Update __init__.py to re-export utils instead of compat
- Update 5 test files and add TestNoCompatLayer regression tests
- All 493 tests pass
This commit is contained in:
Artem Akymenko
2026-07-12 16:20:06 +03:00
parent 985e16f1f8
commit a76d338931
24 changed files with 297 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/predownload_gui.py
+4 -4
View File
@@ -22,7 +22,7 @@ from PyQt6.QtWidgets import (
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -115,7 +115,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = get_metadata("kokoro").voices
voice_list = get_voices("kokoro")
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -463,14 +463,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in get_metadata("kokoro").voices:
for voice in get_voices("kokoro"):
if not try_to_load_from_cache(
repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
):
missing.append(voice)
except Exception:
# If HF missing, report all as missing
return False, list(get_metadata("kokoro").voices)
return False, list(get_voices("kokoro"))
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
abogen/pyqt/gui.py
+2 -2
View File
@@ -86,7 +86,7 @@ from abogen.constants import (
COLORS,
SUBTITLE_FORMATS,
)
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
import threading
from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog
from abogen.voice_profiles import load_profiles
@@ -1873,7 +1873,7 @@ class abogen(QWidget):
for pname in load_profiles().keys():
self.voice_combo.addItem(profile_icon, pname, f"profile:{pname}")
# re-add voices
for v in get_metadata("kokoro").voices:
for v in get_voices("kokoro"):
icon = QIcon()
flag_path = get_resource_path("abogen.assets.flags", f"{v[0]}.png")
if flag_path and os.path.exists(flag_path):
abogen/pyqt/predownload_gui.py
+4 -4
View File
@@ -22,7 +22,7 @@ from PyQt6.QtWidgets import (
from PyQt6.QtCore import QThread, pyqtSignal
from abogen.constants import COLORS
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
from abogen.spacy_utils import SPACY_MODELS
import abogen.hf_tracker
@@ -115,7 +115,7 @@ class PreDownloadWorker(QThread):
self._voices_success = False
return
voice_list = get_metadata("kokoro").voices
voice_list = get_voices("kokoro")
for idx, voice in enumerate(voice_list, start=1):
if self._cancelled:
self._voices_success = False
@@ -463,14 +463,14 @@ class PreDownloadDialog(QDialog):
try:
from huggingface_hub import try_to_load_from_cache
for voice in get_metadata("kokoro").voices:
for voice in get_voices("kokoro"):
if not try_to_load_from_cache(
repo_id="hexgrad/Kokoro-82M", filename=f"voices/{voice}.pt"
):
missing.append(voice)
except Exception:
# If HF missing, report all as missing
return False, list(get_metadata("kokoro").voices)
return False, list(get_voices("kokoro"))
return (len(missing) == 0), missing
def _check_kokoro_model(self) -> bool:
abogen/pyqt/voice_formula_gui.py
+3 -3
View File
@@ -32,7 +32,7 @@ from abogen.constants import (
LANGUAGE_DESCRIPTIONS,
COLORS,
)
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
import re
import platform
from abogen.utils import get_resource_path
@@ -179,7 +179,7 @@ class VoiceMixer(QWidget):
layout.addWidget(QLabel(name), alignment=Qt.AlignmentFlag.AlignCenter)
# Voice name label with gender icon
is_female = self.voice_name in get_metadata("kokoro").voices and self.voice_name[1] == "f"
is_female = self.voice_name in get_voices("kokoro") and self.voice_name[1] == "f"
# Icons layout (flag and gender)
icons_layout = QHBoxLayout()
@@ -772,7 +772,7 @@ class VoiceFormulaDialog(QDialog):
def add_voices(self, initial_state):
first_enabled_voice = None
for voice in get_metadata("kokoro").voices:
for voice in get_voices("kokoro"):
language_code = voice[0] # First character is the language code
matching_voice = next(
(item for item in initial_state if item[0] == voice), None
abogen/subtitle_utils.py
+5 -5
View File
@@ -477,10 +477,10 @@ def validate_voice_name(voice_name):
- is_valid: True if all voices in the name/formula are valid
- invalid_voice_name: The first invalid voice found, or None if all valid
"""
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
# Create case-insensitive lookup set (done once per call)
voice_lookup_lower = {v.lower() for v in get_metadata("kokoro").voices}
voice_lookup_lower = {v.lower() for v in get_voices("kokoro")}
voice_name = voice_name.strip()
# Check if it's a formula (contains *)
@@ -518,7 +518,7 @@ def split_text_by_voice_markers(text, default_voice):
- valid_count: Number of valid voice markers processed
- invalid_count: Number of invalid voice markers skipped
"""
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text))
@@ -560,7 +560,7 @@ def split_text_by_voice_markers(text, default_voice):
# Find the canonical (lowercase) voice name
voice_part_lower = voice_part.strip().lower()
canonical_voice = next(
(v for v in get_metadata("kokoro").voices if v.lower() == voice_part_lower),
(v for v in get_voices("kokoro") if v.lower() == voice_part_lower),
voice_part.strip()
)
normalized_parts.append(f"{canonical_voice}*{weight.strip()}")
@@ -569,7 +569,7 @@ def split_text_by_voice_markers(text, default_voice):
# Find the canonical (lowercase) voice name
voice_name_lower = voice_name.lower()
current_voice = next(
(v for v in get_metadata("kokoro").voices if v.lower() == voice_name_lower),
(v for v in get_voices("kokoro") if v.lower() == voice_name_lower),
voice_name
)
valid_markers += 1
abogen/tts_plugin/__init__.py
+17 -17
View File
@@ -13,7 +13,7 @@ Public modules:
- plugin: Plugin contract (create_engine function signature)
- loader: Plugin discovery and loading
- plugin_manager: Plugin management and engine creation
- compat: Backward compatibility adapter for old create_backend() API
- utils: Direct utility functions (get_voices, create_pipeline, etc.)
Usage:
from abogen.tts_plugin import (
@@ -59,12 +59,12 @@ Usage:
# Plugin Manager
get_plugin_manager,
reset_plugin_manager,
# Compatibility
create_backend,
get_metadata,
is_registered_backend,
resolve_backend_for_voice,
# Utils
get_voices,
get_default_voice,
is_plugin_registered,
resolve_voice_to_plugin,
create_pipeline,
)
"""
@@ -108,14 +108,14 @@ from abogen.tts_plugin.types import (
VoiceSelection,
)
# Plugin Manager and Compatibility
# Plugin Manager and Utils
from abogen.tts_plugin.plugin_manager import get_plugin_manager, reset_plugin_manager
from abogen.tts_plugin.compat import (
create_backend,
from abogen.tts_plugin.utils import (
create_pipeline,
get_default_voice,
get_metadata,
is_registered_backend,
resolve_backend_for_voice,
get_voices,
is_plugin_registered,
resolve_voice_to_plugin,
)
__all__ = [
@@ -161,10 +161,10 @@ __all__ = [
# Plugin Manager
"get_plugin_manager",
"reset_plugin_manager",
# Compatibility
"create_backend",
"get_metadata",
"is_registered_backend",
"resolve_backend_for_voice",
# Utils
"get_voices",
"get_default_voice",
"is_plugin_registered",
"resolve_voice_to_plugin",
"create_pipeline",
]
abogen/tts_plugin/compat.py
-275
View File
@@ -1,275 +0,0 @@
"""TTS Backend Compatibility Adapter
Provides a drop-in replacement for the old `create_backend()` function
that uses the new Plugin Architecture under the hood.
Usage:
# Old way:
from abogen.tts_backend_registry import create_backend
pipeline = create_backend("kokoro", lang_code="a", device="cpu")
# New way (same interface):
from abogen.tts_plugin.compat import create_backend
pipeline = create_backend("kokoro", lang_code="a", device="cpu")
The adapter wraps the new Engine/EngineSession into a callable that
matches the old TTSBackend protocol.
"""
from typing import Any, Callable, Iterable, Iterator, List, Mapping, Optional, Tuple
import numpy as np
from abogen.tts_plugin.engine import Engine, EngineSession
from abogen.tts_plugin.plugin_manager import get_plugin_manager
class CompatBackend:
"""Compatibility wrapper that makes a new Engine look like the old TTSBackend.
This adapter wraps the new Engine/EngineSession into a callable that
matches the old Kokoro pipeline interface:
pipeline(text, voice=..., speed=..., split_pattern=...) -> Iterator[Segment]
"""
def __init__(self, engine: Engine, **engine_kwargs: Any) -> None:
self._engine = engine
self._engine_kwargs = engine_kwargs
self._session: Optional[EngineSession] = None
def _ensure_session(self) -> EngineSession:
"""Ensure we have an active session."""
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 = r"\n+",
**kwargs: Any,
) -> Iterator[Any]:
"""Call the backend like the old Kokoro pipeline.
Returns an iterator of segment-like objects with .graphemes and .audio attributes.
"""
session = self._ensure_session()
# Build synthesis request using the new API types
from abogen.tts_plugin.types import (
AudioFormat,
ParameterValues,
SynthesisRequest,
VoiceSelection,
)
# Convert voice string to VoiceSelection
voice_selection = VoiceSelection(source="builtin", key=voice)
# Convert speed and split_pattern to parameters
parameters = ParameterValues(values={"speed": speed, "split_pattern": split_pattern})
# Create request with default audio format
request = SynthesisRequest(
text=text,
voice=voice_selection,
parameters=parameters,
format=AudioFormat(mime="audio/wav", extension="wav"),
)
# Synthesize
result = session.synthesize(request)
# Convert result to old-style segment iterator
from dataclasses import dataclass
@dataclass
class Segment:
graphemes: str
audio: np.ndarray
# Convert bytes back to numpy array
audio_array = np.frombuffer(result.data, dtype=np.float32)
# The new API returns a single audio result, but the old API returns
# an iterator of segments. We need to split the text and audio accordingly.
# For now, return a single segment with the full text and audio.
yield Segment(
graphemes=text,
audio=audio_array,
)
def dispose(self) -> None:
"""Dispose the session."""
if self._session is not None:
try:
self._session.dispose()
except Exception:
pass
self._session = None
def __del__(self) -> None:
"""Cleanup on garbage collection."""
self.dispose()
def create_backend(backend_id: str, **kwargs: Any) -> Any:
"""Create a TTS backend using the new Plugin Architecture.
This is a drop-in replacement for the old `create_backend()` function
from `abogen.tts_backend_registry`.
Args:
backend_id: The backend/plugin ID (e.g., "kokoro")
**kwargs: Arguments passed to the engine constructor
Returns:
A callable backend that matches the old TTSBackend protocol
Raises:
KeyError: If plugin_id is not found
Exception: If engine creation fails
"""
manager = get_plugin_manager()
engine = manager.create_engine(backend_id, **kwargs)
return CompatBackend(engine, **kwargs)
def get_metadata(backend_id: str) -> Any:
"""Get metadata for a specific backend using the new Plugin Architecture.
This is a drop-in replacement for the old `get_metadata()` function
from `abogen.tts_backend_registry`.
Args:
backend_id: The backend/plugin ID (e.g., "kokoro")
Returns:
TTSBackendMetadata-like object with voices attribute
Raises:
KeyError: If plugin_id is not found
"""
from abogen.tts_backend import TTSBackendMetadata
manager = get_plugin_manager()
plugin_info = manager.get_plugin(backend_id)
if plugin_info is None:
raise KeyError(f"Unknown backend: {backend_id}")
manifest = plugin_info["manifest"]
# Import voice lists from plugin engines
# This avoids creating an engine just to get the voice list
voices: tuple[str, ...] = ()
if backend_id == "kokoro":
try:
from plugins.kokoro.engine import _KOKORO_VOICES
voices = _KOKORO_VOICES
except ImportError:
pass
elif backend_id == "supertonic":
try:
from plugins.supertonic.engine import _SUPERTONIC_VOICES
voices = _SUPERTONIC_VOICES
except ImportError:
pass
return TTSBackendMetadata(
id=manifest.id,
name=manifest.name,
description=manifest.description,
voices=voices,
)
def is_registered_backend(backend_id: str) -> bool:
"""Check if a backend is registered using the new Plugin Architecture.
This is a drop-in replacement for the old `is_registered_backend()` function
from `abogen.tts_backend_registry`.
Args:
backend_id: The backend/plugin ID (e.g., "kokoro")
Returns:
True if the plugin is loaded, False otherwise
"""
manager = get_plugin_manager()
return manager.has_plugin(backend_id)
def resolve_backend_for_voice(
spec: str,
fallback: str = "kokoro",
) -> str:
"""Determine which backend owns the given voice specification.
This is a drop-in replacement for the old `resolve_backend_for_voice()` function
from `abogen.tts_backend_registry`.
Resolution rules:
1. Empty spec -> fallback
2. Kokoro formula (contains '*' or '+') -> "kokoro"
3. Exact voice ID match against registered plugins -> plugin id
4. Unknown voice -> fallback
Args:
spec: Voice specification (e.g., "af_nova", "M1", "af_nova*0.7+am_liam*0.3")
fallback: Fallback backend ID if no match is found
Returns:
The backend/plugin ID that owns the voice
"""
raw = str(spec or "").strip()
if not raw:
return fallback
# Kokoro formula detection
if "*" in raw or "+" in raw:
return "kokoro"
manager = get_plugin_manager()
upper = raw.upper()
# Check each plugin's voices
for plugin_info in manager.list_plugins():
manifest = plugin_info
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)
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:
engine.dispose()
return manifest.id
engine.dispose()
except Exception:
continue
return fallback
def get_default_voice(backend_id: str, fallback: str = "") -> str:
"""Return the first voice of a backend, or fallback if none.
This is a drop-in replacement for the old `get_default_voice()` function
from `abogen.tts_backend_registry`.
Args:
backend_id: The backend/plugin ID (e.g., "kokoro")
fallback: Fallback voice if no voices are available
Returns:
The first voice ID, or fallback if none
"""
try:
metadata = get_metadata(backend_id)
voices = metadata.voices
return voices[0] if voices else fallback
except KeyError:
return fallback
abogen/tts_plugin/utils.py
+157
View File
@@ -0,0 +1,157 @@
"""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, Optional
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*."""
if plugin_id == "kokoro":
from plugins.kokoro.engine import _KOKORO_VOICES
return _KOKORO_VOICES
if plugin_id == "supertonic":
from plugins.supertonic.engine import _SUPERTONIC_VOICES
return _SUPERTONIC_VOICES
return ()
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, **kwargs: Any) -> Pipeline:
"""Create a callable TTS pipeline via the Plugin Architecture.
Returns a :class:`Pipeline` whose ``__call__`` interface matches the
legacy ``TTSBackend`` callable protocol.
"""
manager = get_plugin_manager()
engine = manager.create_engine(plugin_id, **kwargs)
return Pipeline(engine, **kwargs)
abogen/utils.py
+2 -2
View File
@@ -538,9 +538,9 @@ class LoadPipelineThread(Thread):
def run(self):
try:
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
backend = create_backend(
backend = create_pipeline(
"kokoro", lang_code=self.lang_code, device=self.device
)
self.callback(backend, None)
abogen/voice_cache.py
+2 -2
View File
@@ -17,7 +17,7 @@ if LocalEntryNotFoundError is None: # pragma: no cover - fallback for tests
pass
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
_CACHE_LOCK = threading.Lock()
_CACHED_VOICES: Set[str] = set()
@@ -26,7 +26,7 @@ _BOOTSTRAPPED = False
def _normalize_targets(voices: Optional[Iterable[str]]) -> Set[str]:
kokoro_voices = get_metadata("kokoro").voices
kokoro_voices = get_voices("kokoro")
if not voices:
return set(kokoro_voices)
normalized: Set[str] = set()
abogen/voice_formulas.py
+2 -2
View File
@@ -1,7 +1,7 @@
import re
from typing import List, Tuple
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
# Calls parsing and loads the voice to gpu or cpu
@@ -22,7 +22,7 @@ def parse_formula_terms(formula: str) -> List[Tuple[str, float]]:
raise ValueError("Empty voice formula")
terms: List[Tuple[str, float]] = []
kokoro_voices = get_metadata("kokoro").voices
kokoro_voices = get_voices("kokoro")
for segment in formula.split("+"):
part = segment.strip()
if not part:
abogen/voice_profiles.py
+4 -4
View File
@@ -2,7 +2,7 @@ import json
import os
from typing import Any, Dict, Iterable, List, Tuple
from abogen.tts_plugin.compat import get_metadata, is_registered_backend
from abogen.tts_plugin.utils import get_voices, is_plugin_registered
from abogen.utils import get_user_config_path
@@ -69,7 +69,7 @@ def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]:
def _normalize_supertonic_voice(value: Any) -> str:
raw = str(value or "").strip().upper()
supertonic_voices = get_metadata("supertonic").voices
supertonic_voices = get_voices("supertonic")
return raw if raw in supertonic_voices else "M1"
@@ -101,7 +101,7 @@ def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
return {}
provider = str(entry.get("provider") or "kokoro").strip().lower()
if not is_registered_backend(provider):
if not is_plugin_registered(provider):
provider = "kokoro"
language = str(entry.get("language") or "a").strip().lower() or "a"
@@ -135,7 +135,7 @@ def normalize_profile_entry(entry: Any) -> Dict[str, Any]:
def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]:
normalized: List[Tuple[str, float]] = []
kokoro_voices = get_metadata("kokoro").voices
kokoro_voices = get_voices("kokoro")
for item in entries or []:
if isinstance(item, dict):
voice = item.get("id") or item.get("voice")
abogen/webui/conversion_runner.py
+10 -10
View File
@@ -20,7 +20,7 @@ import numpy as np
import soundfile as sf
import static_ffmpeg
from abogen.tts_plugin.compat import get_metadata, is_registered_backend, resolve_backend_for_voice
from abogen.tts_plugin.utils import get_voices, is_plugin_registered, resolve_voice_to_plugin
from abogen.epub3.exporter import build_epub3_package
from abogen.kokoro_text_normalization import ApostropheConfig, normalize_for_pipeline, HAS_NUM2WORDS
from abogen.normalization_settings import (
@@ -40,7 +40,7 @@ from abogen.utils import (
get_user_output_path,
load_config,
)
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
from abogen.voice_cache import ensure_voice_assets
from abogen.voice_formulas import extract_voice_ids, get_new_voice
from abogen.voice_profiles import load_profiles, normalize_profile_entry
@@ -119,7 +119,7 @@ def _formula_from_kokoro_entry(entry: Mapping[str, Any]) -> str:
def _infer_provider_from_spec(value: Any, fallback: str = "kokoro") -> str:
return resolve_backend_for_voice(str(value or ""), fallback=fallback)
return resolve_voice_to_plugin(str(value or ""), fallback=fallback)
class _JobCancelled(Exception):
@@ -568,7 +568,7 @@ def _spec_to_voice_ids(spec: Any) -> Set[str]:
return set(extract_voice_ids(text))
except ValueError:
return set()
if text in get_metadata("kokoro").voices:
if text in get_voices("kokoro"):
return {text}
return set()
@@ -632,7 +632,7 @@ def _collect_required_voice_ids(job: Job) -> Set[str]:
for key in ("resolved_voice", "voice_formula", "voice"):
voices.update(_spec_to_voice_ids(payload.get(key)))
voices.update(get_metadata("kokoro").voices)
voices.update(get_voices("kokoro"))
return voices
@@ -1566,7 +1566,7 @@ def run_conversion_job(job: Job) -> None:
def get_pipeline(provider: str) -> Any:
nonlocal kokoro_cache_ready
provider_norm = str(provider or "kokoro").strip().lower() or "kokoro"
if not is_registered_backend(provider_norm):
if not is_plugin_registered(provider_norm):
provider_norm = "kokoro"
existing = pipelines.get(provider_norm)
@@ -1574,7 +1574,7 @@ def run_conversion_job(job: Job) -> None:
return existing
if provider_norm == "supertonic":
pipelines[provider_norm] = create_backend(
pipelines[provider_norm] = create_pipeline(
"supertonic",
sample_rate=SAMPLE_RATE,
auto_download=True,
@@ -1589,7 +1589,7 @@ def run_conversion_job(job: Job) -> None:
if not disable_gpu:
device = _select_device()
# Create KPipeline instance directly (conforms to TTSBackend protocol)
pipelines[provider_norm] = create_backend(
pipelines[provider_norm] = create_pipeline(
"kokoro",
lang_code=job.language,
device=device
@@ -2441,7 +2441,7 @@ def _load_pipeline(job: Job):
disable_gpu = not job.use_gpu or not cfg.get("use_gpu", True)
provider = str(getattr(job, "tts_provider", "kokoro") or "kokoro").strip().lower()
if provider == "supertonic":
return create_backend(
return create_pipeline(
"supertonic",
sample_rate=SAMPLE_RATE,
auto_download=True,
@@ -2451,7 +2451,7 @@ def _load_pipeline(job: Job):
device = "cpu"
if not disable_gpu:
device = _select_device()
return create_backend("kokoro", lang_code=job.language, device=device)
return create_pipeline("kokoro", lang_code=job.language, device=device)
def _select_device() -> str:
abogen/webui/debug_tts_runner.py
+2 -2
View File
@@ -15,7 +15,7 @@ from abogen.normalization_settings import build_apostrophe_config
from abogen.text_extractor import extract_from_path
from abogen.voice_cache import ensure_voice_assets
from abogen.webui.conversion_runner import SAMPLE_RATE, SPLIT_PATTERN, _select_device, _to_float32, _resolve_voice, _spec_to_voice_ids
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
_MARKER_RE = re.compile(re.escape(MARKER_PREFIX) + r"(?P<code>[A-Z0-9_]+)" + re.escape(MARKER_SUFFIX))
@@ -45,7 +45,7 @@ def _load_pipeline(language: str, use_gpu: bool) -> Any:
device = "cpu"
if use_gpu:
device = _select_device()
return create_backend("kokoro", lang_code=language, device=device)
return create_pipeline("kokoro", lang_code=language, device=device)
def _extract_cases_from_text(text: str) -> List[Tuple[str, str]]:
abogen/webui/routes/api.py
+3 -3
View File
@@ -34,7 +34,7 @@ from abogen.normalization_settings import (
)
from abogen.llm_client import list_models, LLMClientError
from abogen.kokoro_text_normalization import normalize_for_pipeline
from abogen.tts_plugin.compat import is_registered_backend
from abogen.tts_plugin.utils import is_plugin_registered
from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfConfig
from abogen.integrations.calibre_opds import (
CalibreOPDSClient,
@@ -64,7 +64,7 @@ def api_save_voice_profile() -> ResponseReturnValue:
if profile is None:
# Speaker Studio payload format
provider = str(payload.get("provider") or "kokoro").strip().lower()
if not is_registered_backend(provider):
if not is_plugin_registered(provider):
provider = "kokoro"
if provider == "supertonic":
profile = {
@@ -231,7 +231,7 @@ def api_speaker_preview() -> ResponseReturnValue:
use_gpu = settings.get("use_gpu", False)
base_spec, speaker_name = split_profile_spec(voice)
resolved_provider = tts_provider if is_registered_backend(tts_provider) else ""
resolved_provider = tts_provider if is_plugin_registered(tts_provider) else ""
if speaker_name:
entry = normalize_profile_entry(load_profiles().get(speaker_name))
abogen/webui/routes/utils/form.py
+3 -3
View File
@@ -7,7 +7,7 @@ from flask.typing import ResponseReturnValue
from abogen.webui.service import PendingJob, JobStatus
from abogen.webui.routes.utils.service import get_service
from abogen.tts_plugin.compat import is_registered_backend
from abogen.tts_plugin.utils import is_plugin_registered
from abogen.webui.routes.utils.settings import (
load_settings,
coerce_bool,
@@ -33,7 +33,7 @@ from abogen.webui.routes.utils.common import split_profile_spec
from abogen.utils import calculate_text_length
from abogen.voice_profiles import serialize_profiles, normalize_profile_entry
from abogen.chunking import ChunkLevel, build_chunks_for_chapters
from abogen.tts_plugin.compat import get_default_voice
from abogen.tts_plugin.utils import get_default_voice
from abogen.speaker_configs import get_config
from abogen.kokoro_text_normalization import normalize_roman_numeral_titles
from dataclasses import dataclass
@@ -580,7 +580,7 @@ def apply_book_step_form(
# spec (e.g. "speaker:Name" for saved speakers, or a Kokoro mix formula).
# This enables mixed-provider conversions (e.g. narrator=SuperTonic, characters=Kokoro).
provider_value = str(form.get("tts_provider") or "").strip().lower()
if is_registered_backend(provider_value):
if is_plugin_registered(provider_value):
pending.tts_provider = provider_value
# Determine the base speaker selection (saved speaker ref or raw voice).
abogen/webui/routes/utils/preview.py
+4 -4
View File
@@ -78,9 +78,9 @@ def get_preview_pipeline(language: str, device: str) -> Any:
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
pipeline = create_backend("kokoro", lang_code=language, device=device)
pipeline = create_pipeline("kokoro", lang_code=language, device=device)
_preview_pipelines[key] = pipeline
return pipeline
@@ -136,9 +136,9 @@ def generate_preview_audio(
normalized_text = source_text
if provider == "supertonic":
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
pipeline = create_backend("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps)
pipeline = create_pipeline("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps)
segments = pipeline(
normalized_text,
voice=voice_spec,
abogen/webui/routes/utils/settings.py
+1 -1
View File
@@ -7,7 +7,7 @@ from abogen.constants import (
SUBTITLE_FORMATS,
SUPPORTED_SOUND_FORMATS,
)
from abogen.tts_plugin.compat import get_default_voice
from abogen.tts_plugin.utils import get_default_voice
from abogen.normalization_settings import (
DEFAULT_LLM_PROMPT,
environment_llm_defaults,
abogen/webui/routes/utils/voice.py
+5 -5
View File
@@ -18,9 +18,9 @@ from abogen.constants import (
SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
SAMPLE_VOICE_TEXTS,
)
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
from abogen.speaker_configs import list_configs
from abogen.tts_plugin.compat import create_backend
from abogen.tts_plugin.utils import create_pipeline
from abogen.webui.conversion_runner import _select_device, _to_float32, SAMPLE_RATE, SPLIT_PATTERN
_preview_pipeline_lock = threading.RLock()
@@ -285,7 +285,7 @@ def filter_voice_catalog(
def build_voice_catalog() -> List[Dict[str, str]]:
catalog: List[Dict[str, str]] = []
gender_map = {"f": "Female", "m": "Male"}
for voice_id in get_metadata("kokoro").voices:
for voice_id in get_voices("kokoro"):
prefix, _, rest = voice_id.partition("_")
language_code = prefix[0] if prefix else "a"
gender_code = prefix[1] if len(prefix) > 1 else ""
@@ -590,7 +590,7 @@ def template_options() -> Dict[str, Any]:
voice_catalog = build_voice_catalog()
return {
"languages": LANGUAGE_DESCRIPTIONS,
"voices": get_metadata("kokoro").voices,
"voices": get_voices("kokoro"),
"subtitle_formats": SUBTITLE_FORMATS,
"supported_langs_for_subs": SUPPORTED_LANGUAGES_FOR_SUBTITLE_GENERATION,
"output_formats": SUPPORTED_SOUND_FORMATS,
@@ -741,7 +741,7 @@ def get_preview_pipeline(language: str, device: str):
pipeline = _preview_pipelines.get(key)
if pipeline is not None:
return pipeline
pipeline = create_backend("kokoro", lang_code=language, device=device)
pipeline = create_pipeline("kokoro", lang_code=language, device=device)
_preview_pipelines[key] = pipeline
return pipeline
tests/contracts/test_integration.py
+32 -12
View File
@@ -17,7 +17,7 @@ import numpy as np
from abogen.tts_plugin.engine import Engine, EngineSession
from abogen.tts_plugin.errors import EngineError
from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager
from abogen.tts_plugin.compat import CompatBackend, create_backend
from abogen.tts_plugin.utils import Pipeline, create_pipeline
from abogen.tts_plugin.types import (
AudioFormat,
Duration,
@@ -148,8 +148,8 @@ class TestConsumerFlow:
assert result.format.mime == "audio/wav"
assert result.duration.seconds > 0
def test_consumer_flow_via_compat_adapter(self):
"""Verify flow through compatibility adapter matches direct flow."""
def test_consumer_flow_via_pipeline(self):
"""Verify flow through Pipeline utility matches direct flow."""
manager = PluginManager()
# Register mock plugin
@@ -157,9 +157,9 @@ class TestConsumerFlow:
manager._plugins["mock_tts"] = mock_plugin
manager._loaded = True
# Use compat adapter
with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager):
backend = create_backend("mock_tts")
# Use Pipeline utility
with patch("abogen.tts_plugin.utils.get_plugin_manager", return_value=manager):
backend = create_pipeline("mock_tts")
# Call like old TTSBackend
segments = list(backend("Hello world", voice="default", speed=1.0))
@@ -296,8 +296,8 @@ class TestRegression:
manager._loaded = True
# New path: Plugin Manager → Engine → Session → Synthesis
with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager):
new_backend = create_backend("mock_tts")
with patch("abogen.tts_plugin.utils.get_plugin_manager", return_value=manager):
new_backend = create_pipeline("mock_tts")
new_segments = list(new_backend("Hello world", voice="default", speed=1.0))
# Old path: Direct MockEngine (simulating old registry)
@@ -320,15 +320,15 @@ class TestRegression:
# Both should have same format
assert new_segments[0].audio.dtype == np.float32
def test_compat_adapter_matches_old_interface(self):
"""Compat adapter should match old TTSBackend interface."""
def test_pipeline_matches_old_interface(self):
"""Pipeline utility should match old TTSBackend interface."""
manager = PluginManager()
mock_plugin = create_mock_plugin()
manager._plugins["mock_tts"] = mock_plugin
manager._loaded = True
with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager):
backend = create_backend("mock_tts", lang_code="a", device="cpu")
with patch("abogen.tts_plugin.utils.get_plugin_manager", return_value=manager):
backend = create_pipeline("mock_tts", lang_code="a", device="cpu")
# Old interface: pipeline(text, voice=..., speed=..., split_pattern=...)
segments = list(backend(
@@ -398,3 +398,23 @@ class TestPluginManagerIntegration:
# Cache should be empty
assert len(manager._engines) == 0
class TestNoCompatLayer:
"""Regression: confirm the compatibility layer has been removed."""
def test_compat_module_does_not_exist(self):
"""abogen.tts_plugin.compat must not be importable."""
import importlib
with pytest.raises((ImportError, ModuleNotFoundError)):
importlib.import_module("abogen.tts_plugin.compat")
def test_consumers_use_plugin_architecture_directly(self):
"""Key consumers import from abogen.tts_plugin.utils, not compat."""
import inspect, abogen.voice_profiles, abogen.voice_formulas, abogen.voice_cache
for mod in (abogen.voice_profiles, abogen.voice_formulas, abogen.voice_cache):
source = inspect.getsource(mod)
assert "tts_plugin.compat" not in source, (
f"{mod.__name__} still references tts_plugin.compat"
)
tests/contracts/test_plugin_manager_contract.py
+26 -26
View File
@@ -1,10 +1,10 @@
"""Integration tests for Plugin Manager and compatibility adapter."""
"""Integration tests for Plugin Manager and direct utility functions."""
import pytest
from unittest.mock import MagicMock, patch
from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager
from abogen.tts_plugin.compat import CompatBackend, create_backend
from abogen.tts_plugin.utils import Pipeline, create_pipeline
from abogen.tts_plugin.engine import Engine, EngineSession
from abogen.tts_plugin.types import SynthesisRequest, SynthesizedAudio, AudioFormat
@@ -110,27 +110,27 @@ class TestPluginManager:
reset_plugin_manager()
class TestCompatBackend:
"""Test CompatBackend functionality."""
class TestPipeline:
"""Test Pipeline functionality."""
def test_compat_backend_creation(self):
"""CompatBackend can be created."""
def test_pipeline_creation(self):
"""Pipeline can be created."""
engine = FakeEngine()
backend = CompatBackend(engine)
backend = Pipeline(engine)
assert backend is not None
def test_compat_backend_callable(self):
"""CompatBackend is callable like old TTSBackend."""
def test_pipeline_callable(self):
"""Pipeline is callable like old TTSBackend."""
engine = FakeEngine()
backend = CompatBackend(engine)
backend = Pipeline(engine)
# Should be callable
assert callable(backend)
def test_compat_backend_synthesize(self):
"""CompatBackend can synthesize text."""
def test_pipeline_synthesize(self):
"""Pipeline can synthesize text."""
engine = FakeEngine()
backend = CompatBackend(engine)
backend = Pipeline(engine)
# Call the backend
segments = list(backend("Hello world", voice="default", speed=1.0))
@@ -144,10 +144,10 @@ class TestCompatBackend:
assert hasattr(segment, "audio")
assert segment.graphemes == "Hello world"
def test_compat_backend_dispose(self):
"""CompatBackend can be disposed."""
def test_pipeline_dispose(self):
"""Pipeline can be disposed."""
engine = FakeEngine()
backend = CompatBackend(engine)
backend = Pipeline(engine)
# Create a session by calling
list(backend("test"))
@@ -159,33 +159,33 @@ class TestCompatBackend:
backend.dispose()
class TestCreateBackendCompat:
"""Test create_backend compatibility function."""
class TestCreatePipelineCompat:
"""Test create_pipeline utility function."""
def test_create_backend_returns_callable(self):
"""create_backend returns a callable backend."""
def test_create_pipeline_returns_callable(self):
"""create_pipeline returns a callable backend."""
# Mock the plugin manager
with patch("abogen.tts_plugin.compat.get_plugin_manager") as mock_get_manager:
with patch("abogen.tts_plugin.utils.get_plugin_manager") as mock_get_manager:
mock_manager = MagicMock()
mock_get_manager.return_value = mock_manager
mock_engine = FakeEngine()
mock_manager.create_engine.return_value = mock_engine
backend = create_backend("kokoro", lang_code="a", device="cpu")
backend = create_pipeline("kokoro", lang_code="a", device="cpu")
assert callable(backend)
mock_manager.create_engine.assert_called_once_with("kokoro", lang_code="a", device="cpu")
def test_create_backend_raises_for_unknown_plugin(self):
"""create_backend raises KeyError for unknown plugins."""
with patch("abogen.tts_plugin.compat.get_plugin_manager") as mock_get_manager:
def test_create_pipeline_raises_for_unknown_plugin(self):
"""create_pipeline raises KeyError for unknown plugins."""
with patch("abogen.tts_plugin.utils.get_plugin_manager") as mock_get_manager:
mock_manager = MagicMock()
mock_get_manager.return_value = mock_manager
mock_manager.create_engine.side_effect = KeyError("Plugin not found")
with pytest.raises(KeyError):
create_backend("nonexistent")
create_pipeline("nonexistent")
class TestPluginManagerWithFakePlugins:
tests/test_conversion_voice_resolution.py
+2 -2
View File
@@ -1,7 +1,7 @@
from types import SimpleNamespace
from typing import cast
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
from abogen.webui.conversion_runner import (
_chapter_voice_spec,
_chunk_voice_spec,
@@ -49,4 +49,4 @@ def test_voice_collection_includes_formula_components():
voices = _collect_required_voice_ids(job)
assert {"af_nova", "am_liam"}.issubset(voices)
assert voices.issuperset(get_metadata("kokoro").voices)
assert voices.issuperset(get_voices("kokoro"))
tests/test_preview_applies_manual_overrides.py
+5 -5
View File
@@ -30,16 +30,16 @@ def test_preview_applies_manual_override_before_normalization(monkeypatch):
captured["text"] = text
return iter(())
from abogen.tts_plugin import compat
from abogen.tts_plugin import utils
original_create_backend = compat.create_backend
original_create_pipeline = utils.create_pipeline
def _mock_create_backend(backend_id, **kwargs):
def _mock_create_pipeline(backend_id, **kwargs):
if backend_id == "supertonic":
return DummyPipeline(**kwargs)
return original_create_backend(backend_id, **kwargs)
return original_create_pipeline(backend_id, **kwargs)
monkeypatch.setattr(compat, "create_backend", _mock_create_backend)
monkeypatch.setattr(utils, "create_pipeline", _mock_create_pipeline)
try:
preview.generate_preview_audio(
tests/test_voice_cache.py
+2 -2
View File
@@ -3,7 +3,7 @@ from typing import cast
import pytest
from abogen.tts_plugin.compat import get_metadata
from abogen.tts_plugin.utils import get_voices
from abogen.voice_cache import (
LocalEntryNotFoundError,
_CACHED_VOICES,
@@ -66,4 +66,4 @@ def test_collect_required_voice_ids_includes_all():
voices = _collect_required_voice_ids(cast(Job, job))
assert {"af_nova", "am_liam", "am_michael"}.issubset(voices)
assert voices.issuperset(get_metadata("kokoro").voices)
assert voices.issuperset(get_voices("kokoro"))