Merge branch 'main' into add_news

main/xiaozhi-server/config/logger.py

@@ -4,7 +4,7 @@ from loguru import logger
 from config.config_loader import load_config
 from config.settings import check_config_file
 
-SERVER_VERSION = "0.4.1"
+SERVER_VERSION = "0.4.2"
 
 
 def get_module_abbreviation(module_name, module_dict):

main/xiaozhi-server/core/connection.py

@@ -138,6 +138,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
@@ -241,9 +243,12 @@ class ConnectionHandler:
 
     def _initialize_components(self):
         """初始化组件"""
-        self.prompt = self.config["prompt"]
-        self.change_system_prompt(self.prompt)
-        self.logger.bind(tag=TAG).info(f"初始化组件: prompt成功 {self.prompt[:50]}...")
+        if self.config.get("prompt") is not None:
+            self.prompt = self.config["prompt"]
+            self.change_system_prompt(self.prompt)
+            self.logger.bind(tag=TAG).info(
+                f"初始化组件: prompt成功 {self.prompt[:50]}..."
+            )
 
         """初始化本地组件"""
         if self.vad is None:
@@ -811,9 +816,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}"
@@ -824,7 +832,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
@@ -855,13 +863,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

@@ -1,5 +1,4 @@
 import json
-from config.logger import setup_logging
 from core.handle.sendAudioHandle import send_stt_message
 from core.utils.util import remove_punctuation_and_length
 import shutil
@@ -20,7 +19,16 @@ 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")
+        conn.logger.bind(tag=TAG).info(f"客户端音频格式: {format}")
+        conn.audio_format = format
+        conn.asr.set_audio_format(format)
+        conn.welcome_msg["audio_params"] = audio_params
+
     await conn.websocket.send(json.dumps(conn.welcome_msg))
 
 

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

@@ -5,7 +5,7 @@ from core.handle.sendAudioHandle import send_stt_message
 from core.handle.intentHandler import handle_user_intent
 from core.utils.output_counter import check_device_output_limit
 from core.handle.ttsReportHandle import enqueue_tts_report
-from core.providers.tts.base import audio_to_opus_data
+from core.utils.util import audio_to_data
 
 TAG = __name__
 
@@ -111,7 +111,7 @@ async def max_out_size(conn):
     conn.tts_last_text_index = 0
     conn.llm_finish_task = True
     file_path = "config/assets/max_output_size.wav"
-    opus_packets, _ = audio_to_opus_data(file_path)
+    opus_packets, _ = audio_to_data(file_path)
     conn.audio_play_queue.put((opus_packets, text, 0))
     conn.close_after_chat = True
 
@@ -133,7 +133,7 @@ async def check_bind_device(conn):
 
         # 播放提示音
         music_path = "config/assets/bind_code.wav"
-        opus_packets, _ = audio_to_opus_data(music_path)
+        opus_packets, _ = audio_to_data(music_path)
         conn.audio_play_queue.put((opus_packets, text, 0))
 
         # 逐个播放数字
@@ -141,7 +141,7 @@ async def check_bind_device(conn):
             try:
                 digit = conn.bind_code[i]
                 num_path = f"config/assets/bind_code/{digit}.wav"
-                num_packets, _ = audio_to_opus_data(num_path)
+                num_packets, _ = audio_to_data(num_path)
                 conn.audio_play_queue.put((num_packets, None, i + 1))
             except Exception as e:
                 conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
@@ -153,5 +153,5 @@ async def check_bind_device(conn):
         conn.tts_last_text_index = 0
         conn.llm_finish_task = True
         music_path = "config/assets/bind_not_found.wav"
-        opus_packets, _ = audio_to_opus_data(music_path)
+        opus_packets, _ = audio_to_data(music_path)
         conn.audio_play_queue.put((opus_packets, text, 0))

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

@@ -115,7 +115,7 @@ async def send_tts_message(conn, state, text=None):
             stop_tts_notify_voice = conn.config.get(
                 "stop_tts_notify_voice", "config/assets/tts_notify.mp3"
             )
