Migrate existing STT providers to infrastructure layer

src/infrastructure/base/stt_provider_base.py

@@ -114,6 +114,16 @@ class STTProviderBase(ISpeechRecognitionService, ABC):
         """
         pass
 
+    @abstractmethod
+    def get_default_model(self) -> str:
+        """
+        Get the default model for this provider.
+
+        Returns:
+            str: Default model name
+        """
+        pass
+
     def _preprocess_audio(self, audio: 'AudioContent') -> Path:
         """
         Preprocess audio content for transcription.

src/infrastructure/stt/__init__.py

@@ -0,0 +1,15 @@
+"""STT provider implementations."""
+
+from .whisper_provider import WhisperSTTProvider
+from .parakeet_provider import ParakeetSTTProvider
+from .provider_factory import STTProviderFactory, ASRFactory
+from .legacy_compatibility import transcribe_audio, create_audio_content_from_file
+
+__all__ = [
+    'WhisperSTTProvider',
+    'ParakeetSTTProvider',
+    'STTProviderFactory',
+    'ASRFactory',
+    'transcribe_audio',
+    'create_audio_content_from_file'
+]

src/infrastructure/stt/legacy_compatibility.py

@@ -0,0 +1,150 @@
+"""Legacy compatibility functions for STT functionality."""
+
+import logging
+from pathlib import Path
+from typing import Union
+
+from .provider_factory import STTProviderFactory
+from ...domain.models.audio_content import AudioContent
+from ...domain.exceptions import SpeechRecognitionException
+
+logger = logging.getLogger(__name__)
+
+
+def transcribe_audio(audio_path: Union[str, Path], model_name: str = "parakeet") -> str:
+    """
+    Convert audio file to text using specified STT model (legacy interface).
+    
+    This function maintains backward compatibility with the original utils/stt.py interface.
+    
+    Args:
+        audio_path: Path to input audio file
+        model_name: Name of the STT model/provider to use (whisper or parakeet)
+        
+    Returns:
+        str: Transcribed English text
+        
+    Raises:
+        SpeechRecognitionException: If transcription fails
+    """
+    logger.info(f"Starting transcription for: {audio_path} using {model_name} model")
+    
+    try:
+        # Convert path to Path object
+        audio_path = Path(audio_path)
+        
+        if not audio_path.exists():
+            raise SpeechRecognitionException(f"Audio file not found: {audio_path}")
+        
+        # Read audio file and create AudioContent
+        with open(audio_path, 'rb') as f:
+            audio_data = f.read()
+        
+        # Determine audio format from file extension
+        audio_format = audio_path.suffix.lower().lstrip('.')
+        if audio_format not in ['wav', 'mp3', 'flac', 'ogg']:
+            audio_format = 'wav'  # Default fallback
+        
+        # Create AudioContent (we'll use reasonable placeholder values)
+        # The provider will handle the actual audio analysis during preprocessing
+        try:
+            audio_content = AudioContent(
+                data=audio_data,
+                format=audio_format,
+                sample_rate=16000,  # Standard rate for STT
+                duration=max(1.0, len(audio_data) / (16000 * 2)),  # Rough estimate
+                filename=audio_path.name
+            )
+        except ValueError:
+            # If validation fails, try with minimal valid values
+            audio_content = AudioContent(
+                data=audio_data,
+                format=audio_format,
+                sample_rate=16000,
+                duration=1.0,  # Minimum valid duration
+                filename=audio_path.name
+            )
+        
+        # Get the appropriate provider
+        try:
+            provider = STTProviderFactory.create_provider(model_name)
+        except SpeechRecognitionException:
+            # Fallback to any available provider
+            logger.warning(f"Requested provider {model_name} not available, using fallback")
+            provider = STTProviderFactory.create_provider_with_fallback(model_name)
+        
+        # Get the default model for the provider
+        model = provider.get_default_model()
+        
+        # Transcribe audio
+        text_content = provider.transcribe(audio_content, model)
+        result = text_content.text
+        
+        logger.info(f"Transcription completed: {result}")
+        return result
+
+    except Exception as e:
+        logger.error(f"Transcription failed: {str(e)}", exc_info=True)
+        raise SpeechRecognitionException(f"Transcription failed: {str(e)}") from e
+
+
+def create_audio_content_from_file(audio_path: Union[str, Path]) -> AudioContent:
+    """
+    Create AudioContent from an audio file with proper metadata detection.
+    
+    Args:
+        audio_path: Path to the audio file
+        
+    Returns:
+        AudioContent: The audio content object
+        
+    Raises:
+        SpeechRecognitionException: If file cannot be processed
+    """
+    try:
+        from pydub import AudioSegment
+        
+        audio_path = Path(audio_path)
+        
+        # Load audio file to get metadata
+        audio_segment = AudioSegment.from_file(audio_path)
+        
+        # Read raw audio data
+        with open(audio_path, 'rb') as f:
+            audio_data = f.read()
+        
+        # Determine format
+        audio_format = audio_path.suffix.lower().lstrip('.')
+        if audio_format not in ['wav', 'mp3', 'flac', 'ogg']:
+            audio_format = 'wav'
+        
+        # Create AudioContent with actual metadata
+        return AudioContent(
+            data=audio_data,
+            format=audio_format,
+            sample_rate=audio_segment.frame_rate,
+            duration=len(audio_segment) / 1000.0,  # Convert ms to seconds
+            filename=audio_path.name
+        )
+        
+    except ImportError:
+        # Fallback without pydub
+        logger.warning("pydub not available, using placeholder metadata")
+        
+        with open(audio_path, 'rb') as f:
+            audio_data = f.read()
+        
+        audio_format = Path(audio_path).suffix.lower().lstrip('.')
+        if audio_format not in ['wav', 'mp3', 'flac', 'ogg']:
+            audio_format = 'wav'
+        
+        return AudioContent(
+            data=audio_data,
+            format=audio_format,
+            sample_rate=16000,  # Default
+            duration=1.0,  # Placeholder
+            filename=Path(audio_path).name
+        )
+        
+    except Exception as e:
+        raise SpeechRecognitionException(f"Failed to create AudioContent from file: {str(e)}") from e

