diff --git a/abogen/predownload_gui.py b/abogen/predownload_gui.py index be4851c..fb084ea 100644 --- a/abogen/predownload_gui.py +++ b/abogen/predownload_gui.py @@ -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: diff --git a/abogen/pyqt/gui.py b/abogen/pyqt/gui.py index 9bbee41..dce75b5 100644 --- a/abogen/pyqt/gui.py +++ b/abogen/pyqt/gui.py @@ -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): diff --git a/abogen/pyqt/predownload_gui.py b/abogen/pyqt/predownload_gui.py index be4851c..fb084ea 100644 --- a/abogen/pyqt/predownload_gui.py +++ b/abogen/pyqt/predownload_gui.py @@ -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: diff --git a/abogen/pyqt/voice_formula_gui.py b/abogen/pyqt/voice_formula_gui.py index 85e340f..0621fbf 100644 --- a/abogen/pyqt/voice_formula_gui.py +++ b/abogen/pyqt/voice_formula_gui.py @@ -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 diff --git a/abogen/subtitle_utils.py b/abogen/subtitle_utils.py index 24571dd..489ed24 100644 --- a/abogen/subtitle_utils.py +++ b/abogen/subtitle_utils.py @@ -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 diff --git a/abogen/tts_plugin/__init__.py b/abogen/tts_plugin/__init__.py index fbd67ac..b9caa33 100644 --- a/abogen/tts_plugin/__init__.py +++ b/abogen/tts_plugin/__init__.py @@ -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", ] diff --git a/abogen/tts_plugin/compat.py b/abogen/tts_plugin/compat.py deleted file mode 100644 index 252ffbd..0000000 --- a/abogen/tts_plugin/compat.py +++ /dev/null @@ -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 diff --git a/abogen/tts_plugin/utils.py b/abogen/tts_plugin/utils.py new file mode 100644 index 0000000..b39d932 --- /dev/null +++ b/abogen/tts_plugin/utils.py @@ -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) diff --git a/abogen/utils.py b/abogen/utils.py index 0bebc6b..56812e8 100644 --- a/abogen/utils.py +++ b/abogen/utils.py @@ -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) diff --git a/abogen/voice_cache.py b/abogen/voice_cache.py index 181be5d..838eb65 100644 --- a/abogen/voice_cache.py +++ b/abogen/voice_cache.py @@ -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() diff --git a/abogen/voice_formulas.py b/abogen/voice_formulas.py index aa027a5..b4731d7 100644 --- a/abogen/voice_formulas.py +++ b/abogen/voice_formulas.py @@ -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: diff --git a/abogen/voice_profiles.py b/abogen/voice_profiles.py index 5a4cf85..1841348 100644 --- a/abogen/voice_profiles.py +++ b/abogen/voice_profiles.py @@ -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") diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index b90aaca..955fd6d 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -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: diff --git a/abogen/webui/debug_tts_runner.py b/abogen/webui/debug_tts_runner.py index 8cc39ba..63c34dd 100644 --- a/abogen/webui/debug_tts_runner.py +++ b/abogen/webui/debug_tts_runner.py @@ -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[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]]: diff --git a/abogen/webui/routes/api.py b/abogen/webui/routes/api.py index 1c64142..28c1e9b 100644 --- a/abogen/webui/routes/api.py +++ b/abogen/webui/routes/api.py @@ -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)) diff --git a/abogen/webui/routes/utils/form.py b/abogen/webui/routes/utils/form.py index 16f3bf4..39e033f 100644 --- a/abogen/webui/routes/utils/form.py +++ b/abogen/webui/routes/utils/form.py @@ -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). diff --git a/abogen/webui/routes/utils/preview.py b/abogen/webui/routes/utils/preview.py index f96790b..7ad9158 100644 --- a/abogen/webui/routes/utils/preview.py +++ b/abogen/webui/routes/utils/preview.py @@ -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, diff --git a/abogen/webui/routes/utils/settings.py b/abogen/webui/routes/utils/settings.py index 2bd3c3c..589f8a0 100644 --- a/abogen/webui/routes/utils/settings.py +++ b/abogen/webui/routes/utils/settings.py @@ -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, diff --git a/abogen/webui/routes/utils/voice.py b/abogen/webui/routes/utils/voice.py index 439bc0b..80c8960 100644 --- a/abogen/webui/routes/utils/voice.py +++ b/abogen/webui/routes/utils/voice.py @@ -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 diff --git a/tests/contracts/test_integration.py b/tests/contracts/test_integration.py index 2fed590..ff7ee26 100644 --- a/tests/contracts/test_integration.py +++ b/tests/contracts/test_integration.py @@ -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" + ) diff --git a/tests/contracts/test_plugin_manager_contract.py b/tests/contracts/test_plugin_manager_contract.py index 1a92b73..11e35dd 100644 --- a/tests/contracts/test_plugin_manager_contract.py +++ b/tests/contracts/test_plugin_manager_contract.py @@ -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: diff --git a/tests/test_conversion_voice_resolution.py b/tests/test_conversion_voice_resolution.py index b6e1440..1f41900 100644 --- a/tests/test_conversion_voice_resolution.py +++ b/tests/test_conversion_voice_resolution.py @@ -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")) diff --git a/tests/test_preview_applies_manual_overrides.py b/tests/test_preview_applies_manual_overrides.py index 8501fa1..e7e6bdb 100644 --- a/tests/test_preview_applies_manual_overrides.py +++ b/tests/test_preview_applies_manual_overrides.py @@ -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( diff --git a/tests/test_voice_cache.py b/tests/test_voice_cache.py index eac5846..55ade39 100644 --- a/tests/test_voice_cache.py +++ b/tests/test_voice_cache.py @@ -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"))