Merge pull request #175 from k0sm0naft/refactor/tts-backend-interface

refactor: Switch TTSBackend from ABC to Protocol

abogen/tts_backend.py

@@ -1,117 +1,65 @@
 """
-Minimal TTS Backend Interface
+TTS Backend Interface
 
-This module defines a minimal interface for TTS backends to enable future
+This module defines the protocol for TTS backends.
-extensibility while maintaining backward compatibility with existing Kokoro
-implementation.
 """
 
-from abc import ABC, abstractmethod
+from typing import Protocol, List, Dict, Any
-from typing import Any, Iterator, Optional, Union
 
 
-class TTSBackend(ABC):
+class TTSBackend(Protocol):
     """
-    Minimal interface for TTS backends.
+    Protocol for TTS backends.
 
-    This interface is designed to be minimal and focused on the essential
+    All TTS backends must implement this interface to be compatible
-    operations needed for text-to-speech conversion.
+    with the application.
     """
 
-    @abstractmethod
+    def __init__(self, **kwargs) -> None:
-    def __call__(
-        self,
-        text: str,
-        voice: Union[str, Any],
-        speed: float = 1.0,
-        **kwargs: Any
-    ) -> Iterator[Any]:
         """
-        Generate speech segments from text.
+        Initialize the TTS backend.
 
         Args:
-            text: Text to convert to speech
+            **kwargs: Backend-specific configuration parameters
-            voice: Voice specification or object
-            speed: Speed multiplier for speech
-            **kwargs: Additional backend-specific parameters
-            
-        Yields:
-            Speech segments (audio data, timing info, etc.)
         """
-        pass
+        ...
 
-
+    def synthesize(self, text: str, **kwargs) -> bytes:
-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.
+        Synthesize speech from text.
 
         Args:
-            lang_code: Language code for the model
+            text: Text to synthesize
-            repo_id: Repository ID for the Kokoro model
+            **kwargs: Additional parameters for synthesis
-            device: Device to run the model on (cpu, cuda, etc.)
+
+        Returns:
+            Audio data as bytes
         """
-        self.lang_code = lang_code
+        ...
-        self.repo_id = repo_id
-        self.device = device
-        self._pipeline = None
 
-    def _get_pipeline(self):
+    def get_available_voices(self) -> List[str]:
-        """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.
+        Get list of available voices.
 
-        Args:
+        Returns:
-            text: Text to convert to speech
+            List of voice identifiers
-            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,
+    def get_supported_formats(self) -> List[str]:
-            voice=voice,
+        """
-            speed=speed,
+        Get list of supported audio formats.
-            split_pattern=split_pattern,
+
-            **kwargs
+        Returns:
-        )
+            List of supported audio formats
+        """
+        ...
+
+    def get_info(self) -> Dict[str, Any]:
+        """
+        Get backend information.
+
+        Returns:
+            Dictionary with backend information
+        """
+        ...

abogen/webui/conversion_runner.py

@@ -41,7 +41,7 @@ from abogen.utils import (
     load_config,
     load_numpy_kpipeline,
 )
-from abogen.tts_backend import KokoroTTSBackend
+from abogen.tts_backend import TTSBackend
 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
@@ -1595,8 +1595,9 @@ def run_conversion_job(job: Job) -> None:
             device = "cpu"
             if not disable_gpu:
                 device = _select_device()
-            # Create KokoroTTSBackend instance instead of directly instantiating KPipeline
+            _np, KPipeline = load_numpy_kpipeline()
-            pipelines[provider_norm] = KokoroTTSBackend(
+            # Create KPipeline instance directly (conforms to TTSBackend protocol)
+            pipelines[provider_norm] = KPipeline(
                 lang_code=job.language,
                 repo_id="hexgrad/Kokoro-82M",
                 device=device