From 6284c501ed8ad392b5b6e94818e38f69020fca90 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 14:06:40 +0000 Subject: [PATCH] feat: add SuperTonic TTS plugin - Add plugins/supertonic/ with Engine and EngineSession implementations - Reuse existing SupertonicPipeline from abogen.tts_backends.supertonic - Implement VoiceLister capability (M1-M5, F1-F5 voices) - Declare no streaming support via capabilities - Add 28 tests: plugin loading, protocol compliance, lifecycle, voice listing, parameters, errors - All 237 tests pass --- plugins/supertonic/__init__.py | 123 ++++++++++ plugins/supertonic/engine.py | 156 ++++++++++++ tests/test_supertonic_plugin.py | 415 ++++++++++++++++++++++++++++++++ 3 files changed, 694 insertions(+) create mode 100644 plugins/supertonic/__init__.py create mode 100644 plugins/supertonic/engine.py create mode 100644 tests/test_supertonic_plugin.py diff --git a/plugins/supertonic/__init__.py b/plugins/supertonic/__init__.py new file mode 100644 index 0000000..abd92cc --- /dev/null +++ b/plugins/supertonic/__init__.py @@ -0,0 +1,123 @@ +"""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 abogen.tts_backends.supertonic 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 diff --git a/plugins/supertonic/engine.py b/plugins/supertonic/engine.py new file mode 100644 index 0000000..a410af9 --- /dev/null +++ b/plugins/supertonic/engine.py @@ -0,0 +1,156 @@ +"""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" diff --git a/tests/test_supertonic_plugin.py b/tests/test_supertonic_plugin.py new file mode 100644 index 0000000..6696ba8 --- /dev/null +++ b/tests/test_supertonic_plugin.py @@ -0,0 +1,415 @@ +"""Tests for the SuperTonic TTS Plugin. + +These tests verify that the SuperTonic 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, + SynthesizedAudio, + VoiceSelection, +) + + +# ────────────────────────────────────────────────────────────── +# Path fixtures +# ────────────────────────────────────────────────────────────── + +@pytest.fixture +def supertonic_plugin_dir() -> Path: + return Path(__file__).parent.parent / "plugins" / "supertonic" + + +def _supertonic_available() -> bool: + """Check if SuperTonic is available.""" + try: + from supertonic import TTS # 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 TestSuperTonicPluginLoading: + """Test that SuperTonic plugin loads correctly through the Loader.""" + + def test_plugin_loads_successfully(self, supertonic_plugin_dir: Path) -> None: + result = load_plugin_from_dir(supertonic_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, supertonic_plugin_dir: Path) -> None: + result = load_plugin_from_dir(supertonic_plugin_dir) + assert result.success is True + manifest = result.manifest + assert isinstance(manifest, PluginManifest) + assert manifest.id == "supertonic" + assert manifest.name == "SuperTonic" + assert manifest.api_version == "1.0" + + def test_plugin_has_model_requirements(self, supertonic_plugin_dir: Path) -> None: + result = load_plugin_from_dir(supertonic_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, supertonic_plugin_dir: Path) -> None: + result = load_plugin_from_dir(supertonic_plugin_dir) + assert result.success is True + manifest = result.manifest + assert "voice_list" in manifest.capabilities + + def test_plugin_manifest_engine(self, supertonic_plugin_dir: Path) -> None: + result = load_plugin_from_dir(supertonic_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 TestSuperTonicEngineCreation: + """Test that SuperTonic Engine can be created.""" + + @pytest.mark.skipif( + not _supertonic_available(), + reason="SuperTonic not installed" + ) + def test_create_engine(self, supertonic_plugin_dir: Path, host_context: HostContext) -> None: + result = load_plugin_from_dir(supertonic_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 _supertonic_available(), + reason="SuperTonic not installed" + ) + def test_engine_satisfies_protocol(self, supertonic_plugin_dir: Path, host_context: HostContext) -> None: + result = load_plugin_from_dir(supertonic_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 TestSuperTonicEngineProtocol: + """Test SuperTonic Engine protocol compliance using mock pipeline.""" + + def _create_engine_with_mock(self) -> Any: + """Create engine with mock pipeline for testing.""" + from plugins.supertonic.engine import SuperTonicEngine + + class MockSegment: + def __init__(self): + import numpy as np + self.audio = np.zeros(24000, dtype="float32") + + class MockPipeline: + sample_rate = 24000 + + def __call__(self, text, voice, speed, split_pattern=None, total_steps=None): + return [MockSegment()] + + return SuperTonicEngine(MockPipeline()) + + 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="M1"), + 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="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + with pytest.raises(EngineError): + session.synthesize(request) + engine.dispose() + + def test_session_multiple_synthesize(self) -> None: + engine = self._create_engine_with_mock() + session = engine.createSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result1 = session.synthesize(request) + result2 = session.synthesize(request) + assert isinstance(result1.data, bytes) + assert isinstance(result2.data, bytes) + session.dispose() + engine.dispose() + + def test_full_lifecycle(self) -> None: + engine = self._create_engine_with_mock() + + # Create sessions + session1 = engine.createSession() + session2 = engine.createSession() + + # Use sessions + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + assert isinstance(session1.synthesize(request), SynthesizedAudio) + assert isinstance(session2.synthesize(request), SynthesizedAudio) + + # Dispose sessions + session1.dispose() + session2.dispose() + + # Dispose engine + engine.dispose() + + def test_engine_returns_new_session_instances(self) -> None: + engine = self._create_engine_with_mock() + session1 = engine.createSession() + session2 = engine.createSession() + assert session1 is not session2 + session1.dispose() + session2.dispose() + engine.dispose() + + +# ────────────────────────────────────────────────────────────── +# VoiceLister Tests +# ────────────────────────────────────────────────────────────── + +class TestSuperTonicVoiceLister: + """Test SuperTonic VoiceLister capability.""" + + def test_list_voices(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + 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 = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + voices = engine.listVoices("builtin") + for voice in voices: + assert isinstance(voice.tags, tuple) + assert len(voice.tags) > 0 + engine.dispose() + + def test_voices_are_correct_count(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + voices = engine.listVoices("builtin") + assert len(voices) == 10 # M1-M5, F1-F5 + engine.dispose() + + def test_voice_ids_match_expected(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + voices = engine.listVoices("builtin") + voice_ids = [v.id for v in voices] + expected = ["M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5"] + assert voice_ids == expected + engine.dispose() + + def test_male_voices_have_male_tag(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + voices = engine.listVoices("builtin") + male_voices = [v for v in voices if v.id.startswith("M")] + for voice in male_voices: + assert "male" in voice.tags + engine.dispose() + + def test_female_voices_have_female_tag(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + voices = engine.listVoices("builtin") + female_voices = [v for v in voices if v.id.startswith("F")] + for voice in female_voices: + assert "female" in voice.tags + engine.dispose() + + def test_list_voices_after_dispose_raises(self) -> None: + from abogen.tts_plugin.errors import EngineError + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + engine.dispose() + with pytest.raises(EngineError): + engine.listVoices("builtin") + + +# ────────────────────────────────────────────────────────────── +# Parameter Tests +# ────────────────────────────────────────────────────────────── + +class TestSuperTonicParameters: + """Test SuperTonic parameter handling.""" + + def test_speed_parameter(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + session = engine.createSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={"speed": 1.5}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result.data, bytes) + session.dispose() + engine.dispose() + + def test_total_steps_parameter(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + session = engine.createSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={"total_steps": 10}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result.data, bytes) + session.dispose() + engine.dispose() + + def test_default_parameters(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + session = engine.createSession() + request = SynthesisRequest( + text="Hello", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result.data, bytes) + session.dispose() + engine.dispose() + + +# ────────────────────────────────────────────────────────────── +# Error Handling Tests +# ────────────────────────────────────────────────────────────── + +class TestSuperTonicErrorHandling: + """Test SuperTonic error handling.""" + + def test_synthesize_empty_text(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + session = engine.createSession() + request = SynthesisRequest( + text="", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + # Empty text should return empty audio + assert isinstance(result.data, bytes) + session.dispose() + engine.dispose() + + def test_synthesize_whitespace_text(self) -> None: + engine = TestSuperTonicEngineProtocol._create_engine_with_mock(TestSuperTonicEngineProtocol) + session = engine.createSession() + request = SynthesisRequest( + text=" ", + voice=VoiceSelection(source="builtin", key="M1"), + parameters=ParameterValues(values={}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert isinstance(result.data, bytes) + session.dispose() + engine.dispose()