#!/usr/bin/env python3 """ 땡구 음성비서 openWakeWord -> Groq Whisper API -> OpenAI -> HA 제어 -> edge-tts (무료 Microsoft 신경망 TTS) 설치: pip install openwakeword sounddevice openai pygame requests numpy groq edge-tts 실행: python ddanggu_voice.py """ import os, sys, json, time, re, math import subprocess import tempfile, threading, wave, requests from collections import deque import xml.etree.ElementTree as ET import numpy as np from datetime import datetime from urllib.parse import quote # ════════════════════════════════════════ # settings.env 로드 (Windows \r 자동 제거) # ════════════════════════════════════════ _ENV_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "settings.env") if os.path.exists(_ENV_FILE): with open(_ENV_FILE, encoding="utf-8") as _f: for _line in _f: _line = _line.strip().replace("\r", "") if _line and not _line.startswith("#") and "=" in _line: _k, _v = _line.split("=", 1) _k, _v = _k.strip(), _v.strip() if _v: # 빈 값이 아닐 때만 os.environ[_k] = _v # 무조건 덮어쓰기 (settings.env 우선) def _env(key, default=""): """환경변수 읽기. 필수 키가 비어있으면 경고.""" val = os.environ.get(key, default) if not val and not default: print(f"⚠ 설정 누락: {key} — settings.env를 확인하세요") return val # ════════════════════════════════════════ # 설정 (settings.env 에서만 관리 — 여기를 직접 수정하지 마세요) # ════════════════════════════════════════ OPENAI_KEY = _env("OPENAI_KEY") HA_URL = _env("HA_URL") HA_TOKEN = _env("HA_TOKEN") SERVER_URL = _env("SERVER_URL") PERPLEXITY_KEY = _env("PERPLEXITY_KEY") GROQ_KEY = _env("GROQ_KEY") WAKE_THRESHOLD = float(_env("WAKE_THRESHOLD", "0.9")) ENABLE_WAKE_WORD = False RECORD_SECONDS = int(_env("RECORD_SECONDS", "10")) SILENCE_SEC = int(_env("SILENCE_SEC", "1")) SILENCE_DB = int(_env("SILENCE_DB", "28")) TTS_VOICE_KO = _env("TTS_VOICE_KO", "ko-KR-InJoonNeural") TTS_VOICE_RATE = _env("TTS_VOICE_RATE", "+25%") TTS_VOICE = "echo" # OpenAI TTS fallback (설정 불필요) TTS_VOICE_SUPERTONIC = _env("TTS_VOICE_SUPERTONIC", "M1") TTS_SUPERTONIC_STEPS = int(_env("TTS_SUPERTONIC_STEPS", "20")) TTS_SUPERTONIC_SPEED = float(_env("TTS_SUPERTONIC_SPEED", "1.1")) _mic_env = _env("MIC_INDEX", "auto") MIC_INDEX = None if _mic_env == "auto" else int(_mic_env) _MIC_IDX = 1 _MIC_CH = 1 # 기본 기기 (서버에서 로드 성공 시 덮어씀) HA_DEVICES = {} ACT = { "turn_on": "켰", "turn_off": "껐", "toggle": "전환했", "start": "시작했", "stop": "멈췄", "pause": "일시정지했", "return_to_base": "복귀시켰", "set_temperature": "온도 설정했", "set_fan_mode": "팬 모드 설정했", "set_hvac_mode": "모드 설정했", "set_preset_mode": "프리셋 설정했", "set_swing_mode": "스윙 설정했", "set_humidity": "습도 설정했", "set_percentage": "세기 설정했", "set_direction": "방향 설정했", "increase_speed": "세기 올렸", "decrease_speed": "세기 내렸", "oscillate": "회전 설정했", "locate": "위치 알림했", "clean_spot": "스팟 청소 시작��", "set_fan_speed": "흡입력 설정했", "open_cover": "열었", "close_cover": "닫았", "stop_cover": "멈췄", "set_cover_position": "위치 설정했", "lock": "잠갔", "unlock": "열었", "select_source": "입력 변경했", "volume_up": "볼륨 올렸", "volume_down": "볼륨 내렸", "volume_set": "볼륨 설정했", "volume_mute": "음소거했", "media_play": "재생했", "media_pause": "일시정지했", "media_stop": "멈췄", "media_next_track":"다음 트랙했", "media_previous_track":"이전 트랙했", } # GPT action → HA 실제 서비스 매핑 (domain별) # GPT가 자유롭게 action을 만들어도 여기서 HA 서비스로 변환 _ACTION_MAP = { "light": { "set_brightness": ("turn_on", lambda e: {"brightness": min(255, int(int(e.get("brightness", 100)) * 255 / 100))}), "set_color": ("turn_on", lambda e: _resolve_color(e)), "set_color_temp": ("turn_on", lambda e: {"color_temp_kelvin": int(e.get("color_temp_kelvin", e.get("color_temp", 4000)))}), "set_effect": ("turn_on", lambda e: {"effect": e.get("effect", "Off")}), }, "climate": { "set_temperature": ("set_temperature", lambda e: e), "set_mode": ("set_hvac_mode", lambda e: {"hvac_mode": e.get("hvac_mode", e.get("mode", "auto"))}), "set_fan_mode": ("set_fan_mode", lambda e: e), "set_hvac_mode": ("set_hvac_mode", lambda e: e), "set_preset_mode": ("set_preset_mode", lambda e: e), "set_swing_mode": ("set_swing_mode", lambda e: e), }, "fan": { "set_speed": ("set_percentage", lambda e: {"percentage": int(e.get("percentage", e.get("speed", 50)))}), "set_percentage": ("set_percentage", lambda e: e), "increase_speed": ("increase_speed", lambda e: e), "decrease_speed": ("decrease_speed", lambda e: e), "oscillate": ("oscillate", lambda e: e), "set_direction": ("set_direction", lambda e: e), }, "vacuum": { "start": ("start", lambda e: e), "stop": ("stop", lambda e: e), "pause": ("pause", lambda e: e), "return_to_base": ("return_to_base", lambda e: e), "return_home": ("return_to_base", lambda e: e), "dock": ("return_to_base", lambda e: e), "locate": ("locate", lambda e: e), "clean_spot": ("clean_spot", lambda e: e), "set_fan_speed": ("set_fan_speed", lambda e: e), }, "cover": { "open": ("open_cover", lambda e: e), "close": ("close_cover", lambda e: e), "stop": ("stop_cover", lambda e: e), "set_position": ("set_cover_position", lambda e: e), "open_cover": ("open_cover", lambda e: e), "close_cover": ("close_cover", lambda e: e), }, "lock": { "lock": ("lock", lambda e: e), "unlock": ("unlock", lambda e: e), }, "media_player": { "volume_up": ("volume_up", lambda e: e), "volume_down": ("volume_down", lambda e: e), "volume_set": ("volume_set", lambda e: e), "volume_mute": ("volume_mute", lambda e: e), "select_source": ("select_source", lambda e: e), "play": ("media_play", lambda e: e), "media_play": ("media_play", lambda e: e), "media_pause": ("media_pause", lambda e: e), "media_stop": ("media_stop", lambda e: e), "media_next": ("media_next_track",lambda e: e), "media_previous": ("media_previous_track", lambda e: e), }, } # 색상 이름 → HS 색상값 _COLOR_NAME_MAP = { "빨간": (0, 100), "빨간색": (0, 100), "빨강": (0, 100), "red": (0, 100), "주황": (30, 100), "주황색": (30, 100), "오렌지": (30, 100), "orange": (30, 100), "노란": (60, 100), "노란색": (60, 100), "노랑": (60, 100), "yellow": (60, 100), "초록": (120, 100), "초록색": (120, 100), "녹색": (120, 100), "green": (120, 100), "파란": (240, 100), "파란색": (240, 100), "파랑": (240, 100), "blue": (240, 100), "하늘": (200, 80), "하늘색": (200, 80), "cyan": (180, 100), "보라": (270, 100), "보라색": (270, 100), "purple": (270, 100), "분홍": (330, 70), "분홍색": (330, 70), "핑크": (330, 70), "pink": (330, 70), "하양": (0, 0), "하얀색": (0, 0), "흰색": (0, 0), "white": (0, 0), } def _resolve_color(extra): """GPT가 보낸 color 값을 HA hs_color로 변환.""" color = extra.get("color", "") # hs_color가 직접 있으면 그대로 if "hs_color" in extra: return {"hs_color": extra["hs_color"]} if "rgb_color" in extra: return {"rgb_color": extra["rgb_color"]} # 색 이름으로 매핑 for name, hs in _COLOR_NAME_MAP.items(): if name in str(color).lower(): return {"hs_color": list(hs)} # fallback: 그냥 전달 return {"hs_color": [0, 100]} # ════════════════════════════════════════ # 패키지 확인 # ════════════════════════════════════════ def check_packages(): missing = [] for pkg, imp in [ ("openwakeword", "openwakeword"), ("sounddevice", "sounddevice"), ("groq", "groq"), ("openai", "openai"), ("pygame", "pygame"), ("numpy", "numpy"), ("supertonic", "supertonic"), ]: try: __import__(imp) except ImportError: missing.append(pkg) if missing: print("설치 필요:") print(f" pip install {' '.join(missing)}") sys.exit(1) check_packages() import sounddevice as sd import pygame import httpx from openai import OpenAI from groq import Groq as GroqClient # 전역 API 클라이언트 (TCP 연결 재사용 → 응답 빠름) _pplx_http = httpx.Client(timeout=httpx.Timeout(10, connect=3)) _openai_http = httpx.Client(timeout=httpx.Timeout(30, connect=5)) _pplx_client = OpenAI(api_key=PERPLEXITY_KEY, base_url="https://api.perplexity.ai", http_client=_pplx_http) _openai_client = OpenAI(api_key=OPENAI_KEY, http_client=_openai_http) import onnxruntime as _ort _ort.set_default_logger_severity(3) # 경고 억제 # RPi5: GPU provider 스캔 방지 → CPU 전용 _ORT_OPTS = _ort.SessionOptions() _ORT_OPTS.inter_op_num_threads = 1 _ORT_OPTS.intra_op_num_threads = 2 os.environ.setdefault("OMP_NUM_THREADS", "2") os.environ.setdefault("OMP_WAIT_POLICY", "PASSIVE") os.environ.setdefault("CUDA_VISIBLE_DEVICES", "") from openwakeword.model import Model as WakeModel # ════════════════════════════════════════ # 전역 # ════════════════════════════════════════ wake_model = None _phrase_cache = {} # text -> 오디오 파일 경로 (고정 문구 캐시) supertonic_tts = None # supertonic TTS 인스턴스 (None = 비활성화) CACHED_PHRASES = [ "네?", "잘 못 들었어요", "인식 오류가 났어요", "잠시만요", ] # ════════════════════════════════════════ # 로딩 사운드 (subprocess + aplay — pygame/ALSA 충돌 없음) # ════════════════════════════════════════ import signal as _signal _loading_proc = None # aplay subprocess _loading_pgid = None # 프로세스 그룹 ID (확실한 kill용) _LOADING_WAV = os.path.join(os.path.dirname(os.path.abspath(__file__)), "loading_loop.wav") def _kill_aplay_all(): """시스템에 남은 aplay 프로세스를 모두 정리 (안전장치).""" try: subprocess.run(["pkill", "-9", "-f", "aplay.*loading_loop"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2) except Exception: pass def loading_start(): """'잠시만요' TTS 재생 후, 별도 프로세스(aplay)로 로딩음 루프 재생.""" global _loading_proc, _loading_pgid # 혹시 이전 프로세스가 남아있으면 먼저 정리 loading_stop() try: # "잠시만요" 먼저 말하기 (캐시된 고정 문구 → 빠름) try: path, own = _generate_audio("잠시만요") _play_audio(path, own_file=own) except Exception as e: print(f"잠시만요 재생 실패: {e}") # 로딩 루프 시작 if not os.path.exists(_LOADING_WAV): print(f"로딩 WAV 없음: {_LOADING_WAV}") return _loading_proc = subprocess.Popen( ["bash", "-c", f"while true; do aplay -q '{_LOADING_WAV}'; done"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, preexec_fn=os.setsid ) # 시작 직후 pgid 저장 (나중에 프로세스 죽어도 이 값으로 kill 가능) try: _loading_pgid = os.getpgid(_loading_proc.pid) except Exception: _loading_pgid = None except Exception as e: print(f"로딩 사운드 시작 실패: {e}") _loading_proc = None _loading_pgid = None def loading_stop(): """로딩음 정지 — 3단계 확실 kill.""" global _loading_proc, _loading_pgid # 1단계: 프로세스 그룹 kill (bash + aplay 자식 모두) if _loading_pgid: try: os.killpg(_loading_pgid, _signal.SIGKILL) except (ProcessLookupError, PermissionError, OSError): pass _loading_pgid = None # 2단계: proc 객체로 직접 kill (1단계 실패 대비) if _loading_proc: try: _loading_proc.kill() except (ProcessLookupError, PermissionError, OSError): pass try: _loading_proc.wait(timeout=1) except Exception: pass _loading_proc = None # 3단계: 혹시 남은 aplay 좀비까지 정리 _kill_aplay_all() conversation = [] CONV_MAX_TURNS = 10 CONV_TIMEOUT = 1800 # ════════════════════════════════════════ # 기기 로드 # ════════════════════════════════════════ def load_devices(): global HA_DEVICES try: r = requests.get(f"{SERVER_URL}/api/devices", timeout=3) if r.ok: for dev in r.json(): name = (dev.get("name") or "").strip() orig = (dev.get("original_name") or "").strip() eid = dev.get("entity_id", "") dom = dev.get("domain", "") if not eid: continue info = {"entity_id": eid, "domain": dom} if name: HA_DEVICES[name] = info if orig and orig != name: HA_DEVICES[orig] = info print(f"기기 {len(HA_DEVICES)}개 로드됨 (서버)") except Exception as e: print(f"서버 기기 로드 실패 ({e})") # 2차: HA API에서 직접 로드 if not HA_DEVICES and HA_URL and HA_TOKEN: try: headers = {"Authorization": f"Bearer {HA_TOKEN}"} r = requests.get(f"{HA_URL}/api/states", headers=headers, timeout=5) if r.ok: for ent in r.json(): eid = ent.get("entity_id", "") dom = eid.split(".")[0] if "." in eid else "" if dom not in ("light","switch","climate","vacuum","media_player","fan","cover","lock","scene"): continue name = ent.get("attributes", {}).get("friendly_name", "") if name and eid: HA_DEVICES[name] = {"entity_id": eid, "domain": dom} print(f"기기 {len(HA_DEVICES)}개 로드됨 (HA 직접)") except Exception as e2: print(f"HA 직접 로드도 실패 ({e2})") if not HA_DEVICES: print("설정 누락: 기기 목록이 비어있습니다. 서버 또는 HA 연결을 확인하세요.") # ════════════════════════════════════════ # 모델 초기화 - 개선된 버전 # ════════════════════════════════════════ def find_wake_model_path(): """ 현재 스크립트 디렉토리부터 시작하여 .onnx 파일을 재귀 탐색. 우선순위: ./ddaengguya.onnx > ./smarthome/ddaengguya.onnx > 기타 """ script_dir = os.path.dirname(os.path.abspath(__file__)) candidates = [] priority_paths = [ os.path.join(script_dir, "ddaengguya.onnx"), os.path.join(script_dir, "smarthome", "ddaengguya.onnx"), os.path.join(script_dir, "smarthome_extracted", "ddaengguya.onnx"), ] for path in priority_paths: if os.path.exists(path): candidates.append(path) for root, dirs, files in os.walk(script_dir): for file in files: if file.endswith(".onnx"): full_path = os.path.join(root, file) if full_path not in candidates: candidates.append(full_path) return candidates def init_wake_model(): global wake_model, ENABLE_WAKE_WORD print("Wake Word 모델 로딩...") try: import openwakeword as _oww print(" 필수 모델 파일 확인/다운로드 중...") _oww.utils.download_models() print(" 모델 파일 준비 완료") except Exception as e: print(f" 모델 파일 다운로드 실패 (오프라인?): {str(e)[:80]}") model_candidates = find_wake_model_path() if model_candidates: print("모델 검색 중...") for model_path in model_candidates: try: print(f" 시도: {model_path} ... ", end="", flush=True) wake_model = WakeModel( wakeword_models=[model_path], inference_framework="onnx", enable_speex_noise_suppression=False, ) print("로드 성공!") print(f"커스텀 Wake Word: {model_path}") ENABLE_WAKE_WORD = True return except Exception as e: print(f"실패 ({str(e)[:100]})") print("기본 내장 모델 로드 시도...") for framework, label in [("onnx", "onnx 방식"), (None, "기본 방식")]: try: print(f" {label} 시도...") if framework: wake_model = WakeModel( inference_framework=framework, enable_speex_noise_suppression=False, ) else: wake_model = WakeModel( enable_speex_noise_suppression=False, ) print(f"기본 내장 모델 로드 성공 ({label})") ENABLE_WAKE_WORD = True return except Exception as e: print(f" {label} 실패: {str(e)[:100]}") print(f"[경고] 모든 Wake Word 모델 로드 실패") print(f"[경고] 의존성 확인:") print(f" - openwakeword 재설치: pip install --upgrade openwakeword --force-reinstall") print(f" - ONNX Runtime 설치: pip install --upgrade onnxruntime") print(f" - tflite-runtime 설치: pip install tflite-runtime (Linux/Mac)") print(f" - Windows의 경우 Visual C++ 재배포 가능 패키지 필요") print(f"[경고] Wake Word 감지 비활성화 - 자동으로 음성 입력 대기 모드로 실행됩니다.") wake_model = None ENABLE_WAKE_WORD = False # ════════════════════════════════════════ # 오디오 유틸 # ════════════════════════════════════════ def _save_wav(frames, rate=16000): audio_data = np.concatenate(frames, axis=0) tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) with wave.open(tmp.name, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(rate) wf.writeframes(audio_data.tobytes()) return tmp.name def _to_mono(data): """스테레오(또는 다채널) → 모노 변환""" if data.ndim > 1 and data.shape[1] > 1: return data.mean(axis=1, keepdims=True).astype(np.int16) return data def _detect_mic(): """사용 가능한 입력 장치 인덱스와 채널 수 반환""" global _MIC_IDX, _MIC_CH devices = sd.query_devices() idx = MIC_INDEX if idx is None: for i, d in enumerate(devices): if d["max_input_channels"] > 0: idx = i break if idx is None: raise RuntimeError("사용 가능한 마이크가 없습니다") try: sd.check_input_settings(device=idx, channels=1, samplerate=16000) ch = 1 except Exception: ch = int(sd.query_devices(idx)["max_input_channels"]) _MIC_IDX, _MIC_CH = idx, ch name = sd.query_devices(idx)["name"] print(f"마이크 선택: [{idx}] {name} ({ch}ch)") def _db(data): rms = np.sqrt(np.mean(data.astype(np.float32) ** 2)) return 20 * math.log10(max(rms, 1)) # ════════════════════════════════════════ # 녹음 (침묵 감지) # ════════════════════════════════════════ def record_audio(): print("말씀하세요...") RATE, CHUNK = 16000, 1024 MIN_SPEECH_CHUNKS = 4 # ~0.25초 미만 발화는 소음으로 간주 GRACE_CHUNKS = int(RATE / CHUNK * 0.4) frames, silence_start, started = [], None, False speech_chunks = 0 with sd.InputStream(samplerate=RATE, channels=_MIC_CH, dtype="int16", blocksize=CHUNK, device=_MIC_IDX) as st: for i in range(int(RATE / CHUNK * RECORD_SECONDS)): data, _ = st.read(CHUNK) mono = _to_mono(data) db = _db(mono) if db > SILENCE_DB: started = True silence_start = None speech_chunks += 1 frames.append(mono.copy()) elif started: frames.append(mono.copy()) if silence_start is None: silence_start = time.time() elif time.time() - silence_start > SILENCE_SEC: break elif i < GRACE_CHUNKS: frames.append(mono.copy()) if not frames or not started or speech_chunks < MIN_SPEECH_CHUNKS: raise ValueError("음성이 감지되지 않았습니다") return _save_wav(frames) def record_audio_safe(timeout=None): """record_audio()를 스레드로 실행. 타임아웃 또는 마이크 블로킹 시 None 반환.""" timeout = timeout or (RECORD_SECONDS + 3) result = [None] exc = [None] def _run(): try: result[0] = record_audio() except Exception as e: exc[0] = e t = threading.Thread(target=_run, daemon=True) t.start() t.join(timeout=timeout) if t.is_alive(): print("녹음 타임아웃 (마이크 응답 없음)") return None if exc[0]: if isinstance(exc[0], ValueError): return None raise exc[0] return result[0] def passive_listen(timeout_sec=10): """Wake word 없이 음성 대기. 발화 없으면 None 반환.""" print(f"대화 대기 중... ({timeout_sec}초)") RATE, CHUNK = 16000, 1024 frames, silence_start, started = [], None, False try: with sd.InputStream(samplerate=RATE, channels=_MIC_CH, dtype="int16", blocksize=CHUNK, device=_MIC_IDX) as st: for _ in range(int(RATE / CHUNK * timeout_sec)): data, _ = st.read(CHUNK) db = _db(_to_mono(data)) if db > SILENCE_DB: frames.append(_to_mono(data.copy())) started = True silence_start = None elif started: frames.append(_to_mono(data.copy())) if silence_start is None: silence_start = time.time() elif time.time() - silence_start > SILENCE_SEC: break except Exception as e: print(f"passive_listen 오류: {e}") return None if not started or not frames: return None return _save_wav(frames) def record_with_prebuffer(buffer_frames): """ wake word 감지 직후 즉시 호출. buffer_frames: 감지 전 3초 오디오 (mono int16 numpy arrays) 버퍼 + 추가 녹음을 합쳐 WAV 반환. 발화 없으면 None. """ RATE, CHUNK_SZ = 16000, 1024 frames = list(buffer_frames) silence_start = None # 버퍼 마지막 ~1.5초에 발화가 있는지 확인 recent_n = int(RATE / CHUNK_SZ * 1.5) recent = frames[-recent_n:] if len(frames) >= recent_n else frames started = any(_db(f) > SILENCE_DB for f in recent) try: with sd.InputStream(samplerate=RATE, channels=_MIC_CH, dtype="int16", blocksize=CHUNK_SZ, device=_MIC_IDX) as st: for _ in range(int(RATE / CHUNK_SZ * RECORD_SECONDS)): data, _ = st.read(CHUNK_SZ) mono = _to_mono(data) db = _db(mono) if db > SILENCE_DB: started = True silence_start = None frames.append(mono.copy()) elif started: frames.append(mono.copy()) if silence_start is None: silence_start = time.time() elif time.time() - silence_start > SILENCE_SEC: break except Exception as e: print(f"record_with_prebuffer 오류: {e}") if not started or not frames: return None return _save_wav(frames) # ════════════════════════════════════════ # STT # ════════════════════════════════════════ def stt(path): try: client = GroqClient(api_key=GROQ_KEY) with open(path, "rb") as f: result = client.audio.transcriptions.create( model="whisper-large-v3-turbo", file=f, language="ko", response_format="text", ) text = (result if isinstance(result, str) else result.text).strip() except Exception as e: print(f"Groq STT 오류: {e}") text = "" finally: try: os.unlink(path) except: pass return text def _strip_wakeword(text: str) -> str: """STT 결과 정제. 불필요한 구두점 제거만.""" return text.strip(" ,./!?~") # ════════════════════════════════════════ # TTS - Supertonic (오프라인) + edge-tts + OpenAI fallback # ════════════════════════════════════════ _PYGAME_OK = False try: pygame.mixer.init() _PYGAME_OK = True except Exception as _e: print(f"[WARN] pygame 오디오 초기화 실패: {_e} — 스피커 연결 후 재시작 필요") def init_supertonic(): global supertonic_tts try: import supertonic as _st supertonic_tts = _st.TTS() print(f"Supertonic TTS 초기화 완료 (음성: {TTS_VOICE_SUPERTONIC}, {supertonic_tts.sample_rate}Hz)") except Exception as e: print(f"Supertonic 초기화 실패: {e} → edge-tts 사용") supertonic_tts = None def _supertonic_to_wav(text) -> str: style = supertonic_tts.get_voice_style(TTS_VOICE_SUPERTONIC) audio, _ = supertonic_tts.synthesize(text, style, lang="ko", speed=TTS_SUPERTONIC_SPEED, total_steps=TTS_SUPERTONIC_STEPS) sr = supertonic_tts.sample_rate tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) tmp.close() with wave.open(tmp.name, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(sr) wf.writeframes((audio * 32767).astype(np.int16).tobytes()) return tmp.name def _edgetts_to_mp3(text, voice=None) -> str: import asyncio, edge_tts v = voice or TTS_VOICE_KO tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) tmp_path = tmp.name tmp.close() async def _gen(): communicate = edge_tts.Communicate(text, v, rate=TTS_VOICE_RATE) await communicate.save(tmp_path) asyncio.run(_gen()) return tmp_path def _openai_to_mp3(text) -> str: client = _openai_client resp = client.audio.speech.create( model="tts-1", voice=TTS_VOICE, input=text, response_format="mp3" ) tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) tmp.write(resp.content) tmp.close() return tmp.name def _generate_audio(text) -> tuple: """(파일경로, 임시파일여부) 반환. 캐시 히트 시 임시=False.""" if text in _phrase_cache and os.path.exists(_phrase_cache[text]): return _phrase_cache[text], False if supertonic_tts is not None: try: return _supertonic_to_wav(text), True except Exception as e: print(f"Supertonic 실패: {e} → edge-tts fallback") try: return _edgetts_to_mp3(text), True except Exception as e: print(f"edge-tts 실패: {e} → OpenAI TTS fallback") return _openai_to_mp3(text), True def cache_phrases(): global _phrase_cache print("고정 문구 캐시 생성 중...") cache_dir = os.path.join(tempfile.gettempdir(), "ddanggu_cache") os.makedirs(cache_dir, exist_ok=True) for phrase in CACHED_PHRASES: try: if supertonic_tts is not None: ext = ".wav" path = os.path.join(cache_dir, f"{abs(hash(phrase))}{ext}") if not os.path.exists(path): tmp = _supertonic_to_wav(phrase) import shutil shutil.move(tmp, path) else: ext = ".mp3" path = os.path.join(cache_dir, f"{abs(hash(phrase))}{ext}") if not os.path.exists(path): tmp = _edgetts_to_mp3(phrase) import shutil shutil.move(tmp, path) _phrase_cache[phrase] = path print(f" 캐시: {phrase}") except Exception as e: print(f" 캐시 실패 ({phrase}): {e}") def stop_speaking(): if not _PYGAME_OK: return if pygame.mixer.music.get_busy(): pygame.mixer.music.stop() def play_ack(): """웨이크 확인음 "네?" 재생 (동기 — 끝날 때까지 블록)""" path, own = _generate_audio("네?") _play_audio(path, own_file=own) def play_ack_nonblock(): """웨이크 확인음 "네?" 재생 시작만 하고 즉시 리턴 (비블로킹)""" path, own = _generate_audio("네?") try: pygame.mixer.music.load(path) pygame.mixer.music.play() except Exception as e: print(f"ack 재생 오류: {e}") finally: if own: threading.Thread(target=lambda: (time.sleep(3), _safe_unlink(path)), daemon=True).start() def _safe_unlink(path): try: os.unlink(path) except: pass def _play_audio(file_path, own_file=True, timeout=30): if not _PYGAME_OK: if own_file: try: os.unlink(file_path) except: pass return try: pygame.mixer.music.load(file_path) pygame.mixer.music.play() deadline = time.time() + timeout while pygame.mixer.music.get_busy(): if time.time() > deadline: pygame.mixer.music.stop() break time.sleep(0.03) pygame.mixer.music.unload() finally: if own_file: try: os.unlink(file_path) except: pass def speak(text): print(f"땡구: {text}") try: path, own = _generate_audio(text) _play_audio(path, own_file=own) except Exception as e: print(f"TTS 오류: {e}") # ════════════════════════════════════════ # HA API # ════════════════════════════════════════ def ha_call(domain, service, entity_id, extra=None): headers = {"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"} data = {"entity_id": entity_id} if extra: data.update(extra) try: r = requests.post(f"{HA_URL}/api/services/{domain}/{service}", headers=headers, json=data, timeout=5) return r.status_code in (200, 201) except Exception as e: print(f"HA 오류: {e}") return False # ════════════════════════════════════════ # 글로벌 처리 락 (프로세스 간 공유) # /tmp/ddanggu.lock 파일로 ddanggu_voice.py / satellite_server.py 간 중복 응답 방지 # ════════════════════════════════════════ _LOCK_FILE_PATH = os.path.join(tempfile.gettempdir(), "ddanggu.lock") _LOCK_TIMEOUT = 30 def _lock_acquire(source: str) -> bool: try: if os.path.exists(_LOCK_FILE_PATH): with open(_LOCK_FILE_PATH) as _lf: lines = _lf.read().splitlines() if len(lines) >= 2: try: if time.time() - float(lines[1]) < _LOCK_TIMEOUT: return False except ValueError: pass os.unlink(_LOCK_FILE_PATH) fd = os.open(_LOCK_FILE_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY) os.write(fd, f"{source}\n{time.time()}".encode()) os.close(fd) return True except FileExistsError: return False except Exception: return True def _lock_release(): try: if os.path.exists(_LOCK_FILE_PATH): os.unlink(_LOCK_FILE_PATH) except Exception: pass # ════════════════════════════════════════ # 대화 기록 # ════════════════════════════════════════ def add_to_history(role, text): global conversation now = time.time() conversation = [c for c in conversation if now - c["time"] < CONV_TIMEOUT] conversation.append({"role": role, "content": text, "time": now}) if len(conversation) > CONV_MAX_TURNS * 2: conversation = conversation[-(CONV_MAX_TURNS * 2):] def get_history_messages(): return [{"role": c["role"], "content": c["content"]} for c in conversation] # ════════════════════════════════════════ # 텍스트 정제 (TTS용) # ════════════════════════════════════════ def clean_response(text): text = re.sub(r'\[([^\]]*)\]\([^\)]*\)', r'\1', text) text = re.sub(r'\(?https?://\S+\)?', '', text) text = re.sub(r'\([^)]{0,60}\.(?:com|org|net|kr|co\.kr|go\.kr)[^)]*\)', '', text) text = re.sub(r'\([^)]*출처[^)]*\)', '', text) text = re.sub(r'\*{1,3}([^*]*)\*{1,3}', r'\1', text) text = re.sub(r'#{1,6}\s*', '', text) text = re.sub(r'\n\s*\d+\.\s*', ' ', text) text = re.sub(r'\n\s*[-•]\s*', ' ', text) text = re.sub(r'\s+', ' ', text).strip() return text # ════════════════════════════════════════ # 실시간 데이터 수집 (뉴스 / 날씨) # ════════════════════════════════════════ _REQ_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} def fetch_news(query=None): if query: url = f"https://news.google.com/rss/search?q={quote(query)}&hl=ko&gl=KR&ceid=KR:ko" else: url = "https://news.google.com/rss?hl=ko&gl=KR&ceid=KR:ko" try: r = requests.get(url, timeout=5, headers=_REQ_HEADERS) r.raise_for_status() root = ET.fromstring(r.content) items = root.findall('.//item')[:7] headlines = [] for item in items: title = item.find('title') source = item.find('source') pub = item.find('pubDate') t = title.text.strip() if title is not None and title.text else "" s = source.text.strip() if source is not None and source.text else "" p = pub.text.strip() if pub is not None and pub.text else "" if t: line = f"- {t}" if s: line += f" [{s}]" if p: line += f" ({p})" headlines.append(line) return "\n".join(headlines) if headlines else None except Exception as e: print(f"뉴스 fetch 실패: {e}") return None _CITY_COORDS = { "서울": (37.5665, 126.9780), "부산": (35.1796, 129.0756), "대구": (35.8714, 128.6014), "인천": (37.4563, 126.7052), "광주": (35.1595, 126.8526), "대전": (36.3504, 127.3845), "울산": (35.5384, 129.3114), "세종": (36.4800, 127.2890), "제주": (33.4996, 126.5312), "수원": (37.2636, 127.0286), "성남": (37.4449, 127.1388), "고양": (37.6584, 126.8320), "용인": (37.2411, 127.1776), "청주": (36.6424, 127.4890), "천안": (36.8151, 127.1139), "전주": (35.8242, 127.1480), "포항": (36.0190, 129.3435), "창원": (35.2280, 128.6811), } _WMO_DESC = { 0: "맑음", 1: "대체로 맑음", 2: "구름 조금", 3: "흐림", 45: "안개", 48: "안개", 51: "이슬비", 53: "이슬비", 55: "이슬비", 61: "비", 63: "비", 65: "강한 비", 71: "눈", 73: "눈", 75: "강한 눈", 80: "소나기", 81: "소나기", 82: "강한 소나기", 95: "뇌우", 96: "우박 동반 뇌우", 99: "우박 동반 뇌우", } def fetch_weather(city="서울"): try: lat, lon = _CITY_COORDS.get(city, _CITY_COORDS["서울"]) r = requests.get( "https://api.open-meteo.com/v1/forecast" f"?latitude={lat}&longitude={lon}" "¤t=temperature_2m,relative_humidity_2m,apparent_temperature,weather_code" "&daily=temperature_2m_max,temperature_2m_min" "&timezone=Asia%2FSeoul&forecast_days=1", timeout=5 ) r.raise_for_status() data = r.json() cur = data.get("current", {}) daily = data.get("daily", {}) temp = cur.get("temperature_2m", "?") feels = cur.get("apparent_temperature", "?") humidity = cur.get("relative_humidity_2m", "?") code = cur.get("weather_code", 0) desc = _WMO_DESC.get(code, "알 수 없음") max_t_list = daily.get("temperature_2m_max") or [] min_t_list = daily.get("temperature_2m_min") or [] max_t = max_t_list[0] if max_t_list else "?" min_t = min_t_list[0] if min_t_list else "?" return ( f"현재 {city}: {desc}, 기온 {temp}도 (체감 {feels}도), " f"습도 {humidity}%, 최고 {max_t}도 / 최저 {min_t}도" ) except Exception as e: print(f"날씨 fetch 실패: {e}") return None def fetch_perplexity(query): try: resp = _pplx_client.chat.completions.create( model="sonar", messages=[ {"role": "system", "content": ( "너는 음성비서의 검색 엔진이야. 스피커로 읽어줄 답변을 만들어. " "규칙: 3문장 이내 구어체. 마크다운·제목·굵기·번호·출처표시([1] 등) 절대 금지. " "핵심 정보(날짜·수치·이름)만 포함. 한국어로 답해." )}, {"role": "user", "content": query}, ], ) text = (resp.choices[0].message.content or "").strip() if text: text = clean_response(text) # 마크다운·URL 잔여물 제거 return text or None except Exception as e: print(f"Perplexity 검색 실패: {e}") return None # ════════════════════════════════════════ # AI 분석 # ════════════════════════════════════════ def analyze(text): client = _openai_client dev_list = "\n".join(f"- {n}" for n in HA_DEVICES) now = datetime.now().strftime("%Y-%m-%d %H:%M") CTRL_KW = [ # 동작 (기본) "켜","꺼","틀어","끄","올려","내려","높여","낮춰","줄여","키워", "시작","멈춰","중지","정지","일시정지","재개","작동","동작", "설정","바꿔","변경","맞춰","조절","전환", "열어","닫아","잠가","잠금","해제","열림","닫힘", # 조명 "조명","불","전구","밝기","밝게","어둡게","퍼센트", "색","색깔","색온도","무드","따뜻","시원", "램프","간접등","스탠드","무드등","취침등","수면등", "디밍","형광","백열","주광","전구색", # 에어컨/온도/난방 "에어컨","냉방","난방","제습","송풍","바람", "온도","도","시원하게","따뜻하게","춥","더워", "히터","보일러","온풍기","라디에이터","선풍기","써큘레이터", "실내온도","희망온도","자동","쾌적","절전","파워", # 청소/로봇 "청소기","청소","로봇","충전","복귀","도킹", "걸레","물걸레","흡입","먼지", # 공기질 "공기청정기","공청기","정화","환기","환풍","환풍기", "가습기","가습","제습기","습도","미세먼지","공기질", # TV/미디어 "TV","티비","텔레비전","볼륨","채널","음량","소리", "스피커","뮤트","음소거","재생","일시중지","다음","이전", "입력","HDMI","소스","넷플릭스","유튜브", # 플러그/전원 "플러그","콘센트","전원","스위치","멀티탭", # 커튼/블라인드 "커튼","블라인드","롤스크린","차양","암막","셔터", # 도어/보안 "도어락","현관","문","잠금장치","초인종","벨", "CCTV","카메라","감시","녹화","센서","모션", "경보","알람","보안","방범","사이렌", # 가전 일반 "세탁기","세탁","건조기","건조","식기세척기","냉장고", "오븐","전자레인지","밥솥","인덕션","정수기", "다리미","헤어드라이어","비데","변기", # 일반 제어 "모드","세기","단계","풍량","풍속","타이머","예약", "스케줄","루틴","자동화","씬","장면", ] NEWS_KW = ["뉴스","소식","헤드라인","이슈"] WEATHER_KW = ["날씨","기온","몇도","온도","비 올","비올","눈 올","눈올","우산","습도","체감"] is_control = any(kw in text for kw in CTRL_KW) is_news = not is_control and any(kw in text for kw in NEWS_KW) is_weather = not is_control and any(kw in text for kw in WEATHER_KW) is_search = not is_control and not is_news and not is_weather extra_context = "" if is_news: print("뉴스 검색 중...") search_term = re.sub(r'(알려|말해|뉴스|줘|요|해줘|주요|오늘|어제|최신)', '', text).strip() news_query = f"오늘 주요 뉴스 헤드라인" if len(search_term) < 2 else f"{search_term} 최신 뉴스" news = fetch_perplexity(news_query) if news: # Perplexity가 이미 한국어 요약 → GPT 재호출 없이 바로 반환 add_to_history("user", text) add_to_history("assistant", news) return {"type": "chat", "response": news} else: extra_context = ( "\n\n[뉴스 수집 실패]\n" "뉴스를 가져오지 못했다고 솔직히 말하고, 잠시 후 다시 시도해달라고 안내해." ) elif is_weather: print("날씨 검색 중...") city_match = re.search( r'(서울|부산|대구|인천|광주|대전|울산|세종|제주|수원|성남|고양|용인|청주|천안|전주|포항|창원)', text ) city = city_match.group(1) if city_match else "서울" weather = fetch_weather(city) if weather: extra_context = ( f"\n\n[실시간 날씨 - {now} 수집]\n{weather}\n\n" "위 날씨 정보를 자연스러운 구어체 한두 문장으로 전달해." ) else: extra_context = ( "\n\n[날씨 수집 실패]\n" "날씨 정보를 가져오지 못했다고 솔직히 말하고, 잠시 후 다시 시도해달라고 안내해." ) elif is_search: print("웹 검색 중...") search_query = text results = fetch_perplexity(search_query) if results: # Perplexity가 이미 한국어 답변 → GPT 재호출 없이 바로 반환 add_to_history("user", text) add_to_history("assistant", results) return {"type": "chat", "response": results} else: extra_context = ( "\n\n[웹 검색 결과 없음]\n" "검색 결과를 찾지 못했다고 짧게 솔직히 말해." ) system_msg = ( f"너는 스마트홈 AI 음성비서 '땡구'야. 현재 {now}.\n\n" f"기기 목록:\n{dev_list}\n\n" "★ 출력 규칙 (절대 준수) ★\n" "- 순수 JSON만 출력. 코드블록·설명·줄바꿈 없음.\n" "- response 값: 스피커로 읽을 자연스러운 한국어 구어체 텍스트만.\n" "- response 안에 JSON, 중괄호, URL, 출처, 마크다운 절대 금지.\n" "- 일반 대화: 1~3문장 구어체.\n" "- device 값은 반드시 위 기기 목록에 있는 이름 중 하나를 그대로 사용.\n\n" "포맷:\n" "기기 제어 → " '{"type":"control","device":"기기목록의정확한이름","action":"동작","delay_minutes":0,"extra":{파라미터},"response":"구어체응답"}\n\n' "action 종류:\n" "- 공통: turn_on, turn_off, toggle\n" "- 조명(light): set_brightness(extra:{brightness:0~100}), set_color(extra:{color:색이름}), set_color_temp(extra:{color_temp_kelvin:2500~6500}), set_effect(extra:{effect:이름})\n" "- 에어컨(climate): set_temperature(extra:{temperature:숫자}), set_hvac_mode(extra:{hvac_mode:cool|heat|dry|fan_only|auto|off}), set_fan_mode(extra:{fan_mode:모드}), set_preset_mode(extra:{preset_mode:모드}), set_swing_mode(extra:{swing_mode:모드})\n" "- 선풍기(fan): set_percentage(extra:{percentage:0~100}), increase_speed, decrease_speed, oscillate(extra:{oscillating:true/false})\n" "- 청소기(vacuum): start, stop, pause, return_to_base, locate, set_fan_speed(extra:{fan_speed:속도})\n" "- 커튼(cover): open_cover, close_cover, stop_cover, set_cover_position(extra:{position:0~100})\n" "- 잠금(lock): lock, unlock\n" "- 미디어(media_player): media_play, media_pause, media_stop, volume_up, volume_down, volume_set(extra:{volume_level:0~1}), volume_mute(extra:{is_volume_muted:true/false}), select_source(extra:{source:이름}), media_next_track, media_previous_track\n\n" "그 외 → " '{"type":"chat","response":"구어체응답"}' + extra_context ) history = get_history_messages() messages = [{"role": "system", "content": system_msg}] + history + [{"role": "user", "content": text}] try: resp = client.chat.completions.create( model="gpt-4o-mini", max_tokens=5000, temperature=0.3, messages=messages, response_format={"type": "json_object"}, timeout=60, ) choice = resp.choices[0] finish_reason = choice.finish_reason if finish_reason not in ("stop", "length"): print(f"GPT finish_reason: {finish_reason}") raw = (choice.message.content or "").strip() if not raw: return {"type": "chat", "response": "응답을 받지 못했어요. 다시 말씀해 주세요."} try: result = json.loads(raw) except json.JSONDecodeError: m = re.search(r'"response"\s*:\s*"([^"]*)"', raw) response_text = m.group(1) if m else "정확한 정보를 찾지 못했어요." result = {"type": "chat", "response": response_text} result["response"] = clean_response(result.get("response", "")) add_to_history("user", text) add_to_history("assistant", result["response"]) return result except Exception as e: print(f"OpenAI 오류: {e}") return {"type": "chat", "response": "AI 오류가 났어요"} # ════════════════════════════════════════ # 명령 실행 # ════════════════════════════════════════ def execute(cmd): if cmd.get("type") != "control": return cmd.get("response") or "잘 모르겠어요" device = cmd.get("device") action = cmd.get("action", "turn_off") delay = int(cmd.get("delay_minutes") or 0) extra = cmd.get("extra") or {} resp = cmd.get("response", "") if device == "전체": for info in HA_DEVICES.values(): ha_call(info["domain"], "turn_off", info["entity_id"]) return resp or "전체 껐어요" # 기기 매칭: 정확한 이름 → 부분 포함 → fuzzy dev_info = None if device: # 1순위: 정확히 일치 if device in HA_DEVICES: dev_info = HA_DEVICES[device] else: # 2순위: 부분 문자열 매칭 for name, info in HA_DEVICES.items(): if device in name or name in device: dev_info = info break if not dev_info: # 3순위: 핵심 단어 매칭 (GPT가 "스마트 전구"→"전구") dev_words = device.replace("스마트 ", "").replace("스마트", "").strip() for name, info in HA_DEVICES.items(): if dev_words and (dev_words in name or name in dev_words): dev_info = info break if not dev_info: return f"'{device}' 기기를 찾을 수 없어요" eid = dev_info["entity_id"] dom = dev_info["domain"] # GPT action → HA 서비스로 변환 (_ACTION_MAP 사용) ha_action = action ha_extra = extra domain_map = _ACTION_MAP.get(dom, {}) if action in domain_map: ha_action, extra_fn = domain_map[action] ha_extra = extra_fn(extra) elif action in ("turn_on", "turn_off", "toggle"): ha_action = action ha_extra = {} else: # 매핑에 없는 action → 그대로 HA에 전달 시도 ha_action = action ha_extra = extra if delay > 0: def _later(): time.sleep(delay * 60) ha_call(dom, ha_action, eid, ha_extra or None) speak(f"{device} {ACT.get(ha_action, ha_action)}어요") threading.Thread(target=_later, daemon=True).start() h, m = divmod(delay, 60) ts = (f"{h}시간 " if h else "") + (f"{m}분" if m else "") return resp or f"{ts} 후에 {device} {ACT.get(ha_action, '동작')}을게요" ok = ha_call(dom, ha_action, eid, ha_extra or None) return (resp or f"{device} {ACT.get(ha_action, ha_action)}어요") if ok else f"{device} 제어 실패" # ════════════════════════════════════════ # 마이크 목록 # ════════════════════════════════════════ def list_mics(): print("\n마이크 목록:") for i, d in enumerate(sd.query_devices()): ch = d["max_input_channels"] if ch > 0: print(f" [{i}] {d['name']} ({ch}ch)") print() # ════════════════════════════════════════ # 메인 루프 # ════════════════════════════════════════ def run(): print("=" * 50) print(" 땡구 음성비서") print("=" * 50) list_mics() _detect_mic() load_devices() init_wake_model() init_supertonic() cache_phrases() CHUNK = 1280 # openWakeWord 요구 프레임 크기 print() if ENABLE_WAKE_WORD: print(" '땡구야' 라고 불러주세요!") else: print(" 내장 wake word로 대기 중...") print(" Ctrl+C 종료") print("=" * 50) print() def make_stream(): s = sd.InputStream(samplerate=16000, channels=_MIC_CH, dtype="int16", blocksize=CHUNK, device=_MIC_IDX) s.start() return s def destroy_stream(s): try: s.stop() s.close() except: pass def speak_with_wake_detect(resp_text, stream): """TTS 재생 중 wake word 감지. 감지 시 True 반환.""" print(f"땡구: {resp_text}") try: audio_path, own_file = _generate_audio(resp_text) except Exception as e: print(f"TTS 생성 오류: {e}") return False try: pygame.mixer.music.load(audio_path) pygame.mixer.music.play() except Exception as e: print(f"재생 오류: {e}") if own_file: _safe_unlink(audio_path) return False wake_model.reset() chunk_count = 0 RESET_INTERVAL = int(16000 / CHUNK * 3) wake_detected = False while pygame.mixer.music.get_busy(): try: raw_w, _ = stream.read(CHUNK) pred_w = wake_model.predict(_to_mono(raw_w).flatten()) for name, score in pred_w.items(): if score >= WAKE_THRESHOLD: print(f"\nTTS 중 Wake Word! [{name}: {score:.2f}]") pygame.mixer.music.stop() wake_model.reset() wake_detected = True break if wake_detected: break chunk_count += 1 if chunk_count >= RESET_INTERVAL: wake_model.reset() chunk_count = 0 except KeyboardInterrupt: raise except Exception as e: print(f" wake감지 오류: {e}") time.sleep(0.03) pygame.mixer.music.unload() if own_file: _safe_unlink(audio_path) return wake_detected def process_wake_response(response_text, stream, depth=1, max_depth=5): """응답 TTS 재생 + TTS 중 wake 감지 처리 (최대 5차 연속). Linux ALSA는 다중 스트림 불가 → 녹음 전 close, 녹음 후 재생성. 새 stream을 반환.""" if depth > max_depth: return stream if not speak_with_wake_detect(response_text, stream): return stream print(f"TTS 중 Wake Word ({depth}차) → 녹음") destroy_stream(stream) play_ack_nonblock() path = record_audio_safe() if not path: print(f"{depth}차: 녹음 실패") return make_stream() try: text = _strip_wakeword(stt(path)) if not text.strip(): print(f"{depth}차: 음성 인식 실패") return make_stream() print(f"인식: {text}") print("분석 중...") loading_start() try: cmd = analyze(text) finally: loading_stop() print(f" {json.dumps(cmd, ensure_ascii=False)}") new_stream = make_stream() return process_wake_response(execute(cmd), new_stream, depth + 1, max_depth) except KeyboardInterrupt: raise except Exception as e: loading_stop() print(f"{depth}차 연속 처리 오류: {e}") return make_stream() def handle_voice(stream): """wake 감지 후 음성 처리. 스트림 close→녹음→재생성. 새 stream을 반환.""" destroy_stream(stream) play_ack_nonblock() path = record_audio_safe() if not path: speak("잘 못 들었어요") _lock_release() wake_model.reset() print("Wake Word 대기 중...") return make_stream() try: text = _strip_wakeword(stt(path)) if len(text.replace(" ", "")) < 3: text = "" except Exception as e: print(f"STT 오류: {e}") speak("인식 오류가 났어요") _lock_release() wake_model.reset() print("Wake Word 대기 중...") return make_stream() if not text.strip(): speak("잘 못 들었어요") _lock_release() wake_model.reset() print("Wake Word 대기 중...") return make_stream() print(f"인식: {text}") print("분석 중...") loading_start() try: cmd = analyze(text) finally: loading_stop() print(f" {json.dumps(cmd, ensure_ascii=False)}") new_stream = make_stream() new_stream = process_wake_response(execute(cmd), new_stream) _lock_release() wake_model.reset() print("Wake Word 대기 중...") return new_stream stream = make_stream() try: while True: if not ENABLE_WAKE_WORD: print("음성 입력 대기 중... (Wake Word 비활성화)") path = passive_listen(timeout_sec=30) if not path: continue try: text = _strip_wakeword(stt(path)) if len(text.replace(" ", "")) < 3: text = "" except Exception as e: print(f"STT 오류: {e}") speak("인식 오류가 났어요") continue if not text.strip(): speak("잘 못 들었어요") continue print(f"인식: {text}") print("분석 중...") loading_start() try: cmd = analyze(text) finally: loading_stop() print(f" {json.dumps(cmd, ensure_ascii=False)}") speak(execute(cmd)) continue try: raw, _ = stream.read(CHUNK) except KeyboardInterrupt: raise except Exception as e: print(f"스트림 오류: {e}, 재시작...") destroy_stream(stream) time.sleep(0.5) stream = make_stream() continue pred = wake_model.predict(_to_mono(raw).flatten()) if not any(score >= WAKE_THRESHOLD for score in pred.values()): continue for name, score in pred.items(): if score >= WAKE_THRESHOLD: print(f"\nWake Word 감지! [{name}: {float(score.flatten()[0]):.2f}]") break wake_model.reset() if not _lock_acquire("로컬마이크"): print("[락] 다른 곳에서 처리 중 → Wake 무시") continue stop_speaking() stream = handle_voice(stream) except KeyboardInterrupt: print("\n종료") destroy_stream(stream) pygame.mixer.quit() sys.exit(0) if __name__ == "__main__": run()