From 9150a80459ce11a99c89c220e8082860a954da98 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 15:55:43 +0000 Subject: [PATCH] refactor: eliminate remaining legacy dependencies from production code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 1: Replace hardcoded VoiceLister bypass in get_voices() - Use PluginManager → Engine → VoiceLister instead of direct imports - No more hardcoded imports of plugins.kokoro.engine / plugins.supertonic.engine Task 2: Remove SuperTonic Plugin dependency on legacy backend - Create self-contained plugins/supertonic/pipeline.py - Plugin no longer imports from abogen.tts_backends Production code now has zero imports from: - abogen.tts_backend - abogen.tts_backend_registry - abogen.tts_backends --- abogen/tts_plugin/utils.py | 26 +++- plugins/supertonic/__init__.py | 2 +- plugins/supertonic/pipeline.py | 266 +++++++++++++++++++++++++++++++++ 3 files changed, 285 insertions(+), 9 deletions(-) create mode 100644 plugins/supertonic/pipeline.py diff --git a/abogen/tts_plugin/utils.py b/abogen/tts_plugin/utils.py index b39d932..145c3e2 100644 --- a/abogen/tts_plugin/utils.py +++ b/abogen/tts_plugin/utils.py @@ -14,14 +14,24 @@ from abogen.tts_plugin.plugin_manager import get_plugin_manager def get_voices(plugin_id: str) -> tuple[str, ...]: - """Return the voice-id tuple for *plugin_id*.""" - if plugin_id == "kokoro": - from plugins.kokoro.engine import _KOKORO_VOICES - return _KOKORO_VOICES - if plugin_id == "supertonic": - from plugins.supertonic.engine import _SUPERTONIC_VOICES - return _SUPERTONIC_VOICES - return () + """Return the voice-id tuple for *plugin_id*. + + Uses the official Plugin Architecture: PluginManager → Engine → VoiceLister. + """ + manager = get_plugin_manager() + if not manager.has_plugin(plugin_id): + return () + + engine = manager.create_engine(plugin_id) + try: + from abogen.tts_plugin.capabilities import VoiceLister + + if isinstance(engine, VoiceLister): + manifests = engine.listVoices("builtin") + return tuple(v.id for v in manifests) + return () + finally: + engine.dispose() def get_default_voice(plugin_id: str, fallback: str = "") -> str: diff --git a/plugins/supertonic/__init__.py b/plugins/supertonic/__init__.py index abd92cc..51e0ab6 100644 --- a/plugins/supertonic/__init__.py +++ b/plugins/supertonic/__init__.py @@ -33,7 +33,7 @@ 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 + from plugins.supertonic.pipeline import SupertonicPipeline return SupertonicPipeline( sample_rate=sample_rate, diff --git a/plugins/supertonic/pipeline.py b/plugins/supertonic/pipeline.py new file mode 100644 index 0000000..aabf3d7 --- /dev/null +++ b/plugins/supertonic/pipeline.py @@ -0,0 +1,266 @@ +"""SuperTonic Pipeline — self-contained TTS pipeline for the plugin. + +This module provides the SuperTonicPipeline class and supporting utilities +used by the SuperTonic plugin. It is independent of the legacy +abogen.tts_backends module. +""" + +from __future__ import annotations + +import ast +import logging +import re +from typing import Any, Iterable, Iterator, Optional + +import numpy as np + +logger = logging.getLogger(__name__) + + +def _ensure_float32_mono(wav: Any) -> np.ndarray: + arr = np.asarray(wav, dtype="float32") + if arr.ndim == 2: + if arr.shape[0] == 1 and arr.shape[1] > 1: + arr = arr.reshape(-1) + else: + arr = arr[:, 0] + return arr.reshape(-1) + + +def _resample_linear(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray: + if src_rate == dst_rate: + return audio + if audio.size == 0: + return audio + ratio = dst_rate / float(src_rate) + new_len = int(round(audio.size * ratio)) + if new_len <= 1: + return np.zeros(0, dtype="float32") + x_old = np.linspace(0.0, 1.0, num=audio.size, endpoint=False) + x_new = np.linspace(0.0, 1.0, num=new_len, endpoint=False) + return np.interp(x_new, x_old, audio).astype("float32", copy=False) + + +def _split_text( + text: str, *, split_pattern: Optional[str], max_chunk_length: int +) -> list[str]: + stripped = (text or "").strip() + if not stripped: + return [] + parts: list[str] + if split_pattern: + try: + parts = [p.strip() for p in re.split(split_pattern, stripped) if p.strip()] + except re.error: + parts = [stripped] + else: + parts = [stripped] + + result: list[str] = [] + for part in parts: + if len(part) <= max_chunk_length: + result.append(part) + continue + start = 0 + while start < len(part): + end = min(len(part), start + max_chunk_length) + if end < len(part): + ws = part.rfind(" ", start, end) + if ws > start + 40: + end = ws + chunk = part[start:end].strip() + if chunk: + result.append(chunk) + start = end + return result + + +_UNSUPPORTED_CHARS_RE = re.compile( + r"unsupported character\(s\):\s*(\[[^\]]*\])", re.IGNORECASE +) + + +def _parse_unsupported_characters(error: BaseException) -> list[str]: + """Best-effort extraction of unsupported characters from SuperTonic errors.""" + message = " ".join( + str(part) for part in getattr(error, "args", ()) if part is not None + ) or str(error) + match = _UNSUPPORTED_CHARS_RE.search(message) + if not match: + return [] + + raw = match.group(1) + try: + value = ast.literal_eval(raw) + except Exception: + return [] + + if isinstance(value, (list, tuple)): + out: list[str] = [] + for item in value: + if item is None: + continue + s = str(item) + if s: + out.append(s) + return out + + if isinstance(value, str) and value: + return [value] + + return [] + + +def _remove_unsupported_characters(text: str, unsupported: Iterable[str]) -> str: + result = text + for item in unsupported: + if not item: + continue + result = result.replace(item, "") + return result + + +def _configure_supertonic_gpu() -> None: + """Patch supertonic's config to enable GPU acceleration if available.""" + try: + import onnxruntime as ort + + available = ort.get_available_providers() + + providers = [] + if "CUDAExecutionProvider" in available: + providers.append("CUDAExecutionProvider") + providers.append("CPUExecutionProvider") + + import supertonic.config as supertonic_config + import supertonic.loader as supertonic_loader + + supertonic_config.DEFAULT_ONNX_PROVIDERS = providers + supertonic_loader.DEFAULT_ONNX_PROVIDERS = providers + logger.info("Supertonic ONNX providers configured: %s", providers) + except Exception as exc: + logger.warning("Could not configure supertonic GPU providers: %s", exc) + + +class SupertonicSegment: + """A single synthesized audio segment.""" + + __slots__ = ("graphemes", "audio") + + def __init__(self, graphemes: str, audio: np.ndarray) -> None: + self.graphemes = graphemes + self.audio = audio + + +class SupertonicPipeline: + """Minimal adapter that mimics Kokoro's pipeline iteration interface.""" + + def __init__( + self, + *, + sample_rate: int, + auto_download: bool = True, + total_steps: int = 5, + max_chunk_length: int = 300, + ) -> None: + self.sample_rate = int(sample_rate) + self.total_steps = int(total_steps) + self.max_chunk_length = int(max_chunk_length) + + _configure_supertonic_gpu() + + try: + from supertonic import TTS # type: ignore[import-not-found] + except Exception as exc: # pragma: no cover + raise RuntimeError( + "Supertonic is not installed. Install it with `pip install supertonic`." + ) from exc + + self._tts = TTS(auto_download=auto_download) + + def __call__( + self, + text: str, + *, + voice: str, + speed: float, + split_pattern: Optional[str] = None, + total_steps: Optional[int] = None, + ) -> Iterator[SupertonicSegment]: + voice_name = (voice or "").strip() or "M1" + steps = int(total_steps) if total_steps is not None else self.total_steps + steps = max(2, min(15, steps)) + speed_value = float(speed) if speed is not None else 1.0 + speed_value = max(0.7, min(2.0, speed_value)) + + style = self._tts.get_voice_style(voice_name=voice_name) + chunks = _split_text( + text, split_pattern=split_pattern, max_chunk_length=self.max_chunk_length + ) + for chunk in chunks: + chunk_to_speak = chunk + removed: set[str] = set() + last_exc: Exception | None = None + + for attempt in range(3): + try: + wav, duration = self._tts.synthesize( + text=chunk_to_speak, + voice_style=style, + total_steps=steps, + speed=speed_value, + max_chunk_length=self.max_chunk_length, + silence_duration=0.0, + verbose=False, + ) + break + except ValueError as exc: + last_exc = exc + unsupported = _parse_unsupported_characters(exc) + if not unsupported: + raise + + removed.update(unsupported) + sanitized = _remove_unsupported_characters( + chunk_to_speak, unsupported + ).strip() + + if sanitized == chunk_to_speak.strip(): + raise + + chunk_to_speak = sanitized + if not chunk_to_speak: + logger.warning( + "SuperTonic: dropped a chunk after removing unsupported characters: %s", + sorted(removed), + ) + break + + if attempt == 0: + logger.warning( + "SuperTonic: removed unsupported characters %s and retried.", + sorted(removed), + ) + else: + assert last_exc is not None + raise last_exc + + if not chunk_to_speak: + continue + + audio = _ensure_float32_mono(wav) + + src_rate = self.sample_rate + try: + dur = float(duration) + if dur > 0 and audio.size > 0: + inferred = int(round(audio.size / dur)) + if 8000 <= inferred <= 96000: + src_rate = inferred + except Exception: + pass + + if src_rate != self.sample_rate: + audio = _resample_linear(audio, src_rate, self.sample_rate) + + yield SupertonicSegment(graphemes=chunk_to_speak, audio=audio)