Merge pull request #181 from denizsafak/refactor/add-supertonic-backend

feat: add SupertonicBackend implementing TTSBackend protocol

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

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"