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

Merge pull request #2158 from xinnan-tech/future

Browse Source 
Future
This commit is contained in:
CGD
2025-09-05 10:16:39 +08:00
committed by GitHub
parent f0e0b97217 bd8ae27f92
commit 49565b4b3c
14 changed files with 432 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/handle/helloHandle.py
+15 -5
View File
@@ -4,15 +4,15 @@ import random
import asyncio
from core.utils.dialogue import Message
from core.utils.util import audio_to_data
from core.providers.tts.dto.dto import SentenceType
from core.utils.wakeup_word import WakeupWordsConfig
from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message
from core.utils.util import remove_punctuation_and_length, opus_datas_to_wav_bytes
from core.providers.tts.dto.dto import ContentType, SentenceType
from core.providers.tools.device_mcp import (
MCPClient,
send_mcp_initialize_message,
send_mcp_tools_list_request,
)
from core.utils.wakeup_word import WakeupWordsConfig
TAG = __name__
@@ -56,7 +56,16 @@ async def checkWakeupWords(conn, text):
"enable_wakeup_words_response_cache"
]
if not enable_wakeup_words_response_cache or not conn.tts:
# 等待tts初始化，最多等待3秒
start_time = time.time()
while time.time() - start_time < 3:
if conn.tts:
break
await asyncio.sleep(0.1)
else:
return False
if not enable_wakeup_words_response_cache:
return False
_, filtered_text = remove_punctuation_and_length(text)
@@ -81,9 +90,10 @@ async def checkWakeupWords(conn, text):
"text": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦",
}
# 获取音频数据
opus_packets = audio_to_data(response.get("file_path"))
# 播放唤醒词回复
conn.client_abort = False
opus_packets, _ = audio_to_data(response.get("file_path"))
conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}")
await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text"))
@@ -138,4 +148,4 @@ async def wakeupWordsResponse(conn):
finally:
# 确保在任何情况下都释放锁
if _wakeup_response_lock.locked():
_wakeup_response_lock.release()
_wakeup_response_lock.release()
main/xiaozhi-server/core/handle/intentHandler.py
+7 -6
View File
@@ -1,12 +1,12 @@
import json
import asyncio
import uuid
from core.handle.sendAudioHandle import send_stt_message
from core.handle.helloHandle import checkWakeupWords
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import ContentType
import asyncio
from core.utils.dialogue import Message
from core.providers.tts.dto.dto import ContentType
from core.handle.helloHandle import checkWakeupWords
from plugins_func.register import Action, ActionResponse
from core.handle.sendAudioHandle import send_stt_message
from core.utils.util import remove_punctuation_and_length
from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
TAG = __name__
@@ -24,9 +24,10 @@ async def handle_user_intent(conn, text):
pass
# 检查是否有明确的退出命令
filtered_text = remove_punctuation_and_length(text)[1]
_, filtered_text = remove_punctuation_and_length(text)
if await check_direct_exit(conn, filtered_text):
return True
# 检查是否是唤醒词
if await checkWakeupWords(conn, filtered_text):
return True
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+17 -17
View File
@@ -1,12 +1,11 @@
from core.handle.sendAudioHandle import send_stt_message
import time
import json
import asyncio
from core.utils.util import audio_to_data
from core.handle.abortHandle import handleAbortMessage
from core.handle.intentHandler import handle_user_intent
from core.utils.output_counter import check_device_output_limit
from core.handle.abortHandle import handleAbortMessage
import time
import asyncio
import json
from core.handle.sendAudioHandle import SentenceType
from core.utils.util import audio_to_data
from core.handle.sendAudioHandle import send_stt_message, SentenceType
TAG = __name__
@@ -22,7 +21,6 @@ async def handleAudioMessage(conn, audio):
if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
return
if have_voice:
if conn.client_is_speaking:
await handleAbortMessage(conn)
@@ -31,18 +29,16 @@ async def handleAudioMessage(conn, audio):
# 接收音频
await conn.asr.receive_audio(conn, audio, have_voice)
async def resume_vad_detection(conn):
# 等待2秒后恢复VAD检测
await asyncio.sleep(1)
conn.just_woken_up = False
async def startToChat(conn, text):
# 检查输入是否是JSON格式（包含说话人信息）
speaker_name = None
actual_text = text
try:
# 尝试解析JSON格式的输入
if text.strip().startswith('{') and text.strip().endswith('}'):
@@ -51,13 +47,13 @@ async def startToChat(conn, text):
speaker_name = data['speaker']
actual_text = data['content']
conn.logger.bind(tag=TAG).info(f"解析到说话人信息: {speaker_name}")
# 直接使用JSON格式的文本，不解析
actual_text = text
except (json.JSONDecodeError, KeyError):
# 如果解析失败，继续使用原始文本
pass
# 保存说话人信息到连接对象
if speaker_name:
conn.current_speaker = speaker_name
@@ -118,10 +114,12 @@ async def no_voice_close_connect(conn, have_voice):
async def max_out_size(conn):
# 播放超出最大输出字数的提示
conn.client_abort = False
text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！"
await send_stt_message(conn, text)
file_path = "config/assets/max_output_size.wav"
opus_packets, _ = audio_to_data(file_path)
opus_packets = audio_to_data(file_path)
conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text))
conn.close_after_chat = True
@@ -140,7 +138,7 @@ async def check_bind_device(conn):
# 播放提示音
music_path = "config/assets/bind_code.wav"
opus_packets, _ = audio_to_data(music_path)
opus_packets = audio_to_data(music_path)
conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text))
# 逐个播放数字
@@ -148,15 +146,17 @@ 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_data(num_path)
num_packets = audio_to_data(num_path)
conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None))
except Exception as e:
conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
continue
conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
else:
# 播放未绑定提示
conn.client_abort = False
text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
await send_stt_message(conn, text)
music_path = "config/assets/bind_not_found.wav"
opus_packets, _ = audio_to_data(music_path)
opus_packets = audio_to_data(music_path)
conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text))
main/xiaozhi-server/core/handle/sendAudioHandle.py
+73 -35
View File
@@ -1,25 +1,26 @@
import json
import asyncio
import time
from core.providers.tts.dto.dto import SentenceType
import asyncio
from core.utils import textUtils
from core.utils.util import audio_to_data
from core.providers.tts.dto.dto import SentenceType
TAG = __name__
async def sendAudioMessage(conn, sentenceType, audios, text):
# 发送句子开始消息
conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}")
pre_buffer = False
if conn.tts.tts_audio_first_sentence:
conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
conn.tts.tts_audio_first_sentence = False
pre_buffer = True
await send_tts_message(conn, "start", None)
await send_tts_message(conn, "sentence_start", text)
if sentenceType == SentenceType.FIRST:
await send_tts_message(conn, "sentence_start", text)
await sendAudio(conn, audios, pre_buffer)
await sendAudio(conn, audios)
# 发送句子开始消息
if sentenceType is not SentenceType.MIDDLE:
conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}")
# 发送结束消息（如果是最后一个文本）
if conn.llm_finish_task and sentenceType == SentenceType.LAST:
@@ -30,45 +31,83 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
# 播放音频
async def sendAudio(conn, audios, pre_buffer=True):
async def sendAudio(conn, audios, frame_duration=60):
"""
发送单个opus包，支持流控
Args:
conn: 连接对象
opus_packet: 单个opus数据包
pre_buffer: 快速发送音频
frame_duration: 帧时长（毫秒），匹配 Opus 编码
"""
if audios is None or len(audios) == 0:
return
# 流控参数优化
frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码
start_time = time.perf_counter()
play_position = 0
# 仅当第一句话时执行预缓冲
if pre_buffer:
pre_buffer_frames = min(3, len(audios))
for i in range(pre_buffer_frames):
await conn.websocket.send(audios[i])
remaining_audios = audios[pre_buffer_frames:]
else:
remaining_audios = audios
# 播放剩余音频帧
for opus_packet in remaining_audios:
if isinstance(audios, bytes):
if conn.client_abort:
break
return
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 计算预期发送时间
expected_time = start_time + (play_position / 1000)
# 获取或初始化流控状态
if not hasattr(conn, "audio_flow_control"):
conn.audio_flow_control = {
"last_send_time": 0,
"packet_count": 0,
"start_time": time.perf_counter(),
}
flow_control = conn.audio_flow_control
current_time = time.perf_counter()
# 计算预期发送时间
expected_time = flow_control["start_time"] + (
flow_control["packet_count"] * frame_duration / 1000
)
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
await conn.websocket.send(opus_packet)
# 发送数据包
await conn.websocket.send(audios)
play_position += frame_duration
# 更新流控状态
flow_control["packet_count"] += 1
flow_control["last_send_time"] = time.perf_counter()
else:
# 文件型音频走普通播放
start_time = time.perf_counter()
play_position = 0
# 执行预缓冲
pre_buffer_frames = min(3, len(audios))
for i in range(pre_buffer_frames):
await conn.websocket.send(audios[i])
remaining_audios = audios[pre_buffer_frames:]
# 播放剩余音频帧
for opus_packet in remaining_audios:
if conn.client_abort:
break
# 重置没有声音的状态
conn.last_activity_time = time.time() * 1000
# 计算预期发送时间
expected_time = start_time + (play_position / 1000)
current_time = time.perf_counter()
delay = expected_time - current_time
if delay > 0:
await asyncio.sleep(delay)
await conn.websocket.send(opus_packet)
play_position += frame_duration
async def send_tts_message(conn, state, text=None):
"""发送 TTS 状态消息"""
if text is None and state == "sentence_start":
return
message = {"type": "tts", "state": state, "session_id": conn.session_id}
if text is not None:
message["text"] = textUtils.check_emoji(text)
@@ -81,7 +120,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, _ = conn.tts.audio_to_opus_data(stop_tts_notify_voice)
audios = audio_to_data(stop_tts_notify_voice, is_opus=True)
await sendAudio(conn, audios)
# 清除服务端讲话状态
conn.clearSpeakStatus()
@@ -91,18 +130,17 @@ async def send_tts_message(conn, state, text=None):
async def send_stt_message(conn, text):
"""发送 STT 状态消息"""
end_prompt_str = conn.config.get("end_prompt", {}).get("prompt")
if end_prompt_str and end_prompt_str == text:
await send_tts_message(conn, "start")
return
"""发送 STT 状态消息"""
# 解析JSON格式，提取实际的用户说话内容
display_text = text
try:
# 尝试解析JSON格式
if text.strip().startswith('{') and text.strip().endswith('}'):
if text.strip().startswith("{") and text.strip().endswith("}"):
parsed_data = json.loads(text)
if isinstance(parsed_data, dict) and "content" in parsed_data:
# 如果是包含说话人信息的JSON格式，只显示content部分
main/xiaozhi-server/core/providers/asr/base.py
+5 -9
View File
@@ -1,18 +1,18 @@
import os
import io
import wave
import uuid
import json
import time
import queue
import asyncio
import traceback
import threading
import opuslib_next
import json
import io
import time
import concurrent.futures
from abc import ABC, abstractmethod
from config.logger import setup_logging
from typing import Optional, Tuple, List, Dict, Any
from typing import Optional, Tuple, List
from core.handle.receiveAudioHandle import startToChat
from core.handle.reportHandle import enqueue_asr_report
from core.utils.util import remove_punctuation_and_length
@@ -87,11 +87,9 @@ class ASRProviderBase(ABC):
# 预先准备WAV数据
wav_data = None
# 使用连接的声纹识别提供者
if conn.voiceprint_provider and combined_pcm_data:
wav_data = self._pcm_to_wav(combined_pcm_data)
# 定义ASR任务
def run_asr():
start_time = time.monotonic()
@@ -132,8 +130,6 @@ class ASRProviderBase(ABC):
return None
# 使用线程池执行器并行运行
parallel_start_time = time.monotonic()
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
@@ -151,7 +147,7 @@ class ASRProviderBase(ABC):
# 处理结果
raw_text, file_path = results.get("asr", ("", None))
raw_text, _ = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
# 记录识别结果
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
+16 -50
View File
@@ -268,11 +268,7 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
if message.sentence_type == SentenceType.LAST:
try:
logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -422,9 +418,6 @@ class TTSProvider(TTSProviderBase):
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
opus_datas_cache = []
is_first_sentence = True
first_sentence_segment_count = 0 # 添加计数器
try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set():
@@ -445,28 +438,16 @@ class TTSProvider(TTSProviderBase):
self.tts_audio_queue.put(
(SentenceType.FIRST, [], None)
)
elif event_name == "SentenceBegin":
opus_datas_cache = []
elif event_name == "SentenceEnd":
if (
not is_first_sentence
or first_sentence_segment_count > 10
):
# 发送缓存的数据
if self.conn.tts_MessageText:
logger.bind(tag=TAG).info(
f"句子语音生成成功： {self.conn.tts_MessageText}"
)
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText)
)
self.conn.tts_MessageText = None
else:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 第一句话结束后，将标志设置为False
is_first_sentence = False
# 发送缓存的数据
if self.conn.tts_MessageText:
logger.bind(tag=TAG).info(
f"句子语音生成成功： {self.conn.tts_MessageText}"
)
self.tts_audio_queue.put(
(SentenceType.FIRST, [], self.conn.tts_MessageText)
)
self.conn.tts_MessageText = None
elif event_name == "SynthesisCompleted":
logger.bind(tag=TAG).debug(f"会话结束～～")
self._process_before_stop_play_files()
@@ -477,22 +458,7 @@ class TTSProvider(TTSProviderBase):
# 二进制消息（音频数据）
elif isinstance(msg, (bytes, bytearray)):
logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False)
logger.bind(tag=TAG).debug(
f"推送数据到队列里面帧数～～{len(opus_datas)}"
)
if is_first_sentence:
first_sentence_segment_count += 1
if first_sentence_segment_count <= 6:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas, None)
)
else:
opus_datas_cache.extend(opus_datas)
else:
# 后续句子缓存
opus_datas_cache.extend(opus_datas)
self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus)
except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭")
break
@@ -615,11 +581,11 @@ class TTSProvider(TTSProviderBase):
while not synthesis_completed:
msg = await ws.recv()
if isinstance(msg, (bytes, bytearray)):
# 编码为Opus并收集
opus_frames = self.opus_encoder.encode_pcm_to_opus(
msg, False
self.opus_encoder.encode_pcm_to_opus_stream(
msg,
end_of_stream=False,
callback=lambda opus: audio_data.append(opus)
)
audio_data.extend(opus_frames)
elif isinstance(msg, str):
data = json.loads(msg)
header = data.get("header", {})
@@ -650,4 +616,4 @@ class TTSProvider(TTSProviderBase):
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
return []
main/xiaozhi-server/core/providers/tts/base.py
+139 -60
View File
@@ -1,19 +1,22 @@
import os
import re
import queue
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.util import audio_to_data, audio_bytes_to_data
from core.utils.tts import MarkdownCleaner
from core.utils.output_counter import add_device_output
from core.handle.reportHandle import enqueue_tts_report
from core.handle.sendAudioHandle import sendAudioMessage
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
from core.providers.tts.dto.dto import (
TTSMessageDTO,
SentenceType,
@@ -21,8 +24,6 @@ from core.providers.tts.dto.dto import (
InterfaceType,
)
import traceback
TAG = __name__
logger = setup_logging()
@@ -31,7 +32,6 @@ class TTSProviderBase(ABC):
def __init__(self, config, delete_audio_file):
self.interface_type = InterfaceType.NON_STREAM
self.conn = None
self.tts_timeout = 10
self.delete_audio_file = delete_audio_file
self.audio_file_type = "wav"
self.output_file = config.get("output_dir", "tmp/")
@@ -50,11 +50,9 @@ class TTSProviderBase(ABC):
"；",
";",
"：",
"~",
)
self.first_sentence_punctuations = (
"，",
"～",
"~",
"、",
",",
@@ -77,6 +75,75 @@ class TTSProviderBase(ABC):
f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
)
def handle_opus(self, opus_data: bytes):
logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}")
self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None))
def handle_audio_file(self, file_audio: bytes, text):
self.before_stop_play_files.append((file_audio, text))
def to_tts_stream(self, text, opus_handler: Callable[[bytes], None] = None) -> None:
text = MarkdownCleaner.clean_markdown(text)
max_repeat_time = 5
if self.delete_audio_file:
# 需要删除文件的直接转为音频数据
while max_repeat_time > 0:
try:
audio_bytes = asyncio.run(self.text_to_speak(text, None))
if audio_bytes:
self.tts_audio_queue.put((SentenceType.FIRST, None, text))
audio_bytes_to_data_stream(
audio_bytes,
file_type=self.audio_file_type,
is_opus=True,
callback=opus_handler,
)
break
else:
max_repeat_time -= 1
except Exception as e:
logger.bind(tag=TAG).warning(
f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
)
max_repeat_time -= 1
if max_repeat_time > 0:
logger.bind(tag=TAG).info(
f"语音生成成功: {text}，重试{5 - max_repeat_time}次"
)
else:
logger.bind(tag=TAG).error(
f"语音生成失败: {text}，请检查网络或服务是否正常"
)
return None
else:
tmp_file = self.generate_filename()
try:
while not os.path.exists(tmp_file) and max_repeat_time > 0:
try:
asyncio.run(self.text_to_speak(text, tmp_file))
except Exception as e:
logger.bind(tag=TAG).warning(
f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
)
# 未执行成功，删除文件
if os.path.exists(tmp_file):
os.remove(tmp_file)
max_repeat_time -= 1
if max_repeat_time > 0:
logger.bind(tag=TAG).info(
f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次"
)
else:
logger.bind(tag=TAG).error(
f"语音生成失败: {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}")
return None
def to_tts(self, text):
text = MarkdownCleaner.clean_markdown(text)
max_repeat_time = 5
@@ -86,8 +153,12 @@ class TTSProviderBase(ABC):
try:
audio_bytes = asyncio.run(self.text_to_speak(text, None))
if audio_bytes:
audio_datas, _ = audio_bytes_to_data(
audio_bytes, file_type=self.audio_file_type, is_opus=True
audio_datas = []
audio_bytes_to_data_stream(
audio_bytes,
file_type=self.audio_file_type,
is_opus=True,
callback=lambda data: audio_datas.append(data)
)
return audio_datas
else:
@@ -139,13 +210,17 @@ class TTSProviderBase(ABC):
async def text_to_speak(self, text, output_file):
pass
def audio_to_pcm_data(self, audio_file_path):
def audio_to_pcm_data_stream(
self, audio_file_path, callback: Callable[[Any], Any] = None
):
"""音频文件转换为PCM编码"""
return audio_to_data(audio_file_path, is_opus=False)
return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback)
def audio_to_opus_data(self, audio_file_path):
def audio_to_opus_data_stream(
self, audio_file_path, callback: Callable[[Any], Any] = None
):
"""音频文件转换为Opus编码"""
return audio_to_data(audio_file_path, is_opus=True)
return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback)
def tts_one_sentence(
self,
@@ -177,7 +252,6 @@ class TTSProviderBase(ABC):
async def open_audio_channels(self, conn):
self.conn = conn
self.tts_timeout = conn.config.get("tts_timeout", 10)
# tts 消化线程
self.tts_priority_thread = threading.Thread(
target=self.tts_text_priority_thread, daemon=True
@@ -212,30 +286,16 @@ class TTSProviderBase(ABC):
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
if segment_text:
if self.delete_audio_file:
audio_datas = self.to_tts(segment_text)
if audio_datas:
self.tts_audio_queue.put(
(message.sentence_type, audio_datas, segment_text)
)
else:
tts_file = self.to_tts(segment_text)
if tts_file:
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put(
(message.sentence_type, audio_datas, segment_text)
)
self.to_tts_stream(segment_text, opus_handler=self.handle_opus)
elif ContentType.FILE == message.content_type:
self._process_remaining_text()
self._process_remaining_text_stream(opus_handler=self.handle_opus)
tts_file = message.content_file
if tts_file and os.path.exists(tts_file):
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put(
(message.sentence_type, audio_datas, message.content_detail)
self._process_audio_file_stream(
tts_file, callback=self.handle_opus
)
if message.sentence_type == SentenceType.LAST:
self._process_remaining_text()
self._process_remaining_text_stream(opus_handler=self.handle_opus)
self.tts_audio_queue.put(
(message.sentence_type, [], message.content_detail)
)
@@ -249,29 +309,59 @@ class TTSProviderBase(ABC):
continue
def _audio_play_priority_thread(self):
# 需要上报的文本和音频列表
enqueue_text = None
enqueue_audio = None
while not self.conn.stop_event.is_set():
text = None
try:
try:
sentence_type, audio_datas, text = self.tts_audio_queue.get(
timeout=1
timeout=0.1
)
except queue.Empty:
if self.conn.stop_event.is_set():
break
continue
if self.conn.client_abort:
logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
enqueue_text, enqueue_audio = None, []
continue
# 收到下一个文本开始或会话结束时进行上报
if sentence_type is not SentenceType.MIDDLE:
# 重置音频流控状态（新句子开始或者结束）
if hasattr(self.conn, 'audio_flow_control'):
self.conn.audio_flow_control = {
'last_send_time': 0,
'packet_count': 0,
'start_time': time.perf_counter()
}
# 上报TTS数据
if enqueue_text is not None and enqueue_audio is not None:
enqueue_tts_report(self.conn, enqueue_text, enqueue_audio)
enqueue_audio = []
enqueue_text = text
# 收集上报音频数据
if isinstance(audio_datas, bytes) and enqueue_audio is not None:
enqueue_audio.append(audio_datas)
# 发送音频
future = asyncio.run_coroutine_threadsafe(
sendAudioMessage(self.conn, sentence_type, audio_datas, text),
self.conn.loop,
)
future.result()
# 记录输出和报告
if self.conn.max_output_size > 0 and text:
add_device_output(self.conn.headers.get("device-id"), len(text))
enqueue_tts_report(self.conn, text, audio_datas)
except Exception as e:
logger.bind(tag=TAG).error(
f"audio_play_priority priority_thread: {text} {e}"
)
logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}")
async def start_session(self, session_id):
pass
@@ -323,22 +413,21 @@ class TTSProviderBase(ABC):
else:
return None
def _process_audio_file(self, tts_file):
def _process_audio_file_stream(
self, tts_file, callback: Callable[[Any], Any]
) -> None:
"""处理音频文件并转换为指定格式
Args:
tts_file: 音频文件路径
content_detail: 内容详情
Returns:
tuple: (sentence_type, audio_datas, content_detail)
callback: 文件处理函数
"""
if tts_file.endswith(".p3"):
audio_datas, _ = p3.decode_opus_from_file(tts_file)
p3.decode_opus_from_file_stream(tts_file, callback=callback)
elif self.conn.audio_format == "pcm":
audio_datas, _ = self.audio_to_pcm_data(tts_file)
self.audio_to_pcm_data_stream(tts_file, callback=callback)
else:
audio_datas, _ = self.audio_to_opus_data(tts_file)
self.audio_to_opus_data_stream(tts_file, callback=callback)
if (
self.delete_audio_file
@@ -347,7 +436,6 @@ class TTSProviderBase(ABC):
and tts_file.startswith(self.output_file)
):
os.remove(tts_file)
return audio_datas
def _process_before_stop_play_files(self):
for audio_datas, text in self.before_stop_play_files:
@@ -355,7 +443,9 @@ class TTSProviderBase(ABC):
self.before_stop_play_files.clear()
self.tts_audio_queue.put((SentenceType.LAST, [], None))
def _process_remaining_text(self):
def _process_remaining_text_stream(
self, opus_handler: Callable[[bytes], None] = None
):
"""处理剩余的文本并生成语音
Returns:
@@ -366,18 +456,7 @@ class TTSProviderBase(ABC):
if remaining_text:
segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
if segment_text:
if self.delete_audio_file:
audio_datas = self.to_tts(segment_text)
if audio_datas:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, audio_datas, segment_text)
)
else:
tts_file = self.to_tts(segment_text)
audio_datas = self._process_audio_file(tts_file)
self.tts_audio_queue.put(
(SentenceType.MIDDLE, audio_datas, segment_text)
)
self.to_tts_stream(segment_text, opus_handler=opus_handler)
self.processed_chars += len(full_text)
return True
return False
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+6 -39
View File
@@ -4,6 +4,7 @@ import json
import queue
import asyncio
import traceback
from typing import Callable, Any
import websockets
from core.utils.tts import MarkdownCleaner
from config.logger import setup_logging
@@ -266,11 +267,7 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
if message.sentence_type == SentenceType.LAST:
try:
logger.bind(tag=TAG).info("开始结束TTS会话...")
@@ -428,9 +425,6 @@ class TTSProvider(TTSProviderBase):
async def _start_monitor_tts_response(self):
"""监听TTS响应"""
opus_datas_cache = []
is_first_sentence = True
first_sentence_segment_count = 0 # 添加计数器
try:
session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set():
@@ -451,37 +445,14 @@ class TTSProvider(TTSProviderBase):
self.tts_audio_queue.put(
(SentenceType.FIRST, [], self.tts_text)
)
opus_datas_cache = []
first_sentence_segment_count = 0 # 重置计数器
elif (
res.optional.event == EVENT_TTSResponse
and res.header.message_type == AUDIO_ONLY_RESPONSE
):
logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
logger.bind(tag=TAG).debug(
f"推送数据到队列里面帧数～～{len(opus_datas)}"
)
if is_first_sentence:
first_sentence_segment_count += 1
if first_sentence_segment_count <= 6:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas, None)
)
else:
opus_datas_cache.extend(opus_datas)
else:
# 后续句子缓存
opus_datas_cache.extend(opus_datas)
self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus)
elif res.optional.event == EVENT_TTSSentenceEnd:
logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
if not is_first_sentence or first_sentence_segment_count > 10:
# 发送缓存的数据
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 第一句话结束后，将标志设置为False
is_first_sentence = False
elif res.optional.event == EVENT_SessionFinished:
logger.bind(tag=TAG).debug(f"会话结束～～")
self._process_before_stop_play_files()
@@ -655,16 +626,13 @@ class TTSProvider(TTSProviderBase):
)
)
def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False):
opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end)
return opus_datas
def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None):
return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback)
def to_tts(self, text: str) -> list:
"""非流式生成音频数据，用于生成音频及测试场景
Args:
text: 要转换的文本
Returns:
list: 音频数据列表
"""
@@ -741,8 +709,7 @@ class TTSProvider(TTSProviderBase):
res.optional.event == EVENT_TTSResponse
and res.header.message_type == AUDIO_ONLY_RESPONSE
):
opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
audio_data.extend(opus_datas)
self.wav_to_opus_data_audio_raw_stream(res.payload, callback=lambda opus_frame: audio_data.append(opus_frame))
elif res.optional.event == EVENT_SessionFinished:
break
main/xiaozhi-server/core/providers/tts/index_stream.py
+23 -52
View File
@@ -1,10 +1,10 @@
import os
import queue
import asyncio
import traceback
import aiohttp
import requests
import time
import queue
import aiohttp
import asyncio
import requests
import traceback
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -27,15 +27,13 @@ class TTSProvider(TTSProviderBase):
self.api_url = config.get("api_url", "http://8.138.114.124:11996/tts")
self.audio_format = "pcm"
self.before_stop_play_files = []
self.segment_count = 0
# 创建Opus编码器 需注意接口返回的采样率为24000
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=24000, channels=1, frame_size_ms=60
)
# 文本缓冲区和PCM缓冲区
self.text_buffer = ""
# PCM缓冲区
self.pcm_buffer = bytearray()
def tts_text_priority_thread(self):
@@ -48,7 +46,6 @@ class TTSProvider(TTSProviderBase):
self.tts_stop_request = False
self.processed_chars = 0
self.tts_text_buff = []
self.segment_count = 0
self.before_stop_play_files.clear()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
@@ -62,14 +59,11 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
if message.sentence_type == SentenceType.LAST:
# 处理剩余的文本
self._process_remaining_text(True)
self._process_remaining_text_stream(True)
except queue.Empty:
continue
@@ -78,7 +72,7 @@ class TTSProvider(TTSProviderBase):
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
def _process_remaining_text(self, is_last=False):
def _process_remaining_text_stream(self, is_last=False):
"""处理剩余的文本并生成语音
Returns:
bool: 是否成功处理了文本
@@ -143,8 +137,6 @@ class TTSProvider(TTSProviderBase):
return
self.pcm_buffer.clear()
opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text))
# 处理音频流数据
@@ -158,41 +150,22 @@ class TTSProvider(TTSProviderBase):
while len(self.pcm_buffer) >= frame_bytes:
frame = bytes(self.pcm_buffer[:frame_bytes])
del self.pcm_buffer[:frame_bytes]
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=False
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=False,
callback=self.handle_opus
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
opus_datas_cache.extend(opus)
# flush 剩余不足一帧的数据
if self.pcm_buffer:
opus = self.opus_encoder.encode_pcm_to_opus(
bytes(self.pcm_buffer), end_of_stream=True
self.opus_encoder.encode_pcm_to_opus_stream(
bytes(self.pcm_buffer),
end_of_stream=True,
callback=self.handle_opus
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
# 直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
# 后续片段缓存
opus_datas_cache.extend(opus)
self.pcm_buffer.clear()
# 如果不是前10个片段，发送缓存的数据
if self.segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 如果是最后一段，输出音频获取完毕
if is_last:
self._process_before_stop_play_files()
@@ -209,10 +182,8 @@ class TTSProvider(TTSProviderBase):
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
@@ -251,14 +222,14 @@ class TTSProvider(TTSProviderBase):
# 最后一帧可能不足，用0填充
frame = frame + b"\x00" * (frame_bytes - len(frame))
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=(i + frame_bytes >= len(pcm_data)),
callback=lambda opus: opus_datas.append(opus)
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
return []
main/xiaozhi-server/core/providers/tts/linkerai.py
+21 -62
View File
@@ -1,10 +1,10 @@
import os
import queue
import asyncio
import traceback
import aiohttp
import requests
import time
import queue
import aiohttp
import asyncio
import requests
import traceback
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -24,23 +24,15 @@ class TTSProvider(TTSProviderBase):
self.api_url = config.get("api_url")
self.audio_format = "pcm"
self.before_stop_play_files = []
self.segment_count = 0 # 添加片段计数器
# 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60
)
# 添加文本缓冲区
self.text_buffer = ""
# PCM缓冲区
self.pcm_buffer = bytearray()
###################################################################################
# linkerai单流式TTS重写父类的方法--开始
###################################################################################
def tts_text_priority_thread(self):
"""流式文本处理线程"""
while not self.conn.stop_event.is_set():
@@ -51,7 +43,6 @@ class TTSProvider(TTSProviderBase):
self.tts_stop_request = False
self.processed_chars = 0
self.tts_text_buff = []
self.segment_count = 0
self.before_stop_play_files.clear()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
@@ -65,14 +56,10 @@ class TTSProvider(TTSProviderBase):
)
if message.content_file and os.path.exists(message.content_file):
# 先处理文件音频数据
file_audio = self._process_audio_file(message.content_file)
self.before_stop_play_files.append(
(file_audio, message.content_detail)
)
self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail))
if message.sentence_type == SentenceType.LAST:
# 处理剩余的文本
self._process_remaining_text(True)
self._process_remaining_text_stream(True)
except queue.Empty:
continue
@@ -81,7 +68,7 @@ class TTSProvider(TTSProviderBase):
f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
)
def _process_remaining_text(self, is_last=False):
def _process_remaining_text_stream(self, is_last=False):
"""处理剩余的文本并生成语音
Returns:
@@ -124,10 +111,6 @@ class TTSProvider(TTSProviderBase):
finally:
return None
###################################################################################
# linkerai单流式TTS重写父类的方法--结束
###################################################################################
async def text_to_speak(self, text, is_last):
"""流式处理TTS音频，每句只推送一次音频列表"""
await self._tts_request(text, is_last)
@@ -176,8 +159,6 @@ class TTSProvider(TTSProviderBase):
return
self.pcm_buffer.clear()
opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text))
# 兼容 iter_chunked / iter_chunks / iter_any
@@ -194,41 +175,21 @@ class TTSProvider(TTSProviderBase):
frame = bytes(self.pcm_buffer[:frame_bytes])
del self.pcm_buffer[:frame_bytes]
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=False
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=False,
callback=self.handle_opus
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
opus_datas_cache.extend(opus)
# flush 剩余不足一帧的数据
if self.pcm_buffer:
opus = self.opus_encoder.encode_pcm_to_opus(
bytes(self.pcm_buffer), end_of_stream=True
self.opus_encoder.encode_pcm_to_opus_stream(
bytes(self.pcm_buffer),
end_of_stream=True,
callback=self.handle_opus
)
if opus:
if self.segment_count < 10: # 前10个片段直接发送
# 直接发送
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus, None)
)
self.segment_count += 1
else:
# 后续片段缓存
opus_datas_cache.extend(opus)
self.pcm_buffer.clear()
# 如果不是前10个片段，发送缓存的数据
if self.segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None)
)
# 如果是最后一段，输出音频获取完毕
if is_last:
self._process_before_stop_play_files()
@@ -239,10 +200,8 @@ class TTSProvider(TTSProviderBase):
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
@@ -295,14 +254,14 @@ class TTSProvider(TTSProviderBase):
# 最后一帧可能不足，用0填充
frame = frame + b"\x00" * (frame_bytes - len(frame))
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=(i + frame_bytes >= len(pcm_data)),
callback=lambda opus: opus_datas.append(opus)
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
return []
main/xiaozhi-server/core/providers/tts/paddle_speech.py
+33 -44
View File
@@ -1,13 +1,15 @@
import asyncio
import json
import base64
import aiohttp
import numpy as np
import io
import wave
import json
import base64
import asyncio
import websockets
from core.providers.tts.base import TTSProviderBase
import numpy as np
from datetime import datetime
from config.logger import setup_logging
from core.providers.tts.base import TTSProviderBase
TAG = __name__
logger = setup_logging()
@@ -18,11 +20,12 @@ class TTSProvider(TTSProviderBase):
super().__init__(config, delete_audio_file)
self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming")
self.protocol = config.get("protocol", "websocket")
if config.get("private_voice"):
self.spk_id = int(config.get("private_voice"))
else:
self.spk_id = int(config.get("spk_id", "0"))
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
@@ -32,7 +35,21 @@ class TTSProvider(TTSProviderBase):
volume = config.get("volume", 1.0)
self.volume = float(volume) if volume else 1.0
self.save_path = config.get("save_path", "./streaming_tts.wav")
self.delete_audio_file = config.get("delete_audio", True)
if not self.delete_audio_file:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
save_path = config.get("save_path")
if save_path:
if not save_path.endswith('.wav'):
save_path = f"{save_path}_{timestamp}.wav"
else:
other_path = save_path[:-4]
save_path = f"{other_path}_{timestamp}.wav"
self.save_path = save_path
else:
self.save_path = f"./streaming_tts_{timestamp}.wav"
else:
self.save_path = None
async def pcm_to_wav(self, pcm_data: bytes, sample_rate: int = 24000, num_channels: int = 1,
bits_per_sample: int = 16) -> bytes:
@@ -58,43 +75,9 @@ class TTSProvider(TTSProviderBase):
async def text_to_speak(self, text, output_file):
if self.protocol == "websocket":
return await self.text_streaming(text, output_file)
elif self.protocol == "http":
return await self.text(text, output_file)
else:
raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.")
async def text(self, text, output_file):
request_json = {
"text": text,
"spk_id": self.spk_id,
"speed": self.speed,
"volume": self.volume,
"sample_rate": self.sample_rate,
"save_path": self.save_path
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(self.url, json=request_json) as resp:
if resp.status == 200:
resp_json = await resp.json()
if resp_json.get("success"):
data = resp_json["result"]
audio_bytes = base64.b64decode(data["audio"])
if output_file:
with open(output_file, "wb") as file_to_save:
file_to_save.write(audio_bytes)
else:
return audio_bytes
else:
raise Exception(
f"Error: {resp_json.get('message', 'Unknown error')} while processing text: {text}")
else:
raise Exception(
f"HTTP Error: {resp.status} - {await resp.text()} while processing text: {text}")
except Exception as e:
raise Exception(f"Error during TTS HTTP request: {e} while processing text: {text}")
async def text_streaming(self, text, output_file):
try:
# 使用 websockets 异步连接到 WebSocket 服务器
@@ -151,6 +134,12 @@ class TTSProvider(TTSProviderBase):
# 接收结束响应避免服务抛出异常
await ws.recv()
# 根据配置决定是否保存文件
if not self.delete_audio_file and self.save_path:
with open(self.save_path, "wb") as f:
f.write(wav_data)
logger.bind(tag=TAG).info(f"音频文件已保存到: {self.save_path}")
# 返回或保存音频数据
if output_file:
with open(output_file, "wb") as file_to_save:
@@ -159,4 +148,4 @@ class TTSProvider(TTSProviderBase):
return wav_data
except Exception as e:
raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")
raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}")
main/xiaozhi-server/core/providers/vad/silero.py
+2 -4
View File
@@ -34,7 +34,7 @@ class VADProvider(VADProviderBase):
)
# 至少要多少帧才算有语音
self.frame_window_threshold = 1
self.frame_window_threshold = 3
def is_vad(self, conn, opus_packet):
try:
@@ -70,9 +70,7 @@ class VADProvider(VADProviderBase):
# 更新滑动窗口
conn.client_voice_window.append(is_voice)
client_have_voice = (
conn.client_voice_window.count(True) >= self.frame_window_threshold
)
client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold)
# 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话
if conn.client_have_voice and not client_have_voice:
main/xiaozhi-server/core/utils/util.py
+74 -32
View File
@@ -1,16 +1,17 @@
import json
import socket
import subprocess
import re
import os
import json
import copy
import wave
import socket
import requests
import subprocess
import numpy as np
import opuslib_next
from io import BytesIO
from core.utils import p3
import numpy as np
import requests
import opuslib_next
from pydub import AudioSegment
import copy
from typing import Callable, Any
TAG = __name__
emoji_map = {
@@ -211,7 +212,7 @@ def extract_json_from_string(input_string):
return None
def audio_to_data(audio_file_path, is_opus=True):
def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any], Any]=None) -> None:
# 获取文件后缀名
file_type = os.path.splitext(audio_file_path)[1]
if file_type:
@@ -224,33 +225,32 @@ def audio_to_data(audio_file_path, is_opus=True):
# 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
# 音频时长(秒)
duration = len(audio) / 1000.0
# 获取原始PCM数据（16位小端）
raw_data = audio.raw_data
pcm_to_data_stream(raw_data, is_opus, callback)
def audio_to_data(audio_file_path: str, is_opus: bool = True) -> list[bytes]:
"""
将音频文件转换为Opus/PCM编码的帧列表
Args:
audio_file_path: 音频文件路径
is_opus: 是否进行Opus编码
"""
# 获取文件后缀名
file_type = os.path.splitext(audio_file_path)[1]
if file_type:
file_type = file_type.lstrip(".")
# 读取音频文件，-nostdin 参数：不要从标准输入读取数据，否则FFmpeg会阻塞
audio = AudioSegment.from_file(
audio_file_path, format=file_type, parameters=["-nostdin"]
)
# 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配）
audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2)
# 获取原始PCM数据（16位小端）
raw_data = audio.raw_data
return pcm_to_data(raw_data, is_opus), duration
def audio_bytes_to_data(audio_bytes, file_type, is_opus=True):
"""
直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3
"""
if file_type == "p3":
# 直接用p3解码
return p3.decode_opus_from_bytes(audio_bytes)
else:
# 其他格式用pydub
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)
duration = len(audio) / 1000.0
raw_data = audio.raw_data
return pcm_to_data(raw_data, is_opus), duration
def pcm_to_data(raw_data, is_opus=True):
# 初始化Opus编码器
encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO)
@@ -280,6 +280,49 @@ def pcm_to_data(raw_data, is_opus=True):
return datas
def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None:
"""
直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3
"""
if file_type == "p3":
# 直接用p3解码
return p3.decode_opus_from_bytes_stream(audio_bytes, callback)
else:
# 其他格式用pydub
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)
raw_data = audio.raw_data
pcm_to_data_stream(raw_data, is_opus, callback)
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)
# 编码参数
frame_duration = 60 # 60ms per frame
frame_size = int(16000 * frame_duration / 1000) # 960 samples/frame
# 按帧处理所有音频数据（包括最后一帧可能补零）
for i in range(0, len(raw_data), frame_size * 2): # 16bit=2bytes/sample
# 获取当前帧的二进制数据
chunk = raw_data[i : i + frame_size * 2]
# 如果最后一帧不足，补零
if len(chunk) < frame_size * 2:
chunk += b"\x00" * (frame_size * 2 - len(chunk))
if is_opus:
# 转换为numpy数组处理
np_frame = np.frombuffer(chunk, dtype=np.int16)
# 编码Opus数据
frame_data = encoder.encode(np_frame.tobytes(), frame_size)
callback(frame_data)
else:
frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk)
callback(frame_data)
def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
"""
@@ -307,7 +350,6 @@ def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
wf.writeframes(pcm_bytes)
return wav_buffer.getvalue()
def check_vad_update(before_config, new_config):
if (
new_config.get("selected_module") is None
main/xiaozhi-server/core/utils/wakeup_word.py
+1 -1
View File
@@ -137,4 +137,4 @@ class WakeupWordsConfig:
return file_path
except Exception as e:
print(f"生成音频文件路径失败: {e}")
raise
raise