Upload 3 files

stt/stt_factory.py

@@ -0,0 +1,106 @@
+"""
+STT Provider Factory for Flare
+"""
+from typing import Optional
+from stt_interface import STTInterface, STTEngineType
+from logger import log_info, log_error, log_warning, log_debug
+from stt_google import GoogleCloudSTT
+from config_provider import ConfigProvider
+from stt_interface import STTInterface
+
+# Import providers conditionally
+stt_providers = {}
+
+try:
+    from stt_google import GoogleCloudSTT
+    stt_providers['google'] = GoogleCloudSTT
+except ImportError:
+    log_info("⚠️ Google Cloud STT not available")
+
+try:
+    from stt_azure import AzureSTT
+    stt_providers['azure'] = AzureSTT
+except ImportError:
+    log_error("⚠️ Azure STT not available")
+
+try:
+    from stt_flicker import FlickerSTT
+    stt_providers['flicker'] = FlickerSTT
+except ImportError:
+    log_error("⚠️ Flicker STT not available")
+
+class NoSTT(STTInterface):
+    """Dummy STT provider when STT is disabled"""
+    
+    async def start_streaming(self, config) -> None:
+        pass
+    
+    async def stream_audio(self, audio_chunk: bytes):
+        return
+        yield  # Make it a generator
+    
+    async def stop_streaming(self):
+        return None
+    
+    def supports_realtime(self) -> bool:
+        return False
+    
+    def get_supported_languages(self):
+        return []
+    
+    def get_provider_name(self) -> str:
+        return "no_stt"
+        
+class STTFactory:
+    """Factory for creating STT providers"""
+    
+    @staticmethod
+    def create_provider() -> Optional[STTInterface]:
+        """Create STT provider based on configuration"""
+        try:
+            cfg = ConfigProvider.get()
+            stt_provider_config = cfg.global_config.stt_provider
+            stt_engine = stt_provider_config.name
+            
+            log_info(f"🎤 Creating STT provider: {stt_engine}")
+            
+            if stt_engine == "no_stt":
+                return NoSTT()
+            
+            # Get provider class
+            provider_class = stt_providers.get(stt_engine)
+            if not provider_class:
+                log_warning(f"⚠️ STT provider '{stt_engine}' not available")
+                return NoSTT()
+            
+            # Get API key or credentials
+            api_key = stt_provider_config.api_key
+            
+            if not api_key:
+                log_warning(f"⚠️ No API key configured for {stt_engine}")
+                return NoSTT()
+            
+            # Create provider instance
+            if stt_engine == "google":
+                # For Google, api_key is the path to credentials JSON
+                return provider_class(credentials_path=api_key)
+            elif stt_engine == "azure":
+                # For Azure, parse the key format
+                parts = api_key.split('|')
+                if len(parts) != 2:
+                    log_warning("⚠️ Invalid Azure STT key format. Expected: subscription_key|region")
+                    return NoSTT()
+                return provider_class(subscription_key=parts[0], region=parts[1])
+            elif stt_engine == "flicker":
+                return provider_class(api_key=api_key)
+            else:
+                return provider_class(api_key=api_key)
+                
+        except Exception as e:
+            log_error("❌ Failed to create STT provider", e)
+            return NoSTT()
+    
+    @staticmethod
+    def get_available_providers():
+        """Get list of available STT providers"""
+        return list(stt_providers.keys()) + ["no_stt"]

stt/stt_google.py