src/infrastructure/stt/parakeet_provider.py

@@ -0,0 +1,122 @@
+"""Parakeet STT provider implementation."""
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ...domain.models.audio_content import AudioContent
+    from ...domain.models.text_content import TextContent
+
+from ..base.stt_provider_base import STTProviderBase
+from ...domain.exceptions import SpeechRecognitionException
+
+logger = logging.getLogger(__name__)
+
+
+class ParakeetSTTProvider(STTProviderBase):
+    """Parakeet STT provider using NVIDIA NeMo implementation."""
+
+    def __init__(self):
+        """Initialize the Parakeet STT provider."""
+        super().__init__(
+            provider_name="Parakeet",
+            supported_languages=["en"]  # Parakeet primarily supports English
+        )
+        self.model = None
+
+    def _perform_transcription(self, audio_path: Path, model: str) -> str:
+        """
+        Perform transcription using Parakeet.
+
+        Args:
+            audio_path: Path to the preprocessed audio file
+            model: The Parakeet model to use
+
+        Returns:
+            str: The transcribed text
+        """
+        try:
+            # Load model if not already loaded
+            if self.model is None:
+                self._load_model(model)
+
+            logger.info(f"Starting Parakeet transcription with model {model}")
+            
+            # Perform transcription
+            output = self.model.transcribe([str(audio_path)])
+            result = output[0].text if output and len(output) > 0 else ""
+            
+            logger.info("Parakeet transcription completed successfully")
+            return result
+
+        except Exception as e:
+            self._handle_provider_error(e, "transcription")
+
+    def _load_model(self, model_name: str):
+        """
+        Load the Parakeet model.
+
+        Args:
+            model_name: Name of the model to load
+        """
+        try:
+            import nemo.collections.asr as nemo_asr
+            
+            logger.info(f"Loading Parakeet model: {model_name}")
+            
+            # Map model names to actual model identifiers
+            model_mapping = {
+                "parakeet-tdt-0.6b-v2": "nvidia/parakeet-tdt-0.6b-v2",
+                "parakeet-tdt-1.1b": "nvidia/parakeet-tdt-1.1b",
+                "parakeet-ctc-0.6b": "nvidia/parakeet-ctc-0.6b",
+                "default": "nvidia/parakeet-tdt-0.6b-v2"
+            }
+            
+            actual_model_name = model_mapping.get(model_name, model_mapping["default"])
+            
+            self.model = nemo_asr.models.ASRModel.from_pretrained(model_name=actual_model_name)
+            logger.info(f"Parakeet model {model_name} loaded successfully")
+            
+        except ImportError as e:
+            raise SpeechRecognitionException(
+                "nemo_toolkit not available. Please install with: pip install -U 'nemo_toolkit[asr]'"
+            ) from e
+        except Exception as e:
+            raise SpeechRecognitionException(f"Failed to load Parakeet model {model_name}: {str(e)}") from e
+
+    def is_available(self) -> bool:
+        """
+        Check if the Parakeet provider is available.
+
+        Returns:
+            bool: True if nemo_toolkit is available, False otherwise
+        """
+        try:
+            import nemo.collections.asr
+            return True
+        except ImportError:
+            logger.warning("nemo_toolkit not available")
+            return False
+
+    def get_available_models(self) -> list[str]:
+        """
+        Get list of available Parakeet models.
+
+        Returns:
+            list[str]: List of available model names
+        """
+        return [
+            "parakeet-tdt-0.6b-v2",
+            "parakeet-tdt-1.1b", 
+            "parakeet-ctc-0.6b"
+        ]
+
+    def get_default_model(self) -> str:
+        """
+        Get the default model for this provider.
+
+        Returns:
+            str: Default model name
+        """
+        return "parakeet-tdt-0.6b-v2"

