PCM音频模式

main/xiaozhi-server/core/connection.py

@@ -131,6 +131,8 @@ class ConnectionHandler:
             int(self.config.get("close_connection_no_voice_time", 120)) + 60
         )  # 在原来第一道关闭的基础上加60秒，进行二道关闭
 
+        self.audio_format = "opus"
+
     async def handle_connection(self, ws):
         try:
             # 获取并验证headers
@@ -867,7 +869,7 @@ class ConnectionHandler:
                 if future is None:
                     continue
                 text = None
-                opus_datas, text_index, tts_file = [], 0, None
+                audio_datas, text_index, tts_file = [], 0, None
                 try:
                     self.logger.bind(tag=TAG).debug("正在处理TTS任务...")
                     tts_timeout = int(self.config.get("tts_timeout", 10))
@@ -885,9 +887,12 @@ class ConnectionHandler:
                             f"TTS生成：文件路径: {tts_file}"
                         )
                         if os.path.exists(tts_file):
-                            opus_datas, _ = self.tts.audio_to_opus_data(tts_file)
+                            if self.audio_format == "pcm":
+                                audio_datas, _ = self.tts.audio_to_pcm_data(tts_file)
+                            else:
+                                audio_datas, _ = self.tts.audio_to_opus_data(tts_file)
                             # 在这里上报TTS数据（使用文件路径）
-                            enqueue_tts_report(self, 2, text, opus_datas)
+                            enqueue_tts_report(self, 2, text, audio_datas)
                         else:
                             self.logger.bind(tag=TAG).error(
                                 f"TTS出错：文件不存在{tts_file}"
@@ -898,7 +903,7 @@ class ConnectionHandler:
                     self.logger.bind(tag=TAG).error(f"TTS出错: {e}")
                 if not self.client_abort:
                     # 如果没有中途打断就发送语音
-                    self.audio_play_queue.put((opus_datas, text, text_index))
+                    self.audio_play_queue.put((audio_datas, text, text_index))
                 if (
                     self.tts.delete_audio_file
                     and tts_file is not None
@@ -929,13 +934,13 @@ class ConnectionHandler:
             text = None
             try:
                 try:
-                    opus_datas, text, text_index = self.audio_play_queue.get(timeout=1)
+                    audio_datas, text, text_index = self.audio_play_queue.get(timeout=1)
                 except queue.Empty:
                     if self.stop_event.is_set():
                         break
                     continue
                 future = asyncio.run_coroutine_threadsafe(
-                    sendAudioMessage(self, opus_datas, text, text_index), self.loop
+                    sendAudioMessage(self, audio_datas, text, text_index), self.loop
                 )
                 future.result()
             except Exception as e:

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

@@ -21,7 +21,15 @@ WAKEUP_CONFIG = {
 }
 
 
-async def handleHelloMessage(conn):
+async def handleHelloMessage(conn, msg_json):
+    """处理hello消息"""
+    audio_params = msg_json.get("audio_params")
+    if audio_params:
+        format = audio_params.get("format")
+        logger.bind(tag=TAG).info(f"客户端音频格式: {format}")
+        conn.audio_format = format
+        conn.welcome_msg['audio_params'] = audio_params
+
     await conn.websocket.send(json.dumps(conn.welcome_msg))
 
 

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

@@ -22,7 +22,7 @@ async def handleTextMessage(conn, message):
             await conn.websocket.send(message)
             return
         if msg_json["type"] == "hello":
-            await handleHelloMessage(conn)
+            await handleHelloMessage(conn, msg_json)
         elif msg_json["type"] == "abort":
             await handleAbortMessage(conn)
         elif msg_json["type"] == "listen":

main/xiaozhi-server/core/providers/asr/aliyun.py

@@ -159,19 +159,6 @@ class ASRProvider(ASRProviderBase):
         request += "&enable_voice_detection=false"
         return request
     
-    def decode_opus(self, opus_data: List[bytes], session_id: str) -> List[bytes]:
-        """将Opus数据解码为PCM"""
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
 
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""

main/xiaozhi-server/core/providers/asr/baidu.py

@@ -49,22 +49,6 @@ class ASRProvider(ASRProviderBase):
 
         return file_path
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes]) -> bytes:
-        """将Opus音频数据解码为PCM数据"""
-
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
     async def speech_to_text(
         self, opus_data: List[bytes], session_id: str
     ) -> Tuple[Optional[str], Optional[str]]:

main/xiaozhi-server/core/providers/asr/base.py

@@ -1,6 +1,6 @@
 from abc import ABC, abstractmethod
 from typing import Optional, Tuple, List
-
+import opuslib_next
 from config.logger import setup_logging
 
 TAG = __name__
