diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index f08a61ff..0fa8b715 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/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: diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 51782bb9..5da1fa53 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/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)) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index d350f78c..33f94cf6 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/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": diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py index 250d3a66..6ef9cc3d 100644 --- a/main/xiaozhi-server/core/providers/asr/aliyun.py +++ b/main/xiaozhi-server/core/providers/asr/aliyun.py @@ -158,20 +158,7 @@ class ASRProvider(ASRProviderBase): request += "&enable_inverse_text_normalization=true" 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文件""" diff --git a/main/xiaozhi-server/core/providers/asr/baidu.py b/main/xiaozhi-server/core/providers/asr/baidu.py index 94996a59..628c38be 100644 --- a/main/xiaozhi-server/core/providers/asr/baidu.py +++ b/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]]: diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 6d9c2c90..61c1b754 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/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 \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index ce7e05ee..f9df3b3a 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/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): """ diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py index 3724ec2e..373d9c6a 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_local.py +++ b/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文件 diff --git a/main/xiaozhi-server/core/providers/asr/fun_server.py b/main/xiaozhi-server/core/providers/asr/fun_server.py index 501d43bd..2c11b8b3 100644 --- a/main/xiaozhi-server/core/providers/asr/fun_server.py +++ b/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. diff --git a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py index 3e720210..7e3fb6dd 100644 --- a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py +++ b/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: diff --git a/main/xiaozhi-server/core/providers/asr/tencent.py b/main/xiaozhi-server/core/providers/asr/tencent.py index dbdcbe26..bc7c5928 100644 --- a/main/xiaozhi-server/core/providers/asr/tencent.py +++ b/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: diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index e668459c..11960221 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/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 # 获取当前帧的二进制数据 @@ -89,12 +96,16 @@ class TTSProviderBase(ABC): # 如果最后一帧不足，补零 if len(chunk) < frame_size * 2: chunk += b"\x00" * (frame_size * 2 - len(chunk)) + + if is_opus: + # 转换为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) - # 转换为numpy数组处理 - np_frame = np.frombuffer(chunk, dtype=np.int16) - # 编码Opus数据 - opus_data = encoder.encode(np_frame.tobytes(), frame_size) - opus_datas.append(opus_data) + datas.append(frame_data) - return opus_datas, duration + return datas, duration