src/infrastructure/stt/provider_factory.py

@@ -0,0 +1,189 @@
+"""Factory for creating STT provider instances."""
+
+import logging
+from typing import Dict, Type, Optional
+
+from ..base.stt_provider_base import STTProviderBase
+from .whisper_provider import WhisperSTTProvider
+from .parakeet_provider import ParakeetSTTProvider
+from ...domain.exceptions import SpeechRecognitionException
+
+logger = logging.getLogger(__name__)
+
+
+class STTProviderFactory:
+    """Factory for creating STT provider instances with availability checking and fallback logic."""
+
+    _providers: Dict[str, Type[STTProviderBase]] = {
+        "whisper": WhisperSTTProvider,
+        "parakeet": ParakeetSTTProvider
+    }
+
+    _fallback_order = ["whisper", "parakeet"]
+
+    @classmethod
+    def create_provider(cls, provider_name: str) -> STTProviderBase:
+        """
+        Create an STT provider instance by name.
+
+        Args:
+            provider_name: Name of the provider to create
+
+        Returns:
+            STTProviderBase: The created provider instance
+
+        Raises:
+            SpeechRecognitionException: If provider is not available or creation fails
+        """
+        provider_name = provider_name.lower()
+        
+        if provider_name not in cls._providers:
+            raise SpeechRecognitionException(f"Unknown STT provider: {provider_name}")
+
+        provider_class = cls._providers[provider_name]
+        
+        try:
+            provider = provider_class()
+            
+            if not provider.is_available():
+                raise SpeechRecognitionException(f"STT provider {provider_name} is not available")
+            
+            logger.info(f"Created STT provider: {provider_name}")
+            return provider
+            
+        except Exception as e:
+            logger.error(f"Failed to create STT provider {provider_name}: {str(e)}")
+            raise SpeechRecognitionException(f"Failed to create STT provider {provider_name}: {str(e)}") from e
+
+    @classmethod
+    def create_provider_with_fallback(cls, preferred_provider: str) -> STTProviderBase:
+        """
+        Create an STT provider with fallback to other available providers.
+
+        Args:
+            preferred_provider: The preferred provider name
+
+        Returns:
+            STTProviderBase: The created provider instance
+
+        Raises:
+            SpeechRecognitionException: If no providers are available
+        """
+        # Try preferred provider first
+        try:
+            return cls.create_provider(preferred_provider)
+        except SpeechRecognitionException as e:
+            logger.warning(f"Preferred STT provider {preferred_provider} failed: {str(e)}")
+
+        # Try fallback providers
+        for provider_name in cls._fallback_order:
+            if provider_name.lower() == preferred_provider.lower():
+                continue  # Skip the preferred provider we already tried
+                
+            try:
+                logger.info(f"Trying fallback STT provider: {provider_name}")
+                return cls.create_provider(provider_name)
+            except SpeechRecognitionException as e:
+                logger.warning(f"Fallback STT provider {provider_name} failed: {str(e)}")
+                continue
+
+        raise SpeechRecognitionException("No STT providers are available")
+
+    @classmethod
+    def get_available_providers(cls) -> list[str]:
+        """
+        Get list of available STT providers.
+
+        Returns:
+            list[str]: List of available provider names
+        """
+        available = []
+        
+        for provider_name, provider_class in cls._providers.items():
+            try:
+                provider = provider_class()
+                if provider.is_available():
+                    available.append(provider_name)
+            except Exception as e:
+                logger.debug(f"Provider {provider_name} not available: {str(e)}")
+                
+        return available
+
+    @classmethod
+    def get_provider_info(cls, provider_name: str) -> Optional[dict]:
+        """
+        Get information about a specific provider.
+
+        Args:
+            provider_name: Name of the provider
+
+        Returns:
+            Optional[dict]: Provider information or None if not found
+        """
+        provider_name = provider_name.lower()
+        
+        if provider_name not in cls._providers:
+            return None
+
+        provider_class = cls._providers[provider_name]
+        
+        try:
+            provider = provider_class()
+            return {
+                "name": provider.provider_name,
+                "available": provider.is_available(),
+                "supported_languages": provider.supported_languages,
+                "available_models": provider.get_available_models() if provider.is_available() else [],
+                "default_model": provider.get_default_model() if provider.is_available() else None
+            }
+        except Exception as e:
+            logger.debug(f"Failed to get info for provider {provider_name}: {str(e)}")
+            return {
+                "name": provider_name,
+                "available": False,
+                "error": str(e)
+            }
+
+    @classmethod
+    def register_provider(cls, name: str, provider_class: Type[STTProviderBase]) -> None:
+        """
+        Register a new STT provider.
+
+        Args:
+            name: Name of the provider
+            provider_class: The provider class
+        """
+        cls._providers[name.lower()] = provider_class
+        logger.info(f"Registered STT provider: {name}")
+
+
+# Legacy compatibility - create an ASRFactory alias
+class ASRFactory:
+    """Legacy ASRFactory for backward compatibility."""
+    
+    @staticmethod
+    def get_model(model_name: str = "parakeet") -> STTProviderBase:
+        """
+        Get STT provider by model name (legacy interface).
+        
+        Args:
+            model_name: Name of the model/provider to use
+            
+        Returns:
+            STTProviderBase: The provider instance
+        """
+        # Map legacy model names to provider names
+        provider_mapping = {
+            "whisper": "whisper",
+            "parakeet": "parakeet",
+            "faster-whisper": "whisper"
+        }
+        
+        provider_name = provider_mapping.get(model_name.lower(), model_name.lower())
+        
+        try:
+            return STTProviderFactory.create_provider(provider_name)
+        except SpeechRecognitionException:
+            # Fallback to any available provider
+            logger.warning(f"Requested provider {provider_name} not available, using fallback")
+            return STTProviderFactory.create_provider_with_fallback(provider_name)

