291 lines
12 KiB
Python
291 lines
12 KiB
Python
import signal
|
||
import sys
|
||
import time
|
||
import re
|
||
import subprocess
|
||
import queue
|
||
# 导入所有模块
|
||
from tts_module import BaiduOnlineTTS
|
||
from volume_module import VolumeController, detect_audio_control
|
||
from motion_module import RobotMotionController
|
||
from camera_module import CameraModule
|
||
from ark_api_module import ArkAPIController
|
||
from voice_recog_module import VoiceRecogController
|
||
|
||
# -------------------- 1. 基础配置（完全保留原代码1. 基础配置） --------------------
|
||
# 1.1 项目路径与运动模型
|
||
PROJECT_ROOT = "/home/duckpi/open_duck_mini_ws/OPEN_DUCK_MINI/Open_Duck_Mini_Runtime-2"
|
||
sys.path.append(PROJECT_ROOT)
|
||
ONNX_MODEL_PATH = "/home/duckpi/open_duck_mini_ws/OPEN_DUCK_MINI/Open_Duck_Mini-2/BEST_WALK_ONNX_2.onnx"
|
||
|
||
# 1.2 火山方舟API配置
|
||
ARK_API_KEY = "390d517c-129a-41c1-bf3d-458048007b69"
|
||
ARK_MODEL_ID = "doubao-seed-1-6-250615"
|
||
|
||
# 1.3 语音识别与唤醒词配置
|
||
APPID = "1ff50710"
|
||
ACCESS_KEY_ID = "a4f43e95ee0a9518d11befac8d31f1d4"
|
||
ACCESS_KEY_SECRET = "YzQ4NTRhZjc2ZTM4MDA1YjM2MmIyNDEy"
|
||
ACCESS_KEY = "e0EQQBoH0HIVU9KrXsmB7CMlVci+GAs2x0Ejtrdp8CTtZmf25rCLaQ=="
|
||
WAKEUP_WORD_PATH = "/home/duckpi/open_duck_mini_ws/OPEN_DUCK_MINI/resources/xiaohuangya_zh_raspberry-pi_v3_0_0.ppn"
|
||
MODEL_PATH = "/home/duckpi/open_duck_mini_ws/OPEN_DUCK_MINI/resources/porcupine_params_zh.pv"
|
||
|
||
# 1.4 百度在线TTS配置
|
||
BAIDU_TTS_API_KEY = "TnwYZPPvElNushOzfL6vBlUI"
|
||
BAIDU_TTS_SECRET_KEY = "55HeI8VNUMNlkW3t2QRwVtrjumpxjfxk"
|
||
|
||
# 1.5 语音反馈文本配置
|
||
FEEDBACK_TEXT = {
|
||
"wakeup": "你好呀，有什么吩咐",
|
||
"move_forward": "好的，我正在前进",
|
||
"move_backward": "好的，我正在后退",
|
||
"turn_left": "好的，我正在左转",
|
||
"turn_right": "好的，我正在右转",
|
||
"image_recog": "好的，我来识别一下",
|
||
"chat": "好的，我来想想",
|
||
"volume_increase": "音量已增大",
|
||
"volume_decrease": "音量已减小",
|
||
"volume_max": "已调至最大音量",
|
||
"volume_min": "已调至最小音量",
|
||
"unknown": "抱歉，没听懂，请再说一次",
|
||
"api_error": "抱歉，处理请求时出错了"
|
||
}
|
||
|
||
# 1.6 音频参数
|
||
VOLUME_STEP = 10
|
||
MIN_VOLUME = 0
|
||
MAX_VOLUME = 100
|
||
CURRENT_VOLUME = 40
|
||
AUDIO_CONTROL_NAME = None
|
||
|
||
# 1.7 麦克风与扬声器参数（模块内部已定义，此处保留用于一致性）
|
||
SAMPLE_RATE = 16000
|
||
CHANNELS = 1
|
||
SAMPLE_FORMAT = "int16"
|
||
AUDIO_ENCODE = "pcm_s16le"
|
||
LANG = "autodialect"
|
||
INTERACTION_TIMEOUT = 30
|
||
|
||
# -------------------- 2. 全局状态变量（完全保留原代码2. 全局状态变量，用列表传引用） --------------------
|
||
audio_q = queue.Queue()
|
||
last_audio_time = [time.time()] # 列表传引用，供模块修改
|
||
current_text = [""] # 列表传引用，供模块修改
|
||
final_result = [""] # 列表传引用，供模块修改
|
||
is_processing = [False] # 列表传引用，供模块修改
|
||
last_command_time = [time.time()]# 列表传引用，供模块修改
|
||
feedback_playing = False # TTS模块使用的全局变量
|
||
|
||
|
||
# -------------------- 8. 指令解析与执行（完全保留原代码8. 指令解析与执行） --------------------
|
||
def parse_voice_command(command_text: str):
|
||
command_text = command_text.strip().lower()
|
||
if not command_text:
|
||
return ("unknown", {})
|
||
|
||
# 运动指令
|
||
motion_rules = [
|
||
{"keywords": ["前进", "往前走", "向前走"], "action": "move_forward"},
|
||
{"keywords": ["后退", "往后走", "向后退"], "action": "move_backward"},
|
||
{"keywords": ["左转", "向左转", "往左转"], "action": "turn_left"},
|
||
{"keywords": ["右转", "向右转", "往右转"], "action": "turn_right"},
|
||
]
|
||
for rule in motion_rules:
|
||
if any(keyword in command_text for keyword in rule["keywords"]):
|
||
number_match = re.search(r"(\d{1,2})", command_text)
|
||
seconds = int(number_match.group(1)) if number_match else 2
|
||
return ("motion", {"action": rule["action"], "seconds": seconds})
|
||
|
||
# 图像识别指令
|
||
image_keywords = ["是什么", "这是什么", "识别", "看这个", "这东西", "这物体", "辨认"]
|
||
if any(keyword in command_text for keyword in image_keywords):
|
||
prompt = f"请简洁描述图片中的物体，1-2句话说明：{command_text}"
|
||
return ("image_recog", {"prompt": prompt})
|
||
|
||
# 闲聊指令
|
||
chat_keywords = [
|
||
"什么", "怎么", "为什么", "哪里", "多少", "如何", "吗", "呢", "吧",
|
||
"你好", "哈喽", "嗨", "今天", "天气", "时间", "故事", "笑话", "知识"
|
||
]
|
||
exclude_keywords = ["前进", "后退", "左转", "右转", "识别", "音量", "增大", "减小"]
|
||
if len(command_text) >= 2 and any(k in command_text for k in chat_keywords) and not any(k in command_text for k in exclude_keywords):
|
||
return ("chat", {"prompt": command_text})
|
||
|
||
# 音量控制指令
|
||
if any(keyword in command_text for keyword in ["增大音量", "声音大一点", "调大音量"]):
|
||
return ("volume", {"action": "increase"})
|
||
elif any(keyword in command_text for keyword in ["减小音量", "声音小一点", "调小音量"]):
|
||
return ("volume", {"action": "decrease"})
|
||
elif any(keyword in command_text for keyword in ["最大音量", "声音最大"]):
|
||
return ("volume", {"action": "max"})
|
||
elif any(keyword in command_text for keyword in ["最小音量", "声音最小", "静音"]):
|
||
return ("volume", {"action": "min"})
|
||
|
||
# 未知指令
|
||
return ("unknown", {})
|
||
|
||
def execute_command(command_type: str, params: dict, motion_controller, ark_api_controller, volume_controller):
|
||
global is_processing, feedback_playing
|
||
if is_processing[0]:
|
||
tts_controller.speak(FEEDBACK_TEXT["unknown"])
|
||
print("⚠️ 已有指令处理中，请稍后再说")
|
||
return
|
||
is_processing[0] = True
|
||
|
||
try:
|
||
if command_type == "motion":
|
||
motion_controller.execute_motion(params["action"], params["seconds"])
|
||
|
||
elif command_type == "image_recog":
|
||
print(f"\n🔍 触发图像识别，正在拍摄...")
|
||
image_base64 = camera_module.capture_base64()
|
||
if not image_base64:
|
||
tts_controller.speak(FEEDBACK_TEXT["unknown"])
|
||
print("\n" + "="*50)
|
||
print("❌ 图像采集失败，无法识别")
|
||
print("="*50 + "\n")
|
||
return
|
||
ark_api_controller.call_ark_api("image_recog", {"image_base64": image_base64, "prompt": params["prompt"]})
|
||
|
||
elif command_type == "chat":
|
||
print(f"\n💬 触发闲聊，正在思考...")
|
||
ark_api_controller.call_ark_api("chat", {"prompt": params["prompt"]})
|
||
|
||
elif command_type == "volume":
|
||
volume_action = params["action"]
|
||
if volume_action == "increase":
|
||
success = volume_controller.adjust_volume(is_increase=True)
|
||
if success:
|
||
tts_controller.speak(FEEDBACK_TEXT["volume_increase"])
|
||
elif volume_action == "decrease":
|
||
success = volume_controller.adjust_volume(is_increase=False)
|
||
if success:
|
||
tts_controller.speak(FEEDBACK_TEXT["volume_decrease"])
|
||
elif volume_action == "max":
|
||
success = volume_controller.set_system_volume(MAX_VOLUME)
|
||
if success:
|
||
tts_controller.speak(FEEDBACK_TEXT["volume_max"])
|
||
elif volume_action == "min":
|
||
success = volume_controller.set_system_volume(MIN_VOLUME)
|
||
if success:
|
||
tts_controller.speak(FEEDBACK_TEXT["volume_min"])
|
||
|
||
elif command_type == "unknown":
|
||
tts_controller.speak(FEEDBACK_TEXT["unknown"])
|
||
print("\n" + "="*50)
|
||
print(f"❌ 未识别到有效指令，支持：")
|
||
print(f" - 运动：前进3秒、左转2秒 | - 图像识别：这是什么")
|
||
print(f" - 闲聊：今天天气怎么样 | - 音量：增大音量、减小音量")
|
||
print("="*50 + "\n")
|
||
|
||
finally:
|
||
is_processing[0] = False
|
||
|
||
|
||
# -------------------- 11. 主循环（完全保留原代码11. 主循环逻辑） --------------------
|
||
def main():
|
||
global tts_controller, camera_module, AUDIO_CONTROL_NAME, feedback_playing
|
||
|
||
# 初始化各模块（按原代码顺序）
|
||
# 1. 初始化TTS
|
||
try:
|
||
tts_controller = BaiduOnlineTTS(BAIDU_TTS_API_KEY, BAIDU_TTS_SECRET_KEY)
|
||
except Exception as e:
|
||
print(f"❌ TTS初始化失败: {str(e)}")
|
||
sys.exit(1)
|
||
|
||
# 2. 初始化音量控制
|
||
AUDIO_CONTROL_NAME = detect_audio_control()
|
||
volume_controller = VolumeController(
|
||
audio_control_name=AUDIO_CONTROL_NAME,
|
||
current_volume=CURRENT_VOLUME,
|
||
volume_step=VOLUME_STEP,
|
||
min_volume=MIN_VOLUME,
|
||
max_volume=MAX_VOLUME
|
||
)
|
||
|
||
# 3. 初始化运动控制
|
||
motion_controller = RobotMotionController(
|
||
onnx_model_path=ONNX_MODEL_PATH,
|
||
tts_controller=tts_controller,
|
||
feedback_text=FEEDBACK_TEXT
|
||
)
|
||
|
||
# 4. 初始化摄像头
|
||
camera_module = CameraModule()
|
||
|
||
# 5. 初始化API控制器
|
||
ark_api_controller = ArkAPIController(
|
||
ark_api_key=ARK_API_KEY,
|
||
ark_model_id=ARK_MODEL_ID,
|
||
tts_controller=tts_controller,
|
||
feedback_text=FEEDBACK_TEXT
|
||
)
|
||
|
||
# 6. 初始化语音识别
|
||
voice_recog_controller = VoiceRecogController(
|
||
access_key=ACCESS_KEY,
|
||
wakeup_word_path=WAKEUP_WORD_PATH,
|
||
model_path=MODEL_PATH,
|
||
appid=APPID,
|
||
access_key_id=ACCESS_KEY_ID,
|
||
access_key_secret=ACCESS_KEY_SECRET,
|
||
tts_controller=tts_controller,
|
||
feedback_text=FEEDBACK_TEXT
|
||
)
|
||
|
||
# 中断处理（完全保留原逻辑）
|
||
def handle_interrupt(signum, frame):
|
||
print("\n🛑 收到退出信号，正在清理资源...")
|
||
# 停止机器人运动
|
||
if 'motion_controller' in globals() and hasattr(motion_controller, 'rl_walk'):
|
||
motion_controller.rl_walk.last_commands = [0.0, 0.0, 0.0]
|
||
# 停止TTS播放
|
||
global feedback_playing
|
||
feedback_playing = False
|
||
# 停止摄像头与麦克风
|
||
if 'camera_module' in globals() and camera_module.camera:
|
||
camera_module.camera.stop()
|
||
if hasattr(voice_recog_controller, 'stream') and voice_recog_controller.stream and voice_recog_controller.stream.active:
|
||
voice_recog_controller.stream.stop()
|
||
# 关闭TTS资源
|
||
tts_controller.close()
|
||
print("✅ 所有资源清理完成，程序退出")
|
||
sys.exit(0)
|
||
|
||
signal.signal(signal.SIGINT, handle_interrupt)
|
||
|
||
# 强制测试一次语音输出（原逻辑）
|
||
print("\n🔍 正在测试语音输出...")
|
||
tts_controller.speak("系统初始化完成,等待语音唤醒")
|
||
|
||
# 主循环（原逻辑）
|
||
while True:
|
||
if voice_recog_controller.wakeup_listener():
|
||
# 定义指令执行回调函数（关键修复）
|
||
def execute_callback(command_text):
|
||
command_type, params = parse_voice_command(command_text)
|
||
execute_command(command_type, params, motion_controller, ark_api_controller, volume_controller)
|
||
|
||
# 启动WebSocket时传入回调函数
|
||
voice_recog_controller.start_websocket(
|
||
current_text=current_text,
|
||
final_result=final_result,
|
||
last_audio_time=last_audio_time,
|
||
is_processing=is_processing,
|
||
last_command_time=last_command_time,
|
||
execute_callback=execute_callback # 传入回调
|
||
)
|
||
# 重置状态
|
||
last_audio_time[0] = time.time()
|
||
last_command_time[0] = time.time()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# # 确保ffmpeg已安装（原逻辑）
|
||
# try:
|
||
# subprocess.run(["ffmpeg", "--version"], capture_output=True, check=True)
|
||
# except:
|
||
# print("⚠️ 未检测到ffmpeg，正在尝试安装...")
|
||
# subprocess.run(["sudo", "apt-get", "install", "-y", "ffmpeg"], check=True)
|
||
|
||
main() |