-            audios, duration = conn.tts.audio_to_opus_data(stop_tts_notify_voice)
+            audios, _ = conn.tts.audio_to_opus_data(stop_tts_notify_voice)
             await sendAudio(conn, audios)
         # 清除服务端讲话状态
         conn.clearSpeakStatus()

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

@@ -20,7 +20,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

@@ -20,64 +20,78 @@ from core.providers.asr.base import ASRProviderBase
 TAG = __name__
 logger = setup_logging()
 
+
 class AccessToken:
     @staticmethod
     def _encode_text(text):
         encoded_text = parse.quote_plus(text)
-        return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~')
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
 
     @staticmethod
     def _encode_dict(dic):
         keys = dic.keys()
         dic_sorted = [(key, dic[key]) for key in sorted(keys)]
         encoded_text = parse.urlencode(dic_sorted)
-        return encoded_text.replace('+', '%20').replace('*', '%2A').replace('%7E', '~')
+        return encoded_text.replace("+", "%20").replace("*", "%2A").replace("%7E", "~")
 
     @staticmethod
     def create_token(access_key_id, access_key_secret):
-        parameters = {'AccessKeyId': access_key_id,
-                      'Action': 'CreateToken',
-                      'Format': 'JSON',
-                      'RegionId': 'cn-shanghai',
-                      'SignatureMethod': 'HMAC-SHA1',
-                      'SignatureNonce': str(uuid.uuid1()),
-                      'SignatureVersion': '1.0',
-                      'Timestamp': time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-                      'Version': '2019-02-28'}
+        parameters = {
+            "AccessKeyId": access_key_id,
+            "Action": "CreateToken",
+            "Format": "JSON",
+            "RegionId": "cn-shanghai",
+            "SignatureMethod": "HMAC-SHA1",
+            "SignatureNonce": str(uuid.uuid1()),
+            "SignatureVersion": "1.0",
+            "Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            "Version": "2019-02-28",
+        }
         # 构造规范化的请求字符串
         query_string = AccessToken._encode_dict(parameters)
         # print('规范化的请求字符串: %s' % query_string)
         # 构造待签名字符串
-        string_to_sign = 'GET' + '&' + AccessToken._encode_text('/') + '&' + AccessToken._encode_text(query_string)
+        string_to_sign = (
+            "GET"
+            + "&"
+            + AccessToken._encode_text("/")
+            + "&"
+            + AccessToken._encode_text(query_string)
+        )
         # print('待签名的字符串: %s' % string_to_sign)
         # 计算签名
-        secreted_string = hmac.new(bytes(access_key_secret + '&', encoding='utf-8'),
-                                   bytes(string_to_sign, encoding='utf-8'),
-                                   hashlib.sha1).digest()
+        secreted_string = hmac.new(
+            bytes(access_key_secret + "&", encoding="utf-8"),
+            bytes(string_to_sign, encoding="utf-8"),
+            hashlib.sha1,
+        ).digest()
         signature = base64.b64encode(secreted_string)
         # print('签名: %s' % signature)
         # 进行URL编码
         signature = AccessToken._encode_text(signature)
         # print('URL编码后的签名: %s' % signature)
         # 调用服务
-        full_url = 'http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s' % (signature, query_string)
+        full_url = "http://nls-meta.cn-shanghai.aliyuncs.com/?Signature=%s&%s" % (
+            signature,
+            query_string,
+        )
         # print('url: %s' % full_url)
         # 提交HTTP GET请求
         response = requests.get(full_url)
         if response.ok:
             root_obj = response.json()
-            key = 'Token'
+            key = "Token"
             if key in root_obj:
-                token = root_obj[key]['Id']
-                expire_time = root_obj[key]['ExpireTime']
+                token = root_obj[key]["Id"]
+                expire_time = root_obj[key]["ExpireTime"]
                 return token, expire_time
         # print(response.text)
         return None, None
 
 