src/infrastructure/stt/whisper_provider.py

@@ -0,0 +1,154 @@
+"""Whisper STT provider implementation."""
+
+import logging
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ...domain.models.audio_content import AudioContent
+    from ...domain.models.text_content import TextContent
+
+from ..base.stt_provider_base import STTProviderBase
+from ...domain.exceptions import SpeechRecognitionException
+
+logger = logging.getLogger(__name__)
+
+
+class WhisperSTTProvider(STTProviderBase):
+    """Whisper STT provider using faster-whisper implementation."""
+
+    def __init__(self):
+        """Initialize the Whisper STT provider."""
+        super().__init__(
+            provider_name="Whisper",
+            supported_languages=["en", "es", "fr", "de", "it", "pt", "ru", "ja", "ko", "zh"]
+        )
+        self.model = None
+        self._device = None
+        self._compute_type = None
+        self._initialize_device_settings()
+
+    def _initialize_device_settings(self):
+        """Initialize device and compute type settings."""
+        try:
+            import torch
+            self._device = "cuda" if torch.cuda.is_available() else "cpu"
+        except ImportError:
+            # Fallback to CPU if torch is not available
+            self._device = "cpu"
+        
+        self._compute_type = "float16" if self._device == "cuda" else "int8"
+        logger.info(f"Whisper provider initialized with device: {self._device}, compute_type: {self._compute_type}")
+
+    def _perform_transcription(self, audio_path: Path, model: str) -> str:
+        """
+        Perform transcription using Faster Whisper.
+
+        Args:
+            audio_path: Path to the preprocessed audio file
+            model: The Whisper model to use (e.g., 'large-v3', 'medium', 'small')
+
+        Returns:
+            str: The transcribed text
+        """
+        try:
+            # Load model if not already loaded or if model changed
+            if self.model is None or getattr(self.model, 'model_size_or_path', None) != model:
+                self._load_model(model)
+
+            logger.info(f"Starting Whisper transcription with model {model}")
+            
+            # Perform transcription
+            segments, info = self.model.transcribe(
+                str(audio_path),
+                beam_size=5,
+                language="en",  # Can be made configurable
+                task="transcribe"
+            )
+
+            logger.info(f"Detected language '{info.language}' with probability {info.language_probability}")
+
+            # Collect all segments into a single text
+            result_text = ""
+            for segment in segments:
+                result_text += segment.text + " "
+                logger.debug(f"[{segment.start:.2f}s -> {segment.end:.2f}s] {segment.text}")
+
+            result = result_text.strip()
+            logger.info("Whisper transcription completed successfully")
+            return result
+
+        except Exception as e:
+            self._handle_provider_error(e, "transcription")
+
+    def _load_model(self, model_name: str):
+        """
+        Load the Whisper model.
+
+        Args:
+            model_name: Name of the model to load
+        """
+        try:
+            from faster_whisper import WhisperModel as FasterWhisperModel
+            
+            logger.info(f"Loading Whisper model: {model_name}")
+            logger.info(f"Using device: {self._device}, compute_type: {self._compute_type}")
+            
+            self.model = FasterWhisperModel(
+                model_name,
+                device=self._device,
+                compute_type=self._compute_type
+            )
+            
+            logger.info(f"Whisper model {model_name} loaded successfully")
+            
+        except ImportError as e:
+            raise SpeechRecognitionException(
+                "faster-whisper not available. Please install with: pip install faster-whisper"
+            ) from e
+        except Exception as e:
+            raise SpeechRecognitionException(f"Failed to load Whisper model {model_name}: {str(e)}") from e
+
+    def is_available(self) -> bool:
+        """
+        Check if the Whisper provider is available.
+
+        Returns:
+            bool: True if faster-whisper is available, False otherwise
+        """
+        try:
+            import faster_whisper
+            return True
+        except ImportError:
+            logger.warning("faster-whisper not available")
+            return False
+
+    def get_available_models(self) -> list[str]:
+        """
+        Get list of available Whisper models.
+
+        Returns:
+            list[str]: List of available model names
+        """
+        return [
+            "tiny",
+            "tiny.en",
+            "base",
+            "base.en", 
+            "small",
+            "small.en",
+            "medium",
+            "medium.en",
+            "large-v1",
+            "large-v2",
+            "large-v3"
+        ]
+
+    def get_default_model(self) -> str:
+        """
+        Get the default model for this provider.
+
+        Returns:
+            str: Default model name
+        """
+        return "large-v3"

