From 50b4d6872a110a23349b8e75017ce3dfd213600b Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Fri, 3 Jul 2026 01:25:41 +0300 Subject: [PATCH] feat: Add minimal TTSBackend interface for future extensibility - Create TTSBackend abstract base class with minimal contract - Implement KokoroTTSBackend that maintains existing behavior - Update conversion_runner.py to use new interface - No behavioral changes, GUI unchanged, no new features --- abogen/tts_backend.py | 117 ++++++++++++++++++++++++++++++ abogen/webui/conversion_runner.py | 37 +++++----- 2 files changed, 134 insertions(+), 20 deletions(-) create mode 100644 abogen/tts_backend.py diff --git a/abogen/tts_backend.py b/abogen/tts_backend.py new file mode 100644 index 0000000..6dcc506 --- /dev/null +++ b/abogen/tts_backend.py @@ -0,0 +1,117 @@ +""" +Minimal TTS Backend Interface + +This module defines a minimal interface for TTS backends to enable future +extensibility while maintaining backward compatibility with existing Kokoro +implementation. +""" + +from abc import ABC, abstractmethod +from typing import Any, Iterator, Optional, Union + + +class TTSBackend(ABC): + """ + Minimal interface for TTS backends. + + This interface is designed to be minimal and focused on the essential + operations needed for text-to-speech conversion. + """ + + @abstractmethod + def __call__( + self, + text: str, + voice: Union[str, Any], + speed: float = 1.0, + **kwargs: Any + ) -> Iterator[Any]: + """ + Generate speech segments from text. + + Args: + text: Text to convert to speech + voice: Voice specification or object + speed: Speed multiplier for speech + **kwargs: Additional backend-specific parameters + + Yields: + Speech segments (audio data, timing info, etc.) + """ + pass + + +class KokoroTTSBackend(TTSBackend): + """ + Implementation of TTSBackend using Kokoro. + + This class provides the concrete implementation that maintains + the existing behavior while conforming to the TTSBackend interface. + """ + + def __init__(self, lang_code: str, repo_id: str = "hexgrad/Kokoro-82M", device: str = "cpu"): + """ + Initialize Kokoro backend. + + Args: + lang_code: Language code for the model + repo_id: Repository ID for the Kokoro model + device: Device to run the model on (cpu, cuda, etc.) + """ + self.lang_code = lang_code + self.repo_id = repo_id + self.device = device + self._pipeline = None + + def _get_pipeline(self): + """Lazy initialization of the Kokoro pipeline.""" + if self._pipeline is None: + from abogen.utils import load_numpy_kpipeline + _, KPipeline = load_numpy_kpipeline() + try: + self._pipeline = KPipeline( + lang_code=self.lang_code, + repo_id=self.repo_id, + device=self.device + ) + except RuntimeError as e: + if "CUDA" in str(e) and self.device != "cpu": + # Fall back to CPU if CUDA fails + self._pipeline = KPipeline( + lang_code=self.lang_code, + repo_id=self.repo_id, + device="cpu" + ) + else: + raise + return self._pipeline + + def __call__( + self, + text: str, + voice: Union[str, Any], + speed: float = 1.0, + split_pattern: str = r"\n+", + **kwargs: Any + ) -> Iterator[Any]: + """ + Generate speech segments from text using Kokoro. + + Args: + text: Text to convert to speech + voice: Voice specification or object + speed: Speed multiplier for speech + split_pattern: Pattern to split text into segments + **kwargs: Additional parameters passed to the pipeline + + Yields: + Speech segments + """ + pipeline = self._get_pipeline() + return pipeline( + text, + voice=voice, + speed=speed, + split_pattern=split_pattern, + **kwargs + ) diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index f41e624..724c3d0 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -41,6 +41,7 @@ from abogen.utils import ( load_config, load_numpy_kpipeline, ) +from abogen.tts_backend import KokoroTTSBackend from abogen.voice_cache import ensure_voice_assets from abogen.voice_formulas import extract_voice_ids, get_new_voice from abogen.voice_profiles import load_profiles, normalize_profile_entry @@ -1594,16 +1595,12 @@ def run_conversion_job(job: Job) -> None: device = "cpu" if not disable_gpu: device = _select_device() - _np, KPipeline = load_numpy_kpipeline() - # Try to initialize with the selected device; fall back to CPU if CUDA fails - try: - pipelines[provider_norm] = KPipeline(lang_code=job.language, repo_id="hexgrad/Kokoro-82M", device=device) - except RuntimeError as e: - if "CUDA" in str(e) and device != "cpu": - job.add_log(f"CUDA initialization failed, falling back to CPU: {e}", level="warning") - pipelines[provider_norm] = KPipeline(lang_code=job.language, repo_id="hexgrad/Kokoro-82M", device="cpu") - else: - raise + # Create KokoroTTSBackend instance instead of directly instantiating KPipeline + pipelines[provider_norm] = KokoroTTSBackend( + lang_code=job.language, + repo_id="hexgrad/Kokoro-82M", + device=device + ) if not kokoro_cache_ready: _initialize_voice_cache(job) kokoro_cache_ready = True @@ -1644,8 +1641,8 @@ def run_conversion_job(job: Job) -> None: return provider, resolved, cached, speed, steps if provider == "kokoro": - kokoro_pipeline = get_pipeline("kokoro") - choice = _resolve_voice(kokoro_pipeline, resolved, job.use_gpu) + kokoro_backend = get_pipeline("kokoro") + choice = _resolve_voice(kokoro_backend, resolved, job.use_gpu) else: choice = resolved @@ -1774,8 +1771,8 @@ def run_conversion_job(job: Job) -> None: voice_cache: Dict[str, Any] = {} base_provider, base_voice_resolved, _, _ = resolve_voice_target(base_voice_spec) if base_provider == "kokoro" and base_voice_resolved and "*" not in base_voice_resolved: - kokoro_pipeline = get_pipeline("kokoro") - voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_pipeline, base_voice_resolved, job.use_gpu) + kokoro_backend = get_pipeline("kokoro") + voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_backend, base_voice_resolved, job.use_gpu) processed_chars = 0 subtitle_index = 1 current_time = 0.0 @@ -1860,8 +1857,8 @@ def run_conversion_job(job: Job) -> None: total_steps=int(supertonic_steps_override if supertonic_steps_override is not None else getattr(job, "supertonic_total_steps", 5)), ) else: - kokoro_pipeline = get_pipeline("kokoro") - segment_iter = kokoro_pipeline( + kokoro_backend = get_pipeline("kokoro") + segment_iter = kokoro_backend( normalized, voice=voice_choice, speed=float(speed_override if speed_override is not None else job.speed), @@ -1950,8 +1947,8 @@ def run_conversion_job(job: Job) -> None: if chapter_provider == "kokoro": voice_choice = voice_cache.get(chapter_cache_key) if voice_choice is None: - kokoro_pipeline = get_pipeline("kokoro") - voice_choice = _resolve_voice(kokoro_pipeline, chapter_voice_resolved, job.use_gpu) + kokoro_backend = get_pipeline("kokoro") + voice_choice = _resolve_voice(kokoro_backend, chapter_voice_resolved, job.use_gpu) voice_cache[chapter_cache_key] = voice_choice else: voice_choice = chapter_voice_resolved @@ -2095,9 +2092,9 @@ def run_conversion_job(job: Job) -> None: if chunk_provider == "kokoro": chunk_voice_choice = voice_cache.get(chunk_cache_key) if chunk_voice_choice is None: - kokoro_pipeline = get_pipeline("kokoro") + kokoro_backend = get_pipeline("kokoro") chunk_voice_choice = _resolve_voice( - kokoro_pipeline, + kokoro_backend, chunk_voice_resolved, job.use_gpu, )