""" Speech-to-Text module using Deepgram API. Recognizes speech from microphone using streaming WebSocket. Supports Russian (default) and English. """ # Модуль распознавания речи (STT - Speech-to-Text). # Использует Deepgram API через веб-сокеты для потокового распознавания в реальном времени. import asyncio import time import pyaudio import logging from ..core.config import DEEPGRAM_API_KEY, SAMPLE_RATE from deepgram import ( DeepgramClient, DeepgramClientOptions, LiveTranscriptionEvents, LiveOptions, ) import deepgram.clients.common.v1.abstract_sync_websocket as sdk_ws import websockets.sync.client from ..core.audio_manager import get_audio_manager # --- Патч (исправление) для библиотеки websockets --- # По умолчанию Deepgram SDK использует слишком короткий таймаут подключения. # Это часто вызывает ошибки при медленном SSL рукопожатии. # Мы подменяем функцию connect, чтобы увеличить таймаут до 30 секунд. _original_connect = websockets.sync.client.connect def _patched_connect(*args, **kwargs): kwargs.setdefault("open_timeout", 30) kwargs.setdefault("ping_timeout", 30) kwargs.setdefault("close_timeout", 30) print(f"DEBUG: Connecting to Deepgram with timeout={kwargs.get('open_timeout')}s") return _original_connect(*args, **kwargs) # Применяем патч sdk_ws.connect = _patched_connect # Отключаем лишний мусор в логах logging.getLogger("deepgram").setLevel(logging.WARNING) class SpeechRecognizer: """Класс распознавания речи через Deepgram.""" def __init__(self): self.dg_client = None self.pa = None self.stream = None self.transcript = "" def initialize(self): """Инициализация клиента Deepgram и PyAudio.""" if not DEEPGRAM_API_KEY: raise ValueError("DEEPGRAM_API_KEY is not set in environment or config.") print("📦 Инициализация Deepgram STT...") config = DeepgramClientOptions( verbose=logging.WARNING, ) self.dg_client = DeepgramClient(DEEPGRAM_API_KEY, config) self.pa = get_audio_manager().get_pyaudio() print("✅ Deepgram клиент готов") def _get_stream(self): """Открывает аудиопоток PyAudio, если он еще не открыт.""" if self.stream is None: self.stream = self.pa.open( rate=SAMPLE_RATE, channels=1, format=pyaudio.paInt16, input=True, frames_per_buffer=4096, ) return self.stream async def _process_audio(self, dg_connection, timeout_seconds, detection_timeout): """ Асинхронная функция для отправки аудио и получения текста. Args: dg_connection: Активное соединение с Deepgram. timeout_seconds: Общее время прослушивания. detection_timeout: Время ожидания начала речи. """ self.transcript = "" transcript_parts = [] loop = asyncio.get_running_loop() stream = self._get_stream() # События для синхронизации stop_event = asyncio.Event() # Пора останавливаться speech_started_event = asyncio.Event() # Речь обнаружена (VAD) # --- Обработчики событий Deepgram --- def on_transcript(unused_self, result, **kwargs): """Вызывается, когда приходит часть текста.""" sentence = result.channel.alternatives[0].transcript if len(sentence) == 0: return if result.is_final: # Собираем только финальные (подтвержденные) фразы transcript_parts.append(sentence) self.transcript = " ".join(transcript_parts).strip() def on_speech_started(unused_self, speech_started, **kwargs): """Вызывается, когда VAD (Voice Activity Detection) слышит голос.""" loop.call_soon_threadsafe(speech_started_event.set) def on_utterance_end(unused_self, utterance_end, **kwargs): """Вызывается, когда Deepgram решает, что фраза закончилась (пауза).""" loop.call_soon_threadsafe(stop_event.set) def on_error(unused_self, error, **kwargs): print(f"Error: {error}") loop.call_soon_threadsafe(stop_event.set) # Подписываемся на события dg_connection.on(LiveTranscriptionEvents.Transcript, on_transcript) dg_connection.on(LiveTranscriptionEvents.SpeechStarted, on_speech_started) dg_connection.on(LiveTranscriptionEvents.UtteranceEnd, on_utterance_end) dg_connection.on(LiveTranscriptionEvents.Error, on_error) # Параметры распознавания options = LiveOptions( model="nova-2", # Самая быстрая и точная модель language=self.current_lang, smart_format=True, # Расстановка знаков препинания encoding="linear16", channels=1, sample_rate=SAMPLE_RATE, interim_results=True, utterance_end_ms=1000, # Пауза 1.0с считается концом фразы (было 1.2) vad_events=True, ) # --- Задача отправки аудио с буферизацией --- async def send_audio(): chunks_sent = 0 audio_buffer = [] # Буфер для накопления звука во время подключения try: # 1. Сразу начинаем захват звука, не дожидаясь сети! stream.start_stream() print("🎤 Stream started (buffering)...") # 2. Запускаем подключение к Deepgram в фоне (через ThreadPool, т.к. start() блокирующий) # Но в данном SDK start() возвращает bool, он может быть блокирующим. # Deepgram Python SDK v3+ start() делает handshake. connect_future = loop.run_in_executor( None, lambda: dg_connection.start(options) ) # Пока подключаемся, копим данные while not connect_future.done(): if stream.is_active(): data = stream.read(4096, exception_on_overflow=False) audio_buffer.append(data) await asyncio.sleep(0.001) # Проверяем результат подключения if connect_future.result() is False: print("Failed to start Deepgram connection") return print(f"🚀 Connected! Sending buffer ({len(audio_buffer)} chunks)...") # 3. Отправляем накопленный буфер for chunk in audio_buffer: dg_connection.send(chunk) chunks_sent += 1 audio_buffer = None # Освобождаем память # 4. Продолжаем стримить в реальном времени while not stop_event.is_set(): if stream.is_active(): data = stream.read(4096, exception_on_overflow=False) dg_connection.send(data) chunks_sent += 1 if chunks_sent % 50 == 0: print(f".", end="", flush=True) await asyncio.sleep(0.005) except Exception as e: print(f"Audio send error: {e}") finally: if stream.is_active(): stream.stop_stream() print(f"\n🛑 Stream stopped. Chunks sent: {chunks_sent}") sender_task = asyncio.create_task(send_audio()) if False: # dg_connection.start(options) перенесен внутрь send_audio pass try: # 1. Ждем начала речи (если задан detection_timeout) if detection_timeout: try: await asyncio.wait_for( speech_started_event.wait(), timeout=detection_timeout ) except asyncio.TimeoutError: # Если за detection_timeout (5 сек) никто не начал говорить, выходим stop_event.set() # 2. Если речь началась (или таймаута нет), ждем завершения (stop_event) # stop_event сработает либо по UtteranceEnd (пауза), либо по общему таймауту if not stop_event.is_set(): await asyncio.wait_for(stop_event.wait(), timeout=timeout_seconds) except asyncio.TimeoutError: pass # Общий таймаут вышел stop_event.set() await sender_task # Завершаем соединение и ждем последние результаты dg_connection.finish() return self.transcript def listen( self, timeout_seconds: float = 7.0, detection_timeout: float = None, lang: str = "ru", ) -> str: """ Основной метод: слушает микрофон и возвращает текст. Args: timeout_seconds: Максимальная длительность фразы. detection_timeout: Сколько ждать начала речи перед тем как сдаться. lang: Язык ("ru" или "en"). """ if not self.dg_client: self.initialize() self.current_lang = lang print(f"🎙️ Слушаю ({lang})...") last_error = None # Делаем 2 попытки на случай сбоя сети for attempt in range(2): # Создаем новое live подключение для каждой сессии dg_connection = self.dg_client.listen.live.v("1") try: # Запускаем асинхронный процесс обработки transcript = asyncio.run( self._process_audio( dg_connection, timeout_seconds, detection_timeout ) ) final_text = transcript.strip() if transcript else "" if final_text: print(f"📝 Распознано: {final_text}") return final_text else: # Если вернулась пустая строка (тишина), считаем это штатным завершением. # Не нужно повторять попытку, как при ошибке сети. return "" except Exception as e: last_error = e if attempt == 0: print("⚠️ Не удалось подключиться к Deepgram, повторяю...") time.sleep(1) if last_error: print(f"❌ Ошибка STT: {last_error}") else: print("⚠️ Речь не распознана") return "" def cleanup(self): """Очистка ресурсов.""" if self.stream: self.stream.stop_stream() self.stream.close() self.stream = None # self.pa.terminate() - Используем общий менеджер # Глобальный экземпляр _recognizer = None def get_recognizer() -> SpeechRecognizer: global _recognizer if _recognizer is None: _recognizer = SpeechRecognizer() return _recognizer def listen( timeout_seconds: float = 7.0, detection_timeout: float = None, lang: str = "ru" ) -> str: """Внешняя функция для прослушивания.""" return get_recognizer().listen(timeout_seconds, detection_timeout, lang) def cleanup(): """Внешняя функция очистки.""" global _recognizer if _recognizer: _recognizer.cleanup() _recognizer = None