-
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
+        super().__init__()
         """阿里云ASR初始化"""
         # 新增空值判断逻辑
         self.access_key_id = config.get("access_key_id")
@@ -102,28 +116,23 @@ class ASRProvider(ASRProviderBase):
         # 确保输出目录存在
         os.makedirs(self.output_dir, exist_ok=True)
 
-
     def _refresh_token(self):
         """刷新Token并记录过期时间"""
         if self.access_key_id and self.access_key_secret:
             self.token, expire_time_str = AccessToken.create_token(
-                self.access_key_id,
-                self.access_key_secret
+                self.access_key_id, self.access_key_secret
             )
             if not expire_time_str:
                 raise ValueError("无法获取有效的Token过期时间")
 
             try:
-                #统一转换为字符串处理
+                # 统一转换为字符串处理
                 expire_str = str(expire_time_str).strip()
 
                 if expire_str.isdigit():
                     expire_time = datetime.fromtimestamp(int(expire_str))
                 else:
-                    expire_time = datetime.strptime(
-                        expire_str,
-                        "%Y-%m-%dT%H:%M:%SZ"
-                    )
+                    expire_time = datetime.strptime(expire_str, "%Y-%m-%dT%H:%M:%SZ")
                 self.expire_time = expire_time.timestamp() - 60
             except Exception as e:
                 raise ValueError(f"无效的过期时间格式: {expire_str}") from e
@@ -145,9 +154,12 @@ class ASRProvider(ASRProviderBase):
         #              f"过期时间 {datetime.fromtimestamp(self.expire_time)} | "
         #              f"剩余 {remaining:.2f}秒")
         return time.time() > self.expire_time
-    def generate_filename(self, extension=".wav"):
-        return os.path.join(self.output_file, f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}")
 
+    def generate_filename(self, extension=".wav"):
+        return os.path.join(
+            self.output_file,
+            f"tts-{__name__}{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
+        )
 
     def _construct_request_url(self) -> str:
         """构造请求URL，包含参数"""
