diff --git a/abogen/predownload_gui.py b/abogen/predownload_gui.py index cc40933..be4851c 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata from abogen.spacy_utils import SPACY_MODELS import abogen.hf_tracker diff --git a/abogen/pyqt/gui.py b/abogen/pyqt/gui.py index ba8319e..9bbee41 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata import threading from abogen.pyqt.voice_formula_gui import VoiceFormulaDialog from abogen.voice_profiles import load_profiles diff --git a/abogen/pyqt/predownload_gui.py b/abogen/pyqt/predownload_gui.py index cc40933..be4851c 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata from abogen.spacy_utils import SPACY_MODELS import abogen.hf_tracker diff --git a/abogen/pyqt/voice_formula_gui.py b/abogen/pyqt/voice_formula_gui.py index 06980ec..85e340f 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata import re import platform from abogen.utils import get_resource_path diff --git a/abogen/subtitle_utils.py b/abogen/subtitle_utils.py index 41d114b..24571dd 100644 --- a/abogen/subtitle_utils.py +++ b/abogen/subtitle_utils.py @@ -477,7 +477,7 @@ 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_backend_registry import get_metadata + from abogen.tts_plugin.compat import get_metadata # Create case-insensitive lookup set (done once per call) voice_lookup_lower = {v.lower() for v in get_metadata("kokoro").voices} @@ -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_backend_registry import get_metadata + from abogen.tts_plugin.compat import get_metadata voice_splits = list(_VOICE_MARKER_SEARCH_PATTERN.finditer(text)) diff --git a/abogen/tts_plugin/__init__.py b/abogen/tts_plugin/__init__.py index 10c9154..fbd67ac 100644 --- a/abogen/tts_plugin/__init__.py +++ b/abogen/tts_plugin/__init__.py @@ -1,156 +1,170 @@ -"""TTS Plugin Architecture - Public API. - -This package defines the frozen Plugin API for the TTS Plugin Architecture. -All public interfaces are fully defined but contain no business logic. - -Public modules: -- types: Core domain value objects (AudioFormat, Duration, VoiceSelection, etc.) -- errors: Error hierarchy (EngineError and subtypes) -- manifest: Plugin manifest types (PluginManifest, EngineManifest, etc.) -- engine: Engine and EngineSession protocols -- capabilities: Optional capability interfaces (VoiceLister, PreviewGenerator, etc.) -- host_context: HostContext dataclass -- 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 - -Usage: - from abogen.tts_plugin import ( - # Types - AudioFormat, - Duration, - VoiceSelection, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - EngineConfig, - # Errors - EngineError, - ModelNotFoundError, - ModelLoadError, - NetworkError, - InvalidInputError, - ConfigurationError, - CancelledError, - InternalError, - # Manifest - PluginManifest, - EngineManifest, - VoiceSourceManifest, - VoiceManifest, - ParameterManifest, - AudioFormatManifest, - EnumOption, - RequirementManifest, - GpuRequirement, - ModelManifest, - # Engine - Engine, - EngineSession, - # Capabilities - VoiceLister, - PreviewGenerator, - StreamingSynthesizer, - CancelableSession, - # Host Context - HostContext, - HttpClient, - # Plugin Manager - get_plugin_manager, - reset_plugin_manager, - # Compatibility - create_backend, - ) -""" - -from abogen.tts_plugin.capabilities import ( - CancelableSession, - PreviewGenerator, - StreamingSynthesizer, - VoiceLister, -) -from abogen.tts_plugin.engine import Engine, EngineSession -from abogen.tts_plugin.errors import ( - CancelledError, - ConfigurationError, - EngineError, - InternalError, - InvalidInputError, - ModelLoadError, - ModelNotFoundError, - NetworkError, -) -from abogen.tts_plugin.host_context import HttpClient, HostContext -from abogen.tts_plugin.manifest import ( - AudioFormatManifest, - EngineManifest, - EnumOption, - GpuRequirement, - ModelManifest, - ParameterManifest, - PluginManifest, - RequirementManifest, - VoiceManifest, - VoiceSourceManifest, -) -from abogen.tts_plugin.types import ( - AudioFormat, - Duration, - EngineConfig, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - VoiceSelection, -) - -# Plugin Manager and Compatibility -from abogen.tts_plugin.plugin_manager import get_plugin_manager, reset_plugin_manager -from abogen.tts_plugin.compat import create_backend - -__all__ = [ - # Types - "AudioFormat", - "Duration", - "VoiceSelection", - "ParameterValues", - "SynthesisRequest", - "SynthesizedAudio", - "EngineConfig", - # Errors - "EngineError", - "ModelNotFoundError", - "ModelLoadError", - "NetworkError", - "InvalidInputError", - "ConfigurationError", - "CancelledError", - "InternalError", - # Manifest - "PluginManifest", - "EngineManifest", - "VoiceSourceManifest", - "VoiceManifest", - "ParameterManifest", - "AudioFormatManifest", - "EnumOption", - "RequirementManifest", - "GpuRequirement", - "ModelManifest", - # Engine - "Engine", - "EngineSession", - # Capabilities - "VoiceLister", - "PreviewGenerator", - "StreamingSynthesizer", - "CancelableSession", - # Host Context - "HostContext", - "HttpClient", - # Plugin Manager - "get_plugin_manager", - "reset_plugin_manager", - # Compatibility - "create_backend", -] +"""TTS Plugin Architecture - Public API. + +This package defines the frozen Plugin API for the TTS Plugin Architecture. +All public interfaces are fully defined but contain no business logic. + +Public modules: +- types: Core domain value objects (AudioFormat, Duration, VoiceSelection, etc.) +- errors: Error hierarchy (EngineError and subtypes) +- manifest: Plugin manifest types (PluginManifest, EngineManifest, etc.) +- engine: Engine and EngineSession protocols +- capabilities: Optional capability interfaces (VoiceLister, PreviewGenerator, etc.) +- host_context: HostContext dataclass +- 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 + +Usage: + from abogen.tts_plugin import ( + # Types + AudioFormat, + Duration, + VoiceSelection, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + EngineConfig, + # Errors + EngineError, + ModelNotFoundError, + ModelLoadError, + NetworkError, + InvalidInputError, + ConfigurationError, + CancelledError, + InternalError, + # Manifest + PluginManifest, + EngineManifest, + VoiceSourceManifest, + VoiceManifest, + ParameterManifest, + AudioFormatManifest, + EnumOption, + RequirementManifest, + GpuRequirement, + ModelManifest, + # Engine + Engine, + EngineSession, + # Capabilities + VoiceLister, + PreviewGenerator, + StreamingSynthesizer, + CancelableSession, + # Host Context + HostContext, + HttpClient, + # Plugin Manager + get_plugin_manager, + reset_plugin_manager, + # Compatibility + create_backend, + get_metadata, + is_registered_backend, + resolve_backend_for_voice, + get_default_voice, + ) +""" + +from abogen.tts_plugin.capabilities import ( + CancelableSession, + PreviewGenerator, + StreamingSynthesizer, + VoiceLister, +) +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import ( + CancelledError, + ConfigurationError, + EngineError, + InternalError, + InvalidInputError, + ModelLoadError, + ModelNotFoundError, + NetworkError, +) +from abogen.tts_plugin.host_context import HttpClient, HostContext +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + EnumOption, + GpuRequirement, + ModelManifest, + ParameterManifest, + PluginManifest, + RequirementManifest, + VoiceManifest, + VoiceSourceManifest, +) +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + EngineConfig, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + +# Plugin Manager and Compatibility +from abogen.tts_plugin.plugin_manager import get_plugin_manager, reset_plugin_manager +from abogen.tts_plugin.compat import ( + create_backend, + get_default_voice, + get_metadata, + is_registered_backend, + resolve_backend_for_voice, +) + +__all__ = [ + # Types + "AudioFormat", + "Duration", + "VoiceSelection", + "ParameterValues", + "SynthesisRequest", + "SynthesizedAudio", + "EngineConfig", + # Errors + "EngineError", + "ModelNotFoundError", + "ModelLoadError", + "NetworkError", + "InvalidInputError", + "ConfigurationError", + "CancelledError", + "InternalError", + # Manifest + "PluginManifest", + "EngineManifest", + "VoiceSourceManifest", + "VoiceManifest", + "ParameterManifest", + "AudioFormatManifest", + "EnumOption", + "RequirementManifest", + "GpuRequirement", + "ModelManifest", + # Engine + "Engine", + "EngineSession", + # Capabilities + "VoiceLister", + "PreviewGenerator", + "StreamingSynthesizer", + "CancelableSession", + # Host Context + "HostContext", + "HttpClient", + # Plugin Manager + "get_plugin_manager", + "reset_plugin_manager", + # Compatibility + "create_backend", + "get_metadata", + "is_registered_backend", + "resolve_backend_for_voice", + "get_default_voice", +] diff --git a/abogen/tts_plugin/compat.py b/abogen/tts_plugin/compat.py index db4384d..252ffbd 100644 --- a/abogen/tts_plugin/compat.py +++ b/abogen/tts_plugin/compat.py @@ -135,3 +135,141 @@ def create_backend(backend_id: str, **kwargs: Any) -> Any: 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/utils.py b/abogen/utils.py index a9dd93f..0bebc6b 100644 --- a/abogen/utils.py +++ b/abogen/utils.py @@ -538,7 +538,7 @@ class LoadPipelineThread(Thread): def run(self): try: - from abogen.tts_backend_registry import create_backend + from abogen.tts_plugin.compat import create_backend backend = create_backend( "kokoro", lang_code=self.lang_code, device=self.device diff --git a/abogen/voice_cache.py b/abogen/voice_cache.py index 7d93542..181be5d 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata _CACHE_LOCK = threading.Lock() _CACHED_VOICES: Set[str] = set() diff --git a/abogen/voice_formulas.py b/abogen/voice_formulas.py index 6078d9b..aa027a5 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata # Calls parsing and loads the voice to gpu or cpu diff --git a/abogen/voice_profiles.py b/abogen/voice_profiles.py index 374b513..5a4cf85 100644 --- a/abogen/voice_profiles.py +++ b/abogen/voice_profiles.py @@ -1,230 +1,230 @@ -import json -import os -from typing import Any, Dict, Iterable, List, Tuple - -from abogen.tts_backend_registry import get_metadata, is_registered_backend -from abogen.utils import get_user_config_path - - -def _get_profiles_path(): - config_path = get_user_config_path() - config_dir = os.path.dirname(config_path) - return os.path.join(config_dir, "voice_profiles.json") - - -def load_profiles(): - """Load all voice profiles from JSON file.""" - path = _get_profiles_path() - if os.path.exists(path): - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - # always expect abogen_voice_profiles wrapper - if isinstance(data, dict) and "abogen_voice_profiles" in data: - return data["abogen_voice_profiles"] - # fallback: treat as profiles dict - if isinstance(data, dict): - return data - except Exception: - return {} - return {} - - -def save_profiles(profiles): - """Save all voice profiles to JSON file.""" - path = _get_profiles_path() - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w", encoding="utf-8") as f: - # always save with abogen_voice_profiles wrapper - json.dump({"abogen_voice_profiles": profiles}, f, indent=2) - - -def delete_profile(name): - """Remove a profile by name.""" - profiles = load_profiles() - if name in profiles: - del profiles[name] - save_profiles(profiles) - - -def duplicate_profile(src, dest): - """Duplicate an existing profile.""" - profiles = load_profiles() - if src in profiles and dest: - profiles[dest] = profiles[src] - save_profiles(profiles) - - -def export_profiles(export_path): - """Export all profiles to specified JSON file.""" - profiles = load_profiles() - with open(export_path, "w", encoding="utf-8") as f: - json.dump({"abogen_voice_profiles": profiles}, f, indent=2) - - -def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]: - """Return profiles in canonical dictionary form.""" - return load_profiles() - - -def _normalize_supertonic_voice(value: Any) -> str: - raw = str(value or "").strip().upper() - supertonic_voices = get_metadata("supertonic").voices - return raw if raw in supertonic_voices else "M1" - - -def _coerce_supertonic_steps(value: Any) -> int: - try: - steps = int(value) - except (TypeError, ValueError): - return 5 - return max(2, min(15, steps)) - - -def _coerce_supertonic_speed(value: Any) -> float: - try: - speed = float(value) - except (TypeError, ValueError): - return 1.0 - return max(0.7, min(2.0, speed)) - - -def normalize_profile_entry(entry: Any) -> Dict[str, Any]: - """Normalize a stored profile entry. - - Backwards compatible: - - Legacy Kokoro-only entries: {language, voices} - - New entries: include provider. - """ - - if not isinstance(entry, dict): - return {} - - provider = str(entry.get("provider") or "kokoro").strip().lower() - if not is_registered_backend(provider): - provider = "kokoro" - - language = str(entry.get("language") or "a").strip().lower() or "a" - - if provider == "supertonic": - return { - "provider": "supertonic", - "language": language, - "voice": _normalize_supertonic_voice( - entry.get("voice") or entry.get("voice_name") or entry.get("name") - ), - "total_steps": _coerce_supertonic_steps( - entry.get("total_steps") - or entry.get("supertonic_total_steps") - or entry.get("quality") - ), - "speed": _coerce_supertonic_speed( - entry.get("speed") or entry.get("supertonic_speed") - ), - } - - voices = _normalize_voice_entries(entry.get("voices", [])) - if not voices: - return {} - return { - "provider": "kokoro", - "language": language, - "voices": voices, - } - - -def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]: - normalized: List[Tuple[str, float]] = [] - kokoro_voices = get_metadata("kokoro").voices - for item in entries or []: - if isinstance(item, dict): - voice = item.get("id") or item.get("voice") - weight = item.get("weight") - elif isinstance(item, (list, tuple)) and len(item) >= 2: - voice, weight = item[0], item[1] - else: - continue - if voice not in kokoro_voices: - continue - if weight is None: - continue - try: - weight_val = float(weight) - except (TypeError, ValueError): - continue - if weight_val <= 0: - continue - normalized.append((voice, weight_val)) - return normalized - - -def normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]: - """Public helper to normalize voice-weight pairs from arbitrary payloads.""" - - return _normalize_voice_entries(entries) - - -def save_profile(name: str, *, language: str, voices: Iterable) -> None: - """Persist a single profile after validating its data.""" - - name = (name or "").strip() - if not name: - raise ValueError("Profile name is required") - - normalized = _normalize_voice_entries(voices) - if not normalized: - raise ValueError("At least one voice with a weight above zero is required") - - if not language: - language = "a" - - profiles = load_profiles() - profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized} - save_profiles(profiles) - - -def remove_profile(name: str) -> None: - delete_profile(name) - - -def import_profiles_data(data: Dict, *, replace_existing: bool = False) -> List[str]: - """Merge profiles from a dictionary structure and persist them. - - Returns the list of profile names that were added or updated. - """ - - if not isinstance(data, dict): - raise ValueError("Invalid profile payload") - - if "abogen_voice_profiles" in data: - data = data["abogen_voice_profiles"] - - if not isinstance(data, dict): - raise ValueError("Invalid profile payload") - - current = load_profiles() - updated: List[str] = [] - for name, entry in data.items(): - normalized = normalize_profile_entry(entry) - if not normalized: - continue - if name in current and not replace_existing: - # skip duplicates unless explicit replacement is requested - continue - current[name] = normalized - updated.append(name) - - if updated: - save_profiles(current) - return updated - - -def export_profiles_payload(names: Iterable[str] | None = None) -> Dict[str, Dict]: - """Return profiles limited to the provided names for download/export.""" - - profiles = load_profiles() - if names is None: - subset = profiles - else: - subset = {name: profiles[name] for name in names if name in profiles} - return {"abogen_voice_profiles": subset} +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.utils import get_user_config_path + + +def _get_profiles_path(): + config_path = get_user_config_path() + config_dir = os.path.dirname(config_path) + return os.path.join(config_dir, "voice_profiles.json") + + +def load_profiles(): + """Load all voice profiles from JSON file.""" + path = _get_profiles_path() + if os.path.exists(path): + try: + with open(path, "r", encoding="utf-8") as f: + data = json.load(f) + # always expect abogen_voice_profiles wrapper + if isinstance(data, dict) and "abogen_voice_profiles" in data: + return data["abogen_voice_profiles"] + # fallback: treat as profiles dict + if isinstance(data, dict): + return data + except Exception: + return {} + return {} + + +def save_profiles(profiles): + """Save all voice profiles to JSON file.""" + path = _get_profiles_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + # always save with abogen_voice_profiles wrapper + json.dump({"abogen_voice_profiles": profiles}, f, indent=2) + + +def delete_profile(name): + """Remove a profile by name.""" + profiles = load_profiles() + if name in profiles: + del profiles[name] + save_profiles(profiles) + + +def duplicate_profile(src, dest): + """Duplicate an existing profile.""" + profiles = load_profiles() + if src in profiles and dest: + profiles[dest] = profiles[src] + save_profiles(profiles) + + +def export_profiles(export_path): + """Export all profiles to specified JSON file.""" + profiles = load_profiles() + with open(export_path, "w", encoding="utf-8") as f: + json.dump({"abogen_voice_profiles": profiles}, f, indent=2) + + +def serialize_profiles() -> Dict[str, Dict[str, Iterable[Tuple[str, float]]]]: + """Return profiles in canonical dictionary form.""" + return load_profiles() + + +def _normalize_supertonic_voice(value: Any) -> str: + raw = str(value or "").strip().upper() + supertonic_voices = get_metadata("supertonic").voices + return raw if raw in supertonic_voices else "M1" + + +def _coerce_supertonic_steps(value: Any) -> int: + try: + steps = int(value) + except (TypeError, ValueError): + return 5 + return max(2, min(15, steps)) + + +def _coerce_supertonic_speed(value: Any) -> float: + try: + speed = float(value) + except (TypeError, ValueError): + return 1.0 + return max(0.7, min(2.0, speed)) + + +def normalize_profile_entry(entry: Any) -> Dict[str, Any]: + """Normalize a stored profile entry. + + Backwards compatible: + - Legacy Kokoro-only entries: {language, voices} + - New entries: include provider. + """ + + if not isinstance(entry, dict): + return {} + + provider = str(entry.get("provider") or "kokoro").strip().lower() + if not is_registered_backend(provider): + provider = "kokoro" + + language = str(entry.get("language") or "a").strip().lower() or "a" + + if provider == "supertonic": + return { + "provider": "supertonic", + "language": language, + "voice": _normalize_supertonic_voice( + entry.get("voice") or entry.get("voice_name") or entry.get("name") + ), + "total_steps": _coerce_supertonic_steps( + entry.get("total_steps") + or entry.get("supertonic_total_steps") + or entry.get("quality") + ), + "speed": _coerce_supertonic_speed( + entry.get("speed") or entry.get("supertonic_speed") + ), + } + + voices = _normalize_voice_entries(entry.get("voices", [])) + if not voices: + return {} + return { + "provider": "kokoro", + "language": language, + "voices": voices, + } + + +def _normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]: + normalized: List[Tuple[str, float]] = [] + kokoro_voices = get_metadata("kokoro").voices + for item in entries or []: + if isinstance(item, dict): + voice = item.get("id") or item.get("voice") + weight = item.get("weight") + elif isinstance(item, (list, tuple)) and len(item) >= 2: + voice, weight = item[0], item[1] + else: + continue + if voice not in kokoro_voices: + continue + if weight is None: + continue + try: + weight_val = float(weight) + except (TypeError, ValueError): + continue + if weight_val <= 0: + continue + normalized.append((voice, weight_val)) + return normalized + + +def normalize_voice_entries(entries: Iterable) -> List[Tuple[str, float]]: + """Public helper to normalize voice-weight pairs from arbitrary payloads.""" + + return _normalize_voice_entries(entries) + + +def save_profile(name: str, *, language: str, voices: Iterable) -> None: + """Persist a single profile after validating its data.""" + + name = (name or "").strip() + if not name: + raise ValueError("Profile name is required") + + normalized = _normalize_voice_entries(voices) + if not normalized: + raise ValueError("At least one voice with a weight above zero is required") + + if not language: + language = "a" + + profiles = load_profiles() + profiles[name] = {"provider": "kokoro", "language": language, "voices": normalized} + save_profiles(profiles) + + +def remove_profile(name: str) -> None: + delete_profile(name) + + +def import_profiles_data(data: Dict, *, replace_existing: bool = False) -> List[str]: + """Merge profiles from a dictionary structure and persist them. + + Returns the list of profile names that were added or updated. + """ + + if not isinstance(data, dict): + raise ValueError("Invalid profile payload") + + if "abogen_voice_profiles" in data: + data = data["abogen_voice_profiles"] + + if not isinstance(data, dict): + raise ValueError("Invalid profile payload") + + current = load_profiles() + updated: List[str] = [] + for name, entry in data.items(): + normalized = normalize_profile_entry(entry) + if not normalized: + continue + if name in current and not replace_existing: + # skip duplicates unless explicit replacement is requested + continue + current[name] = normalized + updated.append(name) + + if updated: + save_profiles(current) + return updated + + +def export_profiles_payload(names: Iterable[str] | None = None) -> Dict[str, Dict]: + """Return profiles limited to the provided names for download/export.""" + + profiles = load_profiles() + if names is None: + subset = profiles + else: + subset = {name: profiles[name] for name in names if name in profiles} + return {"abogen_voice_profiles": subset} diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 2706a1d..b90aaca 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_backend_registry import get_metadata, is_registered_backend, resolve_backend_for_voice +from abogen.tts_plugin.compat import get_metadata, is_registered_backend, resolve_backend_for_voice 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,8 +40,7 @@ from abogen.utils import ( get_user_output_path, load_config, ) -from abogen.tts_backend_registry import create_backend -from abogen.tts_backend import TTSBackend +from abogen.tts_plugin.compat import create_backend 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 diff --git a/abogen/webui/debug_tts_runner.py b/abogen/webui/debug_tts_runner.py index b5634e8..8cc39ba 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_backend_registry import create_backend +from abogen.tts_plugin.compat import create_backend _MARKER_RE = re.compile(re.escape(MARKER_PREFIX) + r"(?P[A-Z0-9_]+)" + re.escape(MARKER_SUFFIX)) diff --git a/abogen/webui/routes/api.py b/abogen/webui/routes/api.py index 7303420..1c64142 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_backend_registry import is_registered_backend +from abogen.tts_plugin.compat import is_registered_backend from abogen.integrations.audiobookshelf import AudiobookshelfClient, AudiobookshelfConfig from abogen.integrations.calibre_opds import ( CalibreOPDSClient, diff --git a/abogen/webui/routes/utils/form.py b/abogen/webui/routes/utils/form.py index 40cfabd..16f3bf4 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_backend_registry import is_registered_backend +from abogen.tts_plugin.compat import is_registered_backend 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_backend_registry import get_default_voice +from abogen.tts_plugin.compat 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 diff --git a/abogen/webui/routes/utils/settings.py b/abogen/webui/routes/utils/settings.py index 3ffb072..2bd3c3c 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_backend_registry import get_default_voice +from abogen.tts_plugin.compat 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 42a2c13..439bc0b 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata from abogen.speaker_configs import list_configs -from abogen.tts_backend_registry import create_backend +from abogen.tts_plugin.compat import create_backend from abogen.webui.conversion_runner import _select_device, _to_float32, SAMPLE_RATE, SPLIT_PATTERN _preview_pipeline_lock = threading.RLock() diff --git a/tests/test_conversion_voice_resolution.py b/tests/test_conversion_voice_resolution.py index 75f09b6..b6e1440 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata from abogen.webui.conversion_runner import ( _chapter_voice_spec, _chunk_voice_spec, diff --git a/tests/test_voice_cache.py b/tests/test_voice_cache.py index e5ce9a1..eac5846 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_backend_registry import get_metadata +from abogen.tts_plugin.compat import get_metadata from abogen.voice_cache import ( LocalEntryNotFoundError, _CACHED_VOICES,