test_stt_migration.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Test script for STT migration."""
+
+import sys
+import logging
+from pathlib import Path
+
+# Add src to path
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+def test_provider_availability():
+    """Test that providers can be imported and checked for availability."""
+    try:
+        from infrastructure.stt import STTProviderFactory, WhisperSTTProvider, ParakeetSTTProvider
+        
+        print("✓ Successfully imported STT providers")
+        
+        # Test factory
+        available_providers = STTProviderFactory.get_available_providers()
+        print(f"Available providers: {available_providers}")
+        
+        # Test individual providers
+        whisper = WhisperSTTProvider()
+        print(f"Whisper available: {whisper.is_available()}")
+        print(f"Whisper models: {whisper.get_available_models()}")
+        print(f"Whisper default model: {whisper.get_default_model()}")
+        
+        parakeet = ParakeetSTTProvider()
+        print(f"Parakeet available: {parakeet.is_available()}")
+        print(f"Parakeet models: {parakeet.get_available_models()}")
+        print(f"Parakeet default model: {parakeet.get_default_model()}")
+        
+        return True
+        
+    except Exception as e:
+        print(f"✗ Error testing providers: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def test_legacy_compatibility():
+    """Test legacy compatibility functions."""
+    try:
+        from infrastructure.stt import transcribe_audio, ASRFactory
+        
+        print("✓ Successfully imported legacy compatibility functions")
+        
+        # Test ASRFactory
+        try:
+            model = ASRFactory.get_model("whisper")
+            print(f"✓ ASRFactory created model: {model.provider_name}")
+        except Exception as e:
+            print(f"ASRFactory test failed (expected if dependencies missing): {e}")
+        
+        return True
+        
+    except Exception as e:
+        print(f"✗ Error testing legacy compatibility: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+def test_domain_integration():
+    """Test integration with domain models."""
+    try:
+        from domain.models.audio_content import AudioContent
+        from domain.models.text_content import TextContent
+        from domain.exceptions import SpeechRecognitionException
+        
+        print("✓ Successfully imported domain models")
+        
+        # Create test audio content
+        test_audio = AudioContent(
+            data=b"fake audio data for testing",
+            format="wav",
+            sample_rate=16000,
+            duration=1.0,
+            filename="test.wav"
+        )
+        
+        print(f"✓ Created test AudioContent: {test_audio.filename}")
+        
+        return True
+        
+    except Exception as e:
+        print(f"✗ Error testing domain integration: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+if __name__ == "__main__":
+    print("Testing STT migration...")
+    print("=" * 50)
+    
+    tests = [
+        ("Provider Availability", test_provider_availability),
+        ("Legacy Compatibility", test_legacy_compatibility),
+        ("Domain Integration", test_domain_integration)
+    ]
+    
+    results = []
+    for test_name, test_func in tests:
+        print(f"\n{test_name}:")
+        print("-" * 30)
+        result = test_func()
+        results.append((test_name, result))
+    
+    print("\n" + "=" * 50)
+    print("Test Results:")
+    for test_name, result in results:
+        status = "✓ PASS" if result else "✗ FAIL"
+        print(f"{test_name}: {status}")
+    
+    all_passed = all(result for _, result in results)
+    print(f"\nOverall: {'✓ ALL TESTS PASSED' if all_passed else '✗ SOME TESTS FAILED'}")
+    
+    sys.exit(0 if all_passed else 1)