@@ -159,20 +171,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文件"""
         module_name = __name__.split(".")[-1]
@@ -183,7 +181,7 @@ class ASRProvider(ASRProviderBase):
             wf.setnchannels(1)  # 单声道
             wf.setsampwidth(2)  # 16-bit
             wf.setframerate(self.sample_rate)
-            wf.writeframes(b''.join(pcm_data))
+            wf.writeframes(b"".join(pcm_data))
 
         logger.bind(tag=TAG).debug(f"音频文件已保存至: {file_path}")
         return file_path
@@ -193,9 +191,9 @@ class ASRProvider(ASRProviderBase):
         try:
             # 设置HTTP头
             headers = {
-                'X-NLS-Token': self.token,
-                'Content-type': 'application/octet-stream',
-                'Content-Length': str(len(pcm_data))
+                "X-NLS-Token": self.token,
+                "Content-type": "application/octet-stream",
+                "Content-Length": str(len(pcm_data)),
             }
 
             # 创建连接并发送请求
@@ -203,12 +201,12 @@ class ASRProvider(ASRProviderBase):
             request_url = self._construct_request_url()
 
             loop = asyncio.get_event_loop()
-            await loop.run_in_executor(None, lambda: conn.request(
-                method='POST',
-                url=request_url,
-                body=pcm_data,
-                headers=headers
-            ))
+            await loop.run_in_executor(
+                None,
+                lambda: conn.request(
+                    method="POST", url=request_url, body=pcm_data, headers=headers
+                ),
+            )
 
             # 获取响应
             response = await loop.run_in_executor(None, conn.getresponse)
@@ -218,10 +216,10 @@ class ASRProvider(ASRProviderBase):
             # 解析响应
             try:
                 body_json = json.loads(body)
-                status = body_json.get('status')
+                status = body_json.get("status")
 
                 if status == 20000000:
-                    result = body_json.get('result', '')
+                    result = body_json.get("result", "")
                     logger.bind(tag=TAG).debug(f"ASR结果: {result}")
                     return result
                 else:
@@ -236,7 +234,9 @@ class ASRProvider(ASRProviderBase):
             logger.bind(tag=TAG).error(f"ASR请求失败: {e}", exc_info=True)
             return None
 
-    async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
+    async def speech_to_text(
+        self, opus_data: List[bytes], session_id: str
+    ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if self._is_token_expired():
             logger.warning("Token已过期，正在自动刷新...")
@@ -245,8 +245,11 @@ class ASRProvider(ASRProviderBase):
         file_path = None
         try:
             # 解码Opus为PCM
-            pcm_data = self.decode_opus(opus_data, session_id)
-            combined_pcm_data = b''.join(pcm_data)
+            if self.audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data)
+            combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件
             if self.delete_audio_file:
@@ -264,4 +267,4 @@ class ASRProvider(ASRProviderBase):
 
         except Exception as e:
             logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-            return "", file_path
+            return "", file_path

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

@@ -20,6 +20,7 @@ logger = setup_logging()
 
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool = True):
+        super().__init__()
         self.app_id = config.get("app_id")
         self.api_key = config.get("api_key")
         self.secret_key = config.get("secret_key")
@@ -49,22 +50,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]]:
@@ -81,7 +66,10 @@ class ASRProvider(ASRProviderBase):
                 return None, file_path
 
             # 将Opus音频数据解码为PCM
-            pcm_data = self.decode_opus(opus_data)
+            if self.audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data)
             combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件

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__
@@ -8,12 +8,37 @@ logger = setup_logging()
 
 
 class ASRProviderBase(ABC):
+    def __init__(self):
+        self.audio_format = "opus"
+
     @abstractmethod
     def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
         """PCM数据保存为WAV文件"""
         pass
 
     @abstractmethod
-    async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
+    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

@@ -85,11 +85,12 @@ def parse_response(res):
 
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
+        super().__init__()
         self.appid = config.get("appid")
         self.cluster = config.get("cluster")
         self.access_token = config.get("access_token")
-        self.boosting_table_name = config.get("boosting_table_name")
-        self.correct_table_name = config.get("correct_table_name")
+        self.boosting_table_name = config.get("boosting_table_name", "")
+        self.correct_table_name = config.get("correct_table_name", "")
         self.output_dir = config.get("output_dir")
         self.delete_audio_file = delete_audio_file
 
@@ -226,21 +227,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):
         """
@@ -265,7 +251,10 @@ class ASRProvider(ASRProviderBase):
         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)
             combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件

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
 
@@ -35,6 +33,7 @@ class CaptureOutput:
 
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
+        super().__init__()
         self.model_dir = config.get("model_dir")
         self.output_dir = config.get("output_dir")  # 修正配置键名
         self.delete_audio_file = delete_audio_file
@@ -46,7 +45,7 @@ class ASRProvider(ASRProviderBase):
                 model=self.model_dir,
                 vad_kwargs={"max_single_segment_time": 30000},
                 disable_update=True,
-                hub="hf"
+                hub="hf",
                 # device="cuda:0",  # 启用GPU加速
             )
 
@@ -64,27 +63,18 @@ 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]]:
+    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)
+
             combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件
@@ -103,7 +93,9 @@ class ASRProvider(ASRProviderBase):
                 batch_size_s=60,
             )
             text = rich_transcription_postprocess(result[0]["text"])
-            logger.bind(tag=TAG).debug(f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}")
+            logger.bind(tag=TAG).debug(
+                f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
+            )
 
             return text, file_path
 

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

@@ -9,22 +9,29 @@ import wave
 import websockets
 from config.logger import setup_logging
 import asyncio
+
 TAG = __name__
 logger = setup_logging()
 
