Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-07-18 05:40:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: add Kokoro plugin vertical slice

Browse Source 
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.
This commit is contained in:
Artem Akymenko
2026-07-12 16:20:05 +03:00
parent 6eda8516cc
commit d129b0abe8
3 changed files with 548 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/kokoro/__init__.py
+121
View File
@@ -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
plugins/kokoro/engine.py
+186
View File
@@ -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"
tests/test_kokoro_plugin.py
+241
View File
@@ -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()