diff --git a/abogen/tts_plugin/manifest.py b/abogen/tts_plugin/manifest.py index 4f4a42b..bfa10ea 100644 --- a/abogen/tts_plugin/manifest.py +++ b/abogen/tts_plugin/manifest.py @@ -1,186 +1,189 @@ -"""Plugin manifest types for the TTS Plugin Architecture. - -This module contains static metadata types that describe plugins. -These types have no dependencies and are immutable. -""" - -from __future__ import annotations - -from dataclasses import dataclass, field -from typing import Any - - -@dataclass(frozen=True) -class AudioFormatManifest: - """Manifest describing an audio format. - - Attributes: - mime: MIME type of the audio. - extension: File extension. - """ - - mime: str - extension: str - - -@dataclass(frozen=True) -class EnumOption: - """Manifest describing an enum option for a parameter. - - Attributes: - value: The enum value. - label: Human-readable label. - """ - - value: str - label: str - - -@dataclass(frozen=True) -class ParameterManifest: - """Manifest describing a synthesis parameter. - - Attributes: - id: Parameter identifier. - name: Human-readable name. - description: Parameter description. - type: Parameter type ("float", "int", "string", "boolean", "enum"). - default: Default value. - min: Minimum value (optional, for numeric types). - max: Maximum value (optional, for numeric types). - step: Step size (optional, for numeric types). - options: Available options (optional, for enum type). - unit: Unit of measurement (optional). - group: Parameter group (optional). - """ - - id: str - name: str - description: str - type: str - default: Any - min: float | None = None - max: float | None = None - step: float | None = None - options: tuple[EnumOption, ...] = field(default_factory=tuple) - unit: str | None = None - group: str | None = None - - -@dataclass(frozen=True) -class VoiceManifest: - """Manifest describing a voice. - - Attributes: - id: Voice identifier. - name: Human-readable name. - tags: Voice tags (e.g., language, style). - """ - - id: str - name: str - tags: tuple[str, ...] = field(default_factory=tuple) - - -@dataclass(frozen=True) -class VoiceSourceManifest: - """Manifest describing a voice source. - - Attributes: - id: Voice source identifier. - name: Human-readable name. - type: Source type ("list", "speaker_id", "clone", "blend", "generate", "none"). - config: Source-specific configuration. - """ - - id: str - name: str - type: str - config: Any = None - - -@dataclass(frozen=True) -class EngineManifest: - """Manifest describing engine capabilities. - - Attributes: - voiceSources: Available voice sources. - parameters: Available synthesis parameters. - audioFormats: Supported audio formats. - """ - - voiceSources: tuple[VoiceSourceManifest, ...] = field(default_factory=tuple) - parameters: tuple[ParameterManifest, ...] = field(default_factory=tuple) - audioFormats: tuple[AudioFormatManifest, ...] = field(default_factory=tuple) - - -@dataclass(frozen=True) -class GpuRequirement: - """Manifest describing GPU requirements. - - Attributes: - required: Whether GPU is required. - type: GPU type (e.g., "cuda", "rocm"). - memory: Required GPU memory in GB. - """ - - required: bool = False - type: str | None = None - memory: float | None = None - - -@dataclass(frozen=True) -class RequirementManifest: - """Manifest describing plugin requirements. - - Attributes: - gpu: GPU requirements (optional). - memory: Required RAM in GB (optional). - internet: Whether internet is required (optional). - """ - - gpu: GpuRequirement | None = None - memory: float | None = None - internet: bool | None = None - - -@dataclass(frozen=True) -class ModelManifest: - """Manifest describing a model requirement. - - Attributes: - id: Model identifier. - name: Human-readable name. - size: Model size as string (e.g., "100MB", "2GB"). - """ - - id: str - name: str - size: str - - -@dataclass(frozen=True) -class PluginManifest: - """Main manifest for a TTS plugin. - - Attributes: - id: Plugin identifier (unique). - name: Human-readable name. - version: Plugin version. - api_version: API version (semver format: MAJOR.MINOR). - description: Plugin description. - author: Plugin author. - capabilities: List of capability identifiers. - requires: Plugin requirements. - engine: Engine manifest. - """ - - id: str - name: str - version: str - api_version: str - description: str - author: str - capabilities: tuple[str, ...] = field(default_factory=tuple) - requires: RequirementManifest = field(default_factory=RequirementManifest) - engine: EngineManifest = field(default_factory=EngineManifest) +"""Plugin manifest types for the TTS Plugin Architecture. + +This module contains static metadata types that describe plugins. +These types have no dependencies and are immutable. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class AudioFormatManifest: + """Manifest describing an audio format. + + Attributes: + mime: MIME type of the audio. + extension: File extension. + """ + + mime: str + extension: str + + +@dataclass(frozen=True) +class EnumOption: + """Manifest describing an enum option for a parameter. + + Attributes: + value: The enum value. + label: Human-readable label. + """ + + value: str + label: str + + +@dataclass(frozen=True) +class ParameterManifest: + """Manifest describing a synthesis parameter. + + Attributes: + id: Parameter identifier. + name: Human-readable name. + description: Parameter description. + type: Parameter type ("float", "int", "string", "boolean", "enum"). + default: Default value. + min: Minimum value (optional, for numeric types). + max: Maximum value (optional, for numeric types). + step: Step size (optional, for numeric types). + options: Available options (optional, for enum type). + unit: Unit of measurement (optional). + group: Parameter group (optional). + """ + + id: str + name: str + description: str + type: str + default: Any + min: float | None = None + max: float | None = None + step: float | None = None + options: tuple[EnumOption, ...] = field(default_factory=tuple) + unit: str | None = None + group: str | None = None + + +@dataclass(frozen=True) +class VoiceManifest: + """Manifest describing a voice. + + Attributes: + id: Voice identifier. + name: Human-readable name. + tags: Voice tags (e.g., language, style). + """ + + id: str + name: str + tags: tuple[str, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class VoiceSourceManifest: + """Manifest describing a voice source. + + Attributes: + id: Voice source identifier. + name: Human-readable name. + type: Source type ("list", "speaker_id", "clone", "blend", "generate", "none"). + config: Source-specific configuration. + """ + + id: str + name: str + type: str + config: Any = None + + +@dataclass(frozen=True) +class EngineManifest: + """Manifest describing engine capabilities. + + Attributes: + voiceSources: Available voice sources. + parameters: Available synthesis parameters. + audioFormats: Supported audio formats. + """ + + voiceSources: tuple[VoiceSourceManifest, ...] = field(default_factory=tuple) + parameters: tuple[ParameterManifest, ...] = field(default_factory=tuple) + audioFormats: tuple[AudioFormatManifest, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class GpuRequirement: + """Manifest describing GPU requirements. + + Attributes: + required: Whether GPU is required. + type: GPU type (e.g., "cuda", "rocm"). + memory: Required GPU memory in GB. + """ + + required: bool = False + type: str | None = None + memory: float | None = None + + +@dataclass(frozen=True) +class RequirementManifest: + """Manifest describing plugin requirements. + + Attributes: + gpu: GPU requirements (optional). + memory: Required RAM in GB (optional). + internet: Whether internet is required (optional). + """ + + gpu: GpuRequirement | None = None + memory: float | None = None + internet: bool | None = None + + +@dataclass(frozen=True) +class ModelManifest: + """Manifest describing a model requirement. + + Attributes: + id: Model identifier. + name: Human-readable name. + size: Model size as string (e.g., "100MB", "2GB"). + """ + + id: str + name: str + size: str + + +@dataclass(frozen=True) +class PluginManifest: + """Main manifest for a TTS plugin. + + Attributes: + id: Plugin identifier (unique). + name: Human-readable name. + version: Plugin version. + api_version: API version (semver format: MAJOR.MINOR). + description: Plugin description. + author: Plugin author. + capabilities: List of capability identifiers. + requires: Plugin requirements. + engine: Engine manifest. + voices: Optional static voice catalog. None = not declared (use VoiceLister), + empty tuple = explicitly no static voices, non-empty = static catalog. + """ + + id: str + name: str + version: str + api_version: str + description: str + author: str + capabilities: tuple[str, ...] = field(default_factory=tuple) + requires: RequirementManifest = field(default_factory=RequirementManifest) + engine: EngineManifest = field(default_factory=EngineManifest) + voices: tuple[VoiceManifest, ...] | None = None diff --git a/abogen/tts_plugin/utils.py b/abogen/tts_plugin/utils.py index 19d4780..4a05fa9 100644 --- a/abogen/tts_plugin/utils.py +++ b/abogen/tts_plugin/utils.py @@ -6,7 +6,7 @@ calling the Plugin Manager directly. from __future__ import annotations -from typing import Any, Iterator, Optional +from typing import Any, Iterator import numpy as np @@ -17,6 +17,7 @@ def get_voices(plugin_id: str) -> tuple[str, ...]: """Return the voice-id tuple for *plugin_id*. Uses the official Plugin Architecture: PluginManager → Engine → VoiceLister. + First checks plugin manifest for static voice catalog. """ import logging import tempfile @@ -29,6 +30,13 @@ def get_voices(plugin_id: str) -> tuple[str, ...]: if not manager.has_plugin(plugin_id): return () + # Check manifest for static voice catalog + plugin_info = manager.get_plugin(plugin_id) + if plugin_info is not None: + manifest = plugin_info.get("manifest") + if manifest is not None and manifest.voices is not None: + return tuple(v.id for v in manifest.voices) + ctx = HostContext( config_dir=Path(tempfile.gettempdir()), logger=logging.getLogger(f"abogen.utils.{plugin_id}"), @@ -183,12 +191,45 @@ class Pipeline: self.dispose() -def create_pipeline(plugin_id: str, **kwargs: Any) -> Pipeline: +def create_pipeline( + plugin_id: str, + *, + lang_code: str = "a", + device: str = "cpu", +) -> Pipeline: """Create a callable TTS pipeline via the Plugin Architecture. - Returns a :class:`Pipeline` whose ``__call__`` interface matches the - legacy ``TTSBackend`` callable protocol. + Builds a proper HostContext and EngineConfig, then delegates to the + PluginManager to create the engine. Returns a :class:`Pipeline` whose + ``__call__`` interface matches the legacy ``TTSBackend`` callable protocol. + + Args: + plugin_id: Plugin identifier (e.g., "kokoro", "supertonic"). + lang_code: Language code for the engine. + device: Device to use (e.g., "cpu", "cuda:0"). + + Returns: + A callable Pipeline instance. """ + import logging + import tempfile + from pathlib import Path + + from abogen.tts_plugin.host_context import HostContext + from abogen.tts_plugin.types import EngineConfig + manager = get_plugin_manager() - engine = manager.create_engine(plugin_id, **kwargs) - return Pipeline(engine, **kwargs) + + ctx = HostContext( + config_dir=Path(tempfile.gettempdir()), + logger=logging.getLogger(f"abogen.pipeline.{plugin_id}"), + http_client=type("_StubHttpClient", (), { + "get": staticmethod(lambda url, **kw: None), + "post": staticmethod(lambda url, **kw: None), + })(), + ) + + config = EngineConfig(device=device, lang_code=lang_code) + + engine = manager.create_engine(plugin_id, context=ctx, model_path=None, config=config) + return Pipeline(engine) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 955fd6d..4cf7b9c 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -1576,9 +1576,6 @@ def run_conversion_job(job: Job) -> None: if provider_norm == "supertonic": pipelines[provider_norm] = create_pipeline( "supertonic", - sample_rate=SAMPLE_RATE, - auto_download=True, - total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5), ) return pipelines[provider_norm] @@ -1863,17 +1860,17 @@ def run_conversion_job(job: Job) -> None: canceller() graphemes_raw = getattr(segment, "graphemes", "") or "" graphemes = graphemes_raw.strip() - + audio = _to_float32(getattr(segment, "audio", None)) if audio.size == 0: continue - + local_segments += 1 if chapter_sink: chapter_sink.write(audio) if audio_sink: audio_sink.write(audio) - + duration = len(audio) / SAMPLE_RATE processed_chars += len(graphemes) job.processed_characters = processed_chars @@ -1881,11 +1878,11 @@ def run_conversion_job(job: Job) -> None: job.progress = min(processed_chars / job.total_characters, 0.999) else: job.progress = 0.0 if processed_chars == 0 else 0.999 - + preview_text = graphemes or (graphemes_raw[:80] if graphemes_raw else "[silence]") prefix = f"{preview_prefix} · " if preview_prefix else "" job.add_log(f"{prefix}{processed_chars:,}/{job.total_characters or '—'}: {preview_text[:80]}") - + if subtitle_writer and audio_sink and graphemes: subtitle_writer.write_segment( index=subtitle_index, @@ -1894,10 +1891,10 @@ def run_conversion_job(job: Job) -> None: end=current_time + duration, ) subtitle_index += 1 - + if audio_sink: current_time += duration - + except OverflowError as exc: job.add_log( f"Skipped chunk — number too large for TTS conversion: {exc}", @@ -2408,6 +2405,11 @@ def run_conversion_job(job: Job) -> None: # Explicitly release the pipeline and force garbage collection to prevent # memory accumulation in the worker process, which can lead to host lockups. + for p in pipelines.values(): + try: + p.dispose() + except Exception: + pass pipelines.clear() pipeline = None gc.collect() @@ -2443,9 +2445,6 @@ def _load_pipeline(job: Job): if provider == "supertonic": return create_pipeline( "supertonic", - sample_rate=SAMPLE_RATE, - auto_download=True, - total_steps=int(getattr(job, "supertonic_total_steps", 5) or 5), ) device = "cpu" diff --git a/abogen/webui/routes/utils/preview.py b/abogen/webui/routes/utils/preview.py index 7ad9158..739b2ec 100644 --- a/abogen/webui/routes/utils/preview.py +++ b/abogen/webui/routes/utils/preview.py @@ -14,6 +14,17 @@ _preview_pipelines: Dict[Tuple[str, str], Any] = {} _preview_pipeline_lock = threading.Lock() +def clear_preview_pipelines() -> None: + """Dispose all cached preview pipelines and clear the cache.""" + with _preview_pipeline_lock: + for pipeline in _preview_pipelines.values(): + try: + pipeline.dispose() + except Exception: + pass + _preview_pipelines.clear() + + def _select_device() -> str: import platform @@ -138,7 +149,7 @@ def generate_preview_audio( if provider == "supertonic": from abogen.tts_plugin.utils import create_pipeline - pipeline = create_pipeline("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps) + pipeline = create_pipeline("supertonic") segments = pipeline( normalized_text, voice=voice_spec, diff --git a/docs/architecture-amendment-001.md b/docs/architecture-amendment-001.md new file mode 100644 index 0000000..c8cb73e --- /dev/null +++ b/docs/architecture-amendment-001.md @@ -0,0 +1,89 @@ +# Architecture Amendment #1: EngineConfig — `lang_code` field + +**Date:** 2026-07-12 +**Status:** Accepted +**PR:** #12 (Normalize Pipeline Public API) + +## Summary + +Add `lang_code: str = "a"` to `EngineConfig` and update its definition to clarify the architectural contract. + +## Background + +During migration from the old `KokoroBackend` to the Plugin Architecture, the `lang_code` parameter became a dead argument. The old backend read it from `**kwargs` and passed it to `KPipeline(lang_code=...)`. The new `KokoroPlugin.create_engine()` hardcodes `lang_code="a"`, ignoring the config entirely. Callers continued passing `lang_code` to `create_pipeline()`, unaware it had no effect. + +This is a functional regression relative to the pre-Plugin Architecture behavior. + +## Decision + +### 1. Updated EngineConfig definition + +**Before:** +``` +Immutable value object for engine initialization settings. +Contains only engine-specific settings, no resource references. +``` + +**After:** +``` +Immutable configuration of an Engine instance. +Contains parameters that define how a particular Engine instance is +created and that remain constant throughout the lifetime of that Engine. +Plugin implementations may ignore fields that are not applicable to them. +``` + +### 2. New field + +```python +@dataclass(frozen=True) +class EngineConfig: + device: str = "cpu" + lang_code: str = "a" +``` + +### 3. Architectural rules + +- **Fields in EngineConfig are optional unless explicitly required by a plugin.** +- **Plugins MUST ignore unsupported EngineConfig fields.** +- **All parameters that may vary between individual synthesis requests must remain in `SynthesisRequest.parameters`.** + +## Rationale + +Analysis of real TTS engines (Kokoro, Piper, XTTS, Coqui, StyleTTS2, Fish Speech) confirmed: + +| Parameter type | Where it belongs | Example | +|---------------|-----------------|---------| +| Engine instance config (immutable) | `EngineConfig` | `device`, `lang_code` | +| Synthesis parameters (per-request) | `SynthesisRequest.parameters` | `speed`, `split_pattern`, `total_steps` | + +`lang_code` determines the engine's behavior at creation time and cannot be changed during the engine's lifetime. It is not a synthesis parameter. + +## Impact on existing plugins + +| Plugin | `device` | `lang_code` | Notes | +|--------|----------|-------------|-------| +| Kokoro | Reads ✓ | Reads ✓ (was hardcoded, now from config) | Regression fixed | +| SuperTonic | Ignores | Ignores | No change — no language concept | +| Future plugins | May read | May ignore | Field-ignoring rule applies | + +## Contract tests added + +```python +class TestEngineConfigContract: + def test_default_lang_code(self) # EngineConfig().lang_code == "a" + def test_custom_lang_code(self) # EngineConfig(lang_code="j").lang_code == "j" + def test_immutability_lang_code(self) # frozen — cannot reassign + def test_plugins_may_ignore_irrelevant_fields(self) # field-ignoring rule + def test_engine_config_contains_engine_instance_configuration(self) # definition +``` + +## Files changed + +| File | Change | +|------|--------| +| `abogen/tts_plugin/types.py` | Updated docstring, added `lang_code: str = "a"` | +| `plugins/kokoro/__init__.py` | Reads `config.lang_code` instead of hardcoded `"a"` | +| `abogen/tts_plugin/utils.py` | `create_pipeline()` passes `lang_code` to `EngineConfig` | +| `tests/contracts/test_types_contract.py` | 5 new contract tests | +| `tests/contracts/test_plugin_manager_contract.py` | Updated assertion for `lang_code` | +| `tests/test_behavioral_regression.py` | Updated `test_engine_config_defaults` | diff --git a/plugins/kokoro/__init__.py b/plugins/kokoro/__init__.py index 915e054..bdef4b3 100644 --- a/plugins/kokoro/__init__.py +++ b/plugins/kokoro/__init__.py @@ -1,121 +1,178 @@ -"""Kokoro TTS Plugin for the TTS Plugin Architecture. - -This plugin provides a Kokoro-based TTS engine that implements the -Plugin API contract. It wraps the existing Kokoro backend in the -new Engine/EngineSession architecture. - -Exports: - - PLUGIN_MANIFEST: PluginManifest - - MODEL_REQUIREMENTS: list[ModelManifest] - - create_engine: Factory function -""" - -from __future__ import annotations - -from pathlib import Path -from typing import Any - -from abogen.tts_plugin.engine import Engine -from abogen.tts_plugin.host_context import HostContext -from abogen.tts_plugin.manifest import ( - AudioFormatManifest, - EngineManifest, - ModelManifest, - ParameterManifest, - PluginManifest, - RequirementManifest, - VoiceSourceManifest, +"""Kokoro TTS Plugin for the TTS Plugin Architecture. + +This plugin provides a Kokoro-based TTS engine that implements the +Plugin API contract. It wraps the existing Kokoro backend in the +new Engine/EngineSession architecture. + +Exports: + - PLUGIN_MANIFEST: PluginManifest + - MODEL_REQUIREMENTS: list[ModelManifest] + - create_engine: Factory function +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.engine import Engine +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + ModelManifest, + ParameterManifest, + PluginManifest, + RequirementManifest, + VoiceManifest, + VoiceSourceManifest, ) -from abogen.tts_plugin.types import EngineConfig - -from .engine import KokoroEngine - - -def _load_kpipeline() -> Any: - """Lazy-load Kokoro dependencies.""" - from kokoro import KPipeline # type: ignore[import-not-found] - return KPipeline - - -PLUGIN_MANIFEST = PluginManifest( - id="kokoro", - name="Kokoro", - version="0.9.4", - api_version="1.0", - description="Kokoro TTS engine - high quality multilingual text-to-speech", - author="Kokoro Team", - capabilities=("voice_list",), - requires=RequirementManifest( - internet=False, - ), - engine=EngineManifest( - voiceSources=( - VoiceSourceManifest( - id="builtin", - name="Built-in Voices", - type="list", - config={"voices": "See listVoices()"}, - ), - ), - parameters=( - ParameterManifest( - id="speed", - name="Speed", - description="Speech speed multiplier", - type="float", - default=1.0, - min=0.5, - max=2.0, - step=0.1, - ), - ), - audioFormats=( - AudioFormatManifest(mime="audio/wav", extension="wav"), - ), - ), -) - -MODEL_REQUIREMENTS: list[ModelManifest] = [] - - -def create_engine( - context: HostContext, - model_path: Path | None, - config: EngineConfig, -) -> Engine: - """Create a Kokoro engine instance. - - This function is the plugin entry point. It must be atomic: - succeed fully or raise EngineError and clean up. - - Args: - context: Host services (config dir, logger, http client). - model_path: Resolved model path, or None for default. - config: Engine initialization settings (device, etc.). - - Returns: - A fully initialized KokoroEngine instance. - - Raises: - EngineError: On failure. Cleans up partially created resources. - """ - try: - KPipeline = _load_kpipeline() - - # Determine repo_id from model_path or use default - repo_id = "hexgrad/Kokoro-82M" - if model_path is not None: - # If a specific model path is provided, use it as repo_id - repo_id = str(model_path) - - pipeline = KPipeline( - lang_code="a", # Default language code - repo_id=repo_id, - device=config.device, - ) - - engine = KokoroEngine(pipeline, lang_code="a") - return engine - except Exception as e: - from abogen.tts_plugin.errors import EngineError as EngineErrorClass - raise EngineErrorClass(f"Failed to create Kokoro engine: {e}") from e +from abogen.tts_plugin.types import EngineConfig + +from .engine import KokoroEngine + + +def _load_kpipeline() -> Any: + """Lazy-load Kokoro dependencies.""" + from kokoro import KPipeline # type: ignore[import-not-found] + return KPipeline + + +PLUGIN_MANIFEST = PluginManifest( + id="kokoro", + name="Kokoro", + version="0.9.4", + api_version="1.0", + description="Kokoro TTS engine - high quality multilingual text-to-speech", + author="Kokoro Team", + capabilities=("voice_list",), + requires=RequirementManifest( + internet=False, + ), + engine=EngineManifest( + voiceSources=( + VoiceSourceManifest( + id="builtin", + name="Built-in Voices", + type="list", + config={"voices": "See listVoices()"}, + ), + ), + parameters=( + ParameterManifest( + id="speed", + name="Speed", + description="Speech speed multiplier", + type="float", + default=1.0, + min=0.5, + max=2.0, + step=0.1, + ), + ), + audioFormats=( + AudioFormatManifest(mime="audio/wav", extension="wav"), + ), + ), + voices=( + VoiceManifest(id="af_alloy", name="Alloy", tags=("en", "female")), + VoiceManifest(id="af_aoede", name="Aoede", tags=("en", "female")), + VoiceManifest(id="af_bella", name="Bella", tags=("en", "female")), + VoiceManifest(id="af_heart", name="Heart", tags=("en", "female")), + VoiceManifest(id="af_jessica", name="Jessica", tags=("en", "female")), + VoiceManifest(id="af_kore", name="Kore", tags=("en", "female")), + VoiceManifest(id="af_nicole", name="Nicole", tags=("en", "female")), + VoiceManifest(id="af_nova", name="Nova", tags=("en", "female")), + VoiceManifest(id="af_river", name="River", tags=("en", "female")), + VoiceManifest(id="af_sarah", name="Sarah", tags=("en", "female")), + VoiceManifest(id="af_sky", name="Sky", tags=("en", "female")), + VoiceManifest(id="am_adam", name="Adam", tags=("en", "male")), + VoiceManifest(id="am_echo", name="Echo", tags=("en", "male")), + VoiceManifest(id="am_eric", name="Eric", tags=("en", "male")), + VoiceManifest(id="am_fenrir", name="Fenrir", tags=("en", "male")), + VoiceManifest(id="am_liam", name="Liam", tags=("en", "male")), + VoiceManifest(id="am_michael", name="Michael", tags=("en", "male")), + VoiceManifest(id="am_onyx", name="Onyx", tags=("en", "male")), + VoiceManifest(id="am_puck", name="Puck", tags=("en", "male")), + VoiceManifest(id="am_santa", name="Santa", tags=("en", "male")), + VoiceManifest(id="bf_alice", name="Alice", tags=("en", "female")), + VoiceManifest(id="bf_emma", name="Emma", tags=("en", "female")), + VoiceManifest(id="bf_isabella", name="Isabella", tags=("en", "female")), + VoiceManifest(id="bf_lily", name="Lily", tags=("en", "female")), + VoiceManifest(id="bm_daniel", name="Daniel", tags=("en", "male")), + VoiceManifest(id="bm_fable", name="Fable", tags=("en", "male")), + VoiceManifest(id="bm_george", name="George", tags=("en", "male")), + VoiceManifest(id="bm_lewis", name="Lewis", tags=("en", "male")), + VoiceManifest(id="ef_dora", name="Dora", tags=("es", "female")), + VoiceManifest(id="em_alex", name="Alex", tags=("es", "male")), + VoiceManifest(id="em_santa", name="Santa", tags=("es", "male")), + VoiceManifest(id="ff_siwis", name="Siwis", tags=("fr", "female")), + VoiceManifest(id="hf_alpha", name="Alpha", tags=("hi", "female")), + VoiceManifest(id="hf_beta", name="Beta", tags=("hi", "female")), + VoiceManifest(id="hm_omega", name="Omega", tags=("hi", "male")), + VoiceManifest(id="hm_psi", name="Psi", tags=("hi", "male")), + VoiceManifest(id="if_sara", name="Sara", tags=("it", "female")), + VoiceManifest(id="im_nicola", name="Nicola", tags=("it", "male")), + VoiceManifest(id="jf_alpha", name="Alpha", tags=("ja", "female")), + VoiceManifest(id="jf_gongitsune", name="Gongitsune", tags=("ja", "female")), + VoiceManifest(id="jf_nezumi", name="Nezumi", tags=("ja", "female")), + VoiceManifest(id="jf_tebukuro", name="Tebukuro", tags=("ja", "female")), + VoiceManifest(id="jm_kumo", name="Kumo", tags=("ja", "male")), + VoiceManifest(id="pf_dora", name="Dora", tags=("pt", "female")), + VoiceManifest(id="pm_alex", name="Alex", tags=("pt", "male")), + VoiceManifest(id="pm_santa", name="Santa", tags=("pt", "male")), + VoiceManifest(id="zf_xiaobei", name="Xiaobei", tags=("zh", "female")), + VoiceManifest(id="zf_xiaoni", name="Xiaoni", tags=("zh", "female")), + VoiceManifest(id="zf_xiaoxiao", name="Xiaoxiao", tags=("zh", "female")), + VoiceManifest(id="zf_xiaoyi", name="Xiaoyi", tags=("zh", "female")), + VoiceManifest(id="zm_yunjian", name="Yunjian", tags=("zh", "male")), + VoiceManifest(id="zm_yunxi", name="Yunxi", tags=("zh", "male")), + VoiceManifest(id="zm_yunxia", name="Yunxia", tags=("zh", "female")), + VoiceManifest(id="zm_yunyang", name="Yunyang", tags=("zh", "male")), + ), +) + +MODEL_REQUIREMENTS: list[ModelManifest] = [] + + +def create_engine( + context: HostContext, + model_path: Path | None, + config: EngineConfig, +) -> Engine: + """Create a Kokoro engine instance. + + This function is the plugin entry point. It must be atomic: + succeed fully or raise EngineError and clean up. + + Args: + context: Host services (config dir, logger, http client). + model_path: Resolved model path, or None for default. + config: Engine initialization settings (device, etc.). + + Returns: + A fully initialized KokoroEngine instance. + + Raises: + EngineError: On failure. Cleans up partially created resources. + """ + try: + KPipeline = _load_kpipeline() + + # Determine repo_id from model_path or use default + repo_id = "hexgrad/Kokoro-82M" + if model_path is not None: + # If a specific model path is provided, use it as repo_id + repo_id = str(model_path) + + pipeline = KPipeline( + lang_code=config.lang_code, + repo_id=repo_id, + device=config.device, + ) + + engine = KokoroEngine(pipeline) + return engine + except Exception as e: + from abogen.tts_plugin.errors import EngineError as EngineErrorClass + raise EngineErrorClass(f"Failed to create Kokoro engine: {e}") from e diff --git a/plugins/kokoro/engine.py b/plugins/kokoro/engine.py index 8cae628..3bb8fc9 100644 --- a/plugins/kokoro/engine.py +++ b/plugins/kokoro/engine.py @@ -1,186 +1,118 @@ -"""Kokoro Engine adapter for the TTS Plugin Architecture. - -This module adapts the existing Kokoro backend to the new Engine/EngineSession -protocol. It wraps the KokoroBackend without modifying it. -""" - -from __future__ import annotations - -import logging -from typing import Any, Iterator - -import numpy as np - -from abogen.tts_plugin.capabilities import VoiceLister -from abogen.tts_plugin.engine import Engine, EngineSession -from abogen.tts_plugin.errors import EngineError, InvalidInputError -from abogen.tts_plugin.manifest import VoiceManifest -from abogen.tts_plugin.types import ( - AudioFormat, - Duration, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - VoiceSelection, -) - -logger = logging.getLogger(__name__) - -# Kokoro voice list - source of truth -_KOKORO_VOICES = ( - "af_alloy", "af_aoede", "af_bella", "af_heart", "af_jessica", - "af_kore", "af_nicole", "af_nova", "af_river", "af_sarah", - "af_sky", "am_adam", "am_echo", "am_eric", "am_fenrir", - "am_liam", "am_michael", "am_onyx", "am_puck", "am_santa", - "bf_alice", "bf_emma", "bf_isabella", "bf_lily", - "bm_daniel", "bm_fable", "bm_george", "bm_lewis", - "ef_dora", "em_alex", "em_santa", - "ff_siwis", "hf_alpha", "hf_beta", "hm_omega", "hm_psi", - "if_sara", "im_nicola", - "jf_alpha", "jf_gongitsune", "jf_nezumi", "jf_tebukuro", "jm_kumo", - "pf_dora", "pm_alex", "pm_santa", - "zf_xiaobei", "zf_xiaoni", "zf_xiaoxiao", "zf_xiaoyi", - "zm_yunjian", "zm_yunxi", "zm_yunxia", "zm_yunyang", -) - -# Voice display names mapping -_VOICE_DISPLAY_NAMES: dict[str, str] = { - "af_alloy": "Alloy", "af_aoede": "Aoede", "af_bella": "Bella", - "af_heart": "Heart", "af_jessica": "Jessica", "af_kore": "Kore", - "af_nicole": "Nicole", "af_nova": "Nova", "af_river": "River", - "af_sarah": "Sarah", "af_sky": "Sky", "am_adam": "Adam", - "am_echo": "Echo", "am_eric": "Eric", "am_fenrir": "Fenrir", - "am_liam": "Liam", "am_michael": "Michael", "am_onyx": "Onyx", - "am_puck": "Puck", "am_santa": "Santa", "bf_alice": "Alice", - "bf_emma": "Emma", "bf_isabella": "Isabella", "bf_lily": "Lily", - "bm_daniel": "Daniel", "bm_fable": "Fable", "bm_george": "George", - "bm_lewis": "Lewis", "ef_dora": "Dora", "em_alex": "Alex", - "em_santa": "Santa", "ff_siwis": "Siwis", "hf_alpha": "Alpha", - "hf_beta": "Beta", "hm_omega": "Omega", "hm_psi": "Psi", - "if_sara": "Sara", "im_nicola": "Nicola", - "jf_alpha": "Alpha", "jf_gongitsune": "Gongitsune", - "jf_nezumi": "Nezumi", "jf_tebukuro": "Tebukuro", "jm_kumo": "Kumo", - "pf_dora": "Dora", "pm_alex": "Alex", "pm_santa": "Santa", - "zf_xiaobei": "Xiaobei", "zf_xiaoni": "Xiaoni", - "zf_xiaoxiao": "Xiaoxiao", "zf_xiaoyi": "Xiaoyi", - "zm_yunjian": "Yunjian", "zm_yunxi": "Yunxi", - "zm_yunxia": "Yunxia", "zm_yunyang": "Yunyang", -} - -# Sample rate for Kokoro audio -_KOKORO_SAMPLE_RATE = 24000 - - -class KokoroSession: - """EngineSession implementation for Kokoro. - - Owns mutable execution state for synthesis. - NOT thread-safe. - """ - - def __init__(self, pipeline: Any, lang_code: str) -> None: - self._pipeline = pipeline - self._lang_code = lang_code - self._disposed = False - - def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: - """Synthesize audio from text using Kokoro.""" - if self._disposed: - raise EngineError("Session disposed") - - try: - voice = request.voice.key - speed = request.parameters.values.get("speed", 1.0) - split_pattern = request.parameters.values.get("split_pattern", None) - - audio_parts: list[np.ndarray] = [] - for segment in self._pipeline( - request.text, - voice=voice, - speed=speed, - split_pattern=split_pattern, - ): - audio = segment.audio - if hasattr(audio, "numpy"): - audio = audio.numpy() - audio_parts.append(np.asarray(audio, dtype="float32")) - - if not audio_parts: - return SynthesizedAudio( - data=b"", - format=AudioFormat(mime="audio/wav", extension="wav"), - duration=Duration(seconds=0.0), - ) - - combined = np.concatenate(audio_parts).astype("float32", copy=False) - audio_bytes = combined.tobytes() - duration_seconds = len(combined) / _KOKORO_SAMPLE_RATE - - return SynthesizedAudio( - data=audio_bytes, - format=AudioFormat(mime="audio/wav", extension="wav"), - duration=Duration(seconds=duration_seconds), - ) - except EngineError: - raise - except Exception as e: - raise EngineError(f"Synthesis failed: {e}") from e - - def dispose(self) -> None: - """Release session resources. Idempotent.""" - self._disposed = True - - -class KokoroEngine: - """Engine implementation for Kokoro. - - Factory for KokoroSession instances. Stateless and thread-safe. - """ - - def __init__(self, pipeline: Any, lang_code: str) -> None: - self._pipeline = pipeline - self._lang_code = lang_code - self._disposed = False - - def createSession(self) -> KokoroSession: - """Create a new KokoroSession.""" - if self._disposed: - raise EngineError("Engine disposed") - return KokoroSession(self._pipeline, self._lang_code) - - def dispose(self) -> None: - """Release engine resources. Idempotent.""" - self._disposed = True - - def listVoices(self, sourceId: str) -> list[VoiceManifest]: - """List available Kokoro voices. Implements VoiceLister capability.""" - if self._disposed: - raise EngineError("Engine disposed") - return [ - VoiceManifest( - id=voice_id, - name=_VOICE_DISPLAY_NAMES.get(voice_id, voice_id), - tags=(_get_language_tag(voice_id), _get_gender_tag(voice_id)), - ) - for voice_id in _KOKORO_VOICES - ] - - -def _get_language_tag(voice_id: str) -> str: - """Extract language tag from voice ID.""" - prefix = voice_id.split("_")[0] - lang_map = { - "af": "en-us", "am": "en-us", "bf": "en-gb", "bm": "en-gb", - "ef": "es", "em": "es", "ff": "fr", "hf": "hi", "hm": "hi", - "if": "it", "im": "it", "jf": "ja", "jm": "ja", - "pf": "pt", "pm": "pt", "zf": "zh", "zm": "zh", - } - return lang_map.get(prefix, "unknown") - - -def _get_gender_tag(voice_id: str) -> str: - """Extract gender tag from voice ID.""" - prefix = voice_id.split("_")[0] - if prefix.startswith("a") or prefix.startswith("b") or prefix.startswith("e"): - return "female" if prefix[1] == "f" else "male" - return "unknown" +"""Kokoro Engine adapter for the TTS Plugin Architecture. + +This module adapts the existing Kokoro backend to the new Engine/EngineSession +protocol. It wraps the KokoroBackend without modifying it. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import numpy as np + +from abogen.tts_plugin.capabilities import VoiceLister +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import EngineError +from abogen.tts_plugin.manifest import VoiceManifest +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + SynthesisRequest, + SynthesizedAudio, +) + +logger = logging.getLogger(__name__) + +# Sample rate for Kokoro audio +_KOKORO_SAMPLE_RATE = 24000 + + +class KokoroSession: + """EngineSession implementation for Kokoro. + + Owns mutable execution state for synthesis. + NOT thread-safe. + """ + + def __init__(self, pipeline: Any) -> None: + self._pipeline = pipeline + self._disposed = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + """Synthesize audio from text using Kokoro.""" + if self._disposed: + raise EngineError("Session disposed") + + try: + voice = request.voice.key + speed = request.parameters.values.get("speed", 1.0) + split_pattern = request.parameters.values.get("split_pattern", None) + + audio_parts: list[np.ndarray] = [] + for segment in self._pipeline( + request.text, + voice=voice, + speed=speed, + split_pattern=split_pattern, + ): + audio = segment.audio + if hasattr(audio, "numpy"): + audio = audio.numpy() + audio_parts.append(np.asarray(audio, dtype="float32")) + + if not audio_parts: + return SynthesizedAudio( + data=b"", + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=0.0), + ) + + combined = np.concatenate(audio_parts).astype("float32", copy=False) + audio_bytes = combined.tobytes() + duration_seconds = len(combined) / _KOKORO_SAMPLE_RATE + + return SynthesizedAudio( + data=audio_bytes, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=duration_seconds), + ) + except EngineError: + raise + except Exception as e: + raise EngineError(f"Synthesis failed: {e}") from e + + def dispose(self) -> None: + """Release session resources. Idempotent.""" + self._disposed = True + + +class KokoroEngine: + """Engine implementation for Kokoro. + + Factory for KokoroSession instances. Stateless and thread-safe. + """ + + def __init__(self, pipeline: Any) -> None: + self._pipeline = pipeline + self._disposed = False + + def createSession(self) -> KokoroSession: + """Create a new KokoroSession.""" + if self._disposed: + raise EngineError("Engine disposed") + return KokoroSession(self._pipeline) + + def dispose(self) -> None: + """Release engine resources. Idempotent.""" + self._disposed = True + + def listVoices(self, sourceId: str) -> list[VoiceManifest]: + """List available Kokoro voices. Implements VoiceLister capability. + + Note: Static voices are declared in the plugin manifest. + This method is a fallback for dynamic plugins. + """ + if self._disposed: + raise EngineError("Engine disposed") + return [] diff --git a/plugins/supertonic/__init__.py b/plugins/supertonic/__init__.py index 51e0ab6..795117d 100644 --- a/plugins/supertonic/__init__.py +++ b/plugins/supertonic/__init__.py @@ -1,123 +1,136 @@ -"""SuperTonic TTS Plugin for the TTS Plugin Architecture. - -This plugin provides a SuperTonic-based TTS engine that implements the -Plugin API contract. It wraps the existing SuperTonic backend in the -new Engine/EngineSession architecture. - -Exports: - - PLUGIN_MANIFEST: PluginManifest - - MODEL_REQUIREMENTS: list[ModelManifest] - - create_engine: Factory function -""" - -from __future__ import annotations - -from pathlib import Path -from typing import Any - -from abogen.tts_plugin.engine import Engine -from abogen.tts_plugin.host_context import HostContext -from abogen.tts_plugin.manifest import ( - AudioFormatManifest, - EngineManifest, - ModelManifest, - ParameterManifest, - PluginManifest, - RequirementManifest, - VoiceSourceManifest, -) -from abogen.tts_plugin.types import EngineConfig - -from .engine import SuperTonicEngine - - -def _load_supertonic_pipeline(sample_rate: int = 24000, auto_download: bool = True, total_steps: int = 5) -> Any: - """Lazy-load SuperTonic dependencies and create pipeline.""" - from plugins.supertonic.pipeline import SupertonicPipeline - - return SupertonicPipeline( - sample_rate=sample_rate, - auto_download=auto_download, - total_steps=total_steps, - ) - - -PLUGIN_MANIFEST = PluginManifest( - id="supertonic", - name="SuperTonic", - version="0.1.0", - api_version="1.0", - description="SuperTonic TTS engine - fast high-quality text-to-speech", - author="SuperTonic Team", - capabilities=("voice_list",), - requires=RequirementManifest( - internet=False, - ), - engine=EngineManifest( - voiceSources=( - VoiceSourceManifest( - id="builtin", - name="Built-in Voices", - type="list", - config={"voices": "See listVoices()"}, - ), - ), - parameters=( - ParameterManifest( - id="speed", - name="Speed", - description="Speech speed multiplier", - type="float", - default=1.0, - min=0.7, - max=2.0, - step=0.1, - ), - ParameterManifest( - id="total_steps", - name="Quality Steps", - description="Inference steps (higher = better quality, slower)", - type="int", - default=5, - min=2, - max=15, - step=1, - ), - ), - audioFormats=( - AudioFormatManifest(mime="audio/wav", extension="wav"), - ), - ), -) - -MODEL_REQUIREMENTS: list[ModelManifest] = [] - - -def create_engine( - context: HostContext, - model_path: Path | None, - config: EngineConfig, -) -> Engine: - """Create a SuperTonic engine instance. - - This function is the plugin entry point. It must be atomic: - succeed fully or raise EngineError and clean up. - - Args: - context: Host services (config dir, logger, http client). - model_path: Resolved model path, or None for default. - config: Engine initialization settings (device, etc.). - - Returns: - A fully initialized SuperTonicEngine instance. - - Raises: - EngineError: On failure. Cleans up partially created resources. - """ - try: - pipeline = _load_supertonic_pipeline() - engine = SuperTonicEngine(pipeline) - return engine - except Exception as e: - from abogen.tts_plugin.errors import EngineError as EngineErrorClass - raise EngineErrorClass(f"Failed to create SuperTonic engine: {e}") from e +"""SuperTonic TTS Plugin for the TTS Plugin Architecture. + +This plugin provides a SuperTonic-based TTS engine that implements the +Plugin API contract. It wraps the existing SuperTonic backend in the +new Engine/EngineSession architecture. + +Exports: + - PLUGIN_MANIFEST: PluginManifest + - MODEL_REQUIREMENTS: list[ModelManifest] + - create_engine: Factory function +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from abogen.tts_plugin.engine import Engine +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + ModelManifest, + ParameterManifest, + PluginManifest, + RequirementManifest, + VoiceManifest, + VoiceSourceManifest, +) +from abogen.tts_plugin.types import EngineConfig + +from .engine import SuperTonicEngine + + +def _load_supertonic_pipeline() -> Any: + """Lazy-load SuperTonic dependencies and create pipeline.""" + from plugins.supertonic.pipeline import SupertonicPipeline + + return SupertonicPipeline( + sample_rate=24000, + auto_download=True, + total_steps=5, + ) + + +PLUGIN_MANIFEST = PluginManifest( + id="supertonic", + name="SuperTonic", + version="0.1.0", + api_version="1.0", + description="SuperTonic TTS engine - fast high-quality text-to-speech", + author="SuperTonic Team", + capabilities=("voice_list",), + requires=RequirementManifest( + internet=False, + ), + engine=EngineManifest( + voiceSources=( + VoiceSourceManifest( + id="builtin", + name="Built-in Voices", + type="list", + config={"voices": "See listVoices()"}, + ), + ), + parameters=( + ParameterManifest( + id="speed", + name="Speed", + description="Speech speed multiplier", + type="float", + default=1.0, + min=0.7, + max=2.0, + step=0.1, + ), + ParameterManifest( + id="total_steps", + name="Quality Steps", + description="Inference steps (higher = better quality, slower)", + type="int", + default=5, + min=2, + max=15, + step=1, + ), + ), + audioFormats=( + AudioFormatManifest(mime="audio/wav", extension="wav"), + ), + ), + voices=( + VoiceManifest(id="M1", name="Male 1", tags=("male",)), + VoiceManifest(id="M2", name="Male 2", tags=("male",)), + VoiceManifest(id="M3", name="Male 3", tags=("male",)), + VoiceManifest(id="M4", name="Male 4", tags=("male",)), + VoiceManifest(id="M5", name="Male 5", tags=("male",)), + VoiceManifest(id="F1", name="Female 1", tags=("female",)), + VoiceManifest(id="F2", name="Female 2", tags=("female",)), + VoiceManifest(id="F3", name="Female 3", tags=("female",)), + VoiceManifest(id="F4", name="Female 4", tags=("female",)), + VoiceManifest(id="F5", name="Female 5", tags=("female",)), + ), +) + +MODEL_REQUIREMENTS: list[ModelManifest] = [] + + +def create_engine( + context: HostContext, + model_path: Path | None, + config: EngineConfig, +) -> Engine: + """Create a SuperTonic engine instance. + + This function is the plugin entry point. It must be atomic: + succeed fully or raise EngineError and clean up. + + Args: + context: Host services (config dir, logger, http client). + model_path: Resolved model path, or None for default. + config: Engine initialization settings (device, etc.). + + Returns: + A fully initialized SuperTonicEngine instance. + + Raises: + EngineError: On failure. Cleans up partially created resources. + """ + try: + pipeline = _load_supertonic_pipeline() + engine = SuperTonicEngine(pipeline) + return engine + except Exception as e: + from abogen.tts_plugin.errors import EngineError as EngineErrorClass + raise EngineErrorClass(f"Failed to create SuperTonic engine: {e}") from e diff --git a/plugins/supertonic/engine.py b/plugins/supertonic/engine.py index a410af9..f9e0c6f 100644 --- a/plugins/supertonic/engine.py +++ b/plugins/supertonic/engine.py @@ -1,156 +1,125 @@ -"""SuperTonic Engine adapter for the TTS Plugin Architecture. - -This module adapts the existing SuperTonic backend to the new Engine/EngineSession -protocol. It wraps the SupertonicPipeline without modifying it. -""" - -from __future__ import annotations - -import io -import logging -from typing import Any, Optional - -import numpy as np - -from abogen.tts_plugin.capabilities import VoiceLister -from abogen.tts_plugin.engine import Engine, EngineSession -from abogen.tts_plugin.errors import EngineError, InvalidInputError -from abogen.tts_plugin.manifest import VoiceManifest -from abogen.tts_plugin.types import ( - AudioFormat, - Duration, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - VoiceSelection, -) - -logger = logging.getLogger(__name__) - -# SuperTonic voice list - source of truth -_SUPERTONIC_VOICES = ("M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5") - -# Voice display names mapping -_VOICE_DISPLAY_NAMES: dict[str, str] = { - "M1": "Male 1", - "M2": "Male 2", - "M3": "Male 3", - "M4": "Male 4", - "M5": "Male 5", - "F1": "Female 1", - "F2": "Female 2", - "F3": "Female 3", - "F4": "Female 4", - "F5": "Female 5", -} - -# Sample rate for SuperTonic audio -_SUPERTONIC_SAMPLE_RATE = 24000 - - -class SuperTonicSession: - """EngineSession implementation for SuperTonic. - - Owns mutable execution state for synthesis. - NOT thread-safe. - """ - - def __init__(self, pipeline: Any) -> None: - self._pipeline = pipeline - self._disposed = False - - def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: - """Synthesize audio from text using SuperTonic.""" - if self._disposed: - raise EngineError("Session disposed") - - try: - import soundfile as sf - - voice = request.voice.key - speed = float(request.parameters.values.get("speed", 1.0)) - total_steps = request.parameters.values.get("total_steps", None) - split_pattern = request.parameters.values.get("split_pattern", None) - - if total_steps is not None: - total_steps = int(total_steps) - - audio_parts: list[np.ndarray] = [] - for segment in self._pipeline( - request.text, - voice=voice, - speed=speed, - split_pattern=split_pattern, - total_steps=total_steps, - ): - audio_parts.append(segment.audio) - - if not audio_parts: - return SynthesizedAudio( - data=b"", - format=AudioFormat(mime="audio/wav", extension="wav"), - duration=Duration(seconds=0.0), - ) - - combined = np.concatenate(audio_parts).astype("float32", copy=False) - buf = io.BytesIO() - sf.write(buf, combined, self._pipeline.sample_rate, format="WAV") - audio_bytes = buf.getvalue() - duration_seconds = len(combined) / self._pipeline.sample_rate - - return SynthesizedAudio( - data=audio_bytes, - format=AudioFormat(mime="audio/wav", extension="wav"), - duration=Duration(seconds=duration_seconds), - ) - except EngineError: - raise - except Exception as e: - raise EngineError(f"Synthesis failed: {e}") from e - - def dispose(self) -> None: - """Release session resources. Idempotent.""" - self._disposed = True - - -class SuperTonicEngine: - """Engine implementation for SuperTonic. - - Factory for SuperTonicSession instances. Stateless and thread-safe. - """ - - def __init__(self, pipeline: Any) -> None: - self._pipeline = pipeline - self._disposed = False - - def createSession(self) -> SuperTonicSession: - """Create a new SuperTonicSession.""" - if self._disposed: - raise EngineError("Engine disposed") - return SuperTonicSession(self._pipeline) - - def dispose(self) -> None: - """Release engine resources. Idempotent.""" - self._disposed = True - - def listVoices(self, sourceId: str) -> list[VoiceManifest]: - """List available SuperTonic voices. Implements VoiceLister capability.""" - if self._disposed: - raise EngineError("Engine disposed") - return [ - VoiceManifest( - id=voice_id, - name=_VOICE_DISPLAY_NAMES.get(voice_id, voice_id), - tags=(_get_gender_tag(voice_id),), - ) - for voice_id in _SUPERTONIC_VOICES - ] - - -def _get_gender_tag(voice_id: str) -> str: - """Extract gender tag from voice ID.""" - if voice_id.startswith("M"): - return "male" - elif voice_id.startswith("F"): - return "female" - return "unknown" +"""SuperTonic Engine adapter for the TTS Plugin Architecture. + +This module adapts the existing SuperTonic backend to the new Engine/EngineSession +protocol. It wraps the SupertonicPipeline without modifying it. +""" + +from __future__ import annotations + +import io +import logging +from typing import Any + +import numpy as np + +from abogen.tts_plugin.capabilities import VoiceLister +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import EngineError +from abogen.tts_plugin.manifest import VoiceManifest +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + SynthesisRequest, + SynthesizedAudio, +) + +logger = logging.getLogger(__name__) + +# Sample rate for SuperTonic audio +_SUPERTONIC_SAMPLE_RATE = 24000 + + +class SuperTonicSession: + """EngineSession implementation for SuperTonic. + + Owns mutable execution state for synthesis. + NOT thread-safe. + """ + + def __init__(self, pipeline: Any) -> None: + self._pipeline = pipeline + self._disposed = False + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + """Synthesize audio from text using SuperTonic.""" + if self._disposed: + raise EngineError("Session disposed") + + try: + import soundfile as sf + + voice = request.voice.key + speed = float(request.parameters.values.get("speed", 1.0)) + total_steps = request.parameters.values.get("total_steps", None) + split_pattern = request.parameters.values.get("split_pattern", None) + + if total_steps is not None: + total_steps = int(total_steps) + + audio_parts: list[np.ndarray] = [] + for segment in self._pipeline( + request.text, + voice=voice, + speed=speed, + split_pattern=split_pattern, + total_steps=total_steps, + ): + audio_parts.append(segment.audio) + + if not audio_parts: + return SynthesizedAudio( + data=b"", + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=0.0), + ) + + combined = np.concatenate(audio_parts).astype("float32", copy=False) + buf = io.BytesIO() + sf.write(buf, combined, self._pipeline.sample_rate, format="WAV") + audio_bytes = buf.getvalue() + duration_seconds = len(combined) / self._pipeline.sample_rate + + return SynthesizedAudio( + data=audio_bytes, + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=duration_seconds), + ) + except EngineError: + raise + except Exception as e: + raise EngineError(f"Synthesis failed: {e}") from e + + def dispose(self) -> None: + """Release session resources. Idempotent.""" + self._disposed = True + + +class SuperTonicEngine: + """Engine implementation for SuperTonic. + + Factory for SuperTonicSession instances. Stateless and thread-safe. + """ + + def __init__(self, pipeline: Any) -> None: + self._pipeline = pipeline + self._disposed = False + + def createSession(self) -> SuperTonicSession: + """Create a new SuperTonicSession.""" + if self._disposed: + raise EngineError("Engine disposed") + return SuperTonicSession(self._pipeline) + + def dispose(self) -> None: + """Release engine resources. Idempotent.""" + self._disposed = True + + def listVoices(self, sourceId: str) -> list[VoiceManifest]: + """List available SuperTonic voices. Implements VoiceLister capability. + + Note: Static voice catalog is declared in plugin manifest. + This method is retained for VoiceLister interface compliance. + """ + if self._disposed: + raise EngineError("Engine disposed") + return [] diff --git a/tests/test_kokoro_plugin.py b/tests/test_kokoro_plugin.py index 3d617f0..79566a1 100644 --- a/tests/test_kokoro_plugin.py +++ b/tests/test_kokoro_plugin.py @@ -1,188 +1,196 @@ -"""Tests for the Kokoro TTS Plugin. - -These tests verify that the Kokoro plugin: -- Loads correctly through the Plugin Loader -- Has a valid manifest -- Creates a valid Engine -- Satisfies the Engine/EngineSession contract (via EngineContractMixin) -- Implements VoiceLister capability -""" - -from __future__ import annotations - -import logging -from pathlib import Path -from typing import Any - -import pytest - -from abogen.tts_plugin.engine import Engine, EngineSession -from abogen.tts_plugin.host_context import HostContext -from abogen.tts_plugin.loader import load_plugin_from_dir -from abogen.tts_plugin.manifest import PluginManifest -from abogen.tts_plugin.types import ( - AudioFormat, - EngineConfig, - ParameterValues, - SynthesisRequest, - VoiceSelection, -) - -from tests.contracts.engine_contract import EngineContractMixin - - -# ────────────────────────────────────────────────────────────── -# Helpers -# ────────────────────────────────────────────────────────────── - -def _kokoro_available() -> bool: - try: - from kokoro import KPipeline # type: ignore[import-not-found] - return True - except ImportError: - return False - - -def _make_mock_engine() -> Any: - from plugins.kokoro.engine import KokoroEngine - - class MockPipeline: - def __call__(self, text, voice, speed, split_pattern=None): - class MockSegment: - def __init__(self): - self.audio = MockAudio() - class MockAudio: - def numpy(self): - import numpy as np - return np.zeros(24000, dtype="float32") - return [MockSegment()] - - return KokoroEngine(MockPipeline(), "a") - - -# ────────────────────────────────────────────────────────────── -# Fixtures -# ────────────────────────────────────────────────────────────── - -@pytest.fixture -def kokoro_plugin_dir() -> Path: - return Path(__file__).parent.parent / "plugins" / "kokoro" - - -@pytest.fixture -def host_context(tmp_path: Path) -> HostContext: - class FakeHttpClient: - def get(self, url: str, **kwargs: object) -> object: - return None - def post(self, url: str, **kwargs: object) -> object: - return None - - return HostContext( - config_dir=tmp_path, - logger=logging.getLogger("test"), - http_client=FakeHttpClient(), - ) - - -@pytest.fixture -def engine() -> Engine: - return _make_mock_engine() - - -# ────────────────────────────────────────────────────────────── -# Plugin Loading Tests -# ────────────────────────────────────────────────────────────── - -class TestKokoroPluginLoading: - - def test_plugin_loads_successfully(self, kokoro_plugin_dir: Path) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - assert result.manifest is not None - assert result.create_engine is not None - - def test_plugin_has_valid_manifest(self, kokoro_plugin_dir: Path) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - manifest = result.manifest - assert isinstance(manifest, PluginManifest) - assert manifest.id == "kokoro" - assert manifest.name == "Kokoro" - assert manifest.api_version == "1.0" - - def test_plugin_has_model_requirements(self, kokoro_plugin_dir: Path) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - assert result.model_requirements is not None - assert isinstance(result.model_requirements, tuple) - - def test_plugin_manifest_capabilities(self, kokoro_plugin_dir: Path) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - assert "voice_list" in result.manifest.capabilities - - def test_plugin_manifest_engine(self, kokoro_plugin_dir: Path) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - engine_manifest = result.manifest.engine - assert len(engine_manifest.voiceSources) > 0 - assert len(engine_manifest.audioFormats) > 0 - assert len(engine_manifest.parameters) > 0 - - -# ────────────────────────────────────────────────────────────── -# Engine Creation (real backend, skipped if not installed) -# ────────────────────────────────────────────────────────────── - -class TestKokoroEngineCreation: - - @pytest.mark.skipif(not _kokoro_available(), reason="Kokoro not installed") - def test_create_engine(self, kokoro_plugin_dir: Path, host_context: HostContext) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - engine = result.create_engine(host_context, None, EngineConfig()) - assert isinstance(engine, Engine) - engine.dispose() - - @pytest.mark.skipif(not _kokoro_available(), reason="Kokoro not installed") - def test_engine_satisfies_protocol(self, kokoro_plugin_dir: Path, host_context: HostContext) -> None: - result = load_plugin_from_dir(kokoro_plugin_dir) - assert result.success is True - engine = result.create_engine(host_context, None, EngineConfig()) - assert isinstance(engine, Engine) - engine.dispose() - - -# ────────────────────────────────────────────────────────────── -# Engine / Session Contract (inherited from base) -# ────────────────────────────────────────────────────────────── - -class TestKokoroEngineContract(EngineContractMixin): - """Every test from EngineContractMixin runs against KokoroEngine.""" - - @pytest.fixture - def default_voice(self) -> str: - return "af_nova" - - -# ────────────────────────────────────────────────────────────── -# VoiceLister Tests -# ────────────────────────────────────────────────────────────── - -class TestKokoroVoiceLister: - - def test_list_voices(self) -> None: - engine = _make_mock_engine() - voices = engine.listVoices("builtin") - assert len(voices) > 0 - assert all(hasattr(v, "id") for v in voices) - assert all(hasattr(v, "name") for v in voices) - engine.dispose() - - def test_voices_have_tags(self) -> None: - engine = _make_mock_engine() - voices = engine.listVoices("builtin") - for voice in voices: - assert isinstance(voice.tags, tuple) - assert len(voice.tags) > 0 - engine.dispose() +"""Tests for the Kokoro TTS Plugin. + +These tests verify that the Kokoro plugin: +- Loads correctly through the Plugin Loader +- Has a valid manifest +- Creates a valid Engine +- Satisfies the Engine/EngineSession contract (via EngineContractMixin) +- Implements VoiceLister capability +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any + +import pytest + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.host_context import HostContext +from abogen.tts_plugin.loader import load_plugin_from_dir +from abogen.tts_plugin.manifest import PluginManifest +from abogen.tts_plugin.types import ( + AudioFormat, + EngineConfig, + ParameterValues, + SynthesisRequest, + VoiceSelection, +) + +from tests.contracts.engine_contract import EngineContractMixin + + +# ────────────────────────────────────────────────────────────── +# Helpers +# ────────────────────────────────────────────────────────────── + +def _kokoro_available() -> bool: + try: + from kokoro import KPipeline # type: ignore[import-not-found] + return True + except ImportError: + return False + + +def _make_mock_engine() -> Any: + from plugins.kokoro.engine import KokoroEngine + + class MockPipeline: + def __call__(self, text, voice, speed, split_pattern=None): + class MockSegment: + def __init__(self): + self.audio = MockAudio() + class MockAudio: + def numpy(self): + import numpy as np + return np.zeros(24000, dtype="float32") + return [MockSegment()] + + engine = KokoroEngine(MockPipeline()) + + # Override listVoices for testing (real engine reads from manifest) + from abogen.tts_plugin.manifest import VoiceManifest + engine.listVoices = lambda source_id: [ + VoiceManifest(id="test_voice_1", name="Test Voice 1", tags=("en",)), + VoiceManifest(id="test_voice_2", name="Test Voice 2", tags=("es",)), + ] + return engine + + +# ────────────────────────────────────────────────────────────── +# Fixtures +# ────────────────────────────────────────────────────────────── + +@pytest.fixture +def kokoro_plugin_dir() -> Path: + return Path(__file__).parent.parent / "plugins" / "kokoro" + + +@pytest.fixture +def host_context(tmp_path: Path) -> HostContext: + class FakeHttpClient: + def get(self, url: str, **kwargs: object) -> object: + return None + def post(self, url: str, **kwargs: object) -> object: + return None + + return HostContext( + config_dir=tmp_path, + logger=logging.getLogger("test"), + http_client=FakeHttpClient(), + ) + + +@pytest.fixture +def engine() -> Engine: + return _make_mock_engine() + + +# ────────────────────────────────────────────────────────────── +# Plugin Loading Tests +# ────────────────────────────────────────────────────────────── + +class TestKokoroPluginLoading: + + def test_plugin_loads_successfully(self, kokoro_plugin_dir: Path) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + assert result.manifest is not None + assert result.create_engine is not None + + def test_plugin_has_valid_manifest(self, kokoro_plugin_dir: Path) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + manifest = result.manifest + assert isinstance(manifest, PluginManifest) + assert manifest.id == "kokoro" + assert manifest.name == "Kokoro" + assert manifest.api_version == "1.0" + + def test_plugin_has_model_requirements(self, kokoro_plugin_dir: Path) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + assert result.model_requirements is not None + assert isinstance(result.model_requirements, tuple) + + def test_plugin_manifest_capabilities(self, kokoro_plugin_dir: Path) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + assert "voice_list" in result.manifest.capabilities + + def test_plugin_manifest_engine(self, kokoro_plugin_dir: Path) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + engine_manifest = result.manifest.engine + assert len(engine_manifest.voiceSources) > 0 + assert len(engine_manifest.audioFormats) > 0 + assert len(engine_manifest.parameters) > 0 + + +# ────────────────────────────────────────────────────────────── +# Engine Creation (real backend, skipped if not installed) +# ────────────────────────────────────────────────────────────── + +class TestKokoroEngineCreation: + + @pytest.mark.skipif(not _kokoro_available(), reason="Kokoro not installed") + def test_create_engine(self, kokoro_plugin_dir: Path, host_context: HostContext) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + engine = result.create_engine(host_context, None, EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + @pytest.mark.skipif(not _kokoro_available(), reason="Kokoro not installed") + def test_engine_satisfies_protocol(self, kokoro_plugin_dir: Path, host_context: HostContext) -> None: + result = load_plugin_from_dir(kokoro_plugin_dir) + assert result.success is True + engine = result.create_engine(host_context, None, EngineConfig()) + assert isinstance(engine, Engine) + engine.dispose() + + +# ────────────────────────────────────────────────────────────── +# Engine / Session Contract (inherited from base) +# ────────────────────────────────────────────────────────────── + +class TestKokoroEngineContract(EngineContractMixin): + """Every test from EngineContractMixin runs against KokoroEngine.""" + + @pytest.fixture + def default_voice(self) -> str: + return "af_nova" + + +# ────────────────────────────────────────────────────────────── +# VoiceLister Tests +# ────────────────────────────────────────────────────────────── + +class TestKokoroVoiceLister: + + def test_list_voices(self) -> None: + engine = _make_mock_engine() + voices = engine.listVoices("builtin") + assert len(voices) > 0 + assert all(hasattr(v, "id") for v in voices) + assert all(hasattr(v, "name") for v in voices) + engine.dispose() + + def test_voices_have_tags(self) -> None: + engine = _make_mock_engine() + voices = engine.listVoices("builtin") + for voice in voices: + assert isinstance(voice.tags, tuple) + assert len(voice.tags) > 0 + engine.dispose()