| | |
|---|
| | | from core.utils import textUtils |
|---|
| | | from core.utils.util import audio_to_data |
|---|
| | | from core.providers.tts.dto.dto import SentenceType |
|---|
| | | from core.utils.audioRateController import AudioRateController |
|---|
| | | from core.utils.audioRateController import AudioRateController, audio_completion_timeout_sec |
|---|
| | | |
|---|
| | | TAG = __name__ |
|---|
| | | # 音频帧时长（毫秒） |
|---|
| | |
|---|
| | | async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text, sentence_id=None): |
|---|
| | | # 跳过旧句子残留音频 |
|---|
| | | if sentence_id is not None and sentence_id != conn.sentence_id: |
|---|
| | | conn.logger.bind(tag=TAG).warning( |
|---|
| | | f"丢弃过期TTS音频 sentence={sentence_id} current={conn.sentence_id} type={sentenceType}" |
|---|
| | | ) |
|---|
| | | return |
|---|
| | | |
|---|
| | | if conn.tts.tts_audio_first_sentence: |
|---|
| | | conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") |
|---|
| | | conn.tts.tts_audio_first_sentence = False |
|---|
| | | # speak_txt 不走 STT，补一条 tts start，让设备先进入 speaking 再收 opus。 |
|---|
| | | await send_tts_message(conn, "start") |
|---|
| | | |
|---|
| | | if sentenceType == SentenceType.FIRST: |
|---|
| | | # 同一句子的后续消息加入流控队列，其他情况立即发送 |
|---|
| | |
|---|
| | | await sendAudio(conn, audios) |
|---|
| | | # 发送句子开始消息 |
|---|
| | | if sentenceType is not SentenceType.MIDDLE: |
|---|
| | | conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}") |
|---|
| | | sent = (getattr(conn, "audio_flow_control", None) or {}).get("packet_count", 0) |
|---|
| | | conn.logger.bind(tag=TAG).info( |
|---|
| | | f"发送音频消息: {sentenceType}, {text}, opus包={sent}" |
|---|
| | | ) |
|---|
| | | |
|---|
| | | # 发送结束消息（如果是最后一个文本） |
|---|
| | | # 通话需要维持speaking状态 |
|---|
| | |
|---|
| | | async def _wait_for_audio_completion(conn: "ConnectionHandler"): |
|---|
| | | """ |
|---|
| | | 等待音频队列清空并等待预缓冲包播放完成 |
|---|
| | | |
|---|
| | | Args: |
|---|
| | | conn: 连接对象 |
|---|
| | | """ |
|---|
| | | if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: |
|---|
| | | rate_controller = conn.audio_rate_controller |
|---|
| | | conn.logger.bind(tag=TAG).debug( |
|---|
| | | f"等待音频发送完成，队列中还有 {len(rate_controller.queue)} 个包" |
|---|
| | | ) |
|---|
| | | await rate_controller.queue_empty_event.wait() |
|---|
| | | try: |
|---|
| | | queued = len(rate_controller.queue) |
|---|
| | | timeout = audio_completion_timeout_sec(queued, rate_controller.frame_duration) |
|---|
| | | conn.logger.bind(tag=TAG).debug( |
|---|
| | | f"等待音频发送完成，队列中还有 {queued} 个包，timeout={timeout:.1f}s" |
|---|
| | | ) |
|---|
| | | await asyncio.wait_for(rate_controller.queue_empty_event.wait(), timeout=timeout) |
|---|
| | | except asyncio.TimeoutError: |
|---|
| | | leftover = len(rate_controller.queue) if rate_controller else -1 |
|---|
| | | conn.logger.bind(tag=TAG).warning( |
|---|
| | | f"等待音频发送完成超时 leftover={leftover}，继续下发 tts stop" |
|---|
| | | ) |
|---|
| | | |
|---|
| | | # 等待预缓冲包播放完成 |
|---|
| | | # 前N个包直接发送，增加2个网络抖动包，需要额外等待它们在客户端播放完成 |
|---|
| | | frame_duration_ms = rate_controller.frame_duration |
|---|
| | | pre_buffer_playback_time = (PRE_BUFFER_COUNT + 2) * frame_duration_ms / 1000.0 |
|---|
| | | await asyncio.sleep(pre_buffer_playback_time) |
|---|
| | |
|---|
| | | sequence = flow_control.get("sequence", 0) |
|---|
| | | |
|---|
| | | if conn.conn_from_mqtt_gateway: |
|---|
| | | # 计算时间戳（基于播放位置） |
|---|
| | | start_time = time.time() |
|---|
| | | timestamp = int(start_time * 1000) % (2**32) |
|---|
| | | await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence) |
|---|
| | | else: |
|---|
| | | # 直接发送opus数据包 |
|---|
| | | if packet_index == 0: |
|---|
| | | conn.logger.bind(tag=TAG).info(f"开始下发TTS opus，首包bytes={len(opus_packet)}") |
|---|
| | | await conn.websocket.send(opus_packet) |
|---|
| | | |
|---|
| | | # 更新流控状态 |
|---|
