""" Smart Speaker - Main Application Голосовой ассистент с wake word detection, STT, AI и TTS. Flow: 1. Wait for wake word ("Alexandr") 2. Listen to user speech (STT) 3. Send query to AI (Perplexity) 4. Clean response from markdown 5. Speak response (TTS) 6. Loop back to step 1 """ # Главный файл приложения (`main.py`). # Здесь находится основной бесконечный цикл, который связывает все компоненты воедино. import os import queue import re import signal import sys import threading from collections import deque # Для воспроизведения звуков (mp3) try: from pygame import mixer except Exception as exc: mixer = None _MIXER_IMPORT_ERROR = exc else: _MIXER_IMPORT_ERROR = None # Импорт наших модулей from .audio.sound_level import parse_volume_text, set_volume from .audio.stt import cleanup as cleanup_stt from .audio.stt import get_recognizer, listen from .audio.tts import initialize as init_tts from .audio.tts import speak from .audio.wakeword import ( check_wakeword_once, wait_for_wakeword, ) from .audio.wakeword import ( cleanup as cleanup_wakeword, ) from .audio.wakeword import ( stop_monitoring as stop_wakeword_monitoring, ) from .core.ai import ask_ai, ask_ai_stream, translate_text from .core.cleaner import clean_response from .features.alarm import get_alarm_clock from .features.timer import get_timer_manager from .features.weather import get_weather_report # Список стоп-слов, чтобы прервать диалог или остановить ассистента STOP_WORDS = { "стоп", "хватит", "перестань", "замолчи", "прекрати", "тихо", "stop", } def signal_handler(sig, frame): """ Обработчик сигнала Ctrl+C. Позволяет корректно завершить работу программы, освободив ресурсы (микрофон, модели). """ print("\n\n👋 Завершение работы...") cleanup_wakeword() # Остановка Porcupine cleanup_stt() # Остановка Deepgram sys.exit(0) def parse_translation_request(text: str): """ Определяет, является ли фраза запросом на перевод. Пример: "Переведи на английский привет мир" Возвращает словарь: {'source_lang': 'ru', 'target_lang': 'en', 'text': 'привет мир'} Или None, если это не запрос перевода. """ text_lower = text.lower().strip() # Список префиксов команд перевода и соответствующих направлений языков commands = [ ("переведи на английский", "ru", "en"), ("переведи на русский", "en", "ru"), ("переведи с английского", "en", "ru"), ("переведи с русского", "ru", "en"), ("как по-английски", "ru", "en"), ("как по английски", "ru", "en"), ("как по-русски", "en", "ru"), ("как по русски", "en", "ru"), ("translate to english", "ru", "en"), ("translate into english", "ru", "en"), ("translate to russian", "en", "ru"), ("translate into russian", "en", "ru"), ("translate from english", "en", "ru"), ("translate from russian", "ru", "en"), ] for prefix, source_lang, target_lang in commands: if text_lower.startswith(prefix): # Отрезаем команду (префикс), оставляем только текст для перевода rest = text[len(prefix) :].strip() return { "source_lang": source_lang, "target_lang": target_lang, "text": rest, } return None def is_stop_command(text: str) -> bool: """ Проверяет, содержится ли в тексте команда остановки. Удаляет знаки препинания и ищет слова из списка STOP_WORDS. """ text_lower = text.lower() for ch in ",.!?:;": text_lower = text_lower.replace(ch, " ") words = text_lower.split() for word in words: if word in STOP_WORDS: return True return False def main(): """ Основная функция (точка входа). """ print("=" * 50) print("🔊 УМНАЯ КОЛОНКА") print("=" * 50) print("Скажите 'Alexandr' для активации") print("Нажмите Ctrl+C для выхода") print("=" * 50) print() # Устанавливаем перехватчик Ctrl+C signal.signal(signal.SIGINT, signal_handler) # Предварительная инициализация моделей print("⏳ Инициализация моделей...") # Инициализация звуковой системы для эффектов (опционально) ding_sound = None if mixer is None: print( "Warning: pygame mixer not available; sound effects disabled." f" ({_MIXER_IMPORT_ERROR})" ) else: try: mixer.init() except Exception as exc: print(f"Warning: pygame mixer init failed; sound effects disabled. ({exc})") else: ding_sound_path = "assets/sounds/ding.wav" if os.path.exists(ding_sound_path): ding_sound = mixer.Sound(ding_sound_path) ding_sound.set_volume(0.3) else: print(f"⚠️ Звук {ding_sound_path} не найден") get_recognizer().initialize() # Подключение к Deepgram init_tts() # Загрузка нейросети для синтеза речи (Silero) alarm_clock = get_alarm_clock() # Загрузка будильников timer_manager = get_timer_manager() # Загрузка таймеров print() # История чата (храним последние 10 обменов репликами для контекста) chat_history = deque(maxlen=20) # Переменная для хранения последнего ответа ассистента last_response = None # Переменная, указывающая, нужно ли пропускать ожидание wake word # (True = режим диалога, слушаем сразу. False = ждем "Alexandr") skip_wakeword = False # БЕСКОНЕЧНЫЙ ЦИКЛ РАБОТЫ while True: try: # Гарантируем, что микрофон детектора wake word освобожден stop_wakeword_monitoring() # --- Проверка таймеров --- # Проверяем каждую итерацию. Если таймер сработал, он заблокирует выполнение, пока его не выключат. if timer_manager.check_timers(): skip_wakeword = False continue # --- Проверка будильников --- # Проверяем каждую итерацию. Если будильник сработал, он заблокирует выполнение, пока его не выключат. if alarm_clock.check_alarms(): # Если будильник прозвенел и был выключен пользователем, сбрасываем режим диалога skip_wakeword = False continue # --- Шаг 1: Активация --- if not skip_wakeword: # Ожидание фразы "Alexandr". Используем таймаут 1 сек, чтобы часто проверять будильники. detected = wait_for_wakeword(timeout=1.0) # Если время вышло, а фразы не было — начинаем цикл заново (проверяем будильники) if not detected: continue # Воспроизводим звук активации if ding_sound: ding_sound.play() # Фраза услышана! Слушаем команду пользователя (7 секунд тишины макс) user_text = listen(timeout_seconds=7.0) else: # Режим диалога (Follow-up): ждем продолжения речи без "Alexandr" print("👂 Слушаю продолжение диалога (5 сек)...") # Ждем начала речи 5 сек. Если начали говорить, слушаем до 10 сек. user_text = listen(timeout_seconds=10.0, detection_timeout=5.0) if not user_text: # Пользователь промолчал — выходим из режима диалога, засыпаем. skip_wakeword = False continue # --- Шаг 2: Анализ распознанного текста --- if not user_text: # Была активация, но речь не распознана speak("Извините, я вас не расслышал. Попробуйте ещё раз.") skip_wakeword = False # Возвращаемся в режим ожидания имени continue # Проверка на команду "Стоп" if is_stop_command(user_text): print("_" * 50) print("💤 Жду 'Alexandr' для активации...") skip_wakeword = False continue # Проверка на команду "Повтори" / "Еще раз" user_text_lower = user_text.lower().strip() repeat_phrases = [ "еще раз", "повтори", "скажи еще раз", "что ты сказал", "повтори пожалуйста", "александр еще раз", "еще раз александр", "александр повтори", "повтори александр", ] # Проверяем точное совпадение или если фраза начинается с "повтори" (но не "повтори за мной") if user_text_lower in repeat_phrases or ( user_text_lower.startswith("повтори") and "за мной" not in user_text_lower ): if last_response: print(f"🔁 Повторяю: {last_response}") speak(last_response) else: speak("Я еще ничего не говорил.") # После повтора остаемся в диалоге skip_wakeword = True continue # Проверка команд таймера ("поставь таймер на 6 минут") timer_response = timer_manager.parse_command(user_text) if timer_response: clean_timer_response = clean_response(timer_response, language="ru") completed = speak( clean_timer_response, check_interrupt=check_wakeword_once ) last_response = clean_timer_response skip_wakeword = not completed continue # Проверка команд будильника ("поставь будильник на 7") alarm_response = alarm_clock.parse_command(user_text) if alarm_response: clean_alarm_response = clean_response(alarm_response, language="ru") speak(clean_alarm_response) last_response = clean_alarm_response skip_wakeword = False continue # Проверка команды громкости ("громкость 5") if user_text.lower().startswith("громкость"): try: # Убираем слово "громкость" и ищем число vol_str = user_text.lower().replace("громкость", "", 1).strip() level = parse_volume_text(vol_str) if level is not None: if set_volume(level): msg = f"Громкость установлена на {level}" clean_msg = clean_response(msg, language="ru") speak(clean_msg) last_response = clean_msg else: speak("Не удалось установить громкость.") else: speak( "Я не понял число громкости. Скажите число от одного до десяти." ) skip_wakeword = True continue except Exception as e: print(f"❌ Ошибка громкости: {e}") speak("Не удалось изменить громкость.") skip_wakeword = True continue # Проверка команды "Погода" weather_triggers = [ "погода", "погоду", "что на улице", "какая температура", "сколько градусов", "холодно ли", "жарко ли", "нужен ли зонт", "брать ли зонт", "прогноз погоды", ] if any(trigger in user_text.lower() for trigger in weather_triggers): weather_report = get_weather_report() clean_report = clean_response(weather_report, language="ru") speak(clean_report) last_response = clean_report skip_wakeword = True continue # Проверка запроса на перевод translation_request = parse_translation_request(user_text) if translation_request: source_lang = translation_request["source_lang"] target_lang = translation_request["target_lang"] text_to_translate = translation_request["text"] # Если сказано только "переведи на английский", спрашиваем "что перевести?" if not text_to_translate: prompt = ( "Скажи фразу на английском." if source_lang == "en" else "Скажи фразу на русском." ) speak(prompt) # Слушаем саму фразу на нужном языке text_to_translate = listen( timeout_seconds=7.0, detection_timeout=5.0, lang=source_lang ) if not text_to_translate: speak("Я не расслышал текст для перевода.") skip_wakeword = False continue # Выполняем перевод через AI translated_text = translate_text( text_to_translate, source_lang, target_lang ) # Очищаем результат (убираем лишние символы) clean_text = clean_response(translated_text, language=target_lang) # Сохраняем для повтора last_response = clean_text # Озвучиваем перевод на целевом языке completed = speak( clean_text, check_interrupt=check_wakeword_once, language=target_lang, ) stop_wakeword_monitoring() skip_wakeword = True # Остаемся в диалоге if not completed: print("⏹️ Перевод прерван - слушаю следующий вопрос") continue # --- Шаг 3: Запрос к AI (Streaming) --- # Добавляем сообщение пользователя в историю chat_history.append({"role": "user", "content": user_text}) # Очередь для предложений, которые нужно озвучить tts_q = queue.Queue() # Флаг прерывания для worker-а interrupt_event = threading.Event() def tts_worker(): """Фоновый поток, читающий предложения из очереди и озвучивающий их.""" while True: item = tts_q.get() if item is None: # Poison pill (сигнал остановки) tts_q.task_done() break text, lang = item # Если уже было прерывание, просто пропускаем (чистим очередь) if interrupt_event.is_set(): tts_q.task_done() continue # Озвучиваем completed = speak( text, check_interrupt=check_wakeword_once, language=lang ) if not completed: interrupt_event.set() # Сообщаем всем, что нас перебили tts_q.task_done() # Запускаем поток озвучки worker_thread = threading.Thread(target=tts_worker, daemon=True) worker_thread.start() full_response = "" buffer = "" try: # Получаем генератор потока от AI stream_generator = ask_ai_stream(list(chat_history)) print("🤖 AI говорит: ", end="", flush=True) for chunk in stream_generator: # Если в процессе генерации нас перебили (на ранних фразах), прерываем получение if interrupt_event.is_set(): break buffer += chunk full_response += chunk print(chunk, end="", flush=True) # Проверяем на конец предложения (. ! ? + пробел или конец строки) # Эвристика: ищем знаки препинания, после которых идет пробел или перевод строки if re.search(r"[.!?\n]+(?:\s|$)", buffer): # Очищаем и отправляем в очередь clean_chunk = clean_response(buffer, language="ru") if clean_chunk.strip(): tts_q.put((clean_chunk, "ru")) buffer = "" # Отправляем остаток (если есть) if buffer.strip() and not interrupt_event.is_set(): clean_chunk = clean_response(buffer, language="ru") if clean_chunk.strip(): tts_q.put((clean_chunk, "ru")) except Exception as e: print(f"\n❌ Ошибка стриминга: {e}") speak("Произошла ошибка при получении ответа.") # Ждем, пока все договорится # Добавляем poison pill, чтобы поток завершился, когда очередь пуста tts_q.put(None) worker_thread.join() print() # Перенос строки после вывода AI # Добавляем полный ответ AI в историю chat_history.append({"role": "assistant", "content": full_response}) # Сохраняем для "повтори" last_response = clean_response(full_response, language="ru") # После озвучки обязательно закрываем поток микрофона stop_wakeword_monitoring() # Включаем режим диалога (следующий запрос можно говорить без имени) skip_wakeword = True if interrupt_event.is_set(): print("⏹️ Ответ прерван - слушаю следующий вопрос") # Если перебили, цикл перезапустится и skip_wakeword уже True print() print("-" * 30) print() # --- Шаг 6: Конец итерации, возврат в начало цикла --- except KeyboardInterrupt: signal_handler(None, None) except Exception as e: print(f"❌ Ошибка: {e}") speak("Произошла ошибка. Попробуйте ещё раз.") skip_wakeword = False if __name__ == "__main__": main()