feat: 服务端AEC功能

ref: 入口解码为pcm音频数据，vad和asr可直接使用

main/xiaozhi-server/core/handle/helloHandle.py

@@ -56,6 +56,9 @@ async def handleHelloMessage(conn: "ConnectionHandler", msg_json):
             conn.mcp_client = MCPClient()
             # 发送初始化
             asyncio.create_task(send_mcp_initialize_message(conn))
+        if features.get("aec"):
+            conn.logger.bind(tag=TAG).debug("客户端启用了服务端AEC")
+            conn.client_aec = True
 
     await conn.websocket.send(json.dumps(conn.welcome_msg))
 

main/xiaozhi-server/core/handle/receiveAudioHandle.py

@@ -14,9 +14,9 @@ from core.handle.sendAudioHandle import send_stt_message, SentenceType
 TAG = __name__
 
 
-async def handleAudioMessage(conn: "ConnectionHandler", audio):
+async def handleAudioMessage(conn: "ConnectionHandler", pcm_frame):
     # 当前片段是否有人说话
-    have_voice = conn.vad.is_vad(conn, audio)
+    have_voice = conn.vad.is_vad(conn, pcm_frame)
     # 如果设备刚刚被唤醒，短暂忽略VAD检测
     if hasattr(conn, "just_woken_up") and conn.just_woken_up:
         have_voice = False
@@ -24,10 +24,14 @@ async def handleAudioMessage(conn: "ConnectionHandler", audio):
         if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
             conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
         return
+    # 服务端AEC功能需要实时触发打断
+    if conn.client_aec and have_voice:
+        if conn.client_is_speaking and conn.client_listen_mode != "manual":
+            await handleAbortMessage(conn)
     # 设备长时间空闲检测，用于say goodbye
     await no_voice_close_connect(conn, have_voice)
     # 接收音频
-    await conn.asr.receive_audio(conn, audio, have_voice)
+    await conn.asr.receive_audio(conn, pcm_frame, have_voice)
 
 
 async def resume_vad_detection(conn: "ConnectionHandler"):

main/xiaozhi-server/core/handle/reportHandle.py

@@ -50,33 +50,27 @@ async def report(conn: "ConnectionHandler", type, text, opus_data, report_time):
         conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}")
 
 
-def opus_to_wav(conn: "ConnectionHandler", opus_data):
-    """将Opus数据转换为WAV格式的字节流
+def opus_to_wav(conn: "ConnectionHandler", pcm_data):
+    """将PCM数据转换为WAV格式的字节流
 
     Args:
         output_dir: 输出目录（保留参数以保持接口兼容）
-        opus_data: opus音频数据
+        pcm_data: PCM音频数据（可能是列表或bytes）
 
     Returns:
         bytes: WAV格式的音频数据
     """
-    decoder = None
     try:
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        pcm_data = []
+        # 处理可能是列表或bytes的PCM数据
+        if isinstance(pcm_data, list):
+            pcm_data_bytes = b"".join(pcm_data)
+        else:
+            pcm_data_bytes = pcm_data
 
-        for opus_packet in opus_data:
-            try:
-                pcm_frame = decoder.decode(opus_packet, 960)  # 960 samples = 60ms
-                pcm_data.append(pcm_frame)
-            except opuslib_next.OpusError as e:
-                conn.logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        if not pcm_data:
+        if not pcm_data_bytes:
             raise ValueError("没有有效的PCM数据")
 
         # 创建WAV文件头
-        pcm_data_bytes = b"".join(pcm_data)
         num_samples = len(pcm_data_bytes) // 2  # 16-bit samples
 
         # WAV文件头
@@ -97,12 +91,9 @@ def opus_to_wav(conn: "ConnectionHandler", opus_data):
 
         # 返回完整的WAV数据
         return bytes(wav_header) + pcm_data_bytes
-    finally:
-        if decoder is not None:
-            try:
-                del decoder
-            except Exception as e:
-                conn.logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")
+    except Exception as e:
+        conn.logger.bind(tag=TAG).error(f"PCM转WAV失败: {e}", exc_info=True)
+        raise
 
 
 def enqueue_tts_report(conn: "ConnectionHandler", text, opus_data):

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -1,6 +1,7 @@
 import json
 import time
 import asyncio
+import opuslib_next
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -81,13 +82,24 @@ async def _send_to_mqtt_gateway(
     conn: "ConnectionHandler", opus_packet, timestamp, sequence
 ):
     """
-    发送带16字节头部的opus数据包给mqtt_gateway
+    发送带16字节头部的opus数据包给mqtt_gateway，同时缓存音频用于AEC处理
     Args:
         conn: 连接对象
         opus_packet: opus数据包
         timestamp: 时间戳
         sequence: 序列号
     """
+    # 如果启用了服务端AEC，缓存PCM数据用于后续AEC处理
+    if conn.client_aec and timestamp > 0:
+        if not hasattr(conn, "aec_audio_cache"):
+            conn.aec_audio_cache = {}
+            conn.aec_audio_cache_time = {}
+            conn._send_opus_decoder = opuslib_next.Decoder(16000, 1)
+        # 解码opus为PCM后缓存
+        pcm_data = conn._send_opus_decoder.decode(bytes(opus_packet), 960)
+        conn.aec_audio_cache[timestamp] = bytes(pcm_data)
+        conn.aec_audio_cache_time[timestamp] = time.time()
+
     # 为opus数据包添加16字节头部
     header = bytearray(16)
     header[0] = 1  # type