Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2139 from xinnan-tech/py_wakeup_audio

Browse Source 
Py wakeup audio
This commit is contained in:
欣南科技
2025-08-30 18:15:40 +08:00
committed by GitHub
parent 3e1765a26d 76a7ac04c9
commit b4f4995ff9
10 changed files with 159 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/config.yaml
+2
View File
@@ -59,6 +59,8 @@ log:
delete_audio: true
# 没有语音输入多久后断开连接(秒)，默认2分钟，即120秒
close_connection_no_voice_time: 120
# 开场是否回复唤醒词
enable_greeting: true
# 说完话是否开启提示音
enable_stop_tts_notify: false
# 说完话是否开启提示音，音效地址
main/xiaozhi-server/core/handle/helloHandle.py
+1 -2
View File
@@ -8,7 +8,6 @@ from core.providers.tools.device_mcp import (
TAG = __name__
async def handleHelloMessage(conn, msg_json):
"""处理hello消息"""
audio_params = msg_json.get("audio_params")
@@ -29,4 +28,4 @@ async def handleHelloMessage(conn, msg_json):
# 发送mcp消息，获取tools列表
asyncio.create_task(send_mcp_tools_list_request(conn))
await conn.websocket.send(json.dumps(conn.welcome_msg))
await conn.websocket.send(json.dumps(conn.welcome_msg))
main/xiaozhi-server/core/handle/intentHandler.py
+5 -5
View File
@@ -1,11 +1,11 @@
import json
import asyncio
import uuid
import asyncio
from core.utils.dialogue import Message
from core.providers.tts.dto.dto import ContentType
from plugins_func.register import Action, ActionResponse
from core.handle.sendAudioHandle import send_stt_message
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import ContentType
from core.utils.dialogue import Message
from plugins_func.register import Action, ActionResponse
from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
TAG = __name__
@@ -23,7 +23,7 @@ async def handle_user_intent(conn, text):
pass
# 检查是否有明确的退出命令
filtered_text = remove_punctuation_and_length(text)[1]
_, filtered_text = remove_punctuation_and_length(text)
if await check_direct_exit(conn, filtered_text):
return True
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+9 -23
View File
@@ -1,19 +1,21 @@
import time
import json
from core.handle.sendAudioHandle import send_stt_message
from core.handle.abortHandle import handleAbortMessage
from core.handle.intentHandler import handle_user_intent
from core.utils.output_counter import check_device_output_limit
from core.handle.abortHandle import handleAbortMessage
from core.handle.sendAudioHandle import SentenceType
from core.utils.util import audio_to_data_stream
from core.utils.util import play_audio_frames, play_audio_response
from core.handle.sendAudioHandle import send_stt_message, SentenceType
TAG = __name__
async def handleAudioMessage(conn, audio):
# 检查是否在唤醒处理锁定期内
if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
return
# 当前片段是否有人说话
have_voice = conn.vad.is_vad(conn, audio)
if have_voice:
if conn.client_is_speaking:
await handleAbortMessage(conn)
@@ -105,9 +107,7 @@ async def max_out_size(conn):
text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！"
await send_stt_message(conn, text)
file_path = "config/assets/max_output_size.wav"
conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text))
play_audio_frames(conn, file_path)
conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
play_audio_response(conn, {"text": text, "file_path": file_path})
conn.close_after_chat = True
@@ -142,18 +142,4 @@ async def check_bind_device(conn):
text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
await send_stt_message(conn, text)
music_path = "config/assets/bind_not_found.wav"
conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text))
play_audio_frames(conn, music_path)
conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
def play_audio_frames(conn, file_path):
"""播放音频文件并处理发送帧数据"""
def handle_audio_frame(frame_data):
conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None))
audio_to_data_stream(
file_path,
is_opus=True,
callback=handle_audio_frame
)
play_audio_response(conn, {"text": text, "file_path": music_path})
main/xiaozhi-server/core/handle/sendAudioHandle.py
+5 -11
View File
@@ -1,8 +1,8 @@
import json
import asyncio
import time
from core.providers.tts.dto.dto import SentenceType
import asyncio
from core.utils import textUtils
from core.providers.tts.dto.dto import SentenceType
TAG = __name__
@@ -38,7 +38,7 @@ async def sendAudio(conn, audios, pre_buffer=False):
opus_packet: 单个opus数据包
pre_buffer: 快速发送音频
"""
if audios is None:
if audios is None or len(audios) == 0:
return
if isinstance(audios, bytes):
@@ -50,12 +50,8 @@ async def sendAudio(conn, audios, pre_buffer=False):
await conn.websocket.send(audios)
return
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 流控逻辑：确保按60ms的帧时长间隔发送
frame_duration = 60 # 毫秒
# 获取或初始化流控状态
if not hasattr(conn, "audio_flow_control"):
conn.audio_flow_control = {
@@ -64,15 +60,13 @@ async def sendAudio(conn, audios, pre_buffer=False):
"start_time": time.perf_counter(),
}
frame_duration=60
flow_control = conn.audio_flow_control
current_time = time.perf_counter()
# 计算期望的发送时间
# 计算预期发送时间
expected_time = flow_control["start_time"] + (
flow_control["packet_count"] * frame_duration / 1000
)
# 流控延迟
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
main/xiaozhi-server/core/handle/textHandle.py
+18 -16
View File
@@ -1,13 +1,12 @@
import json
import time
import asyncio
from core.utils.util import filter_sensitive_info
from core.handle.abortHandle import handleAbortMessage
from core.handle.helloHandle import handleHelloMessage
from core.handle.receiveAudioHandle import handleAudioMessage
from core.providers.tools.device_mcp import handle_mcp_message
from core.utils.util import remove_punctuation_and_length, filter_sensitive_info
from core.handle.receiveAudioHandle import startToChat, handleAudioMessage
from core.handle.sendAudioHandle import send_stt_message, send_tts_message
from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus
from core.handle.reportHandle import enqueue_asr_report
import asyncio
TAG = __name__
@@ -45,18 +44,21 @@ async def handleTextMessage(conn, message):
conn.client_have_voice = False
conn.asr_audio.clear()
if "text" in msg_json:
conn.last_activity_time = time.time() * 1000
original_text = msg_json["text"] # 保留原始文本
filtered_len, filtered_text = remove_punctuation_and_length(
original_text
)
original_text = msg_json["text"] # 保留设备上传的文本
# 识别是否是唤醒词
is_wakeup_words = filtered_text in conn.config.get("wakeup_words")
if not is_wakeup_words:
# 上报纯文字数据（复用ASR上报功能，但不提供音频数据）
enqueue_asr_report(conn, original_text, [])
# 否则需要LLM对文字内容进行答复
await startToChat(conn, original_text)
is_wakeup_words = original_text in conn.config.get("wakeup_words")
# 是否开启唤醒词回复
enable_greeting = conn.config.get("enable_greeting", True)
if is_wakeup_words and not enable_greeting:
# 如果是唤醒词，且关闭了唤醒词回复，就不用回答
await send_stt_message(conn, original_text)
await send_tts_message(conn, "stop", None)
conn.client_is_speaking = False
else:
# 检测到唤醒词，开始等待后续进行声纹识别
conn.wakeup_mode = True
conn.logger.bind(tag=TAG).info(f"检测到唤醒词~")
elif msg_json["type"] == "iot":
conn.logger.bind(tag=TAG).info(f"收到iot消息：{message}")
if "descriptors" in msg_json:
main/xiaozhi-server/core/providers/asr/base.py
+88 -60
View File
@@ -1,14 +1,14 @@
import os
import io
import wave
import uuid
import json
import time
import queue
import asyncio
import traceback
import threading
import opuslib_next
import json
import io
import time
import concurrent.futures
from abc import ABC, abstractmethod
from config.logger import setup_logging
@@ -53,6 +53,10 @@ class ASRProviderBase(ABC):
# 接收音频
async def receive_audio(self, conn, audio, audio_have_voice):
# 检查是否在唤醒处理锁定期内
if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
return
if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
have_voice = audio_have_voice
else:
@@ -62,6 +66,14 @@ class ASRProviderBase(ABC):
if not have_voice and not conn.client_have_voice:
conn.asr_audio = conn.asr_audio[-10:]
return
# 检查是否处于唤醒模式
if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10:
asr_audio_task = conn.asr_audio.copy()
conn.reset_vad_states()
conn.asr_audio.clear()
await self.handle_voice_stop(conn, asr_audio_task)
if conn.client_voice_stop:
asr_audio_task = conn.asr_audio.copy()
@@ -87,32 +99,13 @@ class ASRProviderBase(ABC):
# 预先准备WAV数据
wav_data = None
# 使用连接的声纹识别提供者
if conn.voiceprint_provider and combined_pcm_data:
wav_data = self._pcm_to_wav(combined_pcm_data)
# 检查是否处于唤醒模式
wakeup_mode = getattr(conn, 'wakeup_mode', False)
# 定义ASR任务
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 定义声纹识别任务
# 声纹识别任务（公共逻辑）
def run_voiceprint():
if not wav_data:
return None
@@ -131,48 +124,83 @@ class ASRProviderBase(ABC):
logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
return None
# 使用线程池执行器并行运行
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
if wakeup_mode and conn.voiceprint_provider and wav_data:
conn.wakeup_mode = False
# 设置处理锁，防止后续音频片段重复处理
conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期
if conn.voiceprint_provider and wav_data:
# 唤醒模式：只执行声纹识别
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor:
voiceprint_future = thread_executor.submit(run_voiceprint)
# 等待两个线程都完成
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, file_path = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
# 记录识别结果
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
speaker_name = voiceprint_result
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
# 构建包含说话人信息的JSON字符串
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
fixed_text = "嘿，你好啊"
enhanced_text = self._build_enhanced_text(fixed_text, speaker_name)
# 使用自定义模块进行上报
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s")
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
else:
# 正常模式：执行声纹识别和ASR
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 使用线程池执行器并行运行
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
if conn.voiceprint_provider and wav_data:
voiceprint_future = thread_executor.submit(run_voiceprint)
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, _ = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
main/xiaozhi-server/core/providers/tts/base.py
+2 -2
View File
@@ -284,8 +284,8 @@ class TTSProviderBase(ABC):
enqueue_audio = []
enqueue_text = text
# 计算音频数据的帧数
if isinstance(audio_datas, bytes):
# 收集上报音频数据
if isinstance(audio_datas, bytes) and enqueue_audio is not None:
enqueue_audio.append(audio_datas)
# 发送音频
main/xiaozhi-server/core/providers/tts/linkerai.py
+1 -5
View File
@@ -1,8 +1,8 @@
import os
import queue
import aiohttp
import asyncio
import traceback
import aiohttp
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -109,10 +109,6 @@ class TTSProvider(TTSProviderBase):
finally:
return None
###################################################################################
# linkerai单流式TTS重写父类的方法--结束
###################################################################################
async def text_to_speak(self, text, is_last):
"""流式处理TTS音频，每句只推送一次音频列表"""
await self._tts_request(text, is_last)
main/xiaozhi-server/core/utils/util.py
+28 -10
View File
@@ -1,16 +1,17 @@
import json
import socket
import subprocess
import re
import os
from io import BytesIO
from typing import Callable, Any
from core.utils import p3
import numpy as np
import requests
import opuslib_next
from pydub import AudioSegment
import json
import copy
import socket
import requests
import subprocess
import numpy as np
import opuslib_next
from io import BytesIO
from core.utils import p3
from pydub import AudioSegment
from typing import Callable, Any
from core.providers.tts.dto.dto import SentenceType
TAG = __name__
emoji_map = {
@@ -273,6 +274,23 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] =
frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
callback(frame_data)
def play_audio_response(conn, response):
"""音频响应处理"""
conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], response.get("text")))
play_audio_frames(conn, response.get("file_path"))
conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
def play_audio_frames(conn, file_path):
"""播放音频文件并处理发送帧数据"""
def handle_audio_frame(frame_data):
conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None))
audio_to_data_stream(
file_path,
is_opus=True,
callback=handle_audio_frame
)
def check_vad_update(before_config, new_config):
if (