diff --git a/abogen/tts_backends/supertonic.py b/abogen/tts_backends/supertonic.py index f225cf6..07dd897 100644 --- a/abogen/tts_backends/supertonic.py +++ b/abogen/tts_backends/supertonic.py @@ -5,7 +5,7 @@ from dataclasses import dataclass import logging import math import re -from typing import Any, Iterable, Iterator, Optional +from typing import Any, Dict, Iterable, Iterator, List, Optional import numpy as np @@ -275,7 +275,97 @@ class SupertonicPipeline: yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio) -def create_supertonic_backend(**kwargs): +class SupertonicBackend: + """Supertonic TTS backend implementing the TTSBackend protocol. + + Encapsulates ``SupertonicPipeline`` as an internal implementation detail. + """ + + @property + def metadata(self) -> "TTSBackendMetadata": + return TTSBackendMetadata( + id="supertonic", + name="SuperTonic", + description="SuperTonic TTS engine", + ) + + def __init__(self, **kwargs: Any) -> None: + self._pipeline = SupertonicPipeline( + sample_rate=kwargs.get("sample_rate", 24000), + auto_download=kwargs.get("auto_download", True), + total_steps=kwargs.get("total_steps", 5), + ) + + def synthesize(self, text: str, **kwargs: Any) -> bytes: + """Synthesize speech and return raw audio bytes (WAV). + + Delegates to the internal :class:`SupertonicPipeline` and concatenates + all produced segments into a single byte buffer. + """ + import io + + import soundfile as sf + + voice = kwargs.get("voice", "M1") + speed = float(kwargs.get("speed", 1.0)) + split_pattern = kwargs.get("split_pattern") + total_steps = kwargs.get("total_steps") + + segments = self._pipeline( + text, + voice=voice, + speed=speed, + split_pattern=split_pattern, + total_steps=total_steps, + ) + + audio_parts: list[np.ndarray] = [] + for seg in segments: + audio_parts.append(seg.audio) + + if not audio_parts: + return b"" + + combined = np.concatenate(audio_parts) + buf = io.BytesIO() + sf.write(buf, combined, self._pipeline.sample_rate, format="WAV") + return buf.getvalue() + + def get_available_voices(self) -> List[str]: + """Return the list of built-in SuperTonic voice identifiers.""" + return list(DEFAULT_SUPERTONIC_VOICES) + + def get_supported_formats(self) -> List[str]: + return ["wav"] + + def get_info(self) -> Dict[str, Any]: + return { + "sample_rate": self._pipeline.sample_rate, + "total_steps": self._pipeline.total_steps, + "max_chunk_length": self._pipeline.max_chunk_length, + "voices": list(DEFAULT_SUPERTONIC_VOICES), + } + + def __call__( + self, + text: str, + *, + voice: str, + speed: float, + split_pattern: Optional[str] = None, + total_steps: Optional[int] = None, + ) -> Iterator[SupertonicSegment]: + """Backward-compatible call interface, delegates to the pipeline.""" + return self._pipeline( + text, + voice=voice, + speed=speed, + split_pattern=split_pattern, + total_steps=total_steps, + ) + + +def create_supertonic_backend(**kwargs: Any) -> SupertonicBackend: """Create a SuperTonic TTS backend instance. Args: @@ -284,13 +374,9 @@ def create_supertonic_backend(**kwargs): total_steps: Inference steps. Defaults to 5. Returns: - SupertonicPipeline instance. + SupertonicBackend instance. """ - return SupertonicPipeline( - sample_rate=kwargs.get("sample_rate", 24000), - auto_download=kwargs.get("auto_download", True), - total_steps=kwargs.get("total_steps", 5), - ) + return SupertonicBackend(**kwargs) from abogen.tts_backend import TTSBackendMetadata diff --git a/tests/test_tts_supertonic_unsupported_chars.py b/tests/test_tts_supertonic_unsupported_chars.py index ecb14dd..3e8cd00 100644 --- a/tests/test_tts_supertonic_unsupported_chars.py +++ b/tests/test_tts_supertonic_unsupported_chars.py @@ -1,6 +1,6 @@ import numpy as np -from abogen.tts_backends.supertonic import SupertonicPipeline +from abogen.tts_backends.supertonic import SupertonicBackend, SupertonicPipeline class _DummyTTS: @@ -26,13 +26,23 @@ class _DummyTTS: return audio, 0.05 -def test_supertonic_pipeline_strips_unsupported_characters_and_retries(): - # Avoid importing/initializing real supertonic by manually constructing the pipeline. +def _make_pipeline() -> SupertonicPipeline: pipeline = SupertonicPipeline.__new__(SupertonicPipeline) pipeline.sample_rate = 24000 pipeline.total_steps = 5 pipeline.max_chunk_length = 1000 pipeline._tts = _DummyTTS() + return pipeline + + +def _make_backend() -> SupertonicBackend: + backend = SupertonicBackend.__new__(SupertonicBackend) + backend._pipeline = _make_pipeline() + return backend + + +def test_supertonic_pipeline_strips_unsupported_characters_and_retries(): + pipeline = _make_pipeline() segs = list(pipeline("Hello • world", voice="M1", speed=1.0)) assert len(segs) == 1 @@ -43,11 +53,56 @@ def test_supertonic_pipeline_strips_unsupported_characters_and_retries(): def test_supertonic_pipeline_drops_chunk_if_only_unsupported_characters(): - pipeline = SupertonicPipeline.__new__(SupertonicPipeline) - pipeline.sample_rate = 24000 - pipeline.total_steps = 5 - pipeline.max_chunk_length = 1000 - pipeline._tts = _DummyTTS() + pipeline = _make_pipeline() segs = list(pipeline("•", voice="M1", speed=1.0)) assert segs == [] + + +# --- SupertonicBackend tests --- + + +def test_backend_metadata(): + backend = _make_backend() + meta = backend.metadata + assert meta.id == "supertonic" + assert meta.name == "SuperTonic" + assert "SuperTonic" in meta.description + + +def test_backend_get_available_voices(): + backend = _make_backend() + voices = backend.get_available_voices() + assert isinstance(voices, list) + assert "M1" in voices + assert "F1" in voices + + +def test_backend_get_supported_formats(): + backend = _make_backend() + formats = backend.get_supported_formats() + assert "wav" in formats + + +def test_backend_get_info(): + backend = _make_backend() + info = backend.get_info() + assert info["sample_rate"] == 24000 + assert info["total_steps"] == 5 + assert isinstance(info["voices"], list) + + +def test_backend_call_delegates_to_pipeline(): + backend = _make_backend() + segs = list(backend("Hello • world", voice="M1", speed=1.0)) + assert len(segs) == 1 + assert segs[0].audio.size > 0 + + +def test_backend_synthesize_returns_wav_bytes(): + backend = _make_backend() + wav_bytes = backend.synthesize("Hello world", voice="M1", speed=1.0) + assert isinstance(wav_bytes, bytes) + assert len(wav_bytes) > 0 + # WAV magic number + assert wav_bytes[:4] == b"RIFF"