From 8b81b918cc33b20ff9d5266096bf87cd9e5c5091 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 13:49:12 +0800 Subject: [PATCH 01/11] =?UTF-8?q?=E7=A7=BB=E9=99=A4=E6=9C=AA=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E7=9A=84total=5Fduration=E8=BF=94=E5=9B=9E=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/handle/helloHandle.py | 2 +- .../core/handle/receiveAudioHandle.py | 6 +++--- main/xiaozhi-server/core/providers/tts/base.py | 8 ++++---- main/xiaozhi-server/core/utils/p3.py | 17 ++--------------- main/xiaozhi-server/core/utils/util.py | 8 ++------ 5 files changed, 12 insertions(+), 29 deletions(-) diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index e4e836b4..c4f55841 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -83,7 +83,7 @@ async def checkWakeupWords(conn, text): # 播放唤醒词回复 conn.client_abort = False - opus_packets, _ = audio_to_data(response.get("file_path")) + 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")) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index e6f39632..174c8b2b 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -140,7 +140,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,7 +148,7 @@ async def check_bind_device(conn): try: digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" - num_packets, _ = audio_to_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}") @@ -158,5 +158,5 @@ async def check_bind_device(conn): 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)) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 40570326..87bb64f6 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -86,7 +86,7 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - audio_datas, _ = audio_bytes_to_data( + audio_datas = audio_bytes_to_data( audio_bytes, file_type=self.audio_file_type, is_opus=True ) return audio_datas @@ -334,11 +334,11 @@ class TTSProviderBase(ABC): tuple: (sentence_type, audio_datas, content_detail) """ if tts_file.endswith(".p3"): - audio_datas, _ = p3.decode_opus_from_file(tts_file) + audio_datas = p3.decode_opus_from_file(tts_file) elif self.conn.audio_format == "pcm": - audio_datas, _ = self.audio_to_pcm_data(tts_file) + audio_datas = self.audio_to_pcm_data(tts_file) else: - audio_datas, _ = self.audio_to_opus_data(tts_file) + audio_datas = self.audio_to_opus_data(tts_file) if ( self.delete_audio_file diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index c75b968e..f196da0b 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -5,10 +5,6 @@ def decode_opus_from_file(input_file): 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 """ opus_datas = [] - total_frames = 0 - sample_rate = 16000 # 文件采样率 - frame_duration_ms = 60 # 帧时长 - frame_size = int(sample_rate * frame_duration_ms / 1000) with open(input_file, 'rb') as f: while True: @@ -26,11 +22,8 @@ def decode_opus_from_file(input_file): raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.") opus_datas.append(opus_data) - total_frames += 1 - # 计算总时长 - total_duration = (total_frames * frame_duration_ms) / 1000.0 - return opus_datas, total_duration + return opus_datas def decode_opus_from_bytes(input_bytes): """ @@ -38,10 +31,6 @@ def decode_opus_from_bytes(input_bytes): """ import io opus_datas = [] - total_frames = 0 - sample_rate = 16000 # 文件采样率 - frame_duration_ms = 60 # 帧时长 - frame_size = int(sample_rate * frame_duration_ms / 1000) f = io.BytesIO(input_bytes) while True: @@ -53,7 +42,5 @@ def decode_opus_from_bytes(input_bytes): if len(opus_data) != data_len: raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") opus_datas.append(opus_data) - total_frames += 1 - total_duration = (total_frames * frame_duration_ms) / 1000.0 - return opus_datas, total_duration \ No newline at end of file + return opus_datas \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index ce6f20ea..641e2655 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -224,12 +224,9 @@ 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 - return pcm_to_data(raw_data, is_opus), duration + return pcm_to_data(raw_data, is_opus) def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): @@ -245,9 +242,8 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): 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 + return pcm_to_data(raw_data, is_opus) def pcm_to_data(raw_data, is_opus=True): From 75f077324704ff5ccf89fd882f1bb109239ec291 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 14:00:41 +0800 Subject: [PATCH 02/11] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86Opus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/utils/opus_encoder_utils.py | 47 ++++++++++++++++++- 1 file changed, 46 insertions(+), 1 deletion(-) diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index 23d26c39..f9154da1 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -7,7 +7,7 @@ import logging import traceback import numpy as np -from typing import List, Optional +from typing import List, Optional, Callable, Any from opuslib_next import Encoder from opuslib_next import constants @@ -103,6 +103,51 @@ class OpusEncoderUtils: return opus_packets + def encode_pcm_to_opus_stream(self, pcm_data: bytes, end_of_stream: bool, callback: Callable[[Any], Any]): + """ + 将PCM数据编码为Opus格式，以流式方式进行处理 + + Args: + pcm_data: PCM字节数据 + end_of_stream: 是否为流的结束, + callback: opus处理方法 + + Returns: + Opus数据包列表 + """ + # 将字节数据转换为short数组 + new_samples = self._convert_bytes_to_shorts(pcm_data) + + # 校验PCM数据 + self._validate_pcm_data(new_samples) + + # 将新数据追加到缓冲区 + self.buffer = np.append(self.buffer, new_samples) + + offset = 0 + + # 处理所有完整帧 + while offset <= len(self.buffer) - self.total_frame_size: + frame = self.buffer[offset : offset + self.total_frame_size] + output = self._encode(frame) + if output: + callback(output) + offset += self.total_frame_size + + # 保留未处理的样本 + self.buffer = self.buffer[offset:] + + # 流结束时处理剩余数据 + if end_of_stream and len(self.buffer) > 0: + # 创建最后一帧并用0填充 + last_frame = np.zeros(self.total_frame_size, dtype=np.int16) + last_frame[: len(self.buffer)] = self.buffer + + output = self._encode(last_frame) + if output: + callback(output) + self.buffer = np.array([], dtype=np.int16) + def _encode(self, frame: np.ndarray) -> Optional[bytes]: """编码一帧音频数据""" try: From faf2890695d1adbfedf2b0e0bfa7b8be0fbfc24b Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Mon, 4 Aug 2025 17:04:46 +0800 Subject: [PATCH 03/11] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/utils/p3.py | 44 ++++++++++++++++++++++++++-- 1 file changed, 41 insertions(+), 3 deletions(-) diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index f196da0b..f2ee3239 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -1,8 +1,9 @@ import struct +from typing import Callable, Any def decode_opus_from_file(input_file): """ - 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 + 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表。 """ opus_datas = [] @@ -25,9 +26,30 @@ def decode_opus_from_file(input_file): return opus_datas +def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): + """ + 从p3文件中解码 Opus 数据，由 callback 处理 Opus 数据包。 + """ + with open(input_file, 'rb') as f: + while True: + # 读取头部（4字节）：[1字节类型，1字节保留，2字节长度] + header = f.read(4) + if not header: + break + + # 解包头部信息 + _, _, data_len = struct.unpack('>BBH', header) + + # 根据头部指定的长度读取 Opus 数据 + opus_data = f.read(data_len) + if len(opus_data) != data_len: + raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.") + + callback(opus_data) + def decode_opus_from_bytes(input_bytes): """ - 从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表以及总时长。 + 从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表。 """ import io opus_datas = [] @@ -43,4 +65,20 @@ def decode_opus_from_bytes(input_bytes): raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") opus_datas.append(opus_data) - return opus_datas \ No newline at end of file + return opus_datas + +def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]): + """ + 从p3二进制数据中解码 Opus 数据，由 callback 处理 Opus 数据包。 + """ + + f = io.BytesIO(input_bytes) + while True: + header = f.read(4) + if not header: + break + _, _, data_len = struct.unpack('>BBH', header) + opus_data = f.read(data_len) + if len(opus_data) != data_len: + raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") + callback(opus_data) From ac1e19f621be1cb7a83a580780c6bb44b249d13c Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 14:10:55 +0800 Subject: [PATCH 04/11] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86opus?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 61 +++--- .../core/providers/tts/aliyun_stream.py | 38 ++-- .../xiaozhi-server/core/providers/tts/base.py | 118 +++++++++-- .../core/utils/audio_flow_control.py | 200 ++++++++++++++++++ 4 files changed, 352 insertions(+), 65 deletions(-) create mode 100644 main/xiaozhi-server/core/utils/audio_flow_control.py diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 0ca36e1e..09c54b2d 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -31,44 +31,51 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 async def sendAudio(conn, audios, pre_buffer=True): - if audios is None or len(audios) == 0: + if audios is None: 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:] + # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出 + if isinstance(audios ,bytes): + await conn.websocket.send(audios) else: - remaining_audios = audios + 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 conn.client_abort: - break + # 播放剩余音频帧 + for opus_packet in remaining_audios: + if conn.client_abort: + break - # 重置没有声音的状态 - conn.last_activity_time = time.time() * 1000 + # 重置没有声音的状态 + 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) + # 计算预期发送时间 + 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) + await conn.websocket.send(opus_packet) - play_position += frame_duration + play_position += frame_duration async def send_tts_message(conn, state, text=None): """发送 TTS 状态消息""" + if text is None: + return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: message["text"] = textUtils.check_emoji(text) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 0bfb367b..d7f16140 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -422,7 +422,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: @@ -458,13 +457,9 @@ class TTSProvider(TTSProviderBase): f"句子语音生成成功： {self.conn.tts_MessageText}" ) self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText) + (SentenceType.MIDDLE, [], 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 elif event_name == "SynthesisCompleted": @@ -477,22 +472,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 @@ -616,10 +596,10 @@ class TTSProvider(TTSProviderBase): 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, False, self.handle_opus ) - audio_data.extend(opus_frames) + # audio_data.extend(opus_frames) elif isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) @@ -651,3 +631,11 @@ class TTSProvider(TTSProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") return [] + + def handle_opus(self, opus_data: bytes): + logger.bind(tag=TAG).debug( + f"推送数据到队列里面帧数～～" + ) + self.tts_audio_queue.put( + (SentenceType.MIDDLE, opus_data, None) + ) \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 87bb64f6..55ce989b 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -5,10 +5,11 @@ import uuid import asyncio import threading from core.utils import p3 -from datetime import datetime +import time from core.utils import textUtils from abc import ABC, abstractmethod from config.logger import setup_logging +from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption 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 @@ -70,6 +71,8 @@ class TTSProviderBase(ABC): self.tts_stop_request = False self.processed_chars = 0 self.is_first_sentence = True + self.flow_controller = FlowControlConfig.create_flow_controller() + self.flow_control_enabled = config.get("enable_flow_control", True) def generate_filename(self, extension=".wav"): return os.path.join( @@ -253,26 +256,115 @@ class TTSProviderBase(ABC): text = None try: try: - sentence_type, audio_datas, text = self.tts_audio_queue.get( - timeout=1 - ) + sentence_type, audio_datas, text = self.tts_audio_queue.get(timeout=1) except queue.Empty: if self.conn.stop_event.is_set(): break continue - 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) + + # 如果启用了流控 + if self.flow_control_enabled: + # 计算音频数据的帧数 + if isinstance(audio_datas, bytes): + frame_count = 1 # 单个字节流作为一帧 + elif isinstance(audio_datas, (list, tuple)): + frame_count = len(audio_datas) + else: + frame_count = 0 + + # 流控检查 + if frame_count > 0: + max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME + wait_start_time = time.time() + retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL + + while not self.flow_controller.can_send_frames(frame_count): + # 检查是否超时或需要停止 + if (time.time() - wait_start_time > max_wait_time or + self.conn.stop_event.is_set() or + self.conn.client_abort): + logger.bind(tag=TAG).warning( + f"流控等待超时或收到停止信号，跳过音频发送: {text}" + ) + break + + # 短暂等待后重试 + time.sleep(retry_interval) + + # 更新设备消费估计（这里假设设备以恒定速率消费） + # 实际应用中可能需要从设备端获取真实的消费情况 + estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) + self.flow_controller.update_device_consumption(estimated_consumption) + else: + # 可以发送，记录发送的帧数 + self.flow_controller.record_sent_frames(frame_count) + + # 发送音频 + future = asyncio.run_coroutine_threadsafe( + self._send_audio_with_flow_control(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) + + # 输出流控状态（调试用） + if frame_count > 10: # 只在较大的音频块时输出状态 + status = self.flow_controller.get_status() + logger.bind(tag=TAG).debug( + f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + f"可用令牌={status['available_tokens']}, 发送文本: {text[:20]}..." + ) + else: + # 没有音频数据，直接发送 + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self.conn.loop, + ) + future.result() + else: + # 未启用流控，直接发送 + 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}" + f"audio_play_priority_thread: {text} {e}" ) + async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text): + """带流控的音频发送方法""" + await sendAudioMessage(self.conn, sentence_type, audio_datas, text) + + # 模拟设备消费（实际应用中应该从设备获取反馈） + if isinstance(audio_datas, bytes): + frame_count = 1 + elif isinstance(audio_datas, (list, tuple)): + frame_count = len(audio_datas) + else: + frame_count = 0 + + if frame_count > 0: + asyncio.create_task(simulate_device_consumption(self.flow_controller, frame_count)) + + # 在类中添加流控制器重置方法 + def reset_flow_controller(self): + """重置流控制器状态，通常在新会话开始时调用""" + if hasattr(self, 'flow_controller'): + self.flow_controller.reset() + logger.bind(tag=TAG).info("流控制器状态已重置") + async def start_session(self, session_id): pass diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py new file mode 100644 index 00000000..74a5585f --- /dev/null +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -0,0 +1,200 @@ +""" +音频流控模块 +包含令牌桶算法和音频流控制器的实现 +""" + +import asyncio +import time +import threading +from collections import deque +from typing import Optional, Dict, Any + + +class TokenBucket: + """令牌桶实现，用于限流控制""" + + def __init__(self, capacity: int, refill_rate: float, initial_tokens: Optional[int] = None): + """ + 初始化令牌桶 + + Args: + capacity: 桶容量（最大令牌数） + refill_rate: 令牌补充速率（每秒补充的令牌数） + initial_tokens: 初始令牌数，默认为桶容量 + """ + self.capacity = capacity + self.refill_rate = refill_rate + self.tokens = initial_tokens if initial_tokens is not None else capacity + self.last_refill_time = time.time() + self.lock = threading.Lock() + + def get_tokens(self, requested_tokens: int = 1) -> bool: + """ + 获取指定数量的令牌 + + Args: + requested_tokens: 请求的令牌数量 + + Returns: + bool: 是否成功获取到令牌 + """ + with self.lock: + self._refill_tokens() + + if self.tokens >= requested_tokens: + self.tokens -= requested_tokens + return True + else: + return False + + def get_available_tokens(self) -> int: + """获取当前可用令牌数""" + with self.lock: + self._refill_tokens() + return int(self.tokens) + + def _refill_tokens(self): + """内部方法：补充令牌""" + current_time = time.time() + time_passed = current_time - self.last_refill_time + tokens_to_add = time_passed * self.refill_rate + + self.tokens = min(self.capacity, self.tokens + tokens_to_add) + self.last_refill_time = current_time + + +class AudioFlowController: + """音频流控制器，基于令牌桶算法控制音频数据发送""" + + def __init__(self, max_device_buffer: int = 3000, refill_rate: float = 20): + """ + 初始化音频流控制器 + + Args: + max_device_buffer: 设备端最大缓冲区大小（Opus帧数） + refill_rate: 令牌补充速率（每秒允许发送的帧数） + """ + self.max_device_buffer = max_device_buffer + self.token_bucket = TokenBucket( + capacity=max_device_buffer, + refill_rate=refill_rate, + initial_tokens=max_device_buffer // 2 # 初始令牌为容量的一半 + ) + self.sent_frames_count = 0 # 已发送帧数计数 + self.device_consumed_frames = 0 # 设备端已消费帧数 + self.pending_queue = deque() # 等待发送的数据队列 + self._lock = threading.Lock() + + def can_send_frames(self, frame_count: int) -> bool: + """ + 检查是否可以发送指定数量的帧 + + Args: + frame_count: 要发送的帧数 + + Returns: + bool: 是否可以发送 + """ + with self._lock: + # 检查设备端缓冲区是否会溢出 + estimated_device_buffer = self.sent_frames_count - self.device_consumed_frames + if estimated_device_buffer + frame_count > self.max_device_buffer: + return False + + # 检查令牌桶是否有足够令牌 + return self.token_bucket.get_tokens(frame_count) + + def update_device_consumption(self, consumed_frames: int): + """ + 更新设备端消费的帧数 + + Args: + consumed_frames: 设备端消费的帧数 + """ + with self._lock: + self.device_consumed_frames += consumed_frames + + def record_sent_frames(self, frame_count: int): + """ + 记录已发送的帧数 + + Args: + frame_count: 发送的帧数 + """ + with self._lock: + self.sent_frames_count += frame_count + + def get_status(self) -> Dict[str, Any]: + """获取流控状态信息""" + with self._lock: + estimated_buffer = self.sent_frames_count - self.device_consumed_frames + return { + "sent_frames": self.sent_frames_count, + "consumed_frames": self.device_consumed_frames, + "estimated_device_buffer": estimated_buffer, + "available_tokens": self.token_bucket.get_available_tokens(), + "pending_queue_size": len(self.pending_queue), + "buffer_usage_percent": (estimated_buffer / self.max_device_buffer) * 100 + } + + def reset(self): + """重置流控状态""" + with self._lock: + self.sent_frames_count = 0 + self.device_consumed_frames = 0 + self.pending_queue.clear() + # 重新初始化令牌桶 + self.token_bucket = TokenBucket( + capacity=self.max_device_buffer, + refill_rate=self.token_bucket.refill_rate, + initial_tokens=self.max_device_buffer // 2 + ) + + +async def simulate_device_consumption(flow_controller: AudioFlowController, frame_count: int): + """ + 模拟设备消费音频帧的过程 + 实际应用中应该根据设备反馈来更新消费情况 + + Args: + flow_controller: 流控制器实例 + frame_count: 消费的帧数 + """ + # 模拟设备播放延迟（60ms per frame） + await asyncio.sleep(frame_count * 0.06) + flow_controller.update_device_consumption(frame_count) + + +# 流控配置常量 +class FlowControlConfig: + """流控配置常量""" + # Opus 编码参数 + OPUS_FRAME_DURATION_MS = 60 # Opus帧时长（毫秒） + OPUS_FRAMES_PER_SECOND = 1000 / OPUS_FRAME_DURATION_MS # 每秒帧数 + + # 默认流控参数 + DEFAULT_MAX_DEVICE_BUFFER = 1000 # 设备端最大缓冲帧数 + DEFAULT_REFILL_RATE = 20 # 默认令牌补充速率（帧/秒） + DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） + DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） + + # 预缓冲参数 + PRE_BUFFER_FRAMES = 3 # 预缓冲帧数 + + @classmethod + def create_flow_controller(cls, max_buffer: Optional[int] = None, + refill_rate: Optional[float] = None) -> AudioFlowController: + """ + 创建流控制器的工厂方法 + + Args: + max_buffer: 最大缓冲区大小，使用默认值如果为None + refill_rate: 令牌补充速率，使用默认值如果为None + + Returns: + AudioFlowController: 配置好的流控制器实例 + """ + return AudioFlowController( + max_device_buffer=max_buffer or cls.DEFAULT_MAX_DEVICE_BUFFER, + refill_rate=refill_rate or cls.DEFAULT_REFILL_RATE + ) \ No newline at end of file From 1b6cd01eafb59cb495bcb4a0b642f859dfcd83c1 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 14:53:52 +0800 Subject: [PATCH 05/11] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=96=B9=E6=B3=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/utils/util.py | 45 ++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 641e2655..088d8962 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -5,6 +5,8 @@ import re import os import wave from io import BytesIO +from typing import Callable, Any + from core.utils import p3 import numpy as np import requests @@ -245,6 +247,23 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): raw_data = audio.raw_data return pcm_to_data(raw_data, is_opus) +def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]): + """ + 直接用音频二进制数据转为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) + return None + def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 @@ -276,6 +295,32 @@ def pcm_to_data(raw_data, is_opus=True): return datas +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): """ From 095fe72db083cd0e4ce0987fbfdf5bda52e38bbd Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Tue, 5 Aug 2025 18:35:37 +0800 Subject: [PATCH 06/11] =?UTF-8?q?=E5=BE=85=E5=AE=8C=E5=96=84=E6=B5=81?= =?UTF-8?q?=E6=8E=A7=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 6 +- .../core/providers/tts/aliyun_stream.py | 10 +-- .../xiaozhi-server/core/providers/tts/base.py | 79 ++++++++++--------- .../core/utils/audio_flow_control.py | 4 +- main/xiaozhi-server/core/utils/util.py | 20 ++++- 5 files changed, 65 insertions(+), 54 deletions(-) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 09c54b2d..bf155ee9 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -16,8 +16,10 @@ async def sendAudioMessage(conn, sentenceType, audios, text): 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) @@ -74,7 +76,7 @@ async def sendAudio(conn, audios, pre_buffer=True): async def send_tts_message(conn, state, text=None): """发送 TTS 状态消息""" - if text is None: + if text is None and state == "sentence_start": return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index d7f16140..5b3938e9 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -630,12 +630,4 @@ class TTSProvider(TTSProviderBase): return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] - - def handle_opus(self, opus_data: bytes): - logger.bind(tag=TAG).debug( - f"推送数据到队列里面帧数～～" - ) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_data, None) - ) \ No newline at end of file + return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 55ce989b..d105a892 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -4,13 +4,15 @@ import queue import uuid import asyncio import threading +from typing import Callable, Any + from core.utils import p3 import time from core.utils import textUtils from abc import ABC, abstractmethod from config.logger import setup_logging from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption -from core.utils.util import audio_to_data, audio_bytes_to_data +from core.utils.util import audio_to_data, audio_bytes_to_data, audio_bytes_to_data_stream, audio_to_data_stream from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report @@ -51,11 +53,9 @@ class TTSProviderBase(ABC): "；", ";", "：", - "~", ) self.first_sentence_punctuations = ( "，", - "～", "~", "、", ",", @@ -80,7 +80,15 @@ class TTSProviderBase(ABC): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) - def to_tts(self, text): + 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 to_tts(self, text, opus_handler=handle_opus) -> None: text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -89,10 +97,13 @@ 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 + self.tts_audio_queue.put( + (SentenceType.FIRST, None, text) ) - return audio_datas + audio_bytes_to_data_stream( + audio_bytes, file_type=self.audio_file_type, is_opus=True, callback=opus_handler + ) + max_repeat_time = 0 else: max_repeat_time -= 1 except Exception as e: @@ -132,8 +143,10 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).error( f"语音生成失败: {text}，请检查网络或服务是否正常" ) - - return tmp_file + self.tts_audio_queue.put( + (SentenceType.FIRST, None, text) + ) + self._process_audio_file(tmp_file, callback=opus_handler) except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") return None @@ -142,13 +155,13 @@ 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, @@ -215,28 +228,12 @@ 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(segment_text, opus_handler=self.handle_opus) elif ContentType.FILE == message.content_type: self._process_remaining_text() 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(tts_file, callback=self.handle_opus) if message.sentence_type == SentenceType.LAST: self._process_remaining_text() self.tts_audio_queue.put( @@ -295,6 +292,11 @@ class TTSProviderBase(ABC): # 实际应用中可能需要从设备端获取真实的消费情况 estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) self.flow_controller.update_device_consumption(estimated_consumption) + status = self.flow_controller.get_status() + logger.bind(tag=TAG).debug( + f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + f"可用令牌={status['available_tokens']}..." + ) else: # 可以发送，记录发送的帧数 self.flow_controller.record_sent_frames(frame_count) @@ -312,11 +314,11 @@ class TTSProviderBase(ABC): enqueue_tts_report(self.conn, text, audio_datas) # 输出流控状态（调试用） - if frame_count > 10: # 只在较大的音频块时输出状态 + if frame_count > 0: # 只在较大的音频块时输出状态 status = self.flow_controller.get_status() logger.bind(tag=TAG).debug( f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - f"可用令牌={status['available_tokens']}, 发送文本: {text[:20]}..." + f"可用令牌={status['available_tokens']}..." ) else: # 没有音频数据，直接发送 @@ -415,7 +417,7 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file): + def _process_audio_file(self, tts_file, callback: Callable[[Any], Any]): """处理音频文件并转换为指定格式 Args: @@ -426,11 +428,11 @@ class TTSProviderBase(ABC): tuple: (sentence_type, audio_datas, content_detail) """ 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 @@ -439,7 +441,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: @@ -466,7 +467,7 @@ class TTSProviderBase(ABC): ) else: tts_file = self.to_tts(segment_text) - audio_datas = self._process_audio_file(tts_file) + audio_datas = self._process_audio_file(tts_file, callback=self.handle_opus) self.tts_audio_queue.put( (SentenceType.MIDDLE, audio_datas, segment_text) ) diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index 74a5585f..ecdb5570 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -173,8 +173,8 @@ class FlowControlConfig: OPUS_FRAMES_PER_SECOND = 1000 / OPUS_FRAME_DURATION_MS # 每秒帧数 # 默认流控参数 - DEFAULT_MAX_DEVICE_BUFFER = 1000 # 设备端最大缓冲帧数 - DEFAULT_REFILL_RATE = 20 # 默认令牌补充速率（帧/秒） + DEFAULT_MAX_DEVICE_BUFFER = 40 # 设备端最大缓冲帧数 + DEFAULT_REFILL_RATE = 5 # 默认令牌补充速率（帧/秒） DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 088d8962..8aa2d712 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -230,6 +230,23 @@ def audio_to_data(audio_file_path, is_opus=True): raw_data = audio.raw_data return pcm_to_data(raw_data, is_opus) +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: + 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 + pcm_to_data_stream(raw_data, is_opus,callback) + def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): """ @@ -247,7 +264,7 @@ def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): raw_data = audio.raw_data return pcm_to_data(raw_data, is_opus) -def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]): +def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: """ 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 """ @@ -262,7 +279,6 @@ def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callab 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) - return None def pcm_to_data(raw_data, is_opus=True): From ce6a711c8e5d619aa535a4d64159d88369a08091 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 10:09:28 +0800 Subject: [PATCH 07/11] =?UTF-8?q?=E8=B0=83=E6=95=B4=E6=B5=81=E6=8E=A7?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 24 +++++-------- .../xiaozhi-server/core/providers/tts/base.py | 36 ++++++++++--------- .../core/utils/audio_flow_control.py | 4 +-- 3 files changed, 30 insertions(+), 34 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 5b3938e9..94806924 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -447,21 +447,15 @@ class TTSProvider(TTSProviderBase): 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, [], self.conn.tts_MessageText) - ) - self.conn.tts_MessageText = 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() diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index d105a892..2e99557b 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -253,7 +253,7 @@ class TTSProviderBase(ABC): text = None try: try: - sentence_type, audio_datas, text = self.tts_audio_queue.get(timeout=1) + sentence_type, audio_datas, text = self.tts_audio_queue.get(timeout=0.1) except queue.Empty: if self.conn.stop_event.is_set(): break @@ -288,15 +288,14 @@ class TTSProviderBase(ABC): # 短暂等待后重试 time.sleep(retry_interval) - # 更新设备消费估计（这里假设设备以恒定速率消费） - # 实际应用中可能需要从设备端获取真实的消费情况 - estimated_consumption = int(retry_interval * FlowControlConfig.DEFAULT_REFILL_RATE) - self.flow_controller.update_device_consumption(estimated_consumption) - status = self.flow_controller.get_status() - logger.bind(tag=TAG).debug( - f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - f"可用令牌={status['available_tokens']}..." - ) + # status = self.flow_controller.get_status() + # logger.bind(tag=TAG).debug( + # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + # f"可用令牌={status['available_tokens']}..." + # f"发送帧数={status['sent_frames']}..." + # f"消费帧数={status['consumed_frames']}..." + # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." + # ) else: # 可以发送，记录发送的帧数 self.flow_controller.record_sent_frames(frame_count) @@ -314,16 +313,19 @@ class TTSProviderBase(ABC): enqueue_tts_report(self.conn, text, audio_datas) # 输出流控状态（调试用） - if frame_count > 0: # 只在较大的音频块时输出状态 - status = self.flow_controller.get_status() - logger.bind(tag=TAG).debug( - f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - f"可用令牌={status['available_tokens']}..." - ) + # if frame_count > 0: # 只在较大的音频块时输出状态 + # status = self.flow_controller.get_status() + # logger.bind(tag=TAG).debug( + # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + # f"可用令牌={status['available_tokens']}..." + # f"发送帧数={status['sent_frames']}..." + # f"消费帧数={status['consumed_frames']}..." + # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." + # ) else: # 没有音频数据，直接发送 future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self._send_audio_with_flow_control(sentence_type, audio_datas, text), self.conn.loop, ) future.result() diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index ecdb5570..102914c5 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -174,9 +174,9 @@ class FlowControlConfig: # 默认流控参数 DEFAULT_MAX_DEVICE_BUFFER = 40 # 设备端最大缓冲帧数 - DEFAULT_REFILL_RATE = 5 # 默认令牌补充速率（帧/秒） + DEFAULT_REFILL_RATE = OPUS_FRAMES_PER_SECOND # 默认令牌补充速率（帧/秒） DEFAULT_MAX_WAIT_TIME = 5.0 # 流控最大等待时间（秒） - DEFAULT_RETRY_INTERVAL = 0.1 # 流控重试间隔（秒） + DEFAULT_RETRY_INTERVAL = 0.06 # 流控重试间隔（秒） # 预缓冲参数 PRE_BUFFER_FRAMES = 3 # 预缓冲帧数 From 903c0e51eeb1f789337e40444e806f3aea674ae4 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 10:13:42 +0800 Subject: [PATCH 08/11] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E5=89=A9=E4=BD=99?= =?UTF-8?q?=E6=96=87=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/providers/tts/base.py | 13 +------------ 1 file changed, 1 insertion(+), 12 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 2e99557b..575e1325 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -461,18 +461,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, callback=self.handle_opus) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, audio_datas, segment_text) - ) + self.to_tts(segment_text, opus_handler=self.handle_opus) self.processed_chars += len(full_text) return True return False From abf1fa0ef4b72411f4059d623718f48b09277f16 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 11:34:57 +0800 Subject: [PATCH 09/11] =?UTF-8?q?=E8=B0=83=E6=95=B4=E5=AD=90=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 12 ++---- .../xiaozhi-server/core/providers/tts/base.py | 28 +++++++------ .../providers/tts/huoshan_double_stream.py | 40 +++++-------------- .../core/providers/tts/linkerai.py | 12 ++---- 4 files changed, 31 insertions(+), 61 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 94806924..52451cba 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -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会话...") @@ -486,7 +482,7 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式TTS处理，用于测试及保存音频文件的场景""" try: # 创建新的事件循环 @@ -591,7 +587,7 @@ class TTSProvider(TTSProviderBase): if isinstance(msg, (bytes, bytearray)): # 编码为Opus并收集 self.opus_encoder.encode_pcm_to_opus_stream( - msg, False, self.handle_opus + msg, False, opus_handler ) # audio_data.extend(opus_frames) elif isinstance(msg, str): @@ -621,7 +617,5 @@ class TTSProvider(TTSProviderBase): loop.run_until_complete(_generate_audio()) loop.close() - return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 575e1325..d55f434e 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -88,7 +88,12 @@ class TTSProviderBase(ABC): (SentenceType.MIDDLE, opus_data, None) ) - def to_tts(self, text, opus_handler=handle_opus) -> 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=handle_opus) -> None: text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -146,7 +151,7 @@ class TTSProviderBase(ABC): self.tts_audio_queue.put( (SentenceType.FIRST, None, text) ) - self._process_audio_file(tmp_file, callback=opus_handler) + 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 @@ -228,14 +233,14 @@ class TTSProviderBase(ABC): self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + 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): - self._process_audio_file(tts_file, callback=self.handle_opus) + 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) ) @@ -419,15 +424,12 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file, callback: Callable[[Any], Any]): + 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"): p3.decode_opus_from_file_stream(tts_file, callback=callback) @@ -450,7 +452,7 @@ 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=handle_opus): """处理剩余的文本并生成语音 Returns: @@ -461,7 +463,7 @@ class TTSProviderBase(ABC): if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: - self.to_tts(segment_text, opus_handler=self.handle_opus) + self.to_tts_stream(segment_text, opus_handler=opus_handler) self.processed_chars += len(full_text) return True return False diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index ea7f18fd..f45b9a96 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -4,6 +4,8 @@ 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 @@ -260,11 +262,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会话...") @@ -452,21 +450,7 @@ class TTSProvider(TTSProviderBase): 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: @@ -642,15 +626,15 @@ 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: + def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: """非流式生成音频数据，用于生成音频及测试场景 Args: text: 要转换的文本 + opus_handler: opus数据处理方法 Returns: list: 音频数据列表 @@ -663,9 +647,6 @@ class TTSProvider(TTSProviderBase): # 生成会话ID session_id = uuid.uuid4().__str__().replace("-", "") - # 存储音频数据 - audio_data = [] - async def _generate_audio(): # 创建新的WebSocket连接 ws_header = { @@ -728,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=opus_handler) elif res.optional.event == EVENT_SessionFinished: break @@ -744,8 +724,6 @@ class TTSProvider(TTSProviderBase): loop.run_until_complete(_generate_audio()) loop.close() - return audio_data - except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") return [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index d1be586c..a9877d43 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -65,14 +65,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 +77,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: @@ -237,7 +233,7 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def to_tts(self, text: str) -> list: + def to_tts_stream(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: From 4d50eeb8303fb4d9514404a303363319c4c72174 Mon Sep 17 00:00:00 2001 From: Chingfeng Li Date: Wed, 6 Aug 2025 14:53:59 +0800 Subject: [PATCH 10/11] =?UTF-8?q?=E4=BF=AE=E6=AD=A3self=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/aliyun_stream.py | 4 +++- .../core/providers/tts/huoshan_double_stream.py | 4 +++- .../core/providers/tts/linkerai.py | 17 ++++++----------- 3 files changed, 12 insertions(+), 13 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 52451cba..4929491c 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -482,8 +482,10 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts_stream(self, text: str, opus_handler=self.handle_opus) -> None: + def to_tts_stream(self, text: str, opus_handler=None) -> None: """非流式TTS处理，用于测试及保存音频文件的场景""" + if opus_handler is None: + opus_handler = self.handle_opus try: # 创建新的事件循环 loop = asyncio.new_event_loop() diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index f45b9a96..118cb4fe 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -629,7 +629,7 @@ class TTSProvider(TTSProviderBase): 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_stream(self, text: str, opus_handler=self.handle_opus) -> None: + def to_tts_stream(self, text: str, opus_handler=None) -> None: """非流式生成音频数据，用于生成音频及测试场景 Args: @@ -639,6 +639,8 @@ class TTSProvider(TTSProviderBase): Returns: list: 音频数据列表 """ + if opus_handler is None: + opus_handler = self.handle_opus try: # 创建事件循环 loop = asyncio.new_event_loop() diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index a9877d43..09d3d82a 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -233,15 +233,15 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def to_tts_stream(self, text: str) -> list: + def to_tts_stream(self, text: str, opus_handler=None) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 Args: text: 要转换的文本 - - Returns: - list: 返回opus编码后的音频数据列表 + opus_handler: opus数据处理方法 """ + if opus_handler is None: + opus_handler = self.handle_opus start_time = time.time() text = MarkdownCleaner.clean_markdown(text) @@ -291,14 +291,9 @@ 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=opus_handler ) - if opus: - opus_datas.extend(opus) - - return opus_datas except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] From d2f29f335e9e831b32ad7c14257a977ba189e63a Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 15 Aug 2025 16:33:46 +0800 Subject: [PATCH 11/11] =?UTF-8?q?update:=20=E9=9F=B3=E9=A2=91=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 4 - main/xiaozhi-server/core/connection.py | 8 +- .../xiaozhi-server/core/handle/helloHandle.py | 109 ------------ .../core/handle/intentHandler.py | 4 - .../core/handle/receiveAudioHandle.py | 43 +++-- .../core/handle/sendAudioHandle.py | 53 +----- main/xiaozhi-server/core/handle/textHandle.py | 17 +- .../core/providers/tts/aliyun_stream.py | 144 +-------------- .../xiaozhi-server/core/providers/tts/base.py | 164 ++++++++---------- .../providers/tts/huoshan_double_stream.py | 115 +----------- .../core/providers/tts/index_stream.py | 112 ++---------- .../core/providers/tts/linkerai.py | 115 +----------- .../core/utils/audio_flow_control.py | 14 -- .../core/utils/opus_encoder_utils.py | 50 +----- main/xiaozhi-server/core/utils/p3.py | 44 +---- main/xiaozhi-server/core/utils/util.py | 95 +--------- main/xiaozhi-server/core/utils/wakeup_word.py | 140 --------------- 17 files changed, 143 insertions(+), 1088 deletions(-) delete mode 100644 main/xiaozhi-server/core/utils/wakeup_word.py diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index ab662798..b3c1b1a8 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -61,10 +61,6 @@ delete_audio: true close_connection_no_voice_time: 120 # TTS请求超时时间(秒) tts_timeout: 10 -# 开启唤醒词加速 -enable_wakeup_words_response_cache: true -# 开场是否回复唤醒词 -enable_greeting: true # 说完话是否开启提示音 enable_stop_tts_notify: false # 说完话是否开启提示音，音效地址 diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 1a792427..2ca2602a 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -664,16 +664,14 @@ class ConnectionHandler: # 更新系统prompt至上下文 self.dialogue.update_system_message(self.prompt) - def chat(self, query, tool_call=False, depth=0): + def chat(self, query, depth=0): self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") self.llm_finish_task = False - if not tool_call: - self.dialogue.put(Message(role="user", content=query)) - # 为最顶层时新建会话ID和发送FIRST请求 if depth == 0: self.sentence_id = str(uuid.uuid4().hex) + self.dialogue.put(Message(role="user", content=query)) self.tts.tts_text_queue.put( TTSMessageDTO( sentence_id=self.sentence_id, @@ -878,7 +876,7 @@ class ConnectionHandler: content=text, ) ) - self.chat(text, tool_call=True, depth=depth + 1) + self.chat(text, depth=depth + 1) elif result.action == Action.NOTFOUND or result.action == Action.ERROR: text = result.response if result.response else result.result self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index c4f55841..de9587b5 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,32 +1,13 @@ -import time import json -import random import asyncio -from core.utils.dialogue import Message -from core.utils.util import audio_to_data -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__ -WAKEUP_CONFIG = { - "refresh_time": 5, - "words": ["你好", "你好啊", "嘿，你好", "嗨"], -} - -# 创建全局的唤醒词配置管理器 -wakeup_words_config = WakeupWordsConfig() - -# 用于防止并发调用wakeupWordsResponse的锁 -_wakeup_response_lock = asyncio.Lock() - async def handleHelloMessage(conn, msg_json): """处理hello消息""" @@ -49,93 +30,3 @@ async def handleHelloMessage(conn, msg_json): asyncio.create_task(send_mcp_tools_list_request(conn)) await conn.websocket.send(json.dumps(conn.welcome_msg)) - - -async def checkWakeupWords(conn, text): - enable_wakeup_words_response_cache = conn.config[ - "enable_wakeup_words_response_cache" - ] - - if not enable_wakeup_words_response_cache or not conn.tts: - return False - - _, filtered_text = remove_punctuation_and_length(text) - if filtered_text not in conn.config.get("wakeup_words"): - return False - - conn.just_woken_up = True - await send_stt_message(conn, text) - - # 获取当前音色 - voice = getattr(conn.tts, "voice", "default") - if not voice: - voice = "default" - - # 获取唤醒词回复配置 - response = wakeup_words_config.get_wakeup_response(voice) - if not response or not response.get("file_path"): - response = { - "voice": "default", - "file_path": "config/assets/wakeup_words.wav", - "time": 0, - "text": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦", - } - - # 播放唤醒词回复 - 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")) - await sendAudioMessage(conn, SentenceType.LAST, [], None) - - # 补充对话 - conn.dialogue.put(Message(role="assistant", content=response.get("text"))) - - # 检查是否需要更新唤醒词回复 - if time.time() - response.get("time", 0) > WAKEUP_CONFIG["refresh_time"]: - if not _wakeup_response_lock.locked(): - asyncio.create_task(wakeupWordsResponse(conn)) - return True - - -async def wakeupWordsResponse(conn): - if not conn.tts or not conn.llm or not conn.llm.response_no_stream: - return - - try: - # 尝试获取锁，如果获取不到就返回 - if not await _wakeup_response_lock.acquire(): - return - - # 生成唤醒词回复 - wakeup_word = random.choice(WAKEUP_CONFIG["words"]) - question = ( - "此刻用户正在和你说```" - + wakeup_word - + "```。\n请你根据以上用户的内容进行20-30字回复。要符合系统设置的角色情感和态度，不要像机器人一样说话。\n" - + "请勿对这条内容本身进行任何解释和回应，请勿返回表情符号，仅返回对用户的内容的回复。" - ) - - result = conn.llm.response_no_stream(conn.config["prompt"], question) - if not result or len(result) == 0: - return - - # 生成TTS音频 - tts_result = await asyncio.to_thread(conn.tts.to_tts, result) - if not tts_result: - return - - # 获取当前音色 - voice = getattr(conn.tts, "voice", "default") - - wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000) - file_path = wakeup_words_config.generate_file_path(voice) - with open(file_path, "wb") as f: - f.write(wav_bytes) - # 更新配置 - wakeup_words_config.update_wakeup_response(voice, file_path, result) - finally: - # 确保在任何情况下都释放锁 - if _wakeup_response_lock.locked(): - _wakeup_response_lock.release() diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 81911f55..58b2cee7 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -2,7 +2,6 @@ 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 from core.utils.dialogue import Message @@ -27,9 +26,6 @@ async def handle_user_intent(conn, text): filtered_text = remove_punctuation_and_length(text)[1] if await check_direct_exit(conn, filtered_text): return True - # 检查是否是唤醒词 - if await checkWakeupWords(conn, filtered_text): - return True if conn.intent_type == "function_call": # 使用支持function calling的聊天方法,不再进行意图分析 diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 174c8b2b..7d46685b 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,12 +1,12 @@ +import time +import asyncio +import json from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit from core.handle.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.utils.util import audio_to_data_stream TAG = __name__ @@ -42,7 +42,7 @@ async def startToChat(conn, text): # 检查输入是否是JSON格式（包含说话人信息） speaker_name = None actual_text = text - + try: # 尝试解析JSON格式的输入 if text.strip().startswith('{') and text.strip().endswith('}'): @@ -51,13 +51,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 @@ -121,8 +121,9 @@ async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) file_path = "config/assets/max_output_size.wav" - opus_packets, _ = audio_to_data(file_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, file_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) conn.close_after_chat = True @@ -140,16 +141,15 @@ async def check_bind_device(conn): # 播放提示音 music_path = "config/assets/bind_code.wav" - opus_packets = audio_to_data(music_path) - conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, music_path) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 try: digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" - num_packets = audio_to_data(num_path) - conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None)) + play_audio_frames(conn, num_path) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue @@ -158,5 +158,18 @@ async def check_bind_device(conn): text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) music_path = "config/assets/bind_not_found.wav" - opus_packets = audio_to_data(music_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, music_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) + + +def play_audio_frames(conn, file_path): + """播放音频文件并处理发送帧数据""" + def handle_audio_frame(frame_data): + conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None)) + + audio_to_data_stream( + file_path, + is_opus=True, + callback=handle_audio_frame + ) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index bf155ee9..c3c50fbc 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,6 +1,4 @@ import json -import asyncio -import time from core.providers.tts.dto.dto import SentenceType from core.utils import textUtils @@ -8,20 +6,18 @@ 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) 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: @@ -32,46 +28,12 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 -async def sendAudio(conn, audios, pre_buffer=True): +async def sendAudio(conn, audios): if audios is None: return # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出 - if isinstance(audios ,bytes): + if isinstance(audios, bytes): await conn.websocket.send(audios) - else: - 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 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): @@ -100,13 +62,12 @@ 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: diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 1bcf4545..86363217 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -5,7 +5,6 @@ from core.handle.helloHandle import handleHelloMessage from core.providers.tools.device_mcp import handle_mcp_message from core.utils.util import remove_punctuation_and_length, filter_sensitive_info from core.handle.receiveAudioHandle import startToChat, handleAudioMessage -from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus from core.handle.reportHandle import enqueue_asr_report import asyncio @@ -51,23 +50,9 @@ async def handleTextMessage(conn, message): filtered_len, filtered_text = remove_punctuation_and_length( original_text ) - # 识别是否是唤醒词 is_wakeup_words = filtered_text in conn.config.get("wakeup_words") - # 是否开启唤醒词回复 - enable_greeting = conn.config.get("enable_greeting", True) - - if is_wakeup_words and not enable_greeting: - # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 - await send_stt_message(conn, original_text) - await send_tts_message(conn, "stop", None) - conn.client_is_speaking = False - elif is_wakeup_words: - conn.just_woken_up = True - # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, "嘿，你好呀", []) - await startToChat(conn, "嘿，你好呀") - else: + if not is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） enqueue_asr_report(conn, original_text, []) # 否则需要LLM对文字内容进行答复 diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 4929491c..32e4d50f 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -216,6 +216,7 @@ class TTSProvider(TTSProviderBase): if message.sentence_type == SentenceType.FIRST: self.conn.client_abort = False + self.reset_flow_controller() if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") @@ -418,8 +419,6 @@ class TTSProvider(TTSProviderBase): async def _start_monitor_tts_response(self): """监听TTS响应""" - is_first_sentence = True - first_sentence_segment_count = 0 # 添加计数器 try: session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): @@ -440,8 +439,6 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], None) ) - elif event_name == "SentenceBegin": - opus_datas_cache = [] elif event_name == "SentenceEnd": # 发送缓存的数据 if self.conn.tts_MessageText: @@ -482,142 +479,3 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts_stream(self, text: str, opus_handler=None) -> None: - """非流式TTS处理，用于测试及保存音频文件的场景""" - if opus_handler is None: - opus_handler = self.handle_opus - try: - # 创建新的事件循环 - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - # 生成会话ID - session_id = uuid.uuid4().hex - # 存储音频数据 - audio_data = [] - - async def _generate_audio(): - # 刷新Token（如果需要） - if self._is_token_expired(): - self._refresh_token() - - # 建立WebSocket连接 - ws = await websockets.connect( - self.ws_url, - additional_headers={"X-NLS-Token": self.token}, - ping_interval=30, - ping_timeout=10, - close_timeout=10, - ) - try: - # 发送StartSynthesis请求 - start_message_id = str(uuid.uuid4().hex) - start_request = { - "header": { - "message_id": start_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "StartSynthesis", - "appkey": self.appkey, - }, - "payload": { - "voice": self.voice, - "format": self.format, - "sample_rate": self.sample_rate, - "volume": self.volume, - "speech_rate": self.speech_rate, - "pitch_rate": self.pitch_rate, - "enable_subtitle": True, - }, - } - await ws.send(json.dumps(start_request)) - - # 等待SynthesisStarted响应 - synthesis_started = False - while not synthesis_started: - msg = await ws.recv() - if isinstance(msg, str): - data = json.loads(msg) - header = data.get("header", {}) - if header.get("name") == "SynthesisStarted": - synthesis_started = True - logger.bind(tag=TAG).debug("TTS合成已启动") - elif header.get("name") == "TaskFailed": - error_info = data.get("payload", {}).get( - "error_info", {} - ) - error_code = error_info.get("error_code") - error_message = error_info.get( - "error_message", "未知错误" - ) - raise Exception( - f"启动合成失败: {error_code} - {error_message}" - ) - - # 发送文本合成请求 - filtered_text = MarkdownCleaner.clean_markdown(text) - run_message_id = str(uuid.uuid4().hex) - run_request = { - "header": { - "message_id": run_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "RunSynthesis", - "appkey": self.appkey, - }, - "payload": {"text": filtered_text}, - } - await ws.send(json.dumps(run_request)) - - # 发送停止合成请求 - stop_message_id = str(uuid.uuid4().hex) - stop_request = { - "header": { - "message_id": stop_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "StopSynthesis", - "appkey": self.appkey, - } - } - await ws.send(json.dumps(stop_request)) - - # 接收音频数据 - synthesis_completed = False - while not synthesis_completed: - msg = await ws.recv() - if isinstance(msg, (bytes, bytearray)): - # 编码为Opus并收集 - self.opus_encoder.encode_pcm_to_opus_stream( - msg, False, opus_handler - ) - # audio_data.extend(opus_frames) - elif isinstance(msg, str): - data = json.loads(msg) - header = data.get("header", {}) - event_name = header.get("name") - if event_name == "SynthesisCompleted": - synthesis_completed = True - logger.bind(tag=TAG).debug("TTS合成完成") - elif event_name == "TaskFailed": - error_info = data.get("payload", {}).get( - "error_info", {} - ) - error_code = error_info.get("error_code") - error_message = error_info.get( - "error_message", "未知错误" - ) - raise Exception( - f"合成失败: {error_code} - {error_message}" - ) - finally: - try: - await ws.close() - except: - pass - - loop.run_until_complete(_generate_audio()) - loop.close() - - except Exception as e: - logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index d55f434e..9e16df97 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -5,14 +5,14 @@ import uuid import asyncio import threading from typing import Callable, Any - from core.utils import p3 import time +from datetime import datetime from core.utils import textUtils from abc import ABC, abstractmethod from config.logger import setup_logging -from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption -from core.utils.util import audio_to_data, audio_bytes_to_data, audio_bytes_to_data_stream, audio_to_data_stream +from core.utils.audio_flow_control import FlowControlConfig +from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report @@ -72,7 +72,6 @@ class TTSProviderBase(ABC): self.processed_chars = 0 self.is_first_sentence = True self.flow_controller = FlowControlConfig.create_flow_controller() - self.flow_control_enabled = config.get("enable_flow_control", True) def generate_filename(self, extension=".wav"): return os.path.join( @@ -93,7 +92,7 @@ class TTSProviderBase(ABC): (file_audio, text) ) - def to_tts_stream(self, text, opus_handler=handle_opus) -> None: + 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: @@ -108,7 +107,7 @@ class TTSProviderBase(ABC): audio_bytes_to_data_stream( audio_bytes, file_type=self.audio_file_type, is_opus=True, callback=opus_handler ) - max_repeat_time = 0 + break else: max_repeat_time -= 1 except Exception as e: @@ -160,11 +159,11 @@ class TTSProviderBase(ABC): async def text_to_speak(self, text, output_file): pass - def audio_to_pcm_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=None): + def audio_to_pcm_data_stream(self, audio_file_path, callback: Callable[[Any], Any] = None): """音频文件转换为PCM编码""" return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback) - def audio_to_opus_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=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) @@ -229,6 +228,7 @@ class TTSProviderBase(ABC): self.tts_text_buff = [] self.is_first_sentence = True self.tts_audio_first_sentence = True + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -254,6 +254,9 @@ 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: @@ -264,108 +267,91 @@ class TTSProviderBase(ABC): break continue - # 如果启用了流控 - if self.flow_control_enabled: - # 计算音频数据的帧数 - if isinstance(audio_datas, bytes): - frame_count = 1 # 单个字节流作为一帧 - elif isinstance(audio_datas, (list, tuple)): - frame_count = len(audio_datas) + if self.conn.client_abort: + logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据") + # 打断时丢弃未上报的音频数据 + enqueue_text, enqueue_audio = None, [] + continue + + # 收到下一个文本开始或会话结束时进行上报 + if sentence_type is not SentenceType.MIDDLE: + # 上报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): + frame_count = 1 # 单个字节流作为一帧 + enqueue_audio.append(audio_datas) + else: + frame_count = 0 + + # 记录输出和报告 + if self.conn.max_output_size > 0 and text: + add_device_output(self.conn.headers.get("device-id"), len(text)) + + # 流控检查 + if frame_count > 0: + max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME + wait_start_time = time.time() + retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL + + while not self.flow_controller.can_send_frames(frame_count): + # 检查是否超时或需要停止 + if (time.time() - wait_start_time > max_wait_time or + self.conn.stop_event.is_set() or + self.conn.client_abort): + logger.bind(tag=TAG).debug("流控等待超时或收到停止信号，跳过音频发送") + break + # 短暂等待后重试 + time.sleep(retry_interval) else: - frame_count = 0 + # 可以发送，记录发送的帧数 + self.flow_controller.record_sent_frames(frame_count) - # 流控检查 - if frame_count > 0: - max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME - wait_start_time = time.time() - retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL - - while not self.flow_controller.can_send_frames(frame_count): - # 检查是否超时或需要停止 - if (time.time() - wait_start_time > max_wait_time or - self.conn.stop_event.is_set() or - self.conn.client_abort): - logger.bind(tag=TAG).warning( - f"流控等待超时或收到停止信号，跳过音频发送: {text}" - ) - break - - # 短暂等待后重试 - time.sleep(retry_interval) - - # status = self.flow_controller.get_status() - # logger.bind(tag=TAG).debug( - # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - # f"可用令牌={status['available_tokens']}..." - # f"发送帧数={status['sent_frames']}..." - # f"消费帧数={status['consumed_frames']}..." - # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." - # ) - else: - # 可以发送，记录发送的帧数 - self.flow_controller.record_sent_frames(frame_count) - - # 发送音频 - future = asyncio.run_coroutine_threadsafe( - self._send_audio_with_flow_control(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) - - # 输出流控状态（调试用） - # if frame_count > 0: # 只在较大的音频块时输出状态 - # status = self.flow_controller.get_status() - # logger.bind(tag=TAG).debug( - # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - # f"可用令牌={status['available_tokens']}..." - # f"发送帧数={status['sent_frames']}..." - # f"消费帧数={status['consumed_frames']}..." - # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." - # ) - else: - # 没有音频数据，直接发送 + # 发送音频 future = asyncio.run_coroutine_threadsafe( self._send_audio_with_flow_control(sentence_type, audio_datas, text), self.conn.loop, ) future.result() + + # 输出流控状态（调试用） + # status = self.flow_controller.get_status() + # logger.bind(tag=TAG).debug( + # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + # f"可用令牌={status['available_tokens']}..." + # f"发送帧数={status['sent_frames']}..." + # f"消费帧数={status['consumed_frames']}..." + # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." + # ) else: - # 未启用流控，直接发送 + # 没有音频数据，直接发送 future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self._send_audio_with_flow_control(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_thread: {text} {e}" ) async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text): - """带流控的音频发送方法""" + """ + 带流控的音频发送方法 模拟设备消费音频帧的过程 + 实际应用中应该根据设备反馈来更新消费情况 + """ await sendAudioMessage(self.conn, sentence_type, audio_datas, text) - # 模拟设备消费（实际应用中应该从设备获取反馈） + # 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步 if isinstance(audio_datas, bytes): - frame_count = 1 - elif isinstance(audio_datas, (list, tuple)): - frame_count = len(audio_datas) - else: - frame_count = 0 - - if frame_count > 0: - asyncio.create_task(simulate_device_consumption(self.flow_controller, frame_count)) + # 模拟设备播放延迟（60ms per frame）, 实际情况可以低一点（50ms），增加使用体验 + await asyncio.sleep(0.055) + self.flow_controller.update_device_consumption(1) # 在类中添加流控制器重置方法 def reset_flow_controller(self): @@ -452,7 +438,7 @@ class TTSProviderBase(ABC): self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def _process_remaining_text_stream(self, opus_handler=handle_opus): + def _process_remaining_text_stream(self, opus_handler: Callable[[bytes], None] = None): """处理剩余的文本并生成语音 Returns: diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index a971215f..ac6301f9 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -5,7 +5,6 @@ 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 @@ -214,6 +213,7 @@ class TTSProvider(TTSProviderBase): if message.sentence_type == SentenceType.FIRST: self.conn.client_abort = False + self.reset_flow_controller() if self.conn.client_abort: try: @@ -426,9 +426,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(): @@ -449,8 +446,6 @@ 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 @@ -459,13 +454,6 @@ class TTSProvider(TTSProviderBase): 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() @@ -641,104 +629,3 @@ class TTSProvider(TTSProviderBase): 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_stream(self, text: str, opus_handler=None) -> None: - """非流式生成音频数据，用于生成音频及测试场景 - - Args: - text: 要转换的文本 - opus_handler: opus数据处理方法 - - Returns: - list: 音频数据列表 - """ - if opus_handler is None: - opus_handler = self.handle_opus - try: - # 创建事件循环 - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - # 生成会话ID - session_id = uuid.uuid4().__str__().replace("-", "") - - async def _generate_audio(): - # 创建新的WebSocket连接 - ws_header = { - "X-Api-App-Key": self.appId, - "X-Api-Access-Key": self.access_token, - "X-Api-Resource-Id": self.resource_id, - "X-Api-Connect-Id": uuid.uuid4(), - } - ws = await websockets.connect( - self.ws_url, additional_headers=ws_header, max_size=1000000000 - ) - - try: - # 启动会话 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_StartSession, sessionId=session_id - ).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_StartSession, speaker=self.voice - ) - await self.send_event(ws, header, optional, payload) - - # 发送文本 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_TaskRequest, sessionId=session_id - ).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_TaskRequest, text=text, speaker=self.voice - ) - await self.send_event(ws, header, optional, payload) - - # 发送结束会话请求 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_FinishSession, sessionId=session_id - ).as_bytes() - payload = str.encode("{}") - await self.send_event(ws, header, optional, payload) - - # 接收音频数据 - while True: - msg = await ws.recv() - res = self.parser_response(msg) - - if ( - res.optional.event == EVENT_TTSResponse - and res.header.message_type == AUDIO_ONLY_RESPONSE - ): - self.wav_to_opus_data_audio_raw_stream(res.payload, callback=opus_handler) - elif res.optional.event == EVENT_SessionFinished: - break - - finally: - # 清理资源 - try: - await ws.close() - except: - pass - - # 运行异步任务 - loop.run_until_complete(_generate_audio()) - loop.close() - - except Exception as e: - logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] diff --git a/main/xiaozhi-server/core/providers/tts/index_stream.py b/main/xiaozhi-server/core/providers/tts/index_stream.py index 5ea9eb36..d562e802 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/main/xiaozhi-server/core/providers/tts/index_stream.py @@ -3,8 +3,6 @@ import queue import asyncio import traceback import aiohttp -import requests -import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -27,15 +25,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,8 +44,8 @@ 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() + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -62,14 +58,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 +71,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 +136,6 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() - opus_datas_cache = [] - self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 处理音频流数据 @@ -158,41 +149,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() @@ -206,59 +178,3 @@ class TTSProvider(TTSProviderBase): await super().close() if hasattr(self, "opus_encoder"): self.opus_encoder.close() - - def to_tts(self, text: str) -> list: - """非流式TTS处理，用于测试及保存音频文件的场景 - - Args: - text: 要转换的文本 - - Returns: - list: 返回opus编码后的音频数据列表 - """ - start_time = time.time() - text = MarkdownCleaner.clean_markdown(text) - - payload = {"text": text, "character": self.character} - - try: - with requests.post(self.api_url, json=payload, timeout=5) as response: - if response.status_code != 200: - logger.bind(tag=TAG).error( - f"TTS请求失败: {response.status_code}, {response.text}" - ) - return [] - - logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒") - - # 使用opus编码器处理PCM数据 - opus_datas = [] - pcm_data = response.content - - # 计算每帧的字节数 - frame_bytes = int( - self.opus_encoder.sample_rate - * self.opus_encoder.channels - * self.opus_encoder.frame_size_ms - / 1000 - * 2 - ) - - # 分帧处理PCM数据 - for i in range(0, len(pcm_data), frame_bytes): - frame = pcm_data[i : i + frame_bytes] - if len(frame) < frame_bytes: - # 最后一帧可能不足，用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)) - ) - if opus: - opus_datas.extend(opus) - - return opus_datas - - except Exception as e: - logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 09d3d82a..a855e8ed 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -3,8 +3,6 @@ import queue import asyncio import traceback import aiohttp -import requests -import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -24,23 +22,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,8 +41,8 @@ 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() + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -172,8 +162,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 @@ -190,41 +178,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() @@ -232,68 +200,3 @@ class TTSProvider(TTSProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) - - def to_tts_stream(self, text: str, opus_handler=None) -> list: - """非流式TTS处理，用于测试及保存音频文件的场景 - - Args: - text: 要转换的文本 - opus_handler: opus数据处理方法 - """ - if opus_handler is None: - opus_handler = self.handle_opus - start_time = time.time() - text = MarkdownCleaner.clean_markdown(text) - - params = { - "tts_text": text, - "spk_id": self.voice, - "frame_duration": 60, - "stream": False, - "target_sr": 16000, - "audio_format": self.audio_format, - "instruct_text": "请生成一段自然流畅的语音", - } - headers = { - "Authorization": f"Bearer {self.access_token}", - "Content-Type": "application/json", - } - - try: - with requests.get( - self.api_url, params=params, headers=headers, timeout=5 - ) as response: - if response.status_code != 200: - logger.bind(tag=TAG).error( - f"TTS请求失败: {response.status_code}, {response.text}" - ) - return [] - - logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒") - - # 使用opus编码器处理PCM数据 - opus_datas = [] - pcm_data = response.content - - # 计算每帧的字节数 - frame_bytes = int( - self.opus_encoder.sample_rate - * self.opus_encoder.channels - * self.opus_encoder.frame_size_ms - / 1000 - * 2 - ) - - # 分帧处理PCM数据 - for i in range(0, len(pcm_data), frame_bytes): - frame = pcm_data[i : i + frame_bytes] - if len(frame) < frame_bytes: - # 最后一帧可能不足，用0填充 - frame = frame + b"\x00" * (frame_bytes - len(frame)) - - self.opus_encoder.encode_pcm_to_opus_stream( - frame, end_of_stream=(i + frame_bytes >= len(pcm_data)), callback=opus_handler - ) - - except Exception as e: - logger.bind(tag=TAG).error(f"TTS请求异常: {e}") diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index 102914c5..98c29676 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -151,20 +151,6 @@ class AudioFlowController: ) -async def simulate_device_consumption(flow_controller: AudioFlowController, frame_count: int): - """ - 模拟设备消费音频帧的过程 - 实际应用中应该根据设备反馈来更新消费情况 - - Args: - flow_controller: 流控制器实例 - frame_count: 消费的帧数 - """ - # 模拟设备播放延迟（60ms per frame） - await asyncio.sleep(frame_count * 0.06) - flow_controller.update_device_consumption(frame_count) - - # 流控配置常量 class FlowControlConfig: """流控配置常量""" diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index f9154da1..8ca406d3 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -5,9 +5,8 @@ Opus编码工具类 import logging import traceback - import numpy as np -from typing import List, Optional, Callable, Any +from typing import Optional, Callable, Any from opuslib_next import Encoder from opuslib_next import constants @@ -56,53 +55,6 @@ class OpusEncoderUtils: self.encoder.reset_state() self.buffer = np.array([], dtype=np.int16) - def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: - """ - 将PCM数据编码为Opus格式 - - Args: - pcm_data: PCM字节数据 - end_of_stream: 是否为流的结束 - - Returns: - Opus数据包列表 - """ - # 将字节数据转换为short数组 - new_samples = self._convert_bytes_to_shorts(pcm_data) - - # 校验PCM数据 - self._validate_pcm_data(new_samples) - - # 将新数据追加到缓冲区 - self.buffer = np.append(self.buffer, new_samples) - - opus_packets = [] - offset = 0 - - # 处理所有完整帧 - while offset <= len(self.buffer) - self.total_frame_size: - frame = self.buffer[offset : offset + self.total_frame_size] - output = self._encode(frame) - if output: - opus_packets.append(output) - offset += self.total_frame_size - - # 保留未处理的样本 - self.buffer = self.buffer[offset:] - - # 流结束时处理剩余数据 - if end_of_stream and len(self.buffer) > 0: - # 创建最后一帧并用0填充 - last_frame = np.zeros(self.total_frame_size, dtype=np.int16) - last_frame[: len(self.buffer)] = self.buffer - - output = self._encode(last_frame) - if output: - opus_packets.append(output) - self.buffer = np.array([], dtype=np.int16) - - return opus_packets - def encode_pcm_to_opus_stream(self, pcm_data: bytes, end_of_stream: bool, callback: Callable[[Any], Any]): """ 将PCM数据编码为Opus格式，以流式方式进行处理 diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index f2ee3239..415e5366 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -1,30 +1,7 @@ +import io import struct from typing import Callable, Any -def decode_opus_from_file(input_file): - """ - 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表。 - """ - opus_datas = [] - - with open(input_file, 'rb') as f: - while True: - # 读取头部（4字节）：[1字节类型，1字节保留，2字节长度] - header = f.read(4) - if not header: - break - - # 解包头部信息 - _, _, data_len = struct.unpack('>BBH', header) - - # 根据头部指定的长度读取 Opus 数据 - opus_data = f.read(data_len) - if len(opus_data) != data_len: - raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.") - - opus_datas.append(opus_data) - - return opus_datas def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): """ @@ -47,25 +24,6 @@ def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): callback(opus_data) -def decode_opus_from_bytes(input_bytes): - """ - 从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表。 - """ - import io - opus_datas = [] - - f = io.BytesIO(input_bytes) - while True: - header = f.read(4) - if not header: - break - _, _, data_len = struct.unpack('>BBH', header) - opus_data = f.read(data_len) - if len(opus_data) != data_len: - raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") - opus_datas.append(opus_data) - - return opus_datas def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]): """ diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 8aa2d712..1eec7327 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -3,10 +3,8 @@ import socket import subprocess import re import os -import wave from io import BytesIO from typing import Callable, Any - from core.utils import p3 import numpy as np import requests @@ -213,23 +211,6 @@ def extract_json_from_string(input_string): return None -def audio_to_data(audio_file_path, is_opus=True): - # 获取文件后缀名 - 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) - 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] @@ -245,25 +226,9 @@ def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any] # 获取原始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) -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) - raw_data = audio.raw_data - return pcm_to_data(raw_data, is_opus) - def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: """ 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 @@ -281,37 +246,7 @@ def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callab pcm_to_data_stream(raw_data, is_opus, callback) -def pcm_to_data(raw_data, is_opus=True): - # 初始化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 - - datas = [] - # 按帧处理所有音频数据（包括最后一帧可能补零） - 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) - else: - frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk) - - datas.append(frame_data) - - return datas - -def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any]=None): +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) @@ -338,32 +273,6 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any]=No 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): - """ - 将opus帧列表解码为wav字节流 - """ - decoder = opuslib_next.Decoder(sample_rate, channels) - pcm_datas = [] - - frame_duration = 60 # ms - frame_size = int(sample_rate * frame_duration / 1000) # 960 - - for opus_frame in opus_datas: - # 解码为PCM（返回bytes，2字节/采样点） - pcm = decoder.decode(opus_frame, frame_size) - pcm_datas.append(pcm) - - pcm_bytes = b"".join(pcm_datas) - - # 写入wav字节流 - wav_buffer = BytesIO() - with wave.open(wav_buffer, "wb") as wf: - wf.setnchannels(channels) - wf.setsampwidth(2) # 16bit - wf.setframerate(sample_rate) - wf.writeframes(pcm_bytes) - return wav_buffer.getvalue() - def check_vad_update(before_config, new_config): if ( diff --git a/main/xiaozhi-server/core/utils/wakeup_word.py b/main/xiaozhi-server/core/utils/wakeup_word.py deleted file mode 100644 index ab58f290..00000000 --- a/main/xiaozhi-server/core/utils/wakeup_word.py +++ /dev/null @@ -1,140 +0,0 @@ -import os -import re -import yaml -import time -import hashlib -import portalocker -from typing import Dict - - -class FileLock: - def __init__(self, file, timeout=5): - self.file = file - self.timeout = timeout - self.start_time = None - - def __enter__(self): - self.start_time = time.time() - while True: - try: - portalocker.lock(self.file, portalocker.LOCK_EX | portalocker.LOCK_NB) - return self.file - except portalocker.LockException: - if time.time() - self.start_time > self.timeout: - raise TimeoutError("获取文件锁超时") - time.sleep(0.1) - - def __exit__(self, exc_type, exc_val, exc_tb): - portalocker.unlock(self.file) - - -class WakeupWordsConfig: - def __init__(self): - self.config_file = "data/.wakeup_words.yaml" - self.assets_dir = "config/assets/wakeup_words" - self._ensure_directories() - self._config_cache = None - self._last_load_time = 0 - self._cache_ttl = 1 # 缓存有效期（秒） - self._lock_timeout = 5 # 文件锁超时时间（秒） - - def _ensure_directories(self): - """确保必要的目录存在""" - os.makedirs(os.path.dirname(self.config_file), exist_ok=True) - os.makedirs(self.assets_dir, exist_ok=True) - - def _load_config(self) -> Dict: - """加载配置文件，使用缓存机制""" - current_time = time.time() - - # 如果缓存有效，直接返回缓存 - if ( - self._config_cache is not None - and current_time - self._last_load_time < self._cache_ttl - ): - return self._config_cache - - try: - with open(self.config_file, "a+") as f: - with FileLock(f, timeout=self._lock_timeout): - f.seek(0) - content = f.read() - config = yaml.safe_load(content) if content else {} - self._config_cache = config - self._last_load_time = current_time - return config - except (TimeoutError, IOError) as e: - print(f"加载配置文件失败: {e}") - return {} - except Exception as e: - print(f"加载配置文件时发生未知错误: {e}") - return {} - - def _save_config(self, config: Dict): - """保存配置到文件，使用文件锁保护""" - try: - with open(self.config_file, "w") as f: - with FileLock(f, timeout=self._lock_timeout): - yaml.dump(config, f, allow_unicode=True) - self._config_cache = config - self._last_load_time = time.time() - except (TimeoutError, IOError) as e: - print(f"保存配置文件失败: {e}") - raise - except Exception as e: - print(f"保存配置文件时发生未知错误: {e}") - raise - - def get_wakeup_response(self, voice: str) -> Dict: - voice = hashlib.md5(voice.encode()).hexdigest() - """获取唤醒词回复配置""" - config = self._load_config() - - if not config or voice not in config: - return None - - # 检查文件大小 - file_path = config[voice]["file_path"] - if not os.path.exists(file_path) or os.stat(file_path).st_size < (15 * 1024): - return None - - return config[voice] - - def update_wakeup_response(self, voice: str, file_path: str, text: str): - """更新唤醒词回复配置""" - try: - # 过滤表情符号 - filtered_text = re.sub(r'[\U0001F600-\U0001F64F\U0001F900-\U0001F9FF]', '', text) - - config = self._load_config() - voice_hash = hashlib.md5(voice.encode()).hexdigest() - config[voice_hash] = { - "voice": voice, - "file_path": file_path, - "time": time.time(), - "text": filtered_text, - } - self._save_config(config) - except Exception as e: - print(f"更新唤醒词回复配置失败: {e}") - raise - - def generate_file_path(self, voice: str) -> str: - """生成音频文件路径，使用voice的哈希值作为文件名""" - try: - # 生成voice的哈希值 - voice_hash = hashlib.md5(voice.encode()).hexdigest() - file_path = os.path.join(self.assets_dir, f"{voice_hash}.wav") - - # 如果文件已存在，先删除 - if os.path.exists(file_path): - try: - os.remove(file_path) - except Exception as e: - print(f"删除已存在的音频文件失败: {e}") - raise - - return file_path - except Exception as e: - print(f"生成音频文件路径失败: {e}") - raise