@@ -17,3 +17,23 @@ class ASRProviderBase(ABC):
     async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         pass
+    
+    def set_audio_format(self, format: str) -> None:
+        """设置音频格式"""
+        self.audio_format = format
+
+    @staticmethod
+    def decode_opus(opus_data: List[bytes]) -> bytes:
+        """将Opus音频数据解码为PCM数据"""
+
+        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
+        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:
+                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
+
+        return pcm_data

main/xiaozhi-server/core/providers/asr/doubao.py

@@ -226,21 +226,6 @@ class ASRProvider(ASRProviderBase):
             logger.bind(tag=TAG).error(f"ASR request failed: {e}", exc_info=True)
             return None
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes], session_id: str) -> List[bytes]:
-
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
     @staticmethod
     def slice_data(data: bytes, chunk_size: int) -> (list, bool):
         """

main/xiaozhi-server/core/providers/asr/fun_local.py

@@ -6,9 +6,7 @@ import io
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
 import uuid
-import opuslib_next
 from core.providers.asr.base import ASRProviderBase
-
 from funasr import AutoModel
 from funasr.utils.postprocess_utils import rich_transcription_postprocess
 
@@ -64,27 +62,16 @@ class ASRProvider(ASRProviderBase):
 
         return file_path
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes], session_id: str) -> List[bytes]:
-
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
     async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
         """语音转文本主处理逻辑"""
         file_path = None
         try:
             # 合并所有opus数据包
-            pcm_data = self.decode_opus(opus_data, session_id)
+            if self.audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data, session_id)
+                
             combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件

main/xiaozhi-server/core/providers/asr/fun_server.py

@@ -44,23 +44,6 @@ class ASRProvider(ASRProviderBase):
 
         return file_path
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes]) -> bytes:
-        """将Opus音频数据解码为PCM数据"""
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
-
-
     async def _receive_responses(self, ws) -> None:
         '''
         Asynchronous generator to receive messages from the WebSocket.

main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py

@@ -97,21 +97,6 @@ class ASRProvider(ASRProviderBase):
 
         return file_path
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes], session_id: str) -> List[bytes]:
-
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
     def read_wave(self, wave_filename: str) -> Tuple[np.ndarray, int]:
         """
         Args:

main/xiaozhi-server/core/providers/asr/tencent.py

@@ -45,22 +45,6 @@ class ASRProvider(ASRProviderBase):
 
         return file_path
 
-    @staticmethod
-    def decode_opus(opus_data: List[bytes]) -> bytes:
-        """将Opus音频数据解码为PCM数据"""
-
-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        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:
-                logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
-
-        return pcm_data
-
     async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if not opus_data:

main/xiaozhi-server/core/providers/tts/base.py

@@ -53,8 +53,15 @@ class TTSProviderBase(ABC):
     async def text_to_speak(self, text, output_file):
         pass
 
+    def audio_to_pcm_data(self, audio_file_path):
+        """音频文件转换为PCM编码"""
+        return self.audio_to_data(audio_file_path, is_opus=False)
+    
     def audio_to_opus_data(self, audio_file_path):
         """音频文件转换为Opus编码"""
+        return self.audio_to_data(audio_file_path, is_opus=True)
+    
+    def audio_to_data(self, audio_file_path, is_opus=True):
         # 获取文件后缀名
         file_type = os.path.splitext(audio_file_path)[1]
         if file_type:
@@ -80,7 +87,7 @@ class TTSProviderBase(ABC):
         frame_duration = 60  # 60ms per frame
         frame_size = int(16000 * frame_duration / 1000)  # 960 samples/frame
 
-        opus_datas = []
+        datas = []
         # 按帧处理所有音频数据（包括最后一帧可能补零）
         for i in range(0, len(raw_data), frame_size * 2):  # 16bit=2bytes/sample
             # 获取当前帧的二进制数据
@@ -90,11 +97,15 @@ class TTSProviderBase(ABC):
             if len(chunk) < frame_size * 2:
                 chunk += b"\x00" * (frame_size * 2 - len(chunk))
             
-            # 转换为numpy数组处理
+            if is_opus:
-            np_frame = np.frombuffer(chunk, dtype=np.int16)
+                # 转换为numpy数组处理
+                np_frame = np.frombuffer(chunk, dtype=np.int16)
+                 # 编码Opus数据
+                frame_data = encoder.encode(np_frame.tobytes(), frame_size)
+            else: 
+                frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
 
-            # 编码Opus数据
-            opus_data = encoder.encode(np_frame.tobytes(), frame_size)
-            opus_datas.append(opus_data)
 
-        return opus_datas, duration
+            datas.append(frame_data)
+
+        return datas, duration