mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 05:40:26 +02:00
feat: add SupertonicBackend implementing TTSBackend protocol
Encapsulate SupertonicPipeline as an internal detail of SupertonicBackend. The factory create_supertonic_backend() now returns a SupertonicBackend instance instead of a raw SupertonicPipeline, satisfying the TTSBackend protocol with metadata, synthesize, get_available_voices, get_supported_formats, and get_info methods. Backward-compatible __call__ delegates to the internal pipeline.
This commit is contained in:
@@ -5,7 +5,7 @@ from dataclasses import dataclass
|
|||||||
import logging
|
import logging
|
||||||
import math
|
import math
|
||||||
import re
|
import re
|
||||||
from typing import Any, Iterable, Iterator, Optional
|
from typing import Any, Dict, Iterable, Iterator, List, Optional
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
@@ -275,7 +275,97 @@ class SupertonicPipeline:
|
|||||||
yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio)
|
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.
|
"""Create a SuperTonic TTS backend instance.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
@@ -284,13 +374,9 @@ def create_supertonic_backend(**kwargs):
|
|||||||
total_steps: Inference steps. Defaults to 5.
|
total_steps: Inference steps. Defaults to 5.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
SupertonicPipeline instance.
|
SupertonicBackend instance.
|
||||||
"""
|
"""
|
||||||
return SupertonicPipeline(
|
return SupertonicBackend(**kwargs)
|
||||||
sample_rate=kwargs.get("sample_rate", 24000),
|
|
||||||
auto_download=kwargs.get("auto_download", True),
|
|
||||||
total_steps=kwargs.get("total_steps", 5),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
from abogen.tts_backend import TTSBackendMetadata
|
from abogen.tts_backend import TTSBackendMetadata
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
from abogen.tts_backends.supertonic import SupertonicPipeline
|
from abogen.tts_backends.supertonic import SupertonicBackend, SupertonicPipeline
|
||||||
|
|
||||||
|
|
||||||
class _DummyTTS:
|
class _DummyTTS:
|
||||||
@@ -26,13 +26,23 @@ class _DummyTTS:
|
|||||||
return audio, 0.05
|
return audio, 0.05
|
||||||
|
|
||||||
|
|
||||||
def test_supertonic_pipeline_strips_unsupported_characters_and_retries():
|
def _make_pipeline() -> SupertonicPipeline:
|
||||||
# Avoid importing/initializing real supertonic by manually constructing the pipeline.
|
|
||||||
pipeline = SupertonicPipeline.__new__(SupertonicPipeline)
|
pipeline = SupertonicPipeline.__new__(SupertonicPipeline)
|
||||||
pipeline.sample_rate = 24000
|
pipeline.sample_rate = 24000
|
||||||
pipeline.total_steps = 5
|
pipeline.total_steps = 5
|
||||||
pipeline.max_chunk_length = 1000
|
pipeline.max_chunk_length = 1000
|
||||||
pipeline._tts = _DummyTTS()
|
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))
|
segs = list(pipeline("Hello • world", voice="M1", speed=1.0))
|
||||||
assert len(segs) == 1
|
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():
|
def test_supertonic_pipeline_drops_chunk_if_only_unsupported_characters():
|
||||||
pipeline = SupertonicPipeline.__new__(SupertonicPipeline)
|
pipeline = _make_pipeline()
|
||||||
pipeline.sample_rate = 24000
|
|
||||||
pipeline.total_steps = 5
|
|
||||||
pipeline.max_chunk_length = 1000
|
|
||||||
pipeline._tts = _DummyTTS()
|
|
||||||
|
|
||||||
segs = list(pipeline("•", voice="M1", speed=1.0))
|
segs = list(pipeline("•", voice="M1", speed=1.0))
|
||||||
assert segs == []
|
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"
|
||||||
|
|||||||
Reference in New Issue
Block a user