feat: add static voice catalog to PluginManifest

- Add  to PluginManifest
  - None = not declared (use VoiceLister fallback)
  - () = explicitly no static voices
  - Non-empty = static catalog available without Engine instantiation

- Update get_voices() to check manifest first, fall back to Engine
- Declare 54 Kokoro voices and 10 SuperTonic voices in manifests
- Remove hardcoded voice lists from engine.py files
- Engine.listVoices() now returns [] (manifest is source of truth)

- Clean up dead create_pipeline() kwargs (sample_rate, auto_download, total_steps)
  - SuperTonic plugin uses internal defaults
  - total_steps is per-request parameter via Pipeline.__call__() kwargs

- Add clear_preview_pipelines() for resource cleanup
- Fix test mocks to override listVoices()
- Update Architecture Amendment #1 doc
This commit is contained in:
Artem Akymenko
2026-07-12 16:20:16 +03:00
parent 5d1e7165bb
commit 735098d7cd
10 changed files with 1101 additions and 979 deletions
+177 -120
View File
@@ -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
+118 -186
View File
@@ -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 []
+136 -123
View File
@@ -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
+125 -156
View File
@@ -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 []