fix: 唤醒机制

This commit is contained in:
Sakura-RanChen
2025-09-02 11:18:04 +08:00
parent b4f4995ff9
commit c428a9ddee
15 changed files with 848 additions and 145 deletions
+55 -85
View File
@@ -53,10 +53,6 @@ class ASRProviderBase(ABC):
# 接收音频
async def receive_audio(self, conn, audio, audio_have_voice):
# 检查是否在唤醒处理锁定期内
if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic():
return
if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime":
have_voice = audio_have_voice
else:
@@ -66,14 +62,6 @@ class ASRProviderBase(ABC):
if not have_voice and not conn.client_have_voice:
conn.asr_audio = conn.asr_audio[-10:]
return
# 检查是否处于唤醒模式
if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10:
asr_audio_task = conn.asr_audio.copy()
conn.reset_vad_states()
conn.asr_audio.clear()
await self.handle_voice_stop(conn, asr_audio_task)
if conn.client_voice_stop:
asr_audio_task = conn.asr_audio.copy()
@@ -102,10 +90,27 @@ class ASRProviderBase(ABC):
if conn.voiceprint_provider and combined_pcm_data:
wav_data = self._pcm_to_wav(combined_pcm_data)
# 检查是否处于唤醒模式
wakeup_mode = getattr(conn, 'wakeup_mode', False)
# 定义ASR任务
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 声纹识别任务(公共逻辑)
# 定义声纹识别任务
def run_voiceprint():
if not wav_data:
return None
@@ -124,83 +129,48 @@ class ASRProviderBase(ABC):
logger.bind(tag=TAG).error(f"声纹识别失败: {e}")
return None
if wakeup_mode and conn.voiceprint_provider and wav_data:
conn.wakeup_mode = False
# 设置处理锁,防止后续音频片段重复处理
conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期
# 使用线程池执行器并行运行
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
# 唤醒模式:只执行声纹识别
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor:
if conn.voiceprint_provider and wav_data:
voiceprint_future = thread_executor.submit(run_voiceprint)
# 等待两个线程都完成
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
speaker_name = voiceprint_result
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, _ = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
# 记录识别结果
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
fixed_text = "嘿,你好啊"
enhanced_text = self._build_enhanced_text(fixed_text, speaker_name)
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
# 构建包含说话人信息的JSON字符串
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s")
# 使用自定义模块进行上报
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
else:
# 正常模式:执行声纹识别和ASR
def run_asr():
start_time = time.monotonic()
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
result = loop.run_until_complete(
self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format)
)
end_time = time.monotonic()
logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s")
return result
finally:
loop.close()
except Exception as e:
end_time = time.monotonic()
logger.bind(tag=TAG).error(f"ASR失败: {e}")
return ("", None)
# 使用线程池执行器并行运行
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor:
asr_future = thread_executor.submit(run_asr)
if conn.voiceprint_provider and wav_data:
voiceprint_future = thread_executor.submit(run_voiceprint)
asr_result = asr_future.result(timeout=15)
voiceprint_result = voiceprint_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": voiceprint_result}
else:
asr_result = asr_future.result(timeout=15)
results = {"asr": asr_result, "voiceprint": None}
# 处理结果
raw_text, _ = results.get("asr", ("", None))
speaker_name = results.get("voiceprint", None)
if raw_text:
logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
if speaker_name:
logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}")
# 性能监控
total_time = time.monotonic() - total_start_time
logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s")
# 检查文本长度
text_len, _ = remove_punctuation_and_length(raw_text)
self.stop_ws_connection()
if text_len > 0:
enhanced_text = self._build_enhanced_text(raw_text, speaker_name)
await startToChat(conn, enhanced_text)
enqueue_asr_report(conn, enhanced_text, asr_audio_task)
except Exception as e:
logger.bind(tag=TAG).error(f"处理语音停止失败: {e}")
@@ -478,3 +478,142 @@ class TTSProvider(TTSProviderBase):
finally:
self._monitor_task = None
def to_tts(self, text: str) -> list:
"""非流式TTS处理,用于测试及保存音频文件的场景"""
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)):
self.opus_encoder.encode_pcm_to_opus_stream(
msg,
end_of_stream=False,
callback=lambda opus: audio_data.append(opus)
)
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()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
+67 -6
View File
@@ -1,21 +1,22 @@
import os
import re
import queue
import time
import uuid
import queue
import asyncio
import threading
from typing import Callable, Any
import traceback
from core.utils import p3
import time
from datetime import datetime
from core.utils import textUtils
from typing import Callable, Any
from abc import ABC, abstractmethod
from config.logger import setup_logging
from core.utils.util import audio_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
from core.handle.sendAudioHandle import sendAudioMessage
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
from core.providers.tts.dto.dto import (
TTSMessageDTO,
SentenceType,
@@ -23,8 +24,6 @@ from core.providers.tts.dto.dto import (
InterfaceType,
)
import traceback
TAG = __name__
logger = setup_logging()
@@ -144,6 +143,68 @@ class TTSProviderBase(ABC):
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
return None
def to_tts(self, text):
text = MarkdownCleaner.clean_markdown(text)
max_repeat_time = 5
if self.delete_audio_file:
# 需要删除文件的直接转为音频数据
while max_repeat_time > 0:
try:
audio_bytes = asyncio.run(self.text_to_speak(text, None))
if audio_bytes:
audio_datas = []
audio_bytes_to_data_stream(
audio_bytes,
file_type=self.audio_file_type,
is_opus=True,
callback=lambda data: audio_datas.append(data)
)
return audio_datas
else:
max_repeat_time -= 1
except Exception as e:
logger.bind(tag=TAG).warning(
f"语音生成失败{5 - max_repeat_time + 1}次: {text},错误: {e}"
)
max_repeat_time -= 1
if max_repeat_time > 0:
logger.bind(tag=TAG).info(
f"语音生成成功: {text},重试{5 - max_repeat_time}"
)
else:
logger.bind(tag=TAG).error(
f"语音生成失败: {text},请检查网络或服务是否正常"
)
return None
else:
tmp_file = self.generate_filename()
try:
while not os.path.exists(tmp_file) and max_repeat_time > 0:
try:
asyncio.run(self.text_to_speak(text, tmp_file))
except Exception as e:
logger.bind(tag=TAG).warning(
f"语音生成失败{5 - max_repeat_time + 1}次: {text},错误: {e}"
)
# 未执行成功,删除文件
if os.path.exists(tmp_file):
os.remove(tmp_file)
max_repeat_time -= 1
if max_repeat_time > 0:
logger.bind(tag=TAG).info(
f"语音生成成功: {text}:{tmp_file},重试{5 - max_repeat_time}"
)
else:
logger.bind(tag=TAG).error(
f"语音生成失败: {text},请检查网络或服务是否正常"
)
return tmp_file
except Exception as e:
logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
return None
@abstractmethod
async def text_to_speak(self, text, output_file):
@@ -628,3 +628,104 @@ 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(self, text: str) -> list:
"""非流式生成音频数据,用于生成音频及测试场景
Args:
text: 要转换的文本
Returns:
list: 音频数据列表
"""
try:
# 创建事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 生成会话ID
session_id = uuid.uuid4().__str__().replace("-", "")
# 存储音频数据
audio_data = []
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=lambda opus_frame: audio_data.append(opus_frame))
elif res.optional.event == EVENT_SessionFinished:
break
finally:
# 清理资源
try:
await ws.close()
except:
pass
# 运行异步任务
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
@@ -1,8 +1,10 @@
import os
import time
import queue
import asyncio
import traceback
import aiohttp
import asyncio
import requests
import traceback
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -177,3 +179,57 @@ 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))
self.opus_encoder.encode_pcm_to_opus_stream(
frame,
end_of_stream=(i + frame_bytes >= len(pcm_data)),
callback=lambda opus: opus_datas.append(opus)
)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
@@ -1,7 +1,9 @@
import os
import time
import queue
import aiohttp
import asyncio
import requests
import traceback
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
@@ -195,3 +197,71 @@ 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(self, text: str) -> list:
"""非流式TTS处理,用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回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=lambda opus: opus_datas.append(opus)
)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
@@ -1,14 +1,15 @@
import asyncio
import json
import base64
import aiohttp
import numpy as np
import io
import wave
import json
import base64
import asyncio
import websockets
from core.providers.tts.base import TTSProviderBase
from config.logger import setup_logging
import numpy as np
from datetime import datetime
from config.logger import setup_logging
from core.providers.tts.base import TTSProviderBase
TAG = __name__
logger = setup_logging()
@@ -74,43 +75,9 @@ class TTSProvider(TTSProviderBase):
async def text_to_speak(self, text, output_file):
if self.protocol == "websocket":
return await self.text_streaming(text, output_file)
elif self.protocol == "http":
return await self.text(text, output_file)
else:
raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.")
async def text(self, text, output_file):
request_json = {
"text": text,
"spk_id": self.spk_id,
"speed": self.speed,
"volume": self.volume,
"sample_rate": self.sample_rate,
"save_path": self.save_path
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(self.url, json=request_json) as resp:
if resp.status == 200:
resp_json = await resp.json()
if resp_json.get("success"):
data = resp_json["result"]
audio_bytes = base64.b64decode(data["audio"])
if output_file:
with open(output_file, "wb") as file_to_save:
file_to_save.write(audio_bytes)
else:
return audio_bytes
else:
raise Exception(
f"Error: {resp_json.get('message', 'Unknown error')} while processing text: {text}")
else:
raise Exception(
f"HTTP Error: {resp.status} - {await resp.text()} while processing text: {text}")
except Exception as e:
raise Exception(f"Error during TTS HTTP request: {e} while processing text: {text}")
async def text_streaming(self, text, output_file):
try:
# 使用 websockets 异步连接到 WebSocket 服务器