@@ -0,0 +1,496 @@
+"""
+Google Cloud Speech-to-Text Implementation
+"""
+import os
+import asyncio
+from typing import AsyncIterator, Optional, List, Any
+from datetime import datetime
+import sys
+import queue
+import threading
+import time
+import traceback
+from logger import log_info, log_error, log_debug, log_warning
+
+# Import Google Cloud Speech only if available
+try:
+    from google.cloud import speech
+    from google.api_core import exceptions
+    GOOGLE_SPEECH_AVAILABLE = True
+except ImportError:
+    GOOGLE_SPEECH_AVAILABLE = False
+    log_info("⚠️ Google Cloud Speech library not installed")
+
+from stt_interface import STTInterface, STTConfig, TranscriptionResult
+
+class GoogleCloudSTT(STTInterface):
+    """Google Cloud Speech-to-Text implementation"""
+    
+    def __init__(self, credentials_path: str):
+        if not GOOGLE_SPEECH_AVAILABLE:
+            raise ImportError("google-cloud-speech library not installed. Run: pip install google-cloud-speech")
+            
+        if credentials_path and os.path.exists(credentials_path):
+            os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = credentials_path
+            log_info(f"✅ Google credentials set from: {credentials_path}")
+            
+            # Test credential'ları
+            try:
+                # Client'ı burada oluşturma, her seferinde yeni instance oluştur
+                test_client = speech.SpeechClient()
+                log_info("🔐 Testing Google credentials...")
+                log_info("✅ Google credentials valid")
+                # Test client'ı kapat
+                if hasattr(test_client, 'transport') and hasattr(test_client.transport, 'close'):
+                    test_client.transport.close()
+            except Exception as e:
+                log_error(f"❌ Google credentials error", error=str(e))
+                raise
+        else:
+            log_error(f"❌ Google credentials path not found: {credentials_path}")
+            raise FileNotFoundError(f"Credentials file not found: {credentials_path}")
+            
+        # Client'ı burada oluşturma, start_streaming'de oluştur
+        self.client = None
+        self.streaming_config = None
+        self.is_streaming = False
+        self.audio_queue = None  # Queue'ları None olarak başlat
+        self.responses_queue = None
+        self.stream_thread = None
+        self.stop_event = threading.Event()
+        self.credentials_path = credentials_path
+        
+        # Session tracking
+        self.session_id = 0
+        self.total_audio_bytes = 0
+        self.total_chunks = 0
+
+    def _reset_session_data(self):
+        """Reset all session-specific data"""
+        # Queue'ları temizle
+        if self.audio_queue:
+            while not self.audio_queue.empty():
+                try:
+                    self.audio_queue.get_nowait()
+                except:
+                    pass
+
+        if self.responses_queue:
+            while not self.responses_queue.empty():
+                try:
+                    self.responses_queue.get_nowait()
+                except:
+                    pass
+        
+        # Counters'ı sıfırla
+        self.total_audio_bytes = 0
+        self.total_chunks = 0
+        
+        # Yeni session ID
+        self.session_id += 1
+        
+        log_info(f"🔄 Google STT session data reset. New session ID: {self.session_id}")    
+    
+    def _create_fresh_queues(self):
+        """Create fresh queue instances"""
+        # Eski queue'ları temizle
+        if self.audio_queue:
+            while not self.audio_queue.empty():
+                try:
+                    self.audio_queue.get_nowait()
+                except:
+                    pass
+                    
+        if self.responses_queue:
+            while not self.responses_queue.empty():
+                try:
+                    self.responses_queue.get_nowait()
+                except:
+                    pass
+        
+        # Yeni queue'lar oluştur
+        self.audio_queue = queue.Queue(maxsize=1000)  # Max size ekle
+        self.responses_queue = queue.Queue(maxsize=100)
+        log_debug("✅ Created fresh queues")
+        
+    async def start_streaming(self, config: dict) -> None:
+        """Initialize streaming session with clean state"""
+        try:
+            # Önce mevcut stream'i temizle
+            if self.is_streaming or self.stream_thread:
+                log_warning("⚠️ Previous stream still active, stopping it first")
+                await self.stop_streaming()
+                # Temizlik için bekle
+                await asyncio.sleep(0.5)
+            
+            # Session verilerini resetle ve ID'yi artır
+            self._reset_session_data()
+            
+            log_info(f"🎤 Starting Google STT streaming session #{self.session_id} with config: {config}")
+            
+            # Fresh queue'lar oluştur
+            self._create_fresh_queues()
+            
+            # Stop event'i temizle
+            self.stop_event.clear()
+            
+            # Yeni client oluştur (TEK SEFER)
+            self.client = speech.SpeechClient()
+            log_info("✅ Created new Google Speech client")
+            
+            # Convert dict to STTConfig if needed
+            if isinstance(config, dict):
+                stt_config = STTConfig(
+                    language=config.get("language", "tr-TR"),
+                    sample_rate=config.get("sample_rate", 16000),
+                    encoding=config.get("encoding", "WEBM_OPUS"),
+                    enable_punctuation=config.get("enable_punctuation", True),
+                    interim_results=config.get("interim_results", True),
+                    single_utterance=config.get("single_utterance", False)
+                )
+            else:
+                stt_config = config
+            
+            recognition_config = speech.RecognitionConfig(
+                encoding=self._get_encoding(stt_config.encoding),
+                sample_rate_hertz=stt_config.sample_rate,
+                language_code=stt_config.language,
+                enable_automatic_punctuation=stt_config.enable_punctuation,
+                model="latest_long",
+                use_enhanced=True
+            )
+            
+            self.streaming_config = speech.StreamingRecognitionConfig(
+                config=recognition_config,
+                interim_results=stt_config.interim_results,
+                single_utterance=stt_config.single_utterance
+            )
+            
+            self.is_streaming = True
+            
+            # Start streaming thread with unique name
+            self.stream_thread = threading.Thread(
+                target=self._run_stream,
+                name=f"GoogleSTT-Session-{self.session_id}"
+            )
+            self.stream_thread.daemon = True  # Daemon thread olarak işaretle
+            self.stream_thread.start()
+            
+            log_info(f"✅ Google STT streaming session #{self.session_id} started successfully")
+            
+        except Exception as e:
+            log_error(f"❌ Failed to start Google STT streaming", error=str(e))
+            self.is_streaming = False
+            self.client = None
+            self._create_fresh_queues()  # Hata durumunda da queue'ları temizle
+            raise
+
+    def _put_result(self, result: TranscriptionResult):
+        """Helper to put result in queue"""
+        try:
+            self.responses_queue.put(result)
+            # Debug log'u kaldırdık
+        except Exception as e:
+            log_error(f"❌ Error queuing result: {e}")
+    
+    def _run_stream(self):
+        """Run the streaming recognition in a separate thread"""
+        try:
+            log_info("🎤 Google STT stream thread started")
+                
+            def request_generator():
+                """Generate streaming requests"""
+                chunk_count = 0
+                total_bytes = 0
+                first_chunk_processed = False
+                
+                while not self.stop_event.is_set():
+                    try:
+                        chunk = self.audio_queue.get(timeout=0.1)
+                        if chunk is None:
+                            log_info("📛 Poison pill received, stopping request generator")
+                            break
+                        
+                        chunk_count += 1
+                        total_bytes += len(chunk)
+                        
+                        # İlk chunk'ta audio format kontrolü
+                        if chunk_count == 1:
+                            log_info(f"📤 First chunk - size: {len(chunk)} bytes")
+                            # Audio header kontrolü (WEBM magic bytes)
+                            if len(chunk) >= 4:
+                                if chunk[:4] == b'\x1a\x45\xdf\xa3':
+                                    log_info("✅ Valid WEBM header detected")
+                                    first_chunk_processed = True
+                                else:
+                                    log_error(f"❌ Invalid audio format, first 4 bytes: {chunk[:4].hex()}")
+                                    # Hatalı format, stream'i durdur
+                                    break
+                        
+                        # İlk chunk geçerliyse devam et
+                        if chunk_count == 1 and not first_chunk_processed:
+                            break
+                        
+                        # Her 100 chunk'ta durum raporu
+                        if chunk_count % 100 == 0:
+                            avg_chunk_size = total_bytes / chunk_count
+                            log_info(f"📤 Progress: {chunk_count} chunks, {total_bytes/1024:.1f}KB total, avg {avg_chunk_size:.0f} bytes/chunk")
+                        
+                        yield speech.StreamingRecognizeRequest(audio_content=chunk)
+                        
+                    except queue.Empty:
+                        continue
+                    except Exception as e:
+                        log_error(f"❌ Error in request generator: {e}")
+                        break
+                
+            # Create streaming client
+            requests = request_generator()
+            
+            log_info("🎤 Creating Google STT streaming client...")
+            
+            try:
+                responses = self.client.streaming_recognize(
+                    self.streaming_config, 
+                    requests, 
+                    timeout=300 # 5 dakika timeout
+                ) 
+                
+                log_info("✅ Google STT streaming client created")
+    
+                # Response timeout kontrolü
+                last_response_time = time.time()
+                RESPONSE_TIMEOUT = 30  # 30 saniye içinde response gelmezse
+                
+                # Process responses
+                response_count = 0
+                empty_response_count = 0
+                
+                for response in responses:
+                    last_response_time = time.time()
+                    response_count += 1
+    
+                    # Response type'ı logla
+                    if response_count == 1:
+                        log_info(f"📨 First response received from Google STT")
+                    
+                    if self.stop_event.is_set():
+                        log_info("🛑 Stop event detected, breaking response loop")
+                        break
+                    
+                    # Response içeriğini kontrol et
+                    if not response.results:
+                        empty_response_count += 1
+                        if empty_response_count == 1:
+                            log_debug("📭 Received empty response (no results)")
+                        continue
+    
+                    for i, result in enumerate(response.results):
+                        log_debug(f"📋 Result {i}: is_final={result.is_final}, alternatives={len(result.alternatives)}")
+                        
+                        if not result.alternatives:
+                            log_debug(f"📋 Result {i} has no alternatives")
+                            continue
+                            
+                        # İlk alternatifi al
+                        alternative = result.alternatives[0]
+                        
+                        # Sadece anlamlı text'leri işle
+                        if alternative.transcript.strip():
+                            # Create transcription result
+                            transcription = TranscriptionResult(
+                                text=alternative.transcript,
+                                is_final=result.is_final,
+                                confidence=alternative.confidence if hasattr(alternative, 'confidence') and alternative.confidence else 0.0,
+                                timestamp=datetime.now().timestamp()
+                            )
+                            
+                            # Put result in queue
+                            self._put_result(transcription)
+                            
+                            # SADECE final result'ları logla
+                            if result.is_final:
+                                log_info(f"🎯 GOOGLE STT FINAL: '{alternative.transcript}'")
+                        else:
+                            log_debug(f"📋 Result {i} has empty transcript")
+                            continue    
+    
+                if time.time() - last_response_time > RESPONSE_TIMEOUT:
+                    log_error(f"❌ No response from Google STT for {RESPONSE_TIMEOUT} seconds")
+    
+                log_info(f"📊 Google STT stream ended. Total responses: {response_count}, Empty: {empty_response_count}")
+                        
+            except Exception as e:
+                error_msg = str(e)
+                
+                # Detaylı hata mesajları
+                if "Exceeded maximum allowed stream duration" in error_msg:
+                    log_warning("⚠️ Stream duration limit exceeded (5 minutes). This is expected for long sessions.")
+                elif "Bad language code" in error_msg:
+                    log_error(f"❌ Invalid language code in STT config. Check locale settings.")
+                elif "invalid_argument" in error_msg:
+                    log_error(f"❌ Invalid STT configuration. Check encoding and sample rate.")
+                elif "Deadline Exceeded" in error_msg:
+                    log_error(f"❌ Google STT response timeout - possibly network issue or slow connection")
+                elif "503" in error_msg or "Service Unavailable" in error_msg:
+                    log_error(f"❌ Google STT service temporarily unavailable. Will retry...")
+                else:
+                    log_error(f"❌ Google STT stream error: {error_msg}")
+                    
+        except Exception as e:
+            log_error(f"❌ Fatal error in STT stream thread", error=str(e), traceback=traceback.format_exc())
+        finally:
+            log_info("🎤 Google STT stream thread ended")
+            # Thread bittiğinde streaming flag'ini kapat
+            self.is_streaming = False
+            
+    async def stream_audio(self, audio_chunk: bytes) -> AsyncIterator[TranscriptionResult]:
+        """Stream audio chunk and get transcription results"""
+        if not self.is_streaming:
+            # Daha detaylı hata mesajı
+            log_error(f"❌ STT not streaming - is_streaming: {self.is_streaming}, thread alive: {self.stream_thread and self.stream_thread.is_alive() if hasattr(self, 'stream_thread') else 'No thread'}")
+            raise RuntimeError("Streaming not started. Call start_streaming() first.")
+        
+        try:
+            # Put audio in queue for streaming thread
+            self.audio_queue.put(audio_chunk)
+            
+            # Check for any results in queue
+            while True:
+                try:
+                    # Non-blocking get from normal queue
+                    result = self.responses_queue.get_nowait()
+                    yield result
+                except queue.Empty:
+                    # No more results in queue
+                    break
+                    
+        except Exception as e:
+            log_error(f"❌ Google STT streaming error", error=str(e))
+            # Stream'i tekrar başlatmayı tetikle
+            self.is_streaming = False
+            raise
+    
+    async def stop_streaming(self) -> Optional[TranscriptionResult]:
+        """Stop streaming and clean up all resources"""
+        if not self.is_streaming and not self.stream_thread:
+            log_debug("Already stopped, nothing to do")
+            return None
+            
+        try:
+            log_info(f"🛑 Stopping Google STT streaming session #{self.session_id}")
+            
+            # Flag'i hemen kapat
+            self.is_streaming = False
+            self.stop_event.set()
+            
+            # Send poison pill to stop request generator
+            if self.audio_queue:
+                try:
+                    self.audio_queue.put(None)
+                except:
+                    pass
+            
+            # Thread'i durdur
+            if self.stream_thread and self.stream_thread.is_alive():
+                log_info("⏳ Waiting for stream thread to finish...")
+                self.stream_thread.join(timeout=5.0)  # 5 saniye bekle
+                
+                if self.stream_thread.is_alive():
+                    log_warning("⚠️ STT thread did not stop gracefully after 5s")
+                    # Thread'i zorla sonlandıramayız Python'da, ama daemon olduğu için
+                    # ana program kapanınca otomatik kapanacak
+                else:
+                    log_info("✅ Stream thread finished")
+            
+            # Final result'ı al
+            final_result = None
+            if self.responses_queue:
+                while not self.responses_queue.empty():
+                    try:
+                        result = self.responses_queue.get_nowait()
+                        if result.is_final:
+                            final_result = result
+                    except:
+                        pass
+            
+            # Client'ı kapat
+            if self.client:
+                try:
+                    # Transport'u kapat
+                    if hasattr(self.client, 'transport') and hasattr(self.client.transport, 'close'):
+                        self.client.transport.close()
+                        log_debug("✅ Client transport closed")
+                    
+                    # gRPC channel'ı kapat
+                    if hasattr(self.client, '_transport') and hasattr(self.client._transport, '_grpc_channel'):
+                        self.client._transport._grpc_channel.close()
+                        log_debug("✅ gRPC channel closed")
+                except Exception as e:
+                    log_warning(f"⚠️ Error closing Google client: {e}")
+                finally:
+                    self.client = None
+            
+            # Queue'ları None yap (yeniden kullanım için fresh queue gerekecek)
+            self.audio_queue = None
+            self.responses_queue = None
+            
+            # Diğer değişkenleri resetle
+            self.stream_thread = None
+            self.streaming_config = None
+            self.stop_event.clear()
+            
+            log_info(f"✅ Google STT streaming session #{self.session_id} stopped and cleaned")
+            return final_result
+            
+        except Exception as e:
+            log_error(f"❌ Error during stop_streaming", error=str(e))
+            # Force cleanup on error
+            self.is_streaming = False
+            self.stream_thread = None
+            self.client = None
+            self.streaming_config = None
+            self.stop_event.clear()
+            self.audio_queue = None
+            self.responses_queue = None
+            return None
+    
+    def supports_realtime(self) -> bool:
+        """Google Cloud STT supports real-time streaming"""
+        return True
+    
+    def get_supported_languages(self) -> List[str]:
+        """Get list of supported language codes"""
+        return [
+            "tr-TR",  # Turkish
+            "en-US",  # English (US)
+            "en-GB",  # English (UK)
+            "de-DE",  # German
+            "fr-FR",  # French
+            "es-ES",  # Spanish
+            "it-IT",  # Italian
+            "pt-BR",  # Portuguese (Brazil)
+            "ru-RU",  # Russian
+            "ja-JP",  # Japanese
+            "ko-KR",  # Korean
+            "zh-CN",  # Chinese (Simplified)
+            "ar-SA",  # Arabic
+        ]
+    
+    def get_provider_name(self) -> str:
+        """Get provider name"""
+        return "google"
+    
+    def _get_encoding(self, encoding_str: str):
+        """Convert encoding string to Google Speech enum"""
+        if not GOOGLE_SPEECH_AVAILABLE:
+            return None
+            
+        encoding_map = {
+            "WEBM_OPUS": speech.RecognitionConfig.AudioEncoding.WEBM_OPUS,
+            "LINEAR16": speech.RecognitionConfig.AudioEncoding.LINEAR16,
+            "FLAC": speech.RecognitionConfig.AudioEncoding.FLAC,
+            "MP3": speech.RecognitionConfig.AudioEncoding.MP3,
+            "OGG_OPUS": speech.RecognitionConfig.AudioEncoding.OGG_OPUS,
+        }
+        return encoding_map.get(encoding_str, speech.RecognitionConfig.AudioEncoding.WEBM_OPUS)

