first commit
This commit is contained in:
203
archive/claw_mic.py
Normal file
203
archive/claw_mic.py
Normal file
@@ -0,0 +1,203 @@
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
import requests
|
||||
import speech_recognition as sr
|
||||
import pygame
|
||||
import io
|
||||
from ctypes import *
|
||||
|
||||
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
|
||||
|
||||
# ALSA 에러 로그 숨기기
|
||||
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
|
||||
def py_error_handler(filename, line, function, err, fmt): pass
|
||||
c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
|
||||
try:
|
||||
asound = cdll.LoadLibrary('libasound.so.2')
|
||||
asound.snd_lib_error_set_handler(c_error_handler)
|
||||
except:
|
||||
pass
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ─── 환경 설정 ───────────────────────────────────────────────
|
||||
VM_IP = "192.168.0.149"
|
||||
OPENCLAW_PORT = 18789
|
||||
|
||||
# 오픈클로 Gateway RPC 엔드포인트
|
||||
# chat.send: 텍스트 메시지 전송
|
||||
CHAT_URL = f"http://{VM_IP}:{OPENCLAW_PORT}/api/v1/chat/completions"
|
||||
# 오디오 첨부 전송 (media.audio enabled 시 Gemini가 STT 처리)
|
||||
AUDIO_URL = f"http://{VM_IP}:{OPENCLAW_PORT}/v1/audio/transcriptions"
|
||||
|
||||
# TTS: openedai-speech 또는 오픈클로 내장 TTS
|
||||
TTS_URL = f"http://{VM_IP}:{OPENCLAW_PORT}/v1/audio/speech"
|
||||
|
||||
WAKE_WORD = "애미나이"
|
||||
MIC_INDEX = 3
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
pygame.mixer.init()
|
||||
|
||||
def send_audio_to_openclaw(audio_data) -> str:
|
||||
"""
|
||||
WAV를 오픈클로에 전송.
|
||||
media.audio.enabled + google provider 설정 시
|
||||
오픈클로가 Gemini로 STT 후 LLM 답변까지 처리.
|
||||
반환값: 오픈클로 텍스트 응답
|
||||
"""
|
||||
try:
|
||||
wav_bytes = audio_data.get_wav_data()
|
||||
files = {
|
||||
'audio': ('cmd.wav', io.BytesIO(wav_bytes), 'audio/wav')
|
||||
}
|
||||
logger.info("오픈클로로 오디오 전송 중...")
|
||||
resp = requests.post(AUDIO_URL, files=files, timeout=30)
|
||||
resp.raise_for_status()
|
||||
logger.info(f"응답 코드: {resp.status_code}")
|
||||
|
||||
# 오픈클로 응답이 JSON인 경우
|
||||
try:
|
||||
data = resp.json()
|
||||
# 오픈클로 응답 구조에 따라 키 조정
|
||||
return data.get("reply") or data.get("message") or data.get("text") or str(data)
|
||||
except Exception:
|
||||
# 텍스트 응답인 경우
|
||||
return resp.text
|
||||
|
||||
except requests.exceptions.ConnectionError:
|
||||
logger.error(f"연결 실패 - {VM_IP}:{OPENCLAW_PORT} 확인 필요")
|
||||
return ""
|
||||
except Exception as e:
|
||||
logger.error(f"오디오 전송 실패: {e}")
|
||||
return ""
|
||||
|
||||
def send_text_to_openclaw(text: str) -> str:
|
||||
"""텍스트를 오픈클로 chat API로 전송 (오디오 전송 안 될 때 폴백)"""
|
||||
try:
|
||||
payload = {"message": text}
|
||||
resp = requests.post(CHAT_URL, json=payload, timeout=30)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return data.get("reply") or data.get("message") or data.get("text") or str(data)
|
||||
except Exception as e:
|
||||
logger.error(f"텍스트 전송 실패: {e}")
|
||||
return ""
|
||||
|
||||
def speak_edge_tts(text: str):
|
||||
"""
|
||||
오픈클로 내장 TTS (Edge TTS, API 키 불필요) 사용.
|
||||
안 되면 Google gTTS로 폴백.
|
||||
"""
|
||||
try:
|
||||
payload = {
|
||||
"model": "tts-1",
|
||||
"input": text,
|
||||
"voice": "alloy",
|
||||
"response_format": "mp3"
|
||||
}
|
||||
resp = requests.post(TTS_URL, json=payload, timeout=15)
|
||||
if resp.status_code == 200:
|
||||
audio_buf = io.BytesIO(resp.content)
|
||||
pygame.mixer.music.load(audio_buf, "mp3")
|
||||
pygame.mixer.music.play()
|
||||
while pygame.mixer.music.get_busy():
|
||||
time.sleep(0.1)
|
||||
logger.info("TTS 재생 완료")
|
||||
return
|
||||
except Exception as e:
|
||||
logger.warning(f"오픈클로 TTS 실패, gTTS 시도: {e}")
|
||||
|
||||
# gTTS 폴백 (로컬 설치: pip install gtts)
|
||||
try:
|
||||
from gtts import gTTS
|
||||
tts = gTTS(text=text, lang='ko')
|
||||
tts.save("/tmp/reply.mp3")
|
||||
pygame.mixer.music.load("/tmp/reply.mp3")
|
||||
pygame.mixer.music.play()
|
||||
while pygame.mixer.music.get_busy():
|
||||
time.sleep(0.1)
|
||||
except Exception as e:
|
||||
logger.error(f"gTTS도 실패: {e}")
|
||||
|
||||
def main():
|
||||
recognizer = sr.Recognizer()
|
||||
recognizer.energy_threshold = 400
|
||||
recognizer.pause_threshold = 1.0
|
||||
|
||||
# 마이크 목록 출력
|
||||
mics = sr.Microphone.list_microphone_names()
|
||||
logger.info(f"사용 가능한 마이크 ({len(mics)}개):")
|
||||
for i, mic in enumerate(mics):
|
||||
logger.info(f" {i}: {mic}")
|
||||
|
||||
# 안전하게 마이크 열기
|
||||
mic = None
|
||||
for test_index in [MIC_INDEX, 0, 1, 2, 3]: # 여러 인덱스 시도
|
||||
try:
|
||||
logger.info(f"마이크 {test_index} 테스트 중...")
|
||||
mic = sr.Microphone(device_index=test_index)
|
||||
with mic as source:
|
||||
test_audio = recognizer.listen(source, timeout=1, phrase_time_limit=1)
|
||||
logger.info(f"✅ 마이크 {test_index} 성공!")
|
||||
break
|
||||
except Exception as e:
|
||||
logger.warning(f"❌ 마이크 {test_index} 실패: {e}")
|
||||
continue
|
||||
|
||||
if mic is None:
|
||||
logger.error("사용 가능한 마이크를 찾을 수 없습니다. 프로그램 종료.")
|
||||
return
|
||||
|
||||
# 소음 보정 (실패해도 무시)
|
||||
try:
|
||||
with mic as source:
|
||||
recognizer.adjust_for_ambient_noise(source, duration=1)
|
||||
logger.info(f"energy_threshold: {recognizer.energy_threshold:.0f}")
|
||||
except Exception as e:
|
||||
logger.warning(f"소음 보정 실패 (무시): {e}")
|
||||
|
||||
logger.info(f"==== '{WAKE_WORD}' 대기 중 ====")
|
||||
|
||||
while True:
|
||||
try:
|
||||
with mic as source:
|
||||
logger.debug("마이크 듣는 중...")
|
||||
audio = recognizer.listen(source, timeout=None, phrase_time_limit=3)
|
||||
|
||||
try:
|
||||
text = recognizer.recognize_google(audio, language='ko-KR')
|
||||
logger.info(f"🎤 인식: '{text}'")
|
||||
|
||||
if WAKE_WORD in text:
|
||||
logger.info("🔔 호출어 감지!")
|
||||
speak_edge_tts("네, 말씀하세요")
|
||||
|
||||
# 명령어 녹음
|
||||
with mic as source:
|
||||
logger.info("명령 녹음 중...")
|
||||
cmd_audio = recognizer.listen(source, timeout=5, phrase_time_limit=10)
|
||||
|
||||
# 오픈클로 전송
|
||||
reply = send_audio_to_openclaw(cmd_audio)
|
||||
if reply:
|
||||
speak_edge_tts(reply)
|
||||
else:
|
||||
speak_edge_tts("응답을 받지 못했어요.")
|
||||
|
||||
except sr.UnknownValueError:
|
||||
pass # 조용히 무시
|
||||
except sr.RequestError as e:
|
||||
logger.error(f"Google STT 오류: {e}")
|
||||
|
||||
except sr.WaitTimeoutError:
|
||||
pass # 타임아웃 무시
|
||||
except Exception as e:
|
||||
logger.error(f"루프 오류: {e}")
|
||||
time.sleep(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
94
archive/mictotxt.py
Normal file
94
archive/mictotxt.py
Normal file
@@ -0,0 +1,94 @@
|
||||
import os, time, logging, requests
|
||||
import speech_recognition as sr
|
||||
from ctypes import *
|
||||
|
||||
# ALSA 에러 억제
|
||||
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
|
||||
def py_error_handler(filename, line, function, err, fmt): pass
|
||||
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
|
||||
c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
|
||||
try:
|
||||
asound = cdll.LoadLibrary('libasound.so.2')
|
||||
asound.snd_lib_error_set_handler(c_error_handler)
|
||||
except: pass
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ===== 설정 =====
|
||||
OPENWEBUI_URL = "http://192.168.0.149:18789"
|
||||
API_KEY = "your-openwebui-api-key" # OpenWebUI → 설정 → API Key
|
||||
MODEL = "gemini-2.0-flash" # 오픈클로에서 쓰는 모델명
|
||||
WAKE_WORD = "미나이"
|
||||
MIC_INDEX = 3
|
||||
# ================
|
||||
|
||||
def stt_google(audio_data):
|
||||
"""Google STT로 음성 → 텍스트"""
|
||||
recognizer = sr.Recognizer()
|
||||
try:
|
||||
return recognizer.recognize_google(audio_data, language='ko-KR')
|
||||
except:
|
||||
return None
|
||||
|
||||
def send_to_openwebui(text):
|
||||
"""텍스트를 OpenWebUI 채팅 API로 전송"""
|
||||
try:
|
||||
headers = {
|
||||
"Authorization": f"Bearer {API_KEY}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
payload = {
|
||||
"model": MODEL,
|
||||
"messages": [{"role": "user", "content": text}]
|
||||
}
|
||||
# OpenAI 호환 엔드포인트 사용
|
||||
res = requests.post(
|
||||
f"{OPENWEBUI_URL}/api/chat/completions",
|
||||
headers=headers,
|
||||
json=payload,
|
||||
timeout=30
|
||||
)
|
||||
response_text = res.json()["choices"][0]["message"]["content"]
|
||||
logger.info(f"AI 응답: {response_text}")
|
||||
return response_text
|
||||
except Exception as e:
|
||||
logger.error(f"전송 실패: {e}")
|
||||
return None
|
||||
|
||||
def main():
|
||||
recognizer = sr.Recognizer()
|
||||
recognizer.energy_threshold = 400
|
||||
recognizer.pause_threshold = 1.0
|
||||
mic = sr.Microphone(device_index=MIC_INDEX)
|
||||
|
||||
logger.info(f"==== '{WAKE_WORD}' 대기 중 ====")
|
||||
|
||||
while True:
|
||||
try:
|
||||
with mic as source:
|
||||
audio = recognizer.listen(source, timeout=None, phrase_time_limit=3)
|
||||
|
||||
text = stt_google(audio)
|
||||
if not text:
|
||||
continue
|
||||
|
||||
if WAKE_WORD in text:
|
||||
logger.info(f"웨이크워드 감지: {text}")
|
||||
logger.info("명령 녹음 중...")
|
||||
|
||||
with mic as source:
|
||||
cmd_audio = recognizer.listen(source, timeout=5, phrase_time_limit=10)
|
||||
|
||||
cmd_text = stt_google(cmd_audio)
|
||||
if cmd_text:
|
||||
logger.info(f"명령: {cmd_text}")
|
||||
send_to_openwebui(cmd_text)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"루프 오류: {e}")
|
||||
time.sleep(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
14
archive/openai.py
Normal file
14
archive/openai.py
Normal file
@@ -0,0 +1,14 @@
|
||||
import openai
|
||||
|
||||
# API 키 설정 (환경 변수 등에서 가져옴)
|
||||
openai.api_key = "YOUR_API_KEY"
|
||||
|
||||
response = openai.ChatCompletion.create(
|
||||
model="gpt-3.5-turbo", # 또는 "gpt-4"
|
||||
messages=[
|
||||
{"role": "system", "content": "You are a helpful assistant."}, # 역할 설정
|
||||
{"role": "user", "content": "Hello!"} # 사용자 입력
|
||||
]
|
||||
)
|
||||
|
||||
print(response.choices[0].message.content)
|
||||
101
archive/smart_mic.py
Normal file
101
archive/smart_mic.py
Normal file
@@ -0,0 +1,101 @@
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
import requests
|
||||
import speech_recognition as sr
|
||||
import pygame
|
||||
from ctypes import *
|
||||
|
||||
# smart_mic.py 맨 위 import 아래 추가
|
||||
import os
|
||||
# 불필요한 로그 억제
|
||||
os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
|
||||
from ctypes import *
|
||||
|
||||
def py_error_handler(filename, line, function, err, fmt): pass
|
||||
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
|
||||
c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
|
||||
try:
|
||||
asound = cdll.LoadLibrary('libasound.so.2')
|
||||
asound.snd_lib_error_set_handler(c_error_handler)
|
||||
except: pass
|
||||
|
||||
|
||||
|
||||
# ALSA 에러 로그 숨기기
|
||||
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
|
||||
def py_error_handler(filename, line, function, err, fmt): pass
|
||||
c_error_handler = ERROR_HANDLER_FUNC(py_error_handler)
|
||||
try:
|
||||
asound = cdll.LoadLibrary('libasound.so.2')
|
||||
asound.snd_lib_error_set_handler(c_error_handler)
|
||||
except: pass
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# [환경 설정]
|
||||
# VM_IP 자리에 오픈클로 VM의 실제 IP를 넣으세요.
|
||||
OPENCLAW_API_URL = "http://192.168.0.149:18789/api/v1/chat"
|
||||
WAKE_WORD = "애미나이"
|
||||
MIC_INDEX = 3 # 현재 확인된 장치 번호
|
||||
|
||||
def send_to_openclaw(audio_data):
|
||||
"""녹음된 명령어를 오픈클로로 슛!"""
|
||||
try:
|
||||
# 파일 저장 확인을 위해 ls -al 에서 보일 이름을 고정합니다.
|
||||
filename = "cmd.wav"
|
||||
with open(filename, "wb") as f:
|
||||
f.write(audio_data.get_wav_data())
|
||||
|
||||
logger.info(f">>> {filename} 생성 완료! 오픈클로 전송 시작...")
|
||||
|
||||
with open(filename, "rb") as f:
|
||||
files = {'file': (filename, f, 'audio/wav')}
|
||||
response = requests.post(OPENCLAW_API_URL, files=files, timeout=20)
|
||||
|
||||
logger.info(f"오픈클로 서버 응답: {response.status_code}")
|
||||
except Exception as e:
|
||||
logger.error(f"전송 실패: {e}")
|
||||
|
||||
def main():
|
||||
recognizer = sr.Recognizer()
|
||||
recognizer.energy_threshold = 400
|
||||
recognizer.pause_threshold = 1.0 # 말이 끝날 때까지 조금 더 넉넉히 기다림
|
||||
|
||||
mic = sr.Microphone(device_index=MIC_INDEX)
|
||||
|
||||
logger.info(f"==== '{WAKE_WORD}' 대기 중 (포트 18789) ====")
|
||||
|
||||
while True:
|
||||
try:
|
||||
with mic as source:
|
||||
# 호출어 인식 단계
|
||||
audio = recognizer.listen(source, timeout=None, phrase_time_limit=3)
|
||||
|
||||
try:
|
||||
text = recognizer.recognize_google(audio, language='ko-KR')
|
||||
|
||||
# "에미나이", "애미나이" 모두 대응
|
||||
if "미나이" in text:
|
||||
logger.info(f"[{text}] 감지됨! 명령을 말씀하세요...")
|
||||
|
||||
# 스피커로 "네" 출력 (연결된 스피커가 있다면)
|
||||
# play_feedback("네")
|
||||
|
||||
with mic as source:
|
||||
logger.info("명령 녹음 중 (5초)...")
|
||||
# 호출어 감지 후 5초간 집중 녹음
|
||||
cmd_audio = recognizer.listen(source, timeout=5, phrase_time_limit=10)
|
||||
|
||||
# 녹음 끝나면 바로 전송 및 파일 생성
|
||||
send_to_openclaw(cmd_audio)
|
||||
|
||||
except sr.UnknownValueError:
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.error(f"루프 오류: {e}")
|
||||
time.sleep(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
310
archive/test_jarvis.py
Normal file
310
archive/test_jarvis.py
Normal file
@@ -0,0 +1,310 @@
|
||||
#!/usr/bin/env python3
|
||||
"""자비스 원격 테스트 — 볼륨 키우고 스피커/마이크/제미나이 확인.
|
||||
|
||||
사용:
|
||||
source ~/jarvis_env/bin/activate
|
||||
python ~/test_jarvis.py # 전체 테스트 (큰 소리 재생)
|
||||
python ~/test_jarvis.py --volume 100 # 볼륨 %
|
||||
python ~/test_jarvis.py --speak-only # 스피커만
|
||||
python ~/test_jarvis.py --mic-only # 마이크 레벨만
|
||||
python ~/test_jarvis.py --gemini-only
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from ctypes import CFUNCTYPE, c_char_p, c_int, cdll
|
||||
|
||||
os.environ["PYGAME_HIDE_SUPPORT_PROMPT"] = "hide"
|
||||
|
||||
ERROR_HANDLER_FUNC = CFUNCTYPE(None, c_char_p, c_int, c_char_p, c_int, c_char_p)
|
||||
|
||||
|
||||
def _py_error_handler(filename, line, function, err, fmt):
|
||||
pass
|
||||
|
||||
|
||||
try:
|
||||
asound = cdll.LoadLibrary("libasound.so.2")
|
||||
asound.snd_lib_error_set_handler(ERROR_HANDLER_FUNC(_py_error_handler))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def set_volume(percent: int) -> None:
|
||||
"""시스템 스피커 볼륨을 percent%로 맞춤 (여러 방식 시도)."""
|
||||
percent = max(0, min(150, int(percent)))
|
||||
print(f"\n[볼륨] {percent}% 로 설정 시도...")
|
||||
|
||||
cmds = [
|
||||
["wpctl", "set-volume", "@DEFAULT_AUDIO_SINK@", f"{percent / 100:.2f}"],
|
||||
["pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{percent}%"],
|
||||
["amixer", "-q", "sset", "Master", f"{min(percent, 100)}%"],
|
||||
["amixer", "-q", "sset", "PCM", f"{min(percent, 100)}%"],
|
||||
["amixer", "-c", "0", "-q", "sset", "Headphone", f"{min(percent, 100)}%"],
|
||||
]
|
||||
ok = False
|
||||
for cmd in cmds:
|
||||
try:
|
||||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
|
||||
if r.returncode == 0:
|
||||
print(f" OK: {' '.join(cmd)}")
|
||||
ok = True
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
|
||||
# 음소거 해제
|
||||
for cmd in (
|
||||
["wpctl", "set-mute", "@DEFAULT_AUDIO_SINK@", "0"],
|
||||
["pactl", "set-sink-mute", "@DEFAULT_SINK@", "0"],
|
||||
["amixer", "-q", "sset", "Master", "unmute"],
|
||||
):
|
||||
try:
|
||||
subprocess.run(cmd, capture_output=True, timeout=5)
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
|
||||
if not ok:
|
||||
print(" 경고: 볼륨 명령을 적용하지 못했습니다. 재생은 시도합니다.")
|
||||
|
||||
|
||||
def show_volume() -> None:
|
||||
for cmd in (
|
||||
["wpctl", "get-volume", "@DEFAULT_AUDIO_SINK@"],
|
||||
["pactl", "get-sink-volume", "@DEFAULT_SINK@"],
|
||||
["amixer", "sget", "Master"],
|
||||
):
|
||||
try:
|
||||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
|
||||
if r.returncode == 0 and r.stdout.strip():
|
||||
print(f" 현재: {r.stdout.strip().splitlines()[0]}")
|
||||
return
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
|
||||
|
||||
def list_devices() -> None:
|
||||
import pyaudio
|
||||
|
||||
pa = pyaudio.PyAudio()
|
||||
print("\n[오디오 장치]")
|
||||
for i in range(pa.get_device_count()):
|
||||
d = pa.get_device_info_by_index(i)
|
||||
if d["maxInputChannels"] > 0 or d["maxOutputChannels"] > 0:
|
||||
print(
|
||||
f" {i}: in={d['maxInputChannels']} out={d['maxOutputChannels']} "
|
||||
f"{d['name']}"
|
||||
)
|
||||
pa.terminate()
|
||||
|
||||
|
||||
def load_api_key() -> str:
|
||||
key = os.environ.get("GOOGLE_API_KEY", "")
|
||||
if key and "여기에_" not in key:
|
||||
return key
|
||||
# jarvis.py 기본값에서 읽기 (이미 키가 들어있는 경우)
|
||||
jarvis = os.path.join(os.path.dirname(os.path.abspath(__file__)), "jarvis.py")
|
||||
if os.path.isfile(jarvis):
|
||||
text = open(jarvis, encoding="utf-8").read()
|
||||
import re
|
||||
|
||||
m = re.search(
|
||||
r'GOOGLE_API_KEY = os\.environ\.get\(\s*"GOOGLE_API_KEY",\s*"([^"]+)"\s*\)',
|
||||
text,
|
||||
)
|
||||
if not m:
|
||||
m = re.search(
|
||||
r'os\.environ\.get\(\s*"GOOGLE_API_KEY",\s*\n\s*"([^"]+)"',
|
||||
text,
|
||||
)
|
||||
if m and "여기에_" not in m.group(1):
|
||||
return m.group(1)
|
||||
raise SystemExit("GOOGLE_API_KEY 를 찾을 수 없습니다. jarvis.py 에 키를 넣으세요.")
|
||||
|
||||
|
||||
def test_speak(volume_pygame: float = 1.0) -> None:
|
||||
from gtts import gTTS
|
||||
import pygame
|
||||
|
||||
print("\n[스피커] TTS 재생 — 카메라로 들리는지 확인해 보세요.")
|
||||
msg = (
|
||||
"안녕하세요. 자비스 스피커 테스트입니다. "
|
||||
"지금 소리가 크게 나고 있습니다. 하나, 둘, 셋."
|
||||
)
|
||||
path = "/tmp/jarvis_test_reply.mp3"
|
||||
gTTS(text=msg, lang="ko").save(path)
|
||||
|
||||
pygame.mixer.init()
|
||||
pygame.mixer.music.set_volume(max(0.0, min(1.0, volume_pygame)))
|
||||
pygame.mixer.music.load(path)
|
||||
pygame.mixer.music.play()
|
||||
while pygame.mixer.music.get_busy():
|
||||
time.sleep(0.1)
|
||||
pygame.mixer.music.unload()
|
||||
print(" 재생 완료.")
|
||||
|
||||
|
||||
def test_beep() -> None:
|
||||
"""네트워크 없이도 스피커 확인용 짧은 비프."""
|
||||
print("\n[스피커] 비프음 재생...")
|
||||
# sox/aplay 또는 speaker-test
|
||||
for cmd in (
|
||||
["speaker-test", "-t", "sine", "-f", "880", "-l", "1", "-c", "2"],
|
||||
[
|
||||
"bash",
|
||||
"-c",
|
||||
"ffmpeg -f lavfi -i 'sine=frequency=880:duration=1' -f wav - "
|
||||
"| aplay -q 2>/dev/null",
|
||||
],
|
||||
):
|
||||
try:
|
||||
r = subprocess.run(cmd, capture_output=True, timeout=8)
|
||||
if r.returncode == 0:
|
||||
print(" 비프 완료.")
|
||||
return
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
print(" 비프 도구 없음 — TTS만 사용합니다.")
|
||||
|
||||
|
||||
def test_mic(seconds: float = 3.0, mic_index: int | None = None) -> None:
|
||||
import numpy as np
|
||||
import pyaudio
|
||||
|
||||
print(f"\n[마이크] {seconds:.0f}초간 입력 레벨 측정 (밖에 있어도 OK)...")
|
||||
pa = pyaudio.PyAudio()
|
||||
|
||||
candidates: list[int | None] = []
|
||||
if mic_index is not None:
|
||||
candidates.append(mic_index)
|
||||
else:
|
||||
for i in range(pa.get_device_count()):
|
||||
d = pa.get_device_info_by_index(i)
|
||||
if d["maxInputChannels"] <= 0:
|
||||
continue
|
||||
name = d["name"]
|
||||
if "pulse" in name or name == "default":
|
||||
candidates.append(i)
|
||||
elif "USB" in name:
|
||||
candidates.append(i)
|
||||
candidates.append(None) # 시스템 기본
|
||||
|
||||
stream = None
|
||||
rate = 16000
|
||||
used_index: int | None = None
|
||||
for idx in candidates:
|
||||
for try_rate in (16000, 44100, 48000):
|
||||
kwargs = dict(
|
||||
format=pyaudio.paInt16,
|
||||
channels=1,
|
||||
rate=try_rate,
|
||||
input=True,
|
||||
frames_per_buffer=1024,
|
||||
)
|
||||
if idx is not None:
|
||||
kwargs["input_device_index"] = idx
|
||||
try:
|
||||
stream = pa.open(**kwargs)
|
||||
rate = try_rate
|
||||
used_index = idx
|
||||
name = (
|
||||
pa.get_device_info_by_index(idx)["name"]
|
||||
if idx is not None
|
||||
else "default"
|
||||
)
|
||||
print(f" 열림: index={idx} ({name}) rate={rate}")
|
||||
break
|
||||
except OSError:
|
||||
continue
|
||||
if stream is not None:
|
||||
break
|
||||
|
||||
if stream is None:
|
||||
pa.terminate()
|
||||
print(" → 마이크를 열 수 없습니다.")
|
||||
return
|
||||
|
||||
peaks = []
|
||||
n = max(1, int(rate / 1024 * seconds))
|
||||
for _ in range(n):
|
||||
data = stream.read(1024, exception_on_overflow=False)
|
||||
arr = np.frombuffer(data, dtype=np.int16).astype(np.float32)
|
||||
peaks.append(float(np.max(np.abs(arr))))
|
||||
stream.stop_stream()
|
||||
stream.close()
|
||||
pa.terminate()
|
||||
|
||||
peak = max(peaks) if peaks else 0
|
||||
avg = sum(peaks) / len(peaks) if peaks else 0
|
||||
print(f" peak={peak:.0f} / 32767 avg={avg:.0f} (device={used_index})")
|
||||
if peak < 200:
|
||||
print(" → 거의 무음. 마이크 연결·권한·MIC_INDEX 확인 필요.")
|
||||
elif peak < 2000:
|
||||
print(" → 입력은 있으나 작음. 마이크 게인 올려보세요.")
|
||||
else:
|
||||
print(" → 마이크 입력 정상으로 보입니다.")
|
||||
|
||||
|
||||
def test_gemini() -> None:
|
||||
import google.generativeai as genai
|
||||
|
||||
print("\n[제미나이] 짧은 텍스트 질의...")
|
||||
key = load_api_key()
|
||||
genai.configure(api_key=key)
|
||||
model_name = os.environ.get("GEMINI_MODEL", "gemini-2.0-flash")
|
||||
model = genai.GenerativeModel(model_name)
|
||||
try:
|
||||
r = model.generate_content("한 문장으로만 답해: 너는 누구야?")
|
||||
print(f" 모델={model_name}")
|
||||
print(f" 답: {(r.text or '').strip()}")
|
||||
except Exception as e:
|
||||
print(f" 실패: {e}")
|
||||
print(" → 모델 이름을 gemini-1.5-flash 등으로 바꿔보세요.")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
p = argparse.ArgumentParser(description="자비스 원격 테스트")
|
||||
p.add_argument("--volume", type=int, default=100, help="시스템 볼륨 %% (기본 100)")
|
||||
p.add_argument("--speak-only", action="store_true")
|
||||
p.add_argument("--mic-only", action="store_true")
|
||||
p.add_argument("--gemini-only", action="store_true")
|
||||
p.add_argument("--mic-index", type=int, default=None)
|
||||
p.add_argument("--no-beep", action="store_true")
|
||||
args = p.parse_args()
|
||||
|
||||
print("=== 자비스 원격 테스트 ===")
|
||||
set_volume(args.volume)
|
||||
show_volume()
|
||||
|
||||
only = args.speak_only or args.mic_only or args.gemini_only
|
||||
if not only:
|
||||
list_devices()
|
||||
|
||||
if args.mic_only:
|
||||
test_mic(mic_index=args.mic_index)
|
||||
return
|
||||
if args.gemini_only:
|
||||
test_gemini()
|
||||
return
|
||||
if args.speak_only:
|
||||
if not args.no_beep:
|
||||
test_beep()
|
||||
test_speak(volume_pygame=1.0)
|
||||
return
|
||||
|
||||
if not args.no_beep:
|
||||
test_beep()
|
||||
test_speak(volume_pygame=1.0)
|
||||
test_mic(mic_index=args.mic_index)
|
||||
test_gemini()
|
||||
print("\n=== 끝 ===")
|
||||
print("카메라로 TTS/비프가 들렸는지 확인해 주세요.")
|
||||
print("다시 크게: python ~/test_jarvis.py --speak-only --volume 100")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user