From d129b0abe84ac004b11fc333a7760344d8c269f2 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 09:54:35 +0000 Subject: [PATCH] feat: add Kokoro plugin vertical slice Implement first TTS plugin using new Plugin Architecture: - plugins/kokoro/: Plugin package with manifest and entry point - plugins/kokoro/engine.py: KokoroEngine and KokoroSession adapters - Wraps existing KokoroBackend without modifying it - Implements VoiceLister capability - Satisfies Engine/EngineSession protocol - Passes all 163 contract tests Tests: - Plugin loading through Plugin Loader - Manifest validation - Engine creation and lifecycle - Session synthesis and dispose - VoiceLister capability 15 new tests for Kokoro plugin. --- plugins/kokoro/__init__.py | 121 ++++++++++++++++++ plugins/kokoro/engine.py | 186 ++++++++++++++++++++++++++++ tests/test_kokoro_plugin.py | 241 ++++++++++++++++++++++++++++++++++++ 3 files changed, 548 insertions(+) create mode 100644 plugins/kokoro/__init__.py create mode 100644 plugins/kokoro/engine.py create mode 100644 tests/test_kokoro_plugin.py diff --git a/plugins/kokoro/__init__.py b/plugins/kokoro/__init__.py new file mode 100644 index 0000000..915e054 --- /dev/null +++ b/plugins/kokoro/__init__.py @@ -0,0 +1,121 @@ +"""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, +) +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 diff --git a/plugins/kokoro/engine.py b/plugins/kokoro/engine.py new file mode 100644 index 0000000..8cae628 --- /dev/null +++ b/plugins/kokoro/engine.py @@ -0,0 +1,186 @@ +"""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" diff --git a/tests/test_kokoro_plugin.py b/tests/test_kokoro_plugin.py new file mode 100644 index 0000000..87d6a6b --- /dev/null +++ b/tests/test_kokoro_plugin.py @@ -0,0 +1,241 @@ +"""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 +- 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, +) + + +# ────────────────────────────────────────────────────────────── +# Path fixtures +# ────────────────────────────────────────────────────────────── + +@pytest.fixture +def kokoro_plugin_dir() -> Path: + return Path(__file__).parent.parent / "plugins" / "kokoro" + + +def _kokoro_available() -> bool: + """Check if Kokoro is available.""" + try: + from kokoro import KPipeline # type: ignore[import-not-found] + return True + except ImportError: + return False + + +@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(), + ) + + +# ────────────────────────────────────────────────────────────── +# Plugin Loading Tests +# ────────────────────────────────────────────────────────────── + +class TestKokoroPluginLoading: + """Test that Kokoro plugin loads correctly through the Loader.""" + + 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 + manifest = result.manifest + assert "voice_list" in 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 Tests +# ────────────────────────────────────────────────────────────── + +class TestKokoroEngineCreation: + """Test that Kokoro Engine can be created.""" + + @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 Protocol Tests (using mock pipeline) +# ────────────────────────────────────────────────────────────── + +class TestKokoroEngineProtocol: + """Test Kokoro Engine protocol compliance using mock pipeline.""" + + def _create_engine_with_mock(self) -> Any: + """Create engine with mock pipeline for testing.""" + 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") + + def test_create_session(self) -> None: + engine = self._create_engine_with_mock() + session = engine.createSession() + assert isinstance(session, EngineSession) + engine.dispose() + + def test_dispose_idempotent(self) -> None: + engine = self._create_engine_with_mock() + engine.dispose() + engine.dispose() # Should not raise + + def test_create_session_after_dispose_raises(self) -> None: + from abogen.tts_plugin.errors import EngineError + engine = self._create_engine_with_mock() + engine.dispose() + with pytest.raises(EngineError): + engine.createSession() + + def test_session_synthesize(self) -> None: + engine = self._create_engine_with_mock() + session = engine.createSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert result.data is not None + assert len(result.data) > 0 + session.dispose() + engine.dispose() + + def test_session_dispose_idempotent(self) -> None: + engine = self._create_engine_with_mock() + session = engine.createSession() + session.dispose() + session.dispose() # Should not raise + engine.dispose() + + def test_session_synthesize_after_dispose_raises(self) -> None: + from abogen.tts_plugin.errors import EngineError + engine = self._create_engine_with_mock() + session = engine.createSession() + session.dispose() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="af_nova"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + with pytest.raises(EngineError): + session.synthesize(request) + engine.dispose() + + +# ────────────────────────────────────────────────────────────── +# VoiceLister Tests +# ────────────────────────────────────────────────────────────── + +class TestKokoroVoiceLister: + """Test Kokoro VoiceLister capability.""" + + def test_list_voices(self) -> None: + engine = TestKokoroEngineProtocol._create_engine_with_mock(TestKokoroEngineProtocol) + 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 = TestKokoroEngineProtocol._create_engine_with_mock(TestKokoroEngineProtocol) + voices = engine.listVoices("builtin") + for voice in voices: + assert isinstance(voice.tags, tuple) + assert len(voice.tags) > 0 + engine.dispose()