stt/stt_interface.py

@@ -0,0 +1,80 @@
+"""
+STT (Speech-to-Text) Interface for Flare
+"""
+from abc import ABC, abstractmethod
+from typing import Optional, Dict, Any, AsyncIterator, List
+from dataclasses import dataclass
+from enum import Enum
+import json
+
+class STTEngineType(Enum):
+    NO_STT = "no_stt"
+    GOOGLE = "google"
+    AZURE = "azure"
+    AMAZON = "amazon"
+    FLICKER = "flicker"
+
+@dataclass
+class STTConfig:
+    """STT configuration parameters"""
+    language: str = "tr-TR"
+    sample_rate: int = 16000
+    encoding: str = "WEBM_OPUS"
+    enable_punctuation: bool = True
+    enable_word_timestamps: bool = False
+    model: str = "latest_long"
+    use_enhanced: bool = True
+    single_utterance: bool = False
+    interim_results: bool = True
+    
+    # Voice Activity Detection
+    vad_enabled: bool = True
+    speech_timeout_ms: int = 2000
+    
+    # Noise reduction
+    noise_reduction_enabled: bool = True
+    noise_reduction_level: int = 2
+
+@dataclass
+class TranscriptionResult:
+    """Result from STT engine"""
+    text: str
+    is_final: bool
+    confidence: float
+    timestamp: float
+    word_timestamps: Optional[List[Dict]] = None
+    language: Optional[str] = None
+    is_interrupt: bool = False
+
+class STTInterface(ABC):
+    """Abstract base class for STT providers"""
+    
+    @abstractmethod
+    async def start_streaming(self, config: STTConfig) -> None:
+        """Start streaming session"""
+        pass
+    
+    @abstractmethod
+    async def stream_audio(self, audio_chunk: bytes) -> AsyncIterator[TranscriptionResult]:
+        """Stream audio chunk and get transcription results"""
+        pass
+    
+    @abstractmethod
+    async def stop_streaming(self) -> Optional[TranscriptionResult]:
+        """Stop streaming and get final result"""
+        pass
+    
+    @abstractmethod
+    def supports_realtime(self) -> bool:
+        """Check if provider supports real-time streaming"""
+        pass
+    
+    @abstractmethod
+    def get_supported_languages(self) -> List[str]:
+        """Get list of supported language codes"""
+        pass
+    
+    @abstractmethod
+    def get_provider_name(self) -> str:
+        """Get provider name for logging"""
+        pass