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

Merge branch 'py_test_error_response' into fix/2075

Browse Source
This commit is contained in:
Sakura-RanChen
2026-02-02 14:49:26 +08:00
committed by GitHub
parent f5b9995f89 ea9778f422
commit 6da32a60c0
61 changed files with 1057 additions and 986 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+20 -3
View File
@@ -87,6 +87,7 @@ class ConnectionHandler:
self.max_output_size = 0
self.chat_history_conf = 0
self.audio_format = "opus"
self.sample_rate = 24000 # 默认采样率，从客户端 hello 消息中动态更新
# 客户端状态相关
self.client_abort = False
@@ -208,6 +209,10 @@ class ConnectionHandler:
self.welcome_msg = self.config["xiaozhi"]
self.welcome_msg["session_id"] = self.session_id
# 从配置中读取采样率
self.sample_rate = self.welcome_msg["audio_params"]["sample_rate"]
self.logger.bind(tag=TAG).info(f"配置输出音频采样率为: {self.sample_rate}")
# 在后台初始化配置和组件（完全不阻塞主循环）
asyncio.create_task(self._background_initialize())
@@ -1184,6 +1189,8 @@ class ConnectionHandler:
if self.tts:
await self.tts.close()
if self.asr:
await self.asr.close()
# 最后关闭线程池（避免阻塞）
if self.executor:
@@ -1232,11 +1239,21 @@ class ConnectionHandler:
f"清理结束: TTS队列大小={self.tts.tts_text_queue.qsize()}, 音频队列大小={self.tts.tts_audio_queue.qsize()}"
)
def reset_vad_states(self):
self.client_audio_buffer = bytearray()
def reset_audio_states(self):
"""
重置所有音频相关状态(VAD + ASR)
"""
# Reset VAD states
self.client_audio_buffer.clear()
self.client_have_voice = False
self.client_voice_stop = False
self.logger.bind(tag=TAG).debug("VAD states reset.")
self.client_voice_window.clear()
self.last_is_voice = False
# Clear ASR buffers
self.asr_audio.clear()
self.logger.bind(tag=TAG).debug("All audio states reset.")
def chat_and_close(self, text):
"""Chat with the user and then close the connection"""
main/xiaozhi-server/core/handle/helloHandle.py
+2 -1
View File
@@ -142,7 +142,8 @@ async def wakeupWordsResponse(conn):
# 获取当前音色
voice = getattr(conn.tts, "voice", "default")
wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000)
# 使用链接的sample_rate
wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=conn.sample_rate)
file_path = wakeup_words_config.generate_file_path(voice)
with open(file_path, "wb") as f:
f.write(wav_bytes)
main/xiaozhi-server/core/handle/receiveAudioHandle.py
-1
View File
@@ -17,7 +17,6 @@ async def handleAudioMessage(conn, audio):
if hasattr(conn, "just_woken_up") and conn.just_woken_up:
have_voice = False
# 设置一个短暂延迟后恢复VAD检测
conn.asr_audio.clear()
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
main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py
+4 -6
View File
@@ -26,10 +26,9 @@ class ListenTextMessageHandler(TextMessageHandler):
f"客户端拾音模式：{conn.client_listen_mode}"
)
if msg_json["state"] == "start":
conn.client_have_voice = True
conn.client_voice_stop = False
# 设备从播放模式切回录音模式,清除所有音频状态和缓冲区
conn.reset_audio_states()
elif msg_json["state"] == "stop":
conn.client_have_voice = True
conn.client_voice_stop = True
if conn.asr.interface_type == InterfaceType.STREAM:
# 流式模式下，发送结束请求
@@ -38,14 +37,13 @@ class ListenTextMessageHandler(TextMessageHandler):
# 非流式模式：直接触发ASR识别
if len(conn.asr_audio) > 0:
asr_audio_task = conn.asr_audio.copy()
conn.asr_audio.clear()
conn.reset_vad_states()
conn.reset_audio_states()
if len(asr_audio_task) > 0:
await conn.asr.handle_voice_stop(conn, asr_audio_task)
elif msg_json["state"] == "detect":
conn.client_have_voice = False
conn.asr_audio.clear()
conn.reset_audio_states()
if "text" in msg_json:
conn.last_activity_time = time.time() * 1000
original_text = msg_json["text"] # 保留原始文本
main/xiaozhi-server/core/providers/asr/aliyun.py
+7 -19
View File
@@ -213,36 +213,24 @@ class ASRProvider(ASRProviderBase):
return None
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if self._is_token_expired():
logger.warning("Token已过期，正在自动刷新...")
self._refresh_token()
file_path = None
try:
# 解码Opus为PCM
if 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:
pass
else:
file_path = self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
# 发送请求并获取文本
text = await self._send_request(combined_pcm_data)
text = await self._send_request(artifacts.pcm_bytes)
if text:
return text, file_path
return text, artifacts.file_path
return "", file_path
return "", artifacts.file_path
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", file_path
return "", None
main/xiaozhi-server/core/providers/asr/aliyun_stream.py
+8 -25
View File
@@ -126,16 +126,8 @@ class ASRProvider(ASRProviderBase):
await super().open_audio_channels(conn)
async def receive_audio(self, conn, audio, audio_have_voice):
# 初始化音频缓存
if not hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
# 存储音频数据
if audio:
conn.asr_audio_for_voiceprint.append(audio)
conn.asr_audio.append(audio)
conn.asr_audio = conn.asr_audio[-10:]
# 先调用父类方法处理基础逻辑
await super().receive_audio(conn, audio, audio_have_voice)
# 只在有声音且没有连接时建立连接（排除正在停止的情况）
if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -204,6 +196,8 @@ class ASRProvider(ASRProviderBase):
"""转发识别结果"""
try:
while not conn.stop_event.is_set():
# 获取当前连接的音频数据
audio_data = conn.asr_audio
try:
response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
result = json.loads(response)
@@ -257,19 +251,12 @@ class ASRProvider(ASRProviderBase):
# 手动模式下，只有在收到stop信号后才触发处理（仅处理一次）
if conn.client_voice_stop:
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
if len(audio_data) > 0:
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, audio_data)
# 清理音频缓存
conn.asr_audio.clear()
conn.reset_vad_states()
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, audio_data)
break
else:
# 自动模式下直接覆盖
self.text = text
conn.reset_vad_states()
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
await self.handle_voice_stop(conn, audio_data)
break
@@ -289,11 +276,7 @@ class ASRProvider(ASRProviderBase):
finally:
# 清理连接的音频缓存
await self._cleanup()
if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'):
conn.asr_audio = []
conn.reset_audio_states()
async def _send_stop_request(self):
"""发送停止识别请求（不关闭连接）"""
@@ -341,7 +324,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
async def speech_to_text(self, opus_data, session_id, audio_format):
async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py
+8 -25
View File
@@ -52,16 +52,8 @@ class ASRProvider(ASRProviderBase):
await super().open_audio_channels(conn)
async def receive_audio(self, conn, audio, audio_have_voice):
# 初始化音频缓存
if not hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
# 存储音频数据
if audio:
conn.asr_audio_for_voiceprint.append(audio)
conn.asr_audio.append(audio)
conn.asr_audio = conn.asr_audio[-10:]
# 先调用父类方法处理基础逻辑
await super().receive_audio(conn, audio, audio_have_voice)
# 只在有声音且没有连接时建立连接
if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -166,6 +158,8 @@ class ASRProvider(ASRProviderBase):
"""转发识别结果"""
try:
while not conn.stop_event.is_set():
# 获取当前连接的音频数据
audio_data = conn.asr_audio
try:
response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
result = json.loads(response)
@@ -214,19 +208,12 @@ class ASRProvider(ASRProviderBase):
# 手动模式下,只有在收到stop信号后才触发处理
if conn.client_voice_stop:
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
if len(audio_data) > 0:
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, audio_data)
# 清理音频缓存
conn.asr_audio.clear()
conn.reset_vad_states()
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, audio_data)
break
else:
# 自动模式下直接覆盖
self.text = text
conn.reset_vad_states()
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
await self.handle_voice_stop(conn, audio_data)
break
@@ -257,11 +244,7 @@ class ASRProvider(ASRProviderBase):
finally:
# 清理连接的音频缓存
await self._cleanup()
if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'):
conn.asr_audio = []
conn.reset_audio_states()
async def _send_stop_request(self):
"""发送停止请求(用于手动模式停止录音)"""
@@ -325,7 +308,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
async def speech_to_text(self, opus_data, session_id, audio_format):
async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
main/xiaozhi-server/core/providers/asr/baidu.py
+8 -19
View File
@@ -30,37 +30,26 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
logger.bind(tag=TAG).warning("音频数据为空！")
return None, None
file_path = None
try:
# 检查配置是否已设置
if not self.app_id or not self.api_key or not self.secret_key:
logger.bind(tag=TAG).error("百度语音识别配置未设置，无法进行识别")
return None, file_path
return None, None
# 将Opus音频数据解码为PCM
if 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:
pass
else:
self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
start_time = time.time()
# 识别本地文件
result = self.client.asr(
combined_pcm_data,
artifacts.pcm_bytes,
"pcm",
16000,
{
@@ -73,13 +62,13 @@ class ASRProvider(ASRProviderBase):
f"百度语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}"
)
result = result["result"][0]
return result, file_path
return result, artifacts.file_path
else:
raise Exception(
f"百度语音识别失败，错误码: {result['err_no']}，错误信息: {result['err_msg']}"
)
return None, file_path
return None, artifacts.file_path
except Exception as e:
logger.bind(tag=TAG).error(f"处理音频时发生错误！{e}", exc_info=True)
return None, file_path
return None, None
main/xiaozhi-server/core/providers/asr/base.py
+144 -30
View File
@@ -8,14 +8,18 @@ import queue
import asyncio
import traceback
import threading
import shutil
import opuslib_next
from abc import ABC, abstractmethod
from config.logger import setup_logging
from typing import Optional, Tuple, List
from typing import Optional, Tuple, List, NamedTuple
from core.providers.asr.dto.dto import InterfaceType
from core.handle.receiveAudioHandle import startToChat
from core.handle.reportHandle import enqueue_asr_report
from core.utils.util import remove_punctuation_and_length
from core.handle.receiveAudioHandle import handleAudioMessage
import tempfile
TAG = __name__
logger = setup_logging()
@@ -57,18 +61,17 @@ class ASRProviderBase(ABC):
conn.asr_audio.append(audio)
else:
# 自动/实时模式：使用VAD检测
have_voice = audio_have_voice
conn.asr_audio.append(audio)
if not have_voice and not conn.client_have_voice:
# 如果没有语音，且之前也没有声音，缓存部分音频
if not audio_have_voice and not conn.client_have_voice:
conn.asr_audio = conn.asr_audio[-10:]
return
# 自动模式下通过VAD检测到语音停止时触发识别
if conn.client_voice_stop:
if conn.asr.interface_type != InterfaceType.STREAM and conn.client_voice_stop:
asr_audio_task = conn.asr_audio.copy()
conn.asr_audio.clear()
conn.reset_vad_states()
conn.reset_audio_states()
if len(asr_audio_task) > 15:
await self.handle_voice_stop(conn, asr_audio_task)
@@ -93,10 +96,14 @@ class ASRProviderBase(ABC):
wav_data = self._pcm_to_wav(combined_pcm_data)
# 定义ASR任务
asr_task = self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
asr_task = self.speech_to_text_wrapper(
asr_audio_task, conn.session_id, conn.audio_format
)
if conn.voiceprint_provider and wav_data:
voiceprint_task = conn.voiceprint_provider.identify_speaker(wav_data, conn.session_id)
voiceprint_task = conn.voiceprint_provider.identify_speaker(
wav_data, conn.session_id
)
# 并发等待两个结果
asr_result, voiceprint_result = await asyncio.gather(
asr_task, voiceprint_task, return_exceptions=True
@@ -159,20 +166,20 @@ class ASRProviderBase(ABC):
if text_len > 0:
# 使用自定义模块进行上报
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
audio_snapshot = asr_audio_task.copy()
enqueue_asr_report(conn, enhanced_text, audio_snapshot)
except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
import traceback
logger.bind(tag=TAG).debug(f"异常详情: {traceback.format_exc()}")
def _build_enhanced_text(self, text: str, speaker_name: Optional[str]) -> str:
"""构建包含说话人信息的文本（仅用于纯文本ASR）"""
if speaker_name and speaker_name.strip():
return json.dumps({
"speaker": speaker_name,
"content": text
}, ensure_ascii=False)
return json.dumps(
{"speaker": speaker_name, "content": text}, ensure_ascii=False
)
else:
return text
@@ -181,23 +188,23 @@ class ASRProviderBase(ABC):
if len(pcm_data) == 0:
logger.bind(tag=TAG).warning("PCM数据为空，无法转换WAV")
return b""
# 确保数据长度是偶数（16位音频）
if len(pcm_data) % 2 != 0:
pcm_data = pcm_data[:-1]
# 创建WAV文件头
wav_buffer = io.BytesIO()
try:
with wave.open(wav_buffer, 'wb') as wav_file:
wav_file.setnchannels(1) # 单声道
wav_file.setsampwidth(2) # 16位
with wave.open(wav_buffer, "wb") as wav_file:
wav_file.setnchannels(1) # 单声道
wav_file.setsampwidth(2) # 16位
wav_file.setframerate(16000) # 16kHz采样率
wav_file.writeframes(pcm_data)
wav_buffer.seek(0)
wav_data = wav_buffer.read()
return wav_data
except Exception as e:
logger.bind(tag=TAG).error(f"WAV转换失败: {e}")
@@ -206,6 +213,44 @@ class ASRProviderBase(ABC):
def stop_ws_connection(self):
pass
async def close(self):
pass
class AudioArtifacts(NamedTuple):
pcm_frames: List[bytes]
"""PCM音频帧列表"""
pcm_bytes: bytes
"""合并后的PCM音频字节数据"""
file_path: Optional[str]
"""WAV文件路径"""
temp_path: Optional[str]
"""临时WAV文件路径"""
def get_current_artifacts(self) -> Optional["ASRProviderBase.AudioArtifacts"]:
return self._current_artifacts
def requires_file(self) -> bool:
"""是否需要文件输入"""
return False
def prefers_temp_file(self) -> bool:
"""是否优先使用临时文件"""
return False
def build_temp_file(self, pcm_bytes: bytes) -> Optional[str]:
try:
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
temp_path = temp_file.name
with wave.open(temp_path, "wb") as wav_file:
wav_file.setnchannels(1)
wav_file.setsampwidth(2)
wav_file.setframerate(16000)
wav_file.writeframes(pcm_bytes)
return temp_path
except Exception as e:
logger.bind(tag=TAG).error(f"临时音频文件生成失败: {e}")
return None
def save_audio_to_file(self, pcm_data: List[bytes], session_id: str) -> str:
"""PCM数据保存为WAV文件"""
module_name = __name__.split(".")[-1]
@@ -220,11 +265,80 @@ class ASRProviderBase(ABC):
return file_path
@abstractmethod
async def speech_to_text(
async def speech_to_text_wrapper(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
file_path = None
temp_path = None
try:
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
combined_pcm_data = b"".join(pcm_data)
free_space = shutil.disk_usage(self.output_dir).free
if free_space < len(combined_pcm_data) * 2:
raise OSError("磁盘空间不足")
if self.requires_file() and self.prefers_temp_file():
temp_path = self.build_temp_file(combined_pcm_data)
if (hasattr(self, "delete_audio_file") and not self.delete_audio_file) or (
self.requires_file() and not self.prefers_temp_file()
):
file_path = self.save_audio_to_file(pcm_data, session_id)
if len(combined_pcm_data) == 0:
artifacts = None
else:
artifacts = ASRProviderBase.AudioArtifacts(
pcm_frames=pcm_data,
pcm_bytes=combined_pcm_data,
file_path=file_path,
temp_path=temp_path,
)
text, _ = await self.speech_to_text(
opus_data, session_id, audio_format, artifacts
)
return text, file_path
except OSError as e:
logger.bind(tag=TAG).error(f"文件操作错误: {e}")
return None, None
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}")
return None, None
finally:
try:
if temp_path and os.path.exists(temp_path):
os.unlink(temp_path)
if (
hasattr(self, "delete_audio_file")
and self.delete_audio_file
and file_path
and os.path.exists(file_path)
):
os.remove(file_path)
except Exception as e:
logger.bind(tag=TAG).error(f"文件清理失败: {e}")
@abstractmethod
async def speech_to_text(
self,
opus_data: List[bytes],
session_id: str,
audio_format="opus",
artifacts: Optional[AudioArtifacts] = None,
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本
:param opus_data: 输入的Opus音频数据
:param session_id: 会话ID
:param audio_format: 音频格式，默认"opus"
:param artifacts: 音频工件，包含PCM数据、文件路径等
:return: 识别结果文本和文件路径（如果有）
"""
pass
@staticmethod
@@ -235,23 +349,23 @@ class ASRProviderBase(ABC):
decoder = opuslib_next.Decoder(16000, 1)
pcm_data = []
buffer_size = 960 # 每次处理960个采样点 (60ms at 16kHz)
for i, opus_packet in enumerate(opus_data):
try:
if not opus_packet or len(opus_packet) == 0:
continue
pcm_frame = decoder.decode(opus_packet, buffer_size)
if pcm_frame and len(pcm_frame) > 0:
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).warning(f"Opus解码错误，跳过数据包 {i}: {e}")
except Exception as e:
logger.bind(tag=TAG).error(f"音频处理错误，数据包 {i}: {e}")
return pcm_data
except Exception as e:
logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}")
return []
main/xiaozhi-server/core/providers/asr/doubao.py
+7 -18
View File
@@ -232,24 +232,13 @@ class ASRProvider(ASRProviderBase):
yield data[offset:data_len], True
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
file_path = None
try:
# 合并所有opus数据包
if 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:
pass
else:
file_path = self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
# 直接使用PCM数据
# 计算分段大小 (单声道, 16bit, 16kHz采样率)
@@ -258,14 +247,14 @@ class ASRProvider(ASRProviderBase):
# 语音识别
start_time = time.time()
text = await self._send_request(combined_pcm_data, segment_size)
text = await self._send_request(artifacts.pcm_bytes, segment_size)
if text:
logger.bind(tag=TAG).debug(
f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
)
return text, file_path
return "", file_path
return text, artifacts.file_path
return "", artifacts.file_path
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", file_path
return "", None
main/xiaozhi-server/core/providers/asr/doubao_stream.py
+7 -35
View File
@@ -60,17 +60,8 @@ class ASRProvider(ASRProviderBase):
await super().open_audio_channels(conn)
async def receive_audio(self, conn, audio, audio_have_voice):
conn.asr_audio.append(audio)
conn.asr_audio = conn.asr_audio[-10:]
# 存储音频数据
if not hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
conn.asr_audio_for_voiceprint.append(audio)
# 当没有音频数据时处理完整语音片段
if conn.client_listen_mode != "manual" and not audio and len(conn.asr_audio_for_voiceprint) > 0:
await self.handle_voice_stop(conn, conn.asr_audio_for_voiceprint)
conn.asr_audio_for_voiceprint = []
# 先调用父类方法处理基础逻辑
await super().receive_audio(conn, audio, audio_have_voice)
# 如果本次有声音，且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
@@ -164,7 +155,7 @@ class ASRProvider(ASRProviderBase):
try:
while self.asr_ws and not conn.stop_event.is_set():
# 获取当前连接的音频数据
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
audio_data = conn.asr_audio
try:
response = await self.asr_ws.recv()
result = self.parse_response(response)
@@ -189,7 +180,6 @@ class ASRProvider(ASRProviderBase):
):
logger.bind(tag=TAG).error(f"识别文本：空")
self.text = ""
conn.reset_vad_states()
if len(audio_data) > 15: # 确保有足够音频数据
await self.handle_voice_stop(conn, audio_data)
break
@@ -200,12 +190,9 @@ class ASRProvider(ASRProviderBase):
if self.enable_multilingual:
continue
if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 0:
if conn.client_listen_mode == "manual" and conn.client_voice_stop and len(audio_data) > 15:
logger.bind(tag=TAG).debug("消息结束收到停止信号，触发处理")
await self.handle_voice_stop(conn, audio_data)
# 清理音频缓存
conn.asr_audio.clear()
conn.reset_vad_states()
break
for utterance in utterances:
@@ -226,14 +213,10 @@ class ASRProvider(ASRProviderBase):
if conn.client_voice_stop and len(audio_data) > 0:
logger.bind(tag=TAG).debug("消息中途收到停止信号，触发处理")
await self.handle_voice_stop(conn, audio_data)
# 清理音频缓存
conn.asr_audio.clear()
conn.reset_vad_states()
break
else:
# 自动模式下直接覆盖
self.text = current_text
conn.reset_vad_states()
if len(audio_data) > 15: # 确保有足够音频数据
await self.handle_voice_stop(conn, audio_data)
break
@@ -262,11 +245,8 @@ class ASRProvider(ASRProviderBase):
await self.asr_ws.close()
self.asr_ws = None
self.is_processing = False
if conn:
if hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'):
conn.asr_audio = []
# 重置所有音频相关状态
conn.reset_audio_states()
def stop_ws_connection(self):
if self.asr_ws:
@@ -408,7 +388,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}")
raise
async def speech_to_text(self, opus_data, session_id, audio_format):
async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
result = self.text
self.text = "" # 清空text
return result, None
@@ -435,11 +415,3 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("Doubao decoder resources released")
except Exception as e:
logger.bind(tag=TAG).debug(f"释放Doubao decoder资源时出错: {e}")
# 清理所有连接的音频缓冲区
if hasattr(self, '_connections'):
for conn in self._connections.values():
if hasattr(conn, 'asr_audio_for_voiceprint'):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, 'asr_audio'):
conn.asr_audio = []
main/xiaozhi-server/core/providers/asr/fun_local.py
+8 -37
View File
@@ -64,39 +64,21 @@ class ASRProvider(ASRProviderBase):
)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
retry_count = 0
while retry_count < MAX_RETRIES:
try:
# 合并所有opus数据包
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
combined_pcm_data = b"".join(pcm_data)
# 检查磁盘空间
if not self.delete_audio_file:
free_space = shutil.disk_usage(self.output_dir).free
if free_space < len(combined_pcm_data) * 2: # 预留2倍空间
raise OSError("磁盘空间不足")
# 判断是否保存为WAV文件
if self.delete_audio_file:
pass
else:
file_path = self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
# 语音识别 - 使用线程池避免阻塞事件循环
start_time = time.time()
result = await asyncio.to_thread(
self.model.generate,
input=combined_pcm_data,
input=artifacts.pcm_bytes,
cache={},
language="auto",
use_itn=True,
@@ -107,7 +89,7 @@ class ASRProvider(ASRProviderBase):
f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text['content']}"
)
return text, file_path
return text, artifacts.file_path
except OSError as e:
retry_count += 1
@@ -115,7 +97,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).error(
f"语音识别失败（已重试{retry_count}次）: {e}", exc_info=True
)
return "", file_path
return "", None
logger.bind(tag=TAG).warning(
f"语音识别失败，正在重试（{retry_count}/{MAX_RETRIES}）: {e}"
)
@@ -123,15 +105,4 @@ class ASRProvider(ASRProviderBase):
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", file_path
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(
f"文件删除失败: {file_path} | 错误: {e}"
)
return "", None
main/xiaozhi-server/core/providers/asr/fun_server.py
+8 -17
View File
@@ -101,7 +101,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug(f"Sent end message: {end_message}")
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""
Convert speech data to text using FunASR.
@@ -109,18 +109,9 @@ class ASRProvider(ASRProviderBase):
:param session_id: Unique session identifier.
:return: Tuple containing recognized text and optional timestamp.
"""
file_path = None
if 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:
pass
else:
file_path = self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
auth_header = {"Authorization": "Bearer; {}".format(self.api_key)}
async with websockets.connect(
self.uri,
@@ -132,7 +123,7 @@ class ASRProvider(ASRProviderBase):
try:
# Use asyncio to handle WebSocket communication
send_task = asyncio.create_task(
self._send_data(ws, combined_pcm_data, session_id)
self._send_data(ws, artifacts.pcm_bytes, session_id)
)
receive_task = asyncio.create_task(self._receive_responses(ws))
@@ -161,14 +152,14 @@ class ASRProvider(ASRProviderBase):
result = lang_tag_filter(result)
return (
result,
file_path,
artifacts.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
return "", artifacts.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
return "", artifacts.file_path
main/xiaozhi-server/core/providers/asr/openai.py
+8 -20
View File
@@ -21,20 +21,16 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus") -> Tuple[Optional[str], Optional[str]]:
def requires_file(self) -> bool:
return True
async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None) -> Tuple[Optional[str], Optional[str]]:
file_path = None
try:
start_time = time.time()
if 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}"
)
if artifacts is None:
return "", None
file_path = artifacts.file_path
logger.bind(tag=TAG).info(f"file path: {file_path}")
headers = {
"Authorization": f"Bearer {self.api_key}",
@@ -71,12 +67,4 @@ class ASRProvider(ASRProviderBase):
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}")
return "", None
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py
+10 -50
View File
@@ -1,5 +1,4 @@
import os
import tempfile
from typing import Optional, Tuple, List
import dashscope
from config.logger import setup_logging
@@ -35,56 +34,25 @@ class ASRProvider(ASRProviderBase):
# 确保输出目录存在
os.makedirs(self.output_dir, exist_ok=True)
def _prepare_audio_file(self, pcm_data: bytes) -> str:
"""将PCM数据转换为WAV文件并返回文件路径"""
try:
import wave
# 创建临时WAV文件
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file:
temp_path = temp_file.name
# 写入WAV格式
with wave.open(temp_path, 'wb') as wav_file:
wav_file.setnchannels(1) # 单声道
wav_file.setsampwidth(2) # 16位
wav_file.setframerate(16000) # 16kHz采样率
wav_file.writeframes(pcm_data)
return temp_path
except Exception as e:
logger.bind(tag=tag).error(f"音频文件准备失败: {e}")
return None
def prefers_temp_file(self) -> bool:
return True
def requires_file(self) -> bool:
return True
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
temp_file_path = None
file_path = None
try:
# 解码音频数据
if audio_format == "pcm":
pcm_data = opus_data
else:
pcm_data = self.decode_opus(opus_data)
combined_pcm_data = b"".join(pcm_data)
if len(combined_pcm_data) == 0:
logger.bind(tag=tag).warning("音频数据为空")
if artifacts is None:
return "", None
# 准备音频文件
temp_file_path = self._prepare_audio_file(combined_pcm_data)
temp_file_path = artifacts.temp_path
file_path = artifacts.file_path
if not temp_file_path:
return "", None
# 保存音频文件（如果需要）
if not self.delete_audio_file:
file_path = self.save_audio_to_file(pcm_data, session_id)
return "", file_path
# 构造请求消息
messages = [
{
@@ -141,11 +109,3 @@ class ASRProvider(ASRProviderBase):
except Exception as e:
logger.bind(tag=tag).error(f"语音识别失败: {e}")
return "", file_path
finally:
# 清理临时文件
if temp_file_path and os.path.exists(temp_file_path):
try:
os.unlink(temp_file_path)
except Exception as e:
logger.bind(tag=tag).warning(f"清理临时文件失败: {e}")
main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py
+7 -20
View File
@@ -120,24 +120,19 @@ class ASRProvider(ASRProviderBase):
samples_float32 = samples_float32 / 32768
return samples_float32, f.getframerate()
def requires_file(self) -> bool:
return True
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
try:
# 保存音频文件
start_time = time.time()
if 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}"
)
if artifacts is None:
return "", None
file_path = artifacts.file_path
# 语音识别
start_time = time.time()
s = self.model.create_stream()
samples, sample_rate = self.read_wave(file_path)
@@ -153,11 +148,3 @@ class ASRProvider(ASRProviderBase):
except Exception as e:
logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
return "", file_path
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
main/xiaozhi-server/core/providers/asr/tencent.py
+7 -18
View File
@@ -32,35 +32,24 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
self, opus_data: List[bytes], session_id: str, audio_format="opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
logger.bind(tag=TAG).warning("音频数据为空！")
return None, None
file_path = None
try:
# 检查配置是否已设置
if not self.secret_id or not self.secret_key:
logger.bind(tag=TAG).error("腾讯云语音识别配置未设置，无法进行识别")
return None, file_path
return None, None
# 将Opus音频数据解码为PCM
if 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:
pass
else:
self.save_audio_to_file(pcm_data, session_id)
if artifacts is None:
return "", None
# 将音频数据转换为Base64编码
base64_audio = base64.b64encode(combined_pcm_data).decode("utf-8")
base64_audio = base64.b64encode(artifacts.pcm_bytes).decode("utf-8")
# 构建请求体
request_body = self._build_request_body(base64_audio)
@@ -77,11 +66,11 @@ class ASRProvider(ASRProviderBase):
f"腾讯云语音识别耗时: {time.time() - start_time:.3f}s | 结果: {result}"
)
return result, file_path
return result, artifacts.file_path
except Exception as e:
logger.bind(tag=TAG).error(f"处理音频时发生错误！{e}", exc_info=True)
return None, file_path
return None, None
def _build_request_body(self, base64_audio: str) -> str:
"""构建请求体"""
main/xiaozhi-server/core/providers/asr/vosk.py
+7 -30
View File
@@ -44,36 +44,21 @@ class ASRProvider(ASRProviderBase):
raise
async def speech_to_text(
self, audio_data: List[bytes], session_id: str, audio_format: str = "opus"
self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
file_path = None
try:
# 检查模型是否加载成功
if not self.model:
logger.bind(tag=TAG).error("VOSK模型未加载，无法进行识别")
return "", None
# 解码音频（如果原始格式是Opus）
if audio_format == "pcm":
pcm_data = audio_data
else:
pcm_data = self.decode_opus(audio_data)
if not pcm_data:
logger.bind(tag=TAG).warning("解码后的PCM数据为空，无法进行识别")
if artifacts is None:
return "", None
# 合并PCM数据
combined_pcm_data = b"".join(pcm_data)
if len(combined_pcm_data) == 0:
if not artifacts.pcm_bytes:
logger.bind(tag=TAG).warning("合并后的PCM数据为空")
return "", None
# 判断是否保存为WAV文件
if not self.delete_audio_file:
file_path = self.save_audio_to_file(pcm_data, session_id)
start_time = time.time()
@@ -81,8 +66,8 @@ class ASRProvider(ASRProviderBase):
chunk_size = 2000
text_result = ""
for i in range(0, len(combined_pcm_data), chunk_size):
chunk = combined_pcm_data[i:i+chunk_size]
for i in range(0, len(artifacts.pcm_bytes), chunk_size):
chunk = artifacts.pcm_bytes[i:i+chunk_size]
if self.recognizer.AcceptWaveform(chunk):
result = json.loads(self.recognizer.Result())
text = result.get('text', '')
@@ -99,16 +84,8 @@ class ASRProvider(ASRProviderBase):
f"VOSK语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text_result.strip()}"
)
return text_result.strip(), file_path
return text_result.strip(), artifacts.file_path
except Exception as e:
logger.bind(tag=TAG).error(f"VOSK语音识别失败: {e}")
return "", None
finally:
# 文件清理逻辑
if self.delete_audio_file and file_path and os.path.exists(file_path):
try:
os.remove(file_path)
logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
except Exception as e:
logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+4 -27
View File
@@ -101,11 +101,6 @@ class ASRProvider(ASRProviderBase):
# 先调用父类方法处理基础逻辑
await super().receive_audio(conn, audio, audio_have_voice)
# 存储音频数据用于声纹识别
if not hasattr(conn, "asr_audio_for_voiceprint"):
conn.asr_audio_for_voiceprint = []
conn.asr_audio_for_voiceprint.append(audio)
# 如果本次有声音，且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
try:
@@ -232,13 +227,8 @@ class ASRProvider(ASRProviderBase):
if status == 2:
if conn.client_listen_mode == "manual":
audio_data = getattr(conn, 'asr_audio_for_voiceprint', [])
if len(audio_data) > 0:
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, audio_data)
# 清理音频缓存
conn.asr_audio.clear()
conn.reset_vad_states()
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
await self.handle_voice_stop(conn, conn.asr_audio)
break
except asyncio.TimeoutError:
@@ -262,13 +252,7 @@ class ASRProvider(ASRProviderBase):
finally:
# 清理连接资源
await self._cleanup()
# 清理连接的音频缓存
if conn:
if hasattr(conn, "asr_audio_for_voiceprint"):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, "asr_audio"):
conn.asr_audio = []
conn.reset_audio_states()
async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]):
"""处理语音停止，发送最后一帧并处理识别结果"""
@@ -334,7 +318,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
async def speech_to_text(self, opus_data, session_id, audio_format):
async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
@@ -363,10 +347,3 @@ class ASRProvider(ASRProviderBase):
except Exception as e:
logger.bind(tag=TAG).debug(f"释放Xunfei decoder资源时出错: {e}")
# 清理所有连接的音频缓冲区
if hasattr(self, "_connections"):
for conn in self._connections.values():
if hasattr(conn, "asr_audio_for_voiceprint"):
conn.asr_audio_for_voiceprint = []
if hasattr(conn, "asr_audio"):
conn.asr_audio = []
main/xiaozhi-server/core/providers/tts/alibl_stream.py
+2 -9
View File
@@ -41,8 +41,6 @@ class TTSProvider(TTSProviderBase):
# 音频参数配置
self.format = config.get("format", "pcm")
sample_rate = config.get("sample_rate", "24000")
self.sample_rate = int(sample_rate) if sample_rate else 24000
volume = config.get("volume", "50")
self.volume = int(volume) if volume else 50
@@ -60,11 +58,6 @@ class TTSProvider(TTSProviderBase):
"X-DashScope-DataInspection": "enable",
}
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=self.sample_rate, channels=1, frame_size_ms=60
)
async def _ensure_connection(self):
"""确保WebSocket连接可用，支持60秒内连接复用"""
try:
@@ -245,7 +238,7 @@ class TTSProvider(TTSProviderBase):
"text_type": "PlainText",
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"volume": self.volume,
"rate": self.rate,
"pitch": self.pitch,
@@ -429,7 +422,7 @@ class TTSProvider(TTSProviderBase):
"text_type": "PlainText",
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"volume": self.volume,
"rate": self.rate,
"pitch": self.pitch,
main/xiaozhi-server/core/providers/tts/aliyun.py
+1 -3
View File
@@ -95,8 +95,6 @@ class TTSProvider(TTSProviderBase):
self.appkey = config.get("appkey")
self.format = config.get("format", "wav")
self.audio_file_type = config.get("format", "wav")
sample_rate = config.get("sample_rate", "16000")
self.sample_rate = int(sample_rate) if sample_rate else 16000
if config.get("private_voice"):
self.voice = config.get("private_voice")
@@ -172,7 +170,7 @@ class TTSProvider(TTSProviderBase):
"token": self.token,
"text": text,
"format": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"voice": self.voice,
"volume": self.volume,
"speech_rate": self.speech_rate,
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
+2 -11
View File
@@ -99,10 +99,6 @@ class TTSProvider(TTSProviderBase):
self.format = config.get("format", "pcm")
self.audio_file_type = config.get("format", "pcm")
# 采样率配置
sample_rate = config.get("sample_rate", "16000")
self.sample_rate = int(sample_rate) if sample_rate else 16000
# 音色配置 - CosyVoice大模型音色
if config.get("private_voice"):
self.voice = config.get("private_voice")
@@ -134,11 +130,6 @@ class TTSProvider(TTSProviderBase):
# 专属tts设置
self.task_id = uuid.uuid4().hex
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
# Token管理
if self.access_key_id and self.access_key_secret:
self._refresh_token()
@@ -344,7 +335,7 @@ class TTSProvider(TTSProviderBase):
"payload": {
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"volume": self.volume,
"speech_rate": self.speech_rate,
"pitch_rate": self.pitch_rate,
@@ -508,7 +499,7 @@ class TTSProvider(TTSProviderBase):
"payload": {
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"volume": self.volume,
"speech_rate": self.speech_rate,
"pitch_rate": self.pitch_rate,
main/xiaozhi-server/core/providers/tts/base.py
+16 -5
View File
@@ -1,17 +1,18 @@
import os
import re
import time
import uuid
import queue
import asyncio
import threading
import traceback
from core.utils import p3
from datetime import datetime
from core.utils import textUtils
from typing import Callable, Any
from abc import ABC, abstractmethod
from config.logger import setup_logging
from core.utils import opus_encoder_utils
from core.utils.tts import MarkdownCleaner
from core.utils.output_counter import add_device_output
from core.handle.reportHandle import enqueue_tts_report
@@ -97,6 +98,8 @@ class TTSProviderBase(ABC):
file_type=self.audio_file_type,
is_opus=True,
callback=opus_handler,
sample_rate=self.conn.sample_rate,
opus_encoder=self.opus_encoder,
)
break
else:
@@ -138,7 +141,7 @@ class TTSProviderBase(ABC):
logger.bind(tag=TAG).error(
f"语音生成失败: {text}，请检查网络或服务是否正常"
)
self.tts_audio_queue.put((SentenceType.FIRST, None, text))
self.tts_audio_queue.put((SentenceType.FIRST, None, text))
self._process_audio_file_stream(tmp_file, callback=opus_handler)
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
@@ -158,7 +161,8 @@ class TTSProviderBase(ABC):
audio_bytes,
file_type=self.audio_file_type,
is_opus=True,
callback=lambda data: audio_datas.append(data)
callback=lambda data: audio_datas.append(data),
sample_rate=self.conn.sample_rate,
)
return audio_datas
else:
@@ -214,13 +218,13 @@ class TTSProviderBase(ABC):
self, audio_file_path, callback: Callable[[Any], Any] = None
):
"""音频文件转换为PCM编码"""
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback)
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=None)
def audio_to_opus_data_stream(
self, audio_file_path, callback: Callable[[Any], Any] = None
):
"""音频文件转换为Opus编码"""
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback)
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback, sample_rate=self.conn.sample_rate, opus_encoder=self.opus_encoder)
def tts_one_sentence(
self,
@@ -252,6 +256,13 @@ class TTSProviderBase(ABC):
async def open_audio_channels(self, conn):
self.conn = conn
# 根据conn的sample_rate创建编码器，如果子类已经创建则不覆盖（IndexTTS接口返回为24kHZ-待重采样处理）
if not hasattr(self, 'opus_encoder') or self.opus_encoder is None:
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=conn.sample_rate, channels=1, frame_size_ms=60
)
# tts 消化线程
self.tts_priority_thread = threading.Thread(
target=self.tts_text_priority_thread, daemon=True
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+37 -36
View File
@@ -154,16 +154,29 @@ class TTSProvider(TTSProviderBase):
self.voice = config.get("private_voice")
else:
self.voice = config.get("speaker")
speech_rate = config.get("speech_rate", "0")
loudness_rate = config.get("loudness_rate", "0")
pitch = config.get("pitch", "0")
self.speech_rate = int(speech_rate) if speech_rate else 0
self.loudness_rate = int(loudness_rate) if loudness_rate else 0
self.pitch = int(pitch) if pitch else 0
# 多情感音色参数
self.emotion = config.get("emotion", "neutral")
emotion_scale = config.get("emotion_scale", "4")
self.emotion_scale = int(emotion_scale) if emotion_scale else 4
# 默认 audio_params 配置
default_audio_params = {
"speech_rate": 0,
"loudness_rate": 0
}
# 默认 additions 配置
default_additions = {
"aigc_metadata": {},
"cache_config": {},
"post_process": {
"pitch": 0
}
}
# 默认 mix_speaker 配置
default_mix_speaker = {}
# 合并用户配置
self.audio_params = {**default_audio_params, **config.get("audio_params", {})}
self.additions = {**default_additions, **config.get("additions", {})}
self.mix_speaker = {**default_mix_speaker, **config.get("mix_speaker", {})}
self.ws_url = config.get("ws_url")
self.authorization = config.get("authorization")
@@ -171,9 +184,7 @@ class TTSProvider(TTSProviderBase):
enable_ws_reuse_value = config.get("enable_ws_reuse", True)
self.enable_ws_reuse = False if str(enable_ws_reuse_value).lower() == 'false' else True
self.tts_text = ""
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
model_key_msg = check_model_key("TTS", self.access_token)
if model_key_msg:
logger.bind(tag=TAG).error(model_key_msg)
@@ -181,6 +192,8 @@ class TTSProvider(TTSProviderBase):
async def open_audio_channels(self, conn):
try:
await super().open_audio_channels(conn)
# 更新 audio_params 中的采样率为实际的 conn.sample_rate
self.audio_params["sample_rate"] = conn.sample_rate
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}")
self.ws = None
@@ -646,20 +659,18 @@ class TTSProvider(TTSProviderBase):
text="",
speaker="",
audio_format="pcm",
audio_sample_rate=16000,
):
audio_params = {
"format": audio_format,
"sample_rate": audio_sample_rate,
"speech_rate": self.speech_rate,
"loudness_rate": self.loudness_rate
# 构建 req_params
req_params = {
"text": text,
"speaker": speaker,
"audio_params": {**self.audio_params, "format": audio_format},
"additions": json.dumps(self.additions)
}
# 如果是多情感音色,添加情感参数
if '_emo_' in self.voice:
if self.emotion:
audio_params["emotion"] = self.emotion
audio_params["emotion_scale"] = self.emotion_scale
# 如果有 mix_speaker 配置，添加到 req_params
if self.mix_speaker:
req_params["mix_speaker"] = self.mix_speaker
return str.encode(
json.dumps(
@@ -667,17 +678,7 @@ class TTSProvider(TTSProviderBase):
"user": {"uid": uid},
"event": event,
"namespace": "BidirectionalTTS",
"req_params": {
"text": text,
"speaker": speaker,
"audio_params": audio_params,
"additions": json.dumps({
"post_process": {
"pitch": self.pitch
}
})
},
"req_params": req_params
}
)
)
main/xiaozhi-server/core/providers/tts/index_stream.py
+14
View File
@@ -174,6 +174,20 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def audio_to_pcm_data_stream(
self, audio_file_path, callback=None
):
"""音频文件转换为PCM编码，使用24kHz采样率"""
from core.utils.util import audio_to_data_stream
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback, sample_rate=24000, opus_encoder=None)
def audio_to_opus_data_stream(
self, audio_file_path, callback=None
):
"""音频文件转换为Opus编码，使用24kHz采样率和自己的编码器"""
from core.utils.util import audio_to_data_stream
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback, sample_rate=24000, opus_encoder=self.opus_encoder)
async def close(self):
"""资源清理"""
await super().close()
main/xiaozhi-server/core/providers/tts/linkerai.py
+3 -8
View File
@@ -25,11 +25,6 @@ class TTSProvider(TTSProviderBase):
self.audio_format = "pcm"
self.before_stop_play_files = []
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
# PCM缓冲区
self.pcm_buffer = bytearray()
@@ -127,7 +122,7 @@ class TTSProvider(TTSProviderBase):
"spk_id": self.voice,
"frame_durition": 60,
"stream": "true",
"target_sr": 16000,
"target_sr": self.conn.sample_rate,
"audio_format": "pcm",
"instruct_text": "请生成一段自然流畅的语音",
}
@@ -136,7 +131,7 @@ class TTSProvider(TTSProviderBase):
"Content-Type": "application/json",
}
# 一帧 PCM 所需字节数：60 ms &times; 16 kHz &times; 1 ch &times; 2 B = 1 920
# 一帧 PCM 所需字节数：60 ms × sample_rate × 1 ch × 2 B
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels # 1
@@ -213,7 +208,7 @@ class TTSProvider(TTSProviderBase):
"spk_id": self.voice,
"frame_duration": 60,
"stream": False,
"target_sr": 16000,
"target_sr": self.conn.sample_rate,
"audio_format": self.audio_format,
"instruct_text": "请生成一段自然流畅的语音",
}
main/xiaozhi-server/core/providers/tts/minimax_httpstream.py
+20 -4
View File
@@ -64,13 +64,17 @@ class TTSProvider(TTSProviderBase):
}
self.audio_file_type = defult_audio_setting.get("format", "pcm")
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=24000, channels=1, frame_size_ms=60
)
# PCM缓冲区
self.pcm_buffer = bytearray()
async def open_audio_channels(self, conn):
"""初始化音频通道,并根据conn.sample_rate更新配置"""
# 调用父类方法
await super().open_audio_channels(conn)
# 更新audio_setting中的采样率为实际的conn.sample_rate
self.audio_setting["sample_rate"] = conn.sample_rate
def tts_text_priority_thread(self):
"""流式文本处理线程"""
while not self.conn.stop_event.is_set():
@@ -212,6 +216,18 @@ class TTSProvider(TTSProviderBase):
try:
data = json.loads(json_str)
# 检查业务层错误
base_resp = data.get("base_resp", {})
status_code = base_resp.get("status_code", 0)
if status_code != 0:
status_msg = base_resp.get("status_msg", "未知错误")
logger.bind(tag=TAG).error(
f"TTS请求失败, 错误码:{status_code}, 错误消息:{status_msg}"
)
self.tts_audio_queue.put((SentenceType.LAST, [], None))
return
status = data.get("data", {}).get("status", 1)
audio_hex = data.get("data", {}).get("audio")
main/xiaozhi-server/core/providers/tts/paddle_speech.py
+1 -4
View File
@@ -25,10 +25,7 @@ class TTSProvider(TTSProviderBase):
self.spk_id = int(config.get("private_voice"))
else:
self.spk_id = int(config.get("spk_id", "0"))
sample_rate = config.get("sample_rate", 24000)
self.sample_rate = float(sample_rate) if sample_rate else 24000
speed = config.get("speed", 1.0)
self.speed = float(speed) if speed else 1.0
main/xiaozhi-server/core/providers/tts/siliconflow.py
-1
View File
@@ -13,7 +13,6 @@ class TTSProvider(TTSProviderBase):
self.voice = config.get("voice")
self.response_format = config.get("response_format", "mp3")
self.audio_file_type = config.get("response_format", "mp3")
self.sample_rate = config.get("sample_rate")
self.speed = float(config.get("speed", 1.0))
self.gain = config.get("gain")
main/xiaozhi-server/core/providers/tts/xunfei_stream.py
+1 -9
View File
@@ -91,9 +91,6 @@ class TTSProvider(TTSProviderBase):
# 音频编码配置
self.format = config.get("format", "raw")
sample_rate = config.get("sample_rate", "24000")
self.sample_rate = int(sample_rate) if sample_rate else 24000
# 口语化配置
self.oral_level = config.get("oral_level", "mid")
@@ -113,11 +110,6 @@ class TTSProvider(TTSProviderBase):
# 序列号管理
self.text_seq = 0
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=self.sample_rate, channels=1, frame_size_ms=60
)
# 验证必需参数
if not all([self.app_id, self.api_key, self.api_secret]):
raise ValueError("讯飞TTS需要配置app_id、api_key和api_secret")
@@ -507,7 +499,7 @@ class TTSProvider(TTSProviderBase):
"rhy": 0,
"audio": {
"encoding": self.format,
"sample_rate": self.sample_rate,
"sample_rate": self.conn.sample_rate,
"channels": 1,
"bit_depth": 16,
"frame_size": 0
main/xiaozhi-server/core/utils/util.py
+33 -16
View File
@@ -227,7 +227,7 @@ def extract_json_from_string(input_string):
def audio_to_data_stream(
audio_file_path, is_opus=True, callback: Callable[[Any], Any] = None
audio_file_path, is_opus=True, callback: Callable[[Any], Any] = None, sample_rate=16000, opus_encoder=None
) -> None:
# 获取文件后缀名
file_type = os.path.splitext(audio_file_path)[1]
@@ -238,12 +238,12 @@ def audio_to_data_stream(
audio_file_path, format=file_type, parameters=["-nostdin"]
)
# 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
# 转换为单声道/指定采样率/16位小端编码（确保与编码器匹配）
audio = audio.set_channels(1).set_frame_rate(sample_rate).set_sample_width(2)
# 获取原始PCM数据（16位小端）
raw_data = audio.raw_data
pcm_to_data_stream(raw_data, is_opus, callback)
pcm_to_data_stream(raw_data, is_opus, callback, sample_rate, opus_encoder)
async def audio_to_data(
@@ -325,7 +325,7 @@ async def audio_to_data(
def audio_bytes_to_data_stream(
audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]
audio_bytes, file_type, is_opus, callback: Callable[[Any], Any], sample_rate=16000, opus_encoder=None
) -> None:
"""
直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3
@@ -338,18 +338,30 @@ def audio_bytes_to_data_stream(
audio = AudioSegment.from_file(
BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"]
)
audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
audio = audio.set_channels(1).set_frame_rate(sample_rate).set_sample_width(2)
raw_data = audio.raw_data
pcm_to_data_stream(raw_data, is_opus, callback)
pcm_to_data_stream(raw_data, is_opus, callback, sample_rate, opus_encoder)
def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None):
# 初始化Opus编码器
encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO)
def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None, sample_rate=16000, opus_encoder=None):
"""
将PCM数据流式编码为Opus或直接输出PCM
Args:
raw_data: PCM原始数据
is_opus: 是否编码为Opus
callback: 回调函数
sample_rate: 采样率
opus_encoder: OpusEncoderUtils对象(推荐提供以保持编码器状态连续)
"""
using_temp_encoder = False
if is_opus and opus_encoder is None:
encoder = opuslib_next.Encoder(sample_rate, 1, opuslib_next.APPLICATION_AUDIO)
using_temp_encoder = True
# 编码参数
frame_duration = 60 # 60ms per frame
frame_size = int(16000 * frame_duration / 1000) # 960 samples/frame
frame_size = int(sample_rate * frame_duration / 1000) # samples/frame
# 按帧处理所有音频数据（包括最后一帧可能补零）
for i in range(0, len(raw_data), frame_size * 2): # 16bit=2bytes/sample
@@ -361,12 +373,17 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] =
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)
callback(frame_data)
if using_temp_encoder:
# 使用临时编码器（仅用于独立音频场景）
np_frame = np.frombuffer(chunk, dtype=np.int16)
frame_data = encoder.encode(np_frame.tobytes(), frame_size)
callback(frame_data)
else:
# 使用外部编码器（TTS流式场景,保持状态连续）
is_last = (i + frame_size * 2 >= len(raw_data))
opus_encoder.encode_pcm_to_opus_stream(chunk, end_of_stream=is_last, callback=callback)
else:
# PCM模式,直接输出
frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
callback(frame_data)