#!/usr/bin/env python3 """ 땡구 음성비서 openWakeWord -> Groq Whisper API -> Gemini 3 Flash -> HA 제어 -> edge-tts (무료 Microsoft 신경망 TTS) 설치: pip install openwakeword sounddevice openai pygame requests numpy groq edge-tts google-generativeai 실행: 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_API_KEY = _env("OPENAI_API_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") OPENWEATHERMAP_KEY = _env("OPENWEATHERMAP_KEY") WAKE_THRESHOLD = float(_env("WAKE_THRESHOLD", "0.9")) ENABLE_WAKE_WORD = False RECORD_SECONDS = int(_env("RECORD_SECONDS", "10")) SILENCE_SEC = float(_env("SILENCE_SEC", "3.0")) SILENCE_DB = int(_env("SILENCE_DB", "28")) SKIP_SECONDS = float(_env("SKIP_SECONDS", "0.7")) MIN_SPEECH_CHUNKS = int(_env("MIN_SPEECH_CHUNKS", "4")) TTS_VOICE_KO = _env("TTS_VOICE_KO", "ko-KR-InJoonNeural") TTS_VOICE_RATE = _env("TTS_VOICE_RATE", "+25%") 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 WEATHER_CITY = _env("WEATHER_CITY", "서울") WEATHER_CITY_OWM = _env("WEATHER_CITY_OWM", "Seoul") # 한국어 도시명 → 영어 도시명 (OpenWeatherMap 호환용) _CITY_NAME_TO_ENG = { "서울": "Seoul", "부산": "Busan", "대구": "Daegu", "인천": "Incheon", "광주": "Gwangju", "대전": "Daejeon", "울산": "Ulsan", "세종": "Sejong", "제주": "Jeju", "수원": "Suwon", "성남": "Seongnam", "고양": "Goyang", "용인": "Yongin", "청주": "Cheongju", "천안": "Cheonan", "전주": "Jeonju", "포항": "Pohang", "창원": "Changwon", "안산": "Ansan", "안양": "Anyang", "남양주": "Namyangju", "화성": "Hwaseong", "의정부": "Uijeongbu", "평택": "Pyeongtaek", "시흥": "Siheung", "파주": "Paju", "김포": "Gimpo", "광명": "Gwangmyeong", "군포": "Gunpo", "하남": "Hanam", "오산": "Osan", "이천": "Icheon", "의왕": "Uiwang", "양주": "Yangju", "구리": "Guri", "여주": "Yeoju", "동두천": "Dongducheon", "과천": "Gwacheon", "강릉": "Gangneung", "원주": "Wonju", "춘천": "Chuncheon", "속초": "Sokcho", "삼척": "Samcheok", "목포": "Mokpo", "여수": "Yeosu", "순천": "Suncheon", "군산": "Gunsan", "익산": "Iksan", "김해": "Gimhae", "진주": "Jinju", "거제": "Geoje", "통영": "Tongyeong" } # 기본 기기 (서버에서 로드 성공 시 덮어씀) 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"), # Perplexity용 ("google-generativeai", "google.generativeai"), ("pygame", "pygame"), ("numpy", "numpy"), ]: 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 # import google.generativeai as genai # Gemini 제거 from openai import OpenAI from groq import Groq as GroqClient # 전역 API 클라이언트 _pplx_http = httpx.Client(timeout=httpx.Timeout(10, connect=3)) _pplx_client = OpenAI(api_key=PERPLEXITY_KEY, base_url="https://api.perplexity.ai", http_client=_pplx_http) _openai_client = OpenAI(api_key=OPENAI_API_KEY) _groq_client = GroqClient(api_key=GROQ_KEY) # genai.configure(api_key=GEMINI_KEY) # Gemini 제거 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 -> 오디오 파일 경로 (고정 문구 캐시) 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(f"말씀하세요... ({RECORD_SECONDS}초 대기)") RATE, CHUNK = 16000, 1024 # 초기 에코/자기소리 방지를 위해 일정 시간 무시 SKIP_CHUNKS = int(RATE / CHUNK * SKIP_SECONDS) # 발화 시작 전 최대 대기 시간 WAIT_FOR_START_SEC = float(RECORD_SECONDS) # 발화 시작 후 침묵 시 종료 시간 SILENCE_SEC_LOCAL = float(SILENCE_SEC) frames, silence_start, started = [], None, False speech_chunks = 0 potential_chunks = 0 # started가 되기 전까지 쌓이는 유효 프레임 수 with sd.InputStream(samplerate=RATE, channels=_MIC_CH, dtype="int16", blocksize=CHUNK, device=_MIC_IDX) as st: max_chunks = int(RATE / CHUNK * (WAIT_FOR_START_SEC + RECORD_SECONDS)) start_wait_begin = time.time() for i in range(max_chunks): data, _ = st.read(CHUNK) if i < SKIP_CHUNKS: continue mono = _to_mono(data) db = _db(mono) if db > SILENCE_DB: potential_chunks += 1 # 3개 이상의 데이터가 들어왔을 때 '진짜 발화'로 간주 if not started and potential_chunks >= 3: print(" (발화 감지) 녹음 중...") started = True if started: silence_start = None speech_chunks += 1 frames.append(mono.copy()) else: # 아직 '진짜'는 아니지만 프레임은 버리지 않고 혹시 모르니 보관 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_LOCAL: print(" (침묵) 녹음 종료") break else: # 시작 대기 중 (소리가 안 나거나 너무 짧은 소리일 때) potential_chunks = 0 # 연속되지 않은 짧은 소리는 리셋 frames = [] # 모아둔 프레임 비움 if time.time() - start_wait_begin > WAIT_FOR_START_SEC: print(" (대기 시간 초과)") break if 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 - edge-tts (primary) # ════════════════════════════════════════ _PYGAME_OK = False try: pygame.mixer.init() _PYGAME_OK = True except Exception as _e: print(f"[WARN] pygame 오디오 초기화 실패: {_e} — 스피커 연결 후 재시작 필요") 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 _clean_for_speech(text): """TTS 발음을 위한 텍스트 전처리 (영어 약어, 기호 등 확장 리스트)""" if not text: return "" # 영어 약어/기호 -> 한글 발음 매핑 (대규모 확장) repls = { # 대기업/브랜드 "SK": "에스케이", "LG": "엘지", "KT": "케이티", "SSG": "에스에스지", "GS": "지에스", "CJ": "씨제이", "NH": "농협", "MG": "새마을금고", "SAMSUNG": "삼성", "APPLE": "애플", # 기술/IT "IT": "아이티", "AI": "에이아이", "PC": "피씨", "TV": "티브이", "SSD": "에스에스디", "CPU": "씨피유", "RAM": "램", "GPU": "지피유", "USB": "유에스비", "HDMI": "에이치디엠아이", "SMS": "문자", "SNS": "에스엔에스", "URL": "유알엘", "API": "에이피아이", "VR": "브이알", "AR": "에이알", "OS": "오에스", "DB": "디비", "AWS": "아마존", "UI": "유아이", "UX": "유엑스", "IoT": "아이오티", "Wi-Fi": "와이파이", "WIFI": "와이파이", "GPT": "지피티", # 경제/금융 "KOSPI": "코스피", "KOSDAQ": "코스닥", "USD": "달러", "KRW": "원", "EUR": "유로", "JPY": "엔", "CNY": "위안", "BTC": "비트코인", "ETH": "이더리움", "IPO": "기업공개", "ETF": "이티에프", "CEO": "최고경영자", "CFO": "재무책임자", "VIP": "브이아이피", # 단위/기호 "%": "퍼센트", "km/h": "킬로미터 퍼 아워", "m/s": "미터 퍼 세컨드", "℃": "도", "°C": "도", "KM": "킬로미터", "CM": "센티미터", "MM": "밀리미터", "KG": "킬로그램", "MG": "밀리그램", "ML": "밀리리터", "VS": "대", "V S": "대", "X": "곱하기", "x": "곱하기", # 미디어/파일 "MP3": "엠피쓰리", "MP4": "엠피포", "WAV": "웨이브", "JPG": "제이피이지", "PNG": "피엔지", "PDF": "피디에프", "DOC": "워드파일", "XLS": "엑셀파일", "PPT": "피피티", } # 정확한 매핑을 위해 대문자 처리 후 치환 (단, 대소문자 섞인 경우 고려) # 여기서는 간단히 순차 치환 for k, v in repls.items(): text = text.replace(k, v) text = text.replace(k.lower(), v) # 기온 고유어(하나 도, 일곱 도 등) 강제 변환 (안전장치) num_map = { "하나": "일", "둘": "이", "셋": "삼", "넷": "사", "다섯": "오", "여섯": "육", "일곱": "칠", "여덟": "팔", "아홉": "구", "열": "십" } def _fix_degree(m): n = m.group(1).strip() # num_map에 있으면 변환, 없으면 그대로 (Python 3.10 호환 방식) val = num_map.get(n, n) return val + " 도" # regex matches e.g. "일곱 도", "하나도" text = re.sub(r'(하나|둘|셋|넷|다섯|여섯|일곱|여덟|아홉|열)\s*도\b', _fix_degree, text) return text def _generate_audio(text) -> tuple: """(파일경로, 임시파일여부) 반환. 캐시 히트 시 임시=False.""" text = _clean_for_speech(text) if text in _phrase_cache and os.path.exists(_phrase_cache[text]): return _phrase_cache[text], False return _edgetts_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: 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 requests.exceptions.ReadTimeout: # 명령 전송 후 응답 대기 중 타임아웃 → 명령은 이미 전달됨 return True 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 def _summarize_weather_with_brain(display_name, current_data, forecast_list=None, query_text=""): """라이브 날씨 데이터를 OpenAI 두뇌에 전달하여 '땡구 스타일' 답변 생성""" try: weather_info = { "current_time": datetime.now().strftime("%Y-%m-%d %H:%M"), "user_query": query_text, "city": display_name, "current": current_data, "forecast_brief": [] } if forecast_list: # 시간/날짜 필터링 (불필요한 데이터 제외하여 속도 향상) import datetime as dt requested_date = None is_past = False # 사용자 쿼리에서 날짜 추출 시도 dm = re.search(r'(\d+)\s*[월\.\-]\s*(\d+)', query_text) if dm: requested_date = f"{datetime.now().year}-{int(dm.group(1)):02d}-{int(dm.group(2)):02d}" elif "모레" in query_text: requested_date = (datetime.now() + dt.timedelta(days=2)).strftime("%Y-%m-%d") elif "내일" in query_text: requested_date = (datetime.now() + dt.timedelta(days=1)).strftime("%Y-%m-%d") elif "어제" in query_text: is_past = True elif "오늘" in query_text or not query_text: requested_date = datetime.now().strftime("%Y-%m-%d") weather_info["is_past_request"] = is_past for item in forecast_list: item_time = item.get("dt_txt", "") if requested_date: if requested_date in item_time: weather_info["forecast_brief"].append({ "time": item_time, "temp": item.get("main", {}).get("temp"), "desc": item.get("weather", [{}])[0].get("description", ""), "pop": item.get("pop", 0) * 100 }) elif not is_past: if len(weather_info["forecast_brief"]) < 8: weather_info["forecast_brief"].append({ "time": item_time, "temp": item.get("main", {}).get("temp"), "desc": item.get("weather", [{}])[0].get("description", ""), "pop": item.get("pop", 0) * 100 }) prompt = ( "너는 간결하고 싹싹한 스마트홈 날씨 비서야. 불필요한 인사는 생략하고 요청받은 시점의 정보를 즉시 전달해.\n" f"데이터: {json.dumps(weather_info, ensure_ascii=False)}\n\n" "[기본 준수 규칙]\n" "0. 과거 요청(is_past_request=true): 과거 날씨 데이터는 없으므로 알 수 없다고 정중히 말해줘.\n" "1. 요청 집중: 사용자가 묻는 시점의 정보만 말해. 묻지 않은 시점의 정보는 언급하지 마.\n" "2. 결론 우선 (No Greeting): 인사말 없이 즉시 해당 시점의 날씨 상태나 기온으로 시작해.\n" "3. 지역명 생략: 주소지(서울, 구 이름 등) 언급은 무조건 생략해.\n" "4. 기온 숫자 읽기 (특급 규칙): 기온은 무조건 한자어(Sino-Korean)로만 말해. 고유어는 절대 금지야.\n" " - 숫자 매핑: 0(영), 1(일), 2(이), 3(삼), 4(사), 5(오), 6(육), 7(칠), 8(팔), 9(구), 10(십)\n" " - 잘못된 예: 다섯 도(X), 열 도(X), 일곱 도(X), 여덟 도(X), 하나 도(X)\n" " - 올바른 예: 오 도(O), 십 도(O), 칠 도(O), 팔 도(O), 일 도(O)\n" "5. 답변 길이 조절: 평상시 딱 2문장, 위험 상황 시 4~5문장.\n" "6. TTS 최적화 (edge-tts): 특수문자 금지, 기온(한자어), 시간(고유어) 규칙 엄수.\n\n" "[말투 및 톤앤매너]\n" "- 싹싹함: 문장 끝은 ~예요, ~하세요, ~바랍니다를 섞어 비서다운 친절함을 유지해.\n" "- 자연스러움: 수치 앞에 약, 정도를 적절히 섞어 기계적인 나열을 방지해." ) resp = _openai_client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "너는 다정한 날씨 안내원 땡구야. 특수기호를 쓰지 않고 한글로만 자연스럽게 말해."}, {"role": "user", "content": prompt} ], temperature=0.7 ) return resp.choices[0].message.content.strip() except Exception as e: print(f"날씨 브레인 요약 실패: {e}") return None def fetch_weather(query_text, city=None): """실시간 날씨 조회. query_text를 통해 '지금' vs '오늘' 구분.""" display_name = city or WEATHER_CITY query_type = "now" if "내일" in query_text: query_type = "tomorrow" elif "오늘" in query_text: query_type = "today" if city: target_owm = _CITY_NAME_TO_ENG.get(city, city) else: target_owm = WEATHER_CITY_OWM api_key = OPENWEATHERMAP_KEY current_url = f"http://api.openweathermap.org/data/2.5/weather?q={target_owm}&appid={api_key}&units=metric&lang=kr" forecast_url = f"http://api.openweathermap.org/data/2.5/forecast?q={target_owm}&appid={api_key}&units=metric&lang=kr" try: r_curr = requests.get(current_url, timeout=5) if r_curr.status_code == 200: current_data = r_curr.json() r_fore = requests.get(forecast_url, timeout=5) forecast_data = r_fore.json().get("list", []) if r_fore.status_code == 200 else [] return _summarize_weather_with_brain(display_name, current_data, forecast_data, query_text) except Exception as e: print(f"Weather 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 # ════════════════════════════════════════ # STT 정규화 # ════════════════════════════════════════ _STT_FIXES = [ # ── 청소기 ────────────────────────────────────────────────── (r'로보\s*청소기', '로봇청소기'), (r'로봇\s+청소기', '로봇청소기'), (r'로봇\s*청소\s*기', '로봇청소기'), # ── 에어컨 ────────────────────────────────────────────────── (r'에어\s*콘', '에어컨'), (r'에어\s+컨', '에어컨'), # ── 공기청정기 (발음 오류 포함) ───────────────────────────── (r'공기\s*청전기|공기\s*청젱기|공기\s*청젱이', '공기청정기'), (r'공기\s*정전기|공기\s*정청기', '공기청정기'), (r'공기\s*청정\s*기', '공기청정기'), (r'공청\s*기', '공기청정기'), # ── 가습기 / 제습기 ───────────────────────────────────────── (r'가\s*습\s*기', '가습기'), (r'제\s*습\s*기', '제습기'), # ── 환풍기 ────────────────────────────────────────────────── (r'환\s*풍\s*기', '환풍기'), # ── TV / 텔레비전 ─────────────────────────────────────────── (r'텔레비\s*젼|텔레\s*비\s*전|텔레비$', '텔레비전'), (r'티\s*비', '티비'), # ── 냉장고 ────────────────────────────────────────────────── (r'냉\s*장\s*고', '냉장고'), # ── 세탁기 / 건조기 ───────────────────────────────────────── (r'세\s*탁\s*기', '세탁기'), (r'건\s*조\s*기', '건조기'), # ── 식기세척기 ────────────────────────────────────────────── (r'식기\s*세\s*척\s*기|식\s*기\s*세척기', '식기세척기'), # ── 전자레인지 ────────────────────────────────────────────── (r'전자\s*레\s*인\s*지|전자\s*레인지', '전자레인지'), (r'전자\s*렌지|전자\s*렌인지', '전자레인지'), # ── 인덕션 ────────────────────────────────────────────────── (r'인\s*덕\s*션|인덕\s*선', '인덕션'), # ── 정수기 ────────────────────────────────────────────────── (r'정\s*수\s*기', '정수기'), # ── 밥솥 ──────────────────────────────────────────────────── (r'밥\s*솥', '밥솥'), # ── 보일러 ────────────────────────────────────────────────── (r'보\s*일\s*러', '보일러'), # ── 도어락 ────────────────────────────────────────────────── (r'도어\s*락|도어\s*록|도\s*어락', '도어락'), # ── 블라인드 / 롤스크린 ───────────────────────────────────── (r'블라\s*인\s*드|블라인\s*드', '블라인드'), (r'롤\s*스\s*크\s*린|롤스\s*크린', '롤스크린'), # ── 선풍기 / 써큘레이터 ───────────────────────────────────── (r'선\s*풍\s*기', '선풍기'), (r'써\s*큘\s*레\s*이\s*터|써큘\s*레이터', '써큘레이터'), # ── 히터 ──────────────────────────────────────────────────── (r'히\s*터', '히터'), # ── 조명 관련 ─────────────────────────────────────────────── (r'간\s*접\s*등', '간접등'), (r'무\s*드\s*등', '무드등'), (r'취\s*침\s*등|수\s*면\s*등', '취침등'), (r'전\s*구', '전구'), (r'전\s*등', '전등'), (r'형\s*광\s*등', '형광등'), (r'스\s*탠\s*드|스탠\s*드', '스탠드'), (r'조\s*명', '조명'), (r'거실\s*등|거실\s*조명', '거실조명'), (r'방\s*등|방\s*조명', '방조명'), (r'주방\s*등|주방\s*조명', '주방조명'), (r'침실\s*등|침실\s*조명', '침실조명'), (r'현관\s*등|현관\s*조명', '현관조명'), (r'화장실\s*등|욕실\s*등', '화장실조명'), # ── 동작어 오타 (STT 받침·모음 혼동) ─────────────────────── (r'껐어\s*줘|거\s*줘(?!요)', '꺼줘'), (r'켜\s*줘요', '켜줘'), (r'틀어\s*줘요', '틀어줘'), (r'꺼\s*줘요', '꺼줘'), (r'올려\s*줘요', '올려줘'), (r'내려\s*줘요', '내려줘'), (r'높여\s*줘요', '높여줘'), (r'낮춰\s*줘요', '낮춰줘'), (r'열어\s*줘요', '열어줘'), (r'닫아\s*줘요', '닫아줘'), (r'시작\s*해\s*줘요', '시작해줘'), (r'멈춰\s*줘요', '멈춰줘'), (r'돌려\s*줘요', '돌려줘'), (r'복귀\s*시켜\s*줘요?', '복귀시켜줘'), (r'켜\s*주세요', '켜줘'), (r'꺼\s*주세요', '꺼줘'), (r'올려\s*주세요', '올려줘'), (r'내려\s*주세요', '내려줘'), (r'열어\s*주세요', '열어줘'), (r'닫아\s*주세요', '닫아줘'), (r'시작\s*해\s*주세요', '시작해줘'), (r'멈춰\s*주세요', '멈춰줘'), # ── 조명 단축칭 ───────────────────────── (r'(불|전등|전구)', '조명'), ] def normalize_text(text): """STT 오타·발음 오류 정규화""" for pattern, replacement in _STT_FIXES: text = re.sub(pattern, replacement, text) return text.strip() # ════════════════════════════════════════ # 룰 기반 - 기기 fuzzy 매칭 # ════════════════════════════════════════ def _match_device_fuzzy(word): """HA_DEVICES에서 기기명 fuzzy 매칭 (동의어 대응 포함)""" if not word: return None, None if word in HA_DEVICES: return word, HA_DEVICES[word] # 조명/전등 등 동의어 기반 선제 매칭 light_syns = ["조명", "불", "전등", "전구", "스위치"] is_light = any(s in word for s in ["조명", "불", "전등", "전구"]) for name, info in HA_DEVICES.items(): # 기본 부분 일치 if word in name or name in word: return name, info # 조명 관련 발화 시 이름에 동의어 포함 여부 확인 if is_light and any(s in name for s in light_syns): return name, info clean = word.replace("스마트 ", "").replace("스마트", "").strip() if clean: for name, info in HA_DEVICES.items(): if clean in name or name in clean: return name, info return None, None # ════════════════════════════════════════ # 룰 기반 - 씬 실행 # ════════════════════════════════════════ def _run_scene(actions): """씬 복합 동작 실행. actions: [(기기명, action, extra?), ...]""" for item in actions: dev_name = item[0] action = item[1] extra = item[2] if len(item) > 2 else {} if dev_name == "전체": for info in HA_DEVICES.values(): ha_call(info["domain"], "turn_off", info["entity_id"]) continue _, dev_info = _match_device_fuzzy(dev_name) if dev_info: ha_call(dev_info["domain"], action, dev_info["entity_id"], extra or None) # ════════════════════════════════════════ # 룰 기반 처리 데이터 # ════════════════════════════════════════ _TIME_PATTERNS = [ # 몇 시 계열 r'몇\s*시\s*(야|에요|이야|지|임|요|예요|인가요|인지|냐|니|래)', r'지금\s*몇\s*시', r'현재\s*몇\s*시', r'지금이\s*몇\s*시', r'몇\s*시\s*됐', r'몇\s*시\s*나', # 시간 알려줘 계열 r'시간\s*(알려|가르쳐|말해)\s*(줘|줘요|주세요|줄래|줄래요)?', r'(현재|지금)\s*시간\s*(알려|가르쳐|말해|어때|어떻게)', r'시간\s*(이\s*)?(어떻게|어때|몇)', r'(현재|지금)\s*시각', r'시각\s*(알려|가르쳐|말해)', # 오전/오후 관련 r'오전이야|오후야|오전인가요|오후인가요', r'오전이에요|오후에요', # 자연스러운 구어체 r'지금\s*몇\s*신지', r'몇\s*시\s*됐\s*(어|어요)', r'시간\s*좀\s*(알려|말해)', r'지금\s*몇\s*분', r'몇\s*분\s*(이야|이에요|이지)', ] _DATE_PATTERNS = [ # 오늘 날짜 계열 r'오늘\s*(날짜|이?며칠|이?무슨\s*날|이?언제)', r'날짜\s*(알려|가르쳐|말해)\s*(줘|줘요|주세요|줄래)?', r'오늘\s*날짜\s*(알려|가르쳐|말해)', # 몇 월 며칠 r'(몇|이)\s*월\s*(몇|이)\s*일', r'오늘\s*(몇|이)\s*월', r'오늘\s*(몇|이)\s*일', r'며칠\s*(이야|이에요|이지|인가요|임)', # 요일 계열 r'(무슨|어떤|몇)\s*요일', r'오늘\s*(이\s*)?(무슨|어떤)\s*요일', r'오늘\s*요일\s*(알려|가르쳐|말해|이야|이에요)', r'요일이\s*(어떻게|뭐야|뭐에요)', # 연도 / 올해 r'올해\s*(몇|이)\s*년', r'(지금|현재)\s*년도', r'(몇|어느)\s*년도', # 자연스러운 구어체 r'오늘이\s*(며칠|몇일|언제)', r'오늘\s*날짜\s*(좀\s*)?(말해|알려)', r'오늘\s*이\s*(무슨\s*날|어떤\s*날)', ] _GREET_RULES = [ # 인사 (r'^(안녕|하이|헬로|헤이|hi|hello|hey)\s*[!~.]*$', '안녕하세요!'), (r'^(안녕하세요|안녕하십니까)\s*[!~.]*$', '안녕하세요! 무엇을 도와드릴까요?'), (r'^(안녕\s*땡구|땡구\s*안녕)\s*[!~.]*$', '안녕하세요!'), (r'(좋은\s*)?(아침|아침이에요|아침이야|굿모닝|good\s*morning)', '좋은 아침이에요!'), (r'(좋은\s*)?(저녁|저녁이에요|저녁이야|굿이브닝)', '좋은 저녁이에요!'), (r'(좋은\s*)?(밤|밤이에요|밤이야|굿나잇|좋은밤)', '좋은 밤 되세요!'), # 감사 (r'고마워|고맙다|감사해|고맙습니다|감사합니다|감사해요|고마워요', '별말씀을요!'), (r'정말\s*(고마워|감사해|고맙다)', '아이구, 별말씀을요!'), (r'수고\s*(했어|했어요|많았어|많았어요)', '감사해요! 또 불러주세요.'), # 칭찬/격려 (r'잘\s*(했어|했어요|하고\s*있어|하고\s*있어요)', '감사해요!'), (r'최고야|최고에요|대단해|대단해요|훌륭해', '감사합니다!'), (r'(너\s*)?똑똑하다|(너\s*)?똑똑해|영리하다|영리해', '과찬이세요!'), (r'(잘\s*)?부탁해|(잘\s*)?부탁드려|(잘\s*)?부탁합니다', '네, 말씀해 주세요!'), # 상태 질문 (r'뭐\s*(해|하고\s*있어|하니|하고\s*있어요|하고\s*있니)', '명령 기다리고 있어요.'), (r'(지금\s*)?뭐\s*(하는\s*중|하는\s*중이야|하고\s*있는\s*중)', '명령 기다리고 있어요.'), (r'(잘\s*)?지내|어떻게\s*지내|잘\s*있어', '네, 잘 있어요! 도와드릴까요?'), (r'기분\s*(어때|어떠세요|좋아)', '네, 좋아요! 뭐 도와드릴까요?'), # 간단 응답 (r'^(네|응|어|오케이|알겠어|알겠어요|알겠습니다|ㅇㅋ)\s*[.!~]*$', '네!'), (r'^(아니|아니요|아니에요)\s*[.!~]*$', '아, 알겠어요!'), (r'^(잠깐|잠시만|잠시만요|조금만\s*기다려)\s*[.!~]*$', '네, 기다릴게요.'), # 칭찬/인격 (r'(넌\s*)?이름이\s*(뭐야|뭐에요|뭐니|뭔데)', '저는 땡구예요!'), (r'(넌\s*)?뭘\s*(할\s*수\s*있어|할\s*수\s*있어요|도와줄\s*수\s*있어)', '스마트홈 제어, 날씨, 검색 등 도와드릴 수 있어요!'), (r'도움\s*(돼|됐어|됐어요)', '도움이 됐다니 기뻐요!'), # 작별 (r'바이|빠이|잘\s*있어|안녕히\s*(계세요|가세요)', '안녕히 계세요!'), (r'수고해|수고하세요|다음에\s*(봐|봐요)', '네, 또 불러주세요!'), ] _SCENE_RULES = [ # ── 취침 ─────────────────────────────────────────────────────────────── (r'자자|자야겠어|취침\s*모드|수면\s*모드|잠\s*잘게|이제\s*잘게' r'|잠자러\s*가|자러\s*가|자려고|이제\s*자|다\s*자|슬슬\s*자|잠\s*자야' r'|굿나잇|good\s*night|나이트', [("전체조명", "turn_off")], "잘 자요! 조명 껐어요."), # ── 외출 ─────────────────────────────────────────────────────────────── (r'나간다|외출\s*(해|할게|모드|할거야)|다녀올게|나갈게|나가요|나갑니다' r'|외출\s*해요|집\s*비워|잠깐\s*나가|나가볼게|나가는\s*중' r'|밖에\s*나가|학교\s*가|회사\s*가|출근', [("전체", "turn_off")], "다녀오세요! 전부 껐어요."), # ── 귀가 ─────────────────────────────────────────────────────────────── (r'집에\s*왔어|들어왔어|귀가|다녀왔어|집에\s*도착|집\s*왔어' r'|들어왔습니다|들어왔어요|집에\s*왔습니다|퇴근\s*했어|퇴근했어요' r'|돌아왔어|돌아왔어요', [("거실조명", "turn_on"), ("거실 조명", "turn_on")], "어서 오세요! 거실 조명 켰어요."), # ── 영화/시네마 ────────────────────────────────────────────────────── (r'영화\s*(볼게|볼거야|시작|모드|보자|틀어줘|켜줘)|시네마\s*모드' r'|넷플릭스|유튜브\s*볼게|티비\s*볼게|TV\s*볼게', [("거실조명", "set_brightness", {"brightness": 15}), ("거실 조명", "set_brightness", {"brightness": 15})], "영화 잘 보세요!"), # ── 독서/공부 ──────────────────────────────────────────────────────── (r'독서\s*모드|공부\s*모드|책\s*읽을게|공부할게|집중\s*모드', [("거실조명", "set_brightness", {"brightness": 80}), ("거실 조명", "set_brightness", {"brightness": 80})], "독서 모드로 설정했어요!"), # ── 기상 ─────────────────────────────────────────────────────────────── (r'일어났어|기상|굿모닝|좋은\s*아침|아침이야|일어났어요|잠\s*깼어' r'|good\s*morning|방금\s*일어났어', [("거실조명", "turn_on"), ("거실 조명", "turn_on")], "좋은 아침이에요! 거실 조명 켰어요."), # ── 파티/손님 ──────────────────────────────────────────────────────── (r'파티\s*모드|손님\s*오셨어|손님\s*왔어|파티\s*시작|홈\s*파티', [("거실조명", "turn_on"), ("거실 조명", "turn_on")], "파티 분위기 설정했어요!"), # ── 로맨틱/분위기 ─────────────────────────────────────────────────── (r'로맨틱\s*모드|분위기\s*있게|무드\s*모드|감성\s*조명|분위기\s*조명', [("거실조명", "set_brightness", {"brightness": 30}), ("거실 조명", "set_brightness", {"brightness": 30})], "분위기 있게 설정했어요!"), # ── 전체 끄기 ──────────────────────────────────────────────────────── (r'전체\s*(꺼줘|껐|끄자|꺼|끄고|끄세요)|다\s*(꺼줘|꺼|끄자|껐)' r'|모두\s*(꺼줘|꺼|끄자)|전부\s*(꺼줘|꺼|끄자|끄세요)' r'|싹\s*다\s*꺼|집에\s*있는거\s*다\s*꺼', [("전체", "turn_off")], "전부 껐어요."), # ── 전체 켜기 ──────────────────────────────────────────────────────── (r'전체\s*(켜줘|켜|틀어줘|켜세요)|다\s*(켜줘|켜|틀어)' r'|모두\s*(켜줘|켜)|전부\s*(켜줘|켜|켜세요)', [("거실조명", "turn_on"), ("거실 조명", "turn_on")], "전체 켰어요."), # ── 에어컨 + 조명 함께 (더운 날) ─────────────────────────────────── (r'너무\s*(더워|더워요|덥다)|더워\s*죽겠어|폭염\s*모드|시원하게\s*해줘', [("에어컨", "turn_on"), ("거실조명", "turn_on"), ("거실 조명", "turn_on")], "에어컨 켰어요. 시원하게 해드릴게요!"), # ── 추울 때 ───────────────────────────────────────────────────────── (r'너무\s*(추워|추워요|춥다)|추워\s*죽겠어|따뜻하게\s*해줘|히팅\s*모드', [("에어컨", "set_hvac_mode", {"hvac_mode": "heat"}), ("보일러", "turn_on")], "난방 켰어요. 따뜻하게 해드릴게요!"), # ── 청소 시작 ──────────────────────────────────────────────────────── (r'청소\s*시작|청소\s*해줘|청소\s*해|청소기\s*돌려줘|청소기\s*돌려' r'|로봇청소기\s*켜줘|청소\s*돌려|청소기\s*시작', [("로봇청소기", "start")], "청소기 돌릴게요!"), # ── 청소기 복귀 ────────────────────────────────────────────────────── (r'청소기\s*(집으로|복귀|돌아와|충전해|충전|충전기로)' r'|로봇청소기\s*(집으로|복귀|돌아와|충전)', [("로봇청소기", "return_to_base")], "청소기 충전 보낼게요!"), # ── 공기 쾌적하게 ─────────────────────────────────────────────────── (r'공기\s*(쾌적하게|좋게|맑게)\s*(해줘)?|공기\s*청정\s*해줘' r'|환기\s*해줘|환기\s*시켜줘', [("공기청정기", "turn_on")], "공기청정기 켰어요!"), # ── 잠자리 준비 (조명 약하게) ─────────────────────────────────────── (r'잠자리\s*준비|자기\s*전|슬슬\s*자야|조명\s*낮춰줘|어둡게\s*해줘' r'|빛\s*줄여줘', [("거실조명", "set_brightness", {"brightness": 10}), ("거실 조명", "set_brightness", {"brightness": 10})], "조명 어둡게 했어요."), ] _SIMPLE_ACTIONS = [ # ── 켜기 ───────────────────────────────────────────────────────────── (r'켜\s*줘|틀어\s*줘|켜\s*줘요|켜\s*주세요|켜\s*봐|켜\s*봐요' r'|틀어\s*줘요|틀어\s*주세요|가동\s*해줘|작동\s*해줘|작동\s*시켜줘' r'|켜라|켜봐|켜|틀어', 'turn_on'), # ── 끄기 ───────────────────────────────────────────────────────────── (r'꺼\s*줘|끄\s*줘|꺼\s*줘요|꺼\s*주세요|끄\s*줘요|끄\s*주세요' r'|꺼라|꺼봐|꺼|끄고|정지\s*시켜|끄세요|꺼져', 'turn_off'), # ── 토글 ───────────────────────────────────────────────────────────── (r'토글|껐다\s*켜|켰다\s*꺼', 'toggle'), # ── 청소기 시작 ────────────────────────────────────────────────────── (r'시작\s*(해줘|해|해요|시켜|시켜줘)?|돌려\s*(줘|줘요)?' r'|돌아\s*가|작동\s*(해|시켜|시작)|청소\s*시작', 'start'), # ── 청소기 복귀 ────────────────────────────────────────────────────── (r'복귀\s*(해|해줘|시켜|시켜줘)?|충전\s*(해|해줘|해요|기로)?' r'|집으로\s*(와|와줘|가|가줘)|돌아\s*(와|와줘)', 'return_to_base'), # ── 정지/멈춤 ──────────────────────────────────────────────────────── (r'멈춰\s*(줘|줘요|주세요)?|중지\s*(해|해줘|해요)?|정지\s*(해|해줘|해요)?' r'|세워\s*줘|스톱|stop', 'stop'), # ── 일시정지 ───────────────────────────────────────────────────────── (r'일시\s*정지\s*(해|해줘|해요)?|잠깐\s*(멈춰|세워)\s*(줘)?' r'|포즈|pause|잠시\s*멈춰', 'pause'), # ── 커버 열기 ──────────────────────────────────────────────────────── (r'열어\s*(줘|줘요|주세요|봐|봐요)?|올려\s*(줘|줘요)?' r'|열어라|열어|올려', 'open_cover'), # ── 커버 닫기 ──────────────────────────────────────────────────────── (r'닫아\s*(줘|줘요|주세요|봐)?|내려\s*(줘|줘요)?' r'|닫아라|닫아|내려', 'close_cover'), # ── 잠금 ───────────────────────────────────────────────────────────── (r'잠가\s*(줘|줘요|주세요)?|잠궈\s*(줘|줘요)?|락\s*(걸어|걸어줘)' r'|잠금\s*(해|해줘)', 'lock'), # ── 잠금 해제 ──────────────────────────────────────────────────────── (r'(잠금\s*)?해제\s*(해|해줘|해요)?|언락|unlock' r'|열어\s*줘(?!\s*요)', 'unlock'), # ── 볼륨 올리기 ────────────────────────────────────────────────────── (r'볼륨\s*(올려|높여|키워|크게)\s*(줘|줘요)?' r'|소리\s*(키워|크게|올려|높여)\s*(줘|줘요)?' r'|더\s*크게\s*(해줘|틀어줘)?|소리\s*좀\s*키워', 'volume_up'), # ── 볼륨 내리기 ────────────────────────────────────────────────────── (r'볼륨\s*(내려|낮춰|줄여|작게)\s*(줘|줘요)?' r'|소리\s*(줄여|낮춰|작게|내려)\s*(줘|줘요)?' r'|더\s*작게\s*(해줘)?|소리\s*좀\s*줄여', 'volume_down'), # ── 음소거 ─────────────────────────────────────────────────────────── (r'음소거\s*(해|해줘|해요)?|뮤트\s*(해|해줘)?|소리\s*(꺼|끄|없애)' r'|조용히\s*(해줘|해|해요)?|mute', 'volume_mute'), # ── 재생 ───────────────────────────────────────────────────────────── (r'재생\s*(해|해줘|해요)?|플레이\s*(해|해줘)?|틀어줘(?!.*꺼)' r'|play|다시\s*재생', 'media_play'), # ── 일시정지(미디어) ───────────────────────────────────────────────── (r'(미디어|음악|영상)\s*일시\s*정지|잠깐\s*멈춰\s*줘', 'media_pause'), # ── 다음 곡/트랙 ────────────────────────────────────────────────────── (r'다음\s*(곡|트랙|노래)\s*(틀어|재생|넘겨|넘어가)?' r'|넥스트|next\s*트랙|다음으로', 'media_next_track'), # ── 이전 곡/트랙 ────────────────────────────────────────────────────── (r'이전\s*(곡|트랙|노래)\s*(틀어|재생|넘겨)?' r'|previous\s*트랙|이전으로|뒤로', 'media_previous_track'), # ── 청소기 위치 찾기 ──────────────────────────────────────────────── (r'청소기\s*(어디\s*있어|위치|찾아|찾아줘)|locate|위치\s*알려줘', 'locate'), # ── 에어컨 팬 전용 ────────────────────────────────────────────────── (r'(에어컨|선풍기)\s*(바람\s*만|팬\s*모드|송풍)', 'fan_only'), ] def rule_analyze(text): """룰 기반 처리. 처리 가능하면 결과 dict 반환, 불가능하면 None.""" now = datetime.now() text = normalize_text(text) # 시간 체크 for pat in _TIME_PATTERNS: if re.search(pat, text): ampm = "오전" if now.hour < 12 else "오후" h12 = now.hour if now.hour <= 12 else now.hour - 12 return {"type": "chat", "response": f"지금 {ampm} {h12}시 {now.minute:02d}분이에요."} # 날짜 체크 for pat in _DATE_PATTERNS: if re.search(pat, text): wd = ["월요일","화요일","수요일","목요일","금요일","토요일","일요일"][now.weekday()] return {"type": "chat", "response": f"오늘은 {now.month}월 {now.day}일 {wd}이에요."} # 인사 / 간단한 말 for pat, reply in _GREET_RULES: if re.search(pat, text, re.IGNORECASE): return {"type": "chat", "response": reply} # 씬/모드 (복합 동작) for pat, actions, reply in _SCENE_RULES: if re.search(pat, text): _run_scene(actions) add_to_history("user", text) add_to_history("assistant", reply) return {"type": "chat", "response": reply} # 단일 기기 제어 (기기명 + 명확한 동작) matched_name, matched_dev = None, None # 1. 텍스트에서 명시적 기기명 찾기 for name in HA_DEVICES: if name in text: matched_name, matched_dev = name, HA_DEVICES[name] break # 2. 기기명이 텍스트에 없으면, 텍스트의 앞부분이 기기명인지 확인 if not matched_dev: words = text.split() if words: target_word = words[0] matched_name, matched_dev = _match_device_fuzzy(target_word) if matched_dev: print(f" (푸지 매칭됨: {target_word} -> {matched_name})") if matched_dev: for pat, action in _SIMPLE_ACTIONS: if re.search(pat, text): print(f" (동작 매칭됨: {action})") ok = ha_call(matched_dev["domain"], action, matched_dev["entity_id"]) verb = ACT.get(action, action) resp = f"{matched_name} {verb}어요." if ok else f"{matched_name} 제어 실패했어요." add_to_history("user", text) add_to_history("assistant", resp) return {"type": "chat", "response": resp} return None # 룰로 처리 불가 → AI로 넘김 # ════════════════════════════════════════ # 공통 헬퍼 # ════════════════════════════════════════ def _build_system_msg(extra_context=""): dev_list = "\n".join(f"- {n}" for n in HA_DEVICES) now = datetime.now().strftime("%Y-%m-%d %H:%M") return ( 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 ) def _build_gemini_history(): return [ {"role": "model" if msg["role"] == "assistant" else "user", "parts": [{"text": msg["content"]}]} for msg in get_history_messages() ] def _parse_gemini_raw(raw): raw = re.sub(r'^```(?:json)?\s*', '', raw, flags=re.MULTILINE) raw = re.sub(r'\s*```$', '', raw, flags=re.MULTILINE).strip() try: return json.loads(raw) except json.JSONDecodeError: m = re.search(r'"response"\s*:\s*"([^"]*)"', raw) return {"type": "chat", "response": m.group(1) if m else "정확한 정보를 찾지 못했어요."} def _save_history(text, result): result["response"] = clean_response(result.get("response", "")) add_to_history("user", text) add_to_history("assistant", result["response"]) def analyze_brain(text, system_msg): """OpenAI(gpt-4o-mini)을 사용하여 요청 분석 및 JSON 응답 생성""" try: messages = [{"role": "system", "content": system_msg}] for msg in get_history_messages(): messages.append({"role": "assistant" if msg["role"] == "assistant" else "user", "content": msg["content"]}) messages.append({"role": "user", "content": text}) resp = _openai_client.chat.completions.create( model="gpt-4o-mini", messages=messages, response_format={"type": "json_object"}, temperature=0.1, ) return json.loads(resp.choices[0].message.content) except Exception as e: print(f"OpenAI Brain 오류: {e}") return {"type": "chat", "response": "분석 중에 문제가 생겼어요."} # ════════════════════════════════════════ # AI 분석 (통합 진입점) # ════════════════════════════════════════ def analyze(text): # 0. STT 정규화 text = normalize_text(text) # 1. 룰 기반 (시간/날짜/인사/씬/단순 제어) result = rule_analyze(text) if result: return result # 2. 뉴스/날씨/검색 분기 (Perplexity 직접) NEWS_KW = ["뉴스","소식","헤드라인","이슈"] WEATHER_KW = ["날씨","기온","몇도","비 올","비올","눈 올","눈올","우산","체감", "미세먼지", "초미세먼지", "공기질", "농도"] CTRL_KW = [ "켜","꺼","틀어","끄","올려","내려","높여","낮춰","줄여","키워", "시작","멈춰","중지","정지","일시정지","재개","작동","동작", "설정","바꿔","변경","맞춰","조절","전환", "열어","닫아","잠가","잠금","해제","열림","닫힘", "조명","불","전구","밝기","밝게","어둡게","퍼센트", "색","색깔","색온도","무드","따뜻","시원", "램프","간접등","스탠드","무드등","취침등","수면등", "디밍","형광","백열","주광","전구색", "에어컨","냉방","난방","제습","송풍","바람", "온도","도","시원하게","따뜻하게","춥","더워", "히터","보일러","온풍기","라디에이터","선풍기","써큘레이터", "실내온도","희망온도","자동","쾌적","절전","파워", "청소기","청소","로봇","충전","복귀","도킹", "걸레","물걸레","흡입","먼지", "공기청정기","공청기","정화","환기","환풍","환풍기", "가습기","가습","제습기","습도","미세먼지","공기질", "TV","티비","텔레비전","볼륨","채널","음량","소리", "스피커","뮤트","음소거","재생","일시중지","다음","이전", "입력","HDMI","소스","넷플릭스","유튜브", "플러그","콘센트","전원","스위치","멀티탭", "커튼","블라인드","롤스크린","차양","암막","셔터", "도어락","현관","문","잠금장치","초인종","벨", "CCTV","카메라","감시","녹화","센서","모션", "경보","알람","보안","방범","사이렌", "세탁기","세탁","건조기","건조","식기세척기","냉장고", "오븐","전자레인지","밥솥","인덕션","정수기", "다리미","헤어드라이어","비데","변기", "모드","세기","단계","풍량","풍속","타이머","예약", "스케줄","루틴","자동화","씬","장면", ] 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 = "오늘 주요 뉴스 헤드라인" if len(search_term) < 2 else f"{search_term} 최신 뉴스" news = fetch_perplexity(news_query) if news: add_to_history("user", text) add_to_history("assistant", news) return {"type": "chat", "response": news} extra_context = "\n\n[뉴스 수집 실패]\n뉴스를 가져오지 못했다고 솔직히 말하고, 잠시 후 다시 시도해달라고 안내해." elif is_weather: print("날씨 검색 중...") city_match = re.search( r'(서울|부산|대구|인천|광주|대전|울산|세종|제주' r'|수원|성남|고양|용인|청주|천안|전주|포항|창원' r'|안산|안양|남양주|화성|의정부|평택|시흥|파주' r'|김포|광명|군포|하남|오산|이천|의왕|양주|구리' r'|여주|동두천|과천|강릉|원주|춘천|속초|삼척' r'|목포|여수|순천|군산|익산|김해|진주|거제|통영)', text ) city = city_match.group(1) if city_match else None # None → WEATHER_CITY 기본값 사용 weather = fetch_weather(text, city) if weather: add_to_history("user", text) add_to_history("assistant", weather) return {"type": "chat", "response": weather} extra_context = "\n\n[날씨 수집 실패]\n날씨 정보를 가져오지 못했다고 솔직히 말하고, 잠시 후 다시 시도해달라고 안내해." elif is_search: print("웹 검색 중...") results = fetch_perplexity(text) if results: add_to_history("user", text) add_to_history("assistant", results) return {"type": "chat", "response": results} extra_context = "\n\n[웹 검색 결과 없음]\n검색 결과를 찾지 못했다고 짧게 솔직히 말해." # 3. 브레인 분석 (OpenAI) print("분석(OpenAI)...") system_msg = _build_system_msg(extra_context) result = analyze_brain(text, system_msg) if result: _save_history(text, result) return result return {"type": "chat", "response": "죄송해요, 이해하지 못했어요."} # ════════════════════════════════════════ # 명령 실행 # ════════════════════════════════════════ 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() 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)}") speak(execute(cmd)) _lock_release() wake_model.reset() print("Wake Word 대기 중...") return make_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()