+
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
-        '''
+        """
         Initialize the ASRProvider with server configuration.
         :param config: Dictionary containing 'host', 'port', and 'is_ssl'.
         :param delete_audio_file: Boolean to indicate whether to delete audio files after processing.
-        '''
-        self.host = config.get('host', 'localhost')
-        self.port = config.get('port', 10095)
-        self.is_ssl = config.get('is_ssl', True)
+        """
+        super().__init__()
+        self.host = config.get("host", "localhost")
+        self.port = config.get("port", 10095)
+        self.is_ssl = config.get("is_ssl", True)
         self.output_dir = config.get("output_dir")
         self.delete_audio_file = delete_audio_file
-        self.uri = f"wss://{self.host}:{self.port}" if self.is_ssl else f"ws://{self.host}:{self.port}"
+        self.uri = (
+            f"wss://{self.host}:{self.port}"
+            if self.is_ssl
+            else f"ws://{self.host}:{self.port}"
+        )
         self.ssl_context = ssl.SSLContext() if self.is_ssl else None
         if self.ssl_context:
             self.ssl_context.check_hostname = False
@@ -44,28 +51,11 @@ 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.
         Yields each message as it is received.
-        '''
+        """
         text = ""
         while True:
             try:
@@ -78,30 +68,35 @@ class ASRProvider(ASRProviderBase):
                 else:
                     text += response_data.get("text", "")
             except asyncio.TimeoutError:
-                logger.bind(tag=TAG).error("Timeout while waiting for response from WebSocket.")
+                logger.bind(tag=TAG).error(
+                    "Timeout while waiting for response from WebSocket."
+                )
                 break
             except websockets.exceptions.ConnectionClosed as e:
                 logger.bind(tag=TAG).error(f"WebSocket connection closed: {e}")
                 break
         return text
+
     async def _send_data(self, ws, pcm_data: bytes, session_id: str) -> tuple:
-        '''
+        """
         Internal method to handle WebSocket communication.
         Reuses the persistent WebSocket connection if available.
         :param pcm_data: PCM audio data to send.
         :param session_id: Unique session identifier.
         :return: Tuple containing recognized text and optional timestamp.
-        '''
+        """
 
         # Send initial configuration message
-        config_message = json.dumps({
-            "mode": "offline",
-            "chunk_size": [5, 10, 5],
-            "chunk_interval": 10,
-            "wav_name": session_id,
-            "is_speaking": True,
-            "itn": False
-        })
+        config_message = json.dumps(
+            {
+                "mode": "offline",
+                "chunk_size": [5, 10, 5],
+                "chunk_interval": 10,
+                "wav_name": session_id,
+                "is_speaking": True,
+                "itn": False,
+            }
+        )
         await ws.send(config_message)
         logger.bind(tag=TAG).debug(f"Sent configuration message: {config_message}")
 
@@ -114,16 +109,20 @@ class ASRProvider(ASRProviderBase):
         await ws.send(end_message)
         logger.bind(tag=TAG).debug(f"Sent end message: {end_message}")
 
-
-    async def speech_to_text(self, opus_data: List[bytes], session_id: str) -> Tuple[Optional[str], Optional[str]]:
-        '''
+    async def speech_to_text(
+        self, opus_data: List[bytes], session_id: str
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """
         Convert speech data to text using FunASR.
         :param opus_data: List of Opus-encoded audio data chunks.
         :param session_id: Unique session identifier.
         :return: Tuple containing recognized text and optional timestamp.
-        '''
+        """
         file_path = None
-        pcm_data = self.decode_opus(opus_data)
+        if self.audio_format == "pcm":
+            pcm_data = opus_data
+        else:
+            pcm_data = self.decode_opus(opus_data)
         combined_pcm_data = b"".join(pcm_data)
 
         # 判断是否保存为WAV文件
@@ -132,16 +131,19 @@ class ASRProvider(ASRProviderBase):
         else:
             file_path = self.save_audio_to_file(pcm_data, session_id)
 
-        async with websockets.connect(self.uri, subprotocols=["binary"], ping_interval=None, ssl=self.ssl_context) as ws:
+        async with websockets.connect(
+            self.uri, subprotocols=["binary"], ping_interval=None, ssl=self.ssl_context
+        ) as ws:
             try:
                 # Use asyncio to handle WebSocket communication
-                send_task = asyncio.create_task(self._send_data(ws, combined_pcm_data, session_id))
+                send_task = asyncio.create_task(
+                    self._send_data(ws, combined_pcm_data, session_id)
+                )
                 receive_task = asyncio.create_task(self._receive_responses(ws))
 
                 # Gather tasks with error handling
                 done, pending = await asyncio.wait(
-                    [send_task, receive_task],
-                    return_when=asyncio.FIRST_EXCEPTION
+                    [send_task, receive_task], return_when=asyncio.FIRST_EXCEPTION
                 )
 
                 # Cancel any pending tasks
@@ -155,11 +157,16 @@ class ASRProvider(ASRProviderBase):
 
                 # Get the result from the receive task
                 result = receive_task.result()
-                return result, file_path  # Return the recognized text and timestamp (if any)
+                return (
+                    result,
+                    file_path,
+                )  # Return the recognized text and timestamp (if any)
 
             except websockets.exceptions.ConnectionClosed as e:
                 logger.bind(tag=TAG).error(f"WebSocket connection closed: {e}")
                 return "", file_path
             except Exception as e:
-                logger.bind(tag=TAG).error(f"Error during speech-to-text conversion: {e}", exc_info=True)
-                return "", file_path
+                logger.bind(tag=TAG).error(
+                    f"Error during speech-to-text conversion: {e}", exc_info=True
+                )
+                return "", file_path

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

@@ -37,6 +37,7 @@ class CaptureOutput:
 
 class ASRProvider(ASRProviderBase):
     def __init__(self, config: dict, delete_audio_file: bool):
+        super().__init__()
         self.model_dir = config.get("model_dir")
         self.output_dir = config.get("output_dir")
         self.delete_audio_file = delete_audio_file
@@ -97,21 +98,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:
@@ -144,7 +130,10 @@ class ASRProvider(ASRProviderBase):
         try:
             # 保存音频文件
             start_time = time.time()
-            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)
             file_path = self.save_audio_to_file(pcm_data, session_id)
             logger.bind(tag=TAG).debug(
                 f"音频文件保存耗时: {time.time() - start_time:.3f}s | 路径: {file_path}"

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

@@ -17,12 +17,14 @@ from config.logger import setup_logging
 TAG = __name__
 logger = setup_logging()
 
+
 class ASRProvider(ASRProviderBase):
     API_URL = "https://asr.tencentcloudapi.com"
     API_VERSION = "2019-06-14"
     FORMAT = "pcm"  # 支持的音频格式：pcm, wav, mp3
 
     def __init__(self, config: dict, delete_audio_file: bool = True):
+        super().__init__()
         self.secret_id = config.get("secret_id")
         self.secret_key = config.get("secret_key")
         self.output_dir = config.get("output_dir")
@@ -45,23 +47,9 @@ 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]]:
+    async def speech_to_text(
+        self, opus_data: List[bytes], session_id: str
+    ) -> Tuple[Optional[str], Optional[str]]:
         """将语音数据转换为文本"""
         if not opus_data:
             logger.bind(tag=TAG).warn("音频数据为空！")
@@ -75,7 +63,10 @@ class ASRProvider(ASRProviderBase):
                 return None, file_path
 
             # 将Opus音频数据解码为PCM
-            pcm_data = self.decode_opus(opus_data)
+            if self.audio_format == "pcm":
+                pcm_data = opus_data
+            else:
+                pcm_data = self.decode_opus(opus_data)
             combined_pcm_data = b"".join(pcm_data)
 
             # 判断是否保存为WAV文件
@@ -85,7 +76,7 @@ class ASRProvider(ASRProviderBase):
                 self.save_audio_to_file(pcm_data, session_id)
 
             # 将音频数据转换为Base64编码
-            base64_audio = base64.b64encode(combined_pcm_data).decode('utf-8')
+            base64_audio = base64.b64encode(combined_pcm_data).decode("utf-8")
 
             # 构建请求体
             request_body = self._build_request_body(base64_audio)
@@ -98,7 +89,9 @@ class ASRProvider(ASRProviderBase):
             result = self._send_request(request_body, timestamp, authorization)
 
             if result:
-                logger.bind(tag=TAG).debug(f"腾讯云语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}")
+                logger.bind(tag=TAG).debug(
+                    f"腾讯云语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}"
+                )
 
             return result, file_path
 
@@ -115,7 +108,7 @@ class ASRProvider(ASRProviderBase):
             "SourceType": 1,  # 音频数据来源为语音文件
             "VoiceFormat": self.FORMAT,  # 音频格式
             "Data": base64_audio,  # Base64编码的音频数据
-            "DataLen": len(base64_audio)  # 数据长度
+            "DataLen": len(base64_audio),  # 数据长度
         }
         return json.dumps(request_map)
 
@@ -148,9 +141,11 @@ class ASRProvider(ASRProviderBase):
             action = "SentenceRecognition"  # 接口名称
 
             # 构建规范头部信息，注意顺序和格式
-            canonical_headers = f"content-type:{content_type.lower()}\n" + \
-                               f"host:{host.lower()}\n" + \
-                               f"x-tc-action:{action.lower()}\n"
+            canonical_headers = (
+                f"content-type:{content_type.lower()}\n"
+                + f"host:{host.lower()}\n"
+                + f"x-tc-action:{action.lower()}\n"
+            )
 
             signed_headers = "content-type;host;x-tc-action"
 
@@ -158,21 +153,25 @@ class ASRProvider(ASRProviderBase):
             payload_hash = self._sha256_hex(request_body)
 
             # 构建规范请求字符串
-            canonical_request = f"{http_request_method}\n" + \
-                               f"{canonical_uri}\n" + \
-                               f"{canonical_query_string}\n" + \
-                               f"{canonical_headers}\n" + \
-                               f"{signed_headers}\n" + \
-                               f"{payload_hash}"
+            canonical_request = (
+                f"{http_request_method}\n"
+                + f"{canonical_uri}\n"
+                + f"{canonical_query_string}\n"
+                + f"{canonical_headers}\n"
+                + f"{signed_headers}\n"
+                + f"{payload_hash}"
+            )
 
             # 计算规范请求的哈希值
             hashed_canonical_request = self._sha256_hex(canonical_request)
 
             # 构建待签名字符串
-            string_to_sign = f"{algorithm}\n" + \
-                            f"{timestamp}\n" + \
-                            f"{credential_scope}\n" + \
-                            f"{hashed_canonical_request}"
+            string_to_sign = (
+                f"{algorithm}\n"
+                + f"{timestamp}\n"
+                + f"{credential_scope}\n"
+                + f"{hashed_canonical_request}"
+            )
 
             # 计算签名密钥
             secret_date = self._hmac_sha256(f"TC3{self.secret_key}", date)
@@ -180,13 +179,17 @@ class ASRProvider(ASRProviderBase):
             secret_signing = self._hmac_sha256(secret_service, "tc3_request")
 
             # 计算签名
-            signature = self._bytes_to_hex(self._hmac_sha256(secret_signing, string_to_sign))
+            signature = self._bytes_to_hex(
+                self._hmac_sha256(secret_signing, string_to_sign)
+            )
 
             # 构建授权头
-            authorization = f"{algorithm} " + \
-                           f"Credential={self.secret_id}/{credential_scope}, " + \
-                           f"SignedHeaders={signed_headers}, " + \
-                           f"Signature={signature}"
+            authorization = (
+                f"{algorithm} "
+                + f"Credential={self.secret_id}/{credential_scope}, "
+                + f"SignedHeaders={signed_headers}, "
+                + f"Signature={signature}"
+            )
 
             return timestamp, authorization
 
@@ -194,7 +197,9 @@ class ASRProvider(ASRProviderBase):
             logger.bind(tag=TAG).error(f"生成认证头失败: {e}", exc_info=True)
             raise RuntimeError(f"生成认证头失败: {e}")
 
-    def _send_request(self, request_body: str, timestamp: str, authorization: str) -> Optional[str]:
+    def _send_request(
+        self, request_body: str, timestamp: str, authorization: str
+    ) -> Optional[str]:
         """发送请求到腾讯云API"""
         headers = {
             "Content-Type": "application/json; charset=utf-8",
@@ -203,7 +208,7 @@ class ASRProvider(ASRProviderBase):
             "X-TC-Action": "SentenceRecognition",
             "X-TC-Version": self.API_VERSION,
             "X-TC-Timestamp": timestamp,
-            "X-TC-Region": "ap-shanghai"
+            "X-TC-Region": "ap-shanghai",
         }
 
         try:
@@ -234,16 +239,16 @@ class ASRProvider(ASRProviderBase):
 
     def _sha256_hex(self, data: str) -> str:
         """计算字符串的SHA256哈希值"""
-        digest = hashlib.sha256(data.encode('utf-8')).digest()
+        digest = hashlib.sha256(data.encode("utf-8")).digest()
         return self._bytes_to_hex(digest)
 
     def _hmac_sha256(self, key, data: str) -> bytes:
         """计算HMAC-SHA256"""
         if isinstance(key, str):
-            key = key.encode('utf-8')
+            key = key.encode("utf-8")
 
-        return hmac.new(key, data.encode('utf-8'), hashlib.sha256).digest()
+        return hmac.new(key, data.encode("utf-8"), hashlib.sha256).digest()
 
     def _bytes_to_hex(self, bytes_data: bytes) -> str:
         """字节数组转十六进制字符串"""
-        return ''.join(f"{b:02x}" for b in bytes_data)
+        return "".join(f"{b:02x}" for b in bytes_data)

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

@@ -3,7 +3,7 @@ from config.logger import setup_logging
 import os
 from abc import ABC, abstractmethod
 from core.utils.tts import MarkdownCleaner
-from core.utils.util import audio_to_opus_data
+from core.utils.util import audio_to_data
 
 TAG = __name__
 logger = setup_logging()
@@ -53,5 +53,10 @@ class TTSProviderBase(ABC):
     async def text_to_speak(self, text, output_file):
         pass
 
+    def audio_to_pcm_data(self, audio_file_path):
+        """音频文件转换为PCM编码"""
+        return audio_to_data(audio_file_path, is_opus=False)
+
     def audio_to_opus_data(self, audio_file_path):
-        return audio_to_opus_data(audio_file_path)
+        """音频文件转换为Opus编码"""
+        return audio_to_data(audio_file_path, is_opus=True)

main/xiaozhi-server/core/utils/util.py

@@ -862,8 +862,7 @@ def analyze_emotion(text):
     return top_emotions[0]  # 如果都不在优先级列表里，返回第一个
 
 
-def audio_to_opus_data(audio_file_path):
-    """音频文件转换为Opus编码"""
+def audio_to_data(audio_file_path, is_opus=True):
     # 获取文件后缀名
     file_type = os.path.splitext(audio_file_path)[1]
     if file_type:
@@ -889,7 +888,7 @@ def audio_to_opus_data(audio_file_path):
     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
         # 获取当前帧的二进制数据
@@ -899,14 +898,17 @@ def audio_to_opus_data(audio_file_path):
         if len(chunk) < frame_size * 2:
             chunk += b"\x00" * (frame_size * 2 - len(chunk))
 
-        # 转换为numpy数组处理
-        np_frame = np.frombuffer(chunk, dtype=np.int16)
+        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)
 
-        # 编码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
 
 
 def check_vad_update(before_config, new_config):