Digital_Human/qwen3tts.py

from __future__ import annotations
import time
import numpy as np
import soundfile as sf
import resampy
import asyncio
import os
import queue
import json
import hashlib
import re
from queue import Queue
from io import BytesIO
from threading import Thread, Event
from enum import Enum
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from basereal import BaseReal

from logger import logger

class State(Enum):
    RUNNING = 0
    PAUSE = 1

class Qwen3TTS:
    def __init__(self, opt, parent: BaseReal):
        self.opt = opt
        self.parent = parent
        
        self.fps = opt.fps  # 20 ms per frame
        self.sample_rate = 16000
        self.chunk = self.sample_rate // self.fps  # 320 samples per chunk (20ms * 16000 / 1000)
        self.input_stream = BytesIO()
        
        self.msgqueue = Queue()
        self.high_priority_queue = Queue()
        self.image_description_queue = Queue()  # 超优先级队列，用于图像描述
        self.is_playing_image_description = False  # 标记是否正在播放图像描述
        self.state = State.RUNNING
        self.interrupted_messages = []
        self.current_msg = None
        self.current_msg_progress = 0
        self.interrupt_flag = Event()
        
        # VoxCPM2 配置（替代 Qwen3-TTS API）
        self.voxcpm2_model_path = getattr(opt, 'VOXCPM2_MODEL_PATH', 'VoxCPM2')
        self.ref_audio_path = getattr(opt, 'VOXCPM2_REF_WAV', 'voice_output.wav')
        self.ref_text = getattr(opt, 'VOXCPM2_REF_TEXT', '你好，买水果，卖水果，新鲜的水果。')
        self.cfg_value = getattr(opt, 'CFG_VALUE', 2.5)
        self.inference_timesteps = getattr(opt, 'INFERENCE_TIMESTEPS', 8)
        
        # 加载 VoxCPM2 模型
        self.voxcpm2_model = None
        self._load_voxcpm2_model()
        
        # 预生成音频配置
        self.pre_gen_dir = "./data/pre_generated_tts"
        self.audio_map = self._load_audio_map()
        self.use_pre_gen = len(self.audio_map) > 0
        
        # Image_Analysis 音频目录配置（修改为 wav/wav 子目录）
        self.image_analysis_dir = "/mnt/nvme1data/Digital_Human/Image_Analysis/wav/wav"
        self.played_audio_files = set()  # 已播放的音频文件集合
        
        # 预生成音频循环播放配置
        self.pre_gen_index = 0  # 当前播放索引
        self.is_playing_pre_gen_loop = False  # 是否在循环播放预生成音频
        
        logger.info(f"Qwen3TTS 初始化完成:")
        logger.info(f"  VoxCPM2 模型路径：{self.voxcpm2_model_path}")
        logger.info(f"  参考音频：{self.ref_audio_path}")
        logger.info(f"  参考文本：{self.ref_text}")
        logger.info(f"  CFG 值：{self.cfg_value}")
        logger.info(f"  推理步数：{self.inference_timesteps}")
        logger.info(f"  预生成音频：{len(self.audio_map)} 条")
        logger.info(f"  是否使用预生成：{self.use_pre_gen}")
        logger.info(f"  Image_Analysis 目录：{self.image_analysis_dir}")
        logger.info(f"  预生成循环播放：启用")
    
    def _load_voxcpm2_model(self):
        """加载 VoxCPM2 模型（单例模式）"""
        try:
            import torch
            from voxcpm import VoxCPM
            
            # 禁用 TorchDynamo，避免 scaled_dot_product_attention 兼容性错误
            torch._dynamo.config.suppress_errors = True
            torch.compiler.disable()
            
            logger.info(f"🔄 加载 VoxCPM2 模型: {self.voxcpm2_model_path}")
            logger.info("   禁用 TorchDynamo 以避免兼容性问题")
            
            # 清理显存
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
                logger.info(f"   清理显存完成")
            
            # 加载模型
            self.voxcpm2_model = VoxCPM.from_pretrained(
                self.voxcpm2_model_path,
                load_denoiser=False,
                optimize=False
            )
            
            logger.info("✅ VoxCPM2 模型加载成功")
            
        except Exception as e:
            logger.error(f"❌ VoxCPM2 模型加载失败: {e}")
            logger.error("   请检查模型路径是否正确，以及 voxcpm 包是否已安装")
            raise
    
    def reset_interrupt_flag(self):
        """重置打断标志"""
        self.interrupt_flag.clear()
    
    def set_interrupt_flag(self):
        """设置打断标志"""
        self.interrupt_flag.set()
    
    def flush_talk(self):
        """停止当前播放并清空待处理的消息队列"""
        with self.msgqueue.mutex:
            remaining_msgs = list(self.msgqueue.queue)
            if remaining_msgs:
                self.interrupted_messages.extend(remaining_msgs)
            self.msgqueue.queue.clear()
        
        if self.current_msg:
            self.interrupted_messages.append(self.current_msg)
        
        self.current_msg = None
        self.state = State.PAUSE
        
        if hasattr(self, 'input_stream') and hasattr(self.input_stream, 'seek') and hasattr(self.input_stream, 'truncate'):
            self.input_stream.seek(0)
            self.input_stream.truncate()
    
    def resume_interrupted(self):
        """恢复播放被中断的消息"""
        if self.interrupted_messages:
            with self.msgqueue.mutex:
                for msg in self.interrupted_messages:
                    self.msgqueue.put(msg)
            self.interrupted_messages.clear()
            self.state = State.RUNNING
            return True
        return False
    
    def _load_audio_map(self) -> dict:
        """加载预生成音频映射表"""
        map_path = os.path.join(self.pre_gen_dir, "audio_map.json")
        if os.path.exists(map_path):
            try:
                with open(map_path, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception as e:
                logger.warning(f"加载预生成音频映射表失败: {e}")
        return {}
    
    def _get_pre_generated_audio(self, text: str):
        """获取预生成的音频数据，如果不存在返回 None"""
        if not self.use_pre_gen:
            return None
            
        text_hash = hashlib.md5(text.encode('utf-8')).hexdigest()[:16]
        audio_path = os.path.join(self.pre_gen_dir, f"{text_hash}.wav")
            
        if not os.path.exists(audio_path):
            return None
            
        try:
            # 读取音频文件
            stream, sample_rate = sf.read(audio_path, dtype='float32')
                
            # 转为单声道
            if stream.ndim > 1:
                stream = stream[:, 0]
                
            # 重采样到 16kHz
            if sample_rate != self.sample_rate and stream.shape[0] > 0:
                stream = resampy.resample(x=stream, sr_orig=sample_rate, sr_new=self.sample_rate)
                
            return stream
        except Exception as e:
            logger.warning(f"读取预生成音频失败：{e}")
            return None
        
    def _get_newest_image_analysis_audio(self):
        """获取最新的未播放的图片分析音频文件"""
        if not os.path.exists(self.image_analysis_dir):
            return None
            
        # 获取所有 wav 文件并按修改时间排序
        try:
            audio_files = []
            all_files = os.listdir(self.image_analysis_dir)
            logger.info(f"🔍 Image_Analysis 目录共有 {len(all_files)} 个文件")
            logger.info(f"   已播放文件数：{len(self.played_audio_files)}")
            
            for f in all_files:
                if f.endswith('.wav'):
                    file_path = os.path.join(self.image_analysis_dir, f)
                    
                    # 【修改】尝试从文件名中提取时间戳（格式：hifi_clone_YYYYMMDD_HHMMSS.wav）
                    match = re.search(r'hifi_clone_(\d{8}_\d{6})', f)
                    if match:
                        # 从文件名解析时间
                        time_str = match.group(1)
                        file_timestamp = time.mktime(time.strptime(time_str, "%Y%m%d_%H%M%S"))
                        logger.info(f"   📁 {f} - 文件名时间：{time.strftime('%H:%M:%S', time.localtime(file_timestamp))}, 已播放：{f in self.played_audio_files}")
                    else:
                        # 如果文件名没有时间戳，使用文件修改时间
                        file_timestamp = os.path.getmtime(file_path)
                        logger.info(f"   📁 {f} - 修改时间：{time.strftime('%H:%M:%S', time.localtime(file_timestamp))}, 已播放：{f in self.played_audio_files}")
                    
                    is_played = f in self.played_audio_files
                    if not is_played:
                        audio_files.append((file_path, file_timestamp, f))
            
            if not audio_files:
                logger.info("   ⚠️ 没有未播放的音频文件")
                return None
                
            # 按时间戳排序，最新的在前
            audio_files.sort(key=lambda x: x[1], reverse=True)
            
            newest_file = audio_files[0][2]
            logger.info(f"   ✅ 发现最新未播放文件：{newest_file}")
            
            # 返回最新的文件
            return audio_files[0]
        except Exception as e:
            logger.error(f"获取图片分析音频失败：{e}", exc_info=True)
            return None
        
    def _get_next_pre_gen_audio(self):
        """获取下一条预生成音频（循环播放）"""
        if not self.use_pre_gen or len(self.audio_map) == 0:
            return None
            
        try:
            # 获取所有预生成音频文件
            audio_files = sorted([
                f for f in os.listdir(self.pre_gen_dir) 
                if f.endswith('.wav')
            ])
                
            if not audio_files:
                return None
                
            # 循环索引
            audio_file = audio_files[self.pre_gen_index % len(audio_files)]
            self.pre_gen_index += 1
                
            # 读取音频
            audio_path = os.path.join(self.pre_gen_dir, audio_file)
            stream, sample_rate = sf.read(audio_path, dtype='float32')
                
            # 转为单声道
            if stream.ndim > 1:
                stream = stream[:, 0]
                
            # 重采样到 16kHz
            if sample_rate != self.sample_rate and stream.shape[0] > 0:
                stream = resampy.resample(x=stream, sr_orig=sample_rate, sr_new=self.sample_rate)
                
            return stream
        except Exception as e:
            logger.error(f"获取预生成音频失败：{e}")
            return None
        
    def _play_image_analysis_audio(self):
        """播放图片分析音频（次高优先级）"""
        result = self._get_newest_image_analysis_audio()
            
        if result is None:
            return False
            
        file_path, file_timestamp, filename = result
            
        try:
            read_start = time.time()  # 【时间戳】开始读取文件
            logger.info(f"🎵 播放图片分析音频：{filename}")
            logger.info(f"   音频文件路径：{file_path}")
            logger.info(f"   音频修改时间：{file_timestamp}")
                
            # 读取音频文件
            stream, sample_rate = sf.read(file_path, dtype='float32')
            read_end = time.time()  # 【时间戳】读取完成
                
            logger.info(f"   音频原始信息：sample_rate={sample_rate}, shape={stream.shape}, duration={stream.shape[0]/sample_rate:.2f}s")
            logger.info(f"   ⏱️ 文件读取耗时：{(read_end - read_start)*1000:.1f}毫秒")
                
            # 转为单声道
            if stream.ndim > 1:
                stream = stream[:, 0]
                logger.info(f"   转换为单声道")
                
            # 重采样到 16kHz
            resample_start = time.time()  # 【时间戳】开始重采样
            if sample_rate != self.sample_rate and stream.shape[0] > 0:
                stream = resampy.resample(x=stream, sr_orig=sample_rate, sr_new=self.sample_rate)
                logger.info(f"   重采样到 16kHz")
            resample_end = time.time()  # 【时间戳】重采样完成
            logger.info(f"   ⏱️ 重采样耗时：{(resample_end - resample_start)*1000:.1f}毫秒")
                
            logger.info(f"   处理后音频信息：shape={stream.shape}, chunks={stream.shape[0]//self.chunk}")
                
            # 播放音频流
            eventpoint = {'status': 'start', 'text': '图片描述'}
            logger.info(f"▶️ 开始调用 _play_audio_stream")
            completed = self._play_audio_stream(stream, f"[图片描述] {filename}", eventpoint)
            logger.info(f"◀️ _play_audio_stream 返回，completed={completed}")
                
            # 标记为已播放
            self.played_audio_files.add(filename)
                
            if completed:
                logger.info(f"✅ 图片分析音频播放完成：{filename}")
            else:
                logger.info(f"⚠️ 图片分析音频被中断：{filename}")
            return completed
                
        except Exception as e:
            logger.error(f"播放图片分析音频失败：{e}", exc_info=True)
            return False
    
    def put_msg_txt(self, msg: str, datainfo: dict = {}):
        """将文本消息放入队列"""
        if len(msg) > 0:
            if len(msg) > 100:
                sentences = re.split(r'([。！？.!?])', msg)
                parts = []
                for i in range(0, len(sentences) - 1, 2):
                    sentence = sentences[i]
                    punctuation = sentences[i + 1] if i + 1 < len(sentences) else ''
                    if sentence.strip():
                        parts.append(sentence.strip() + punctuation)
                
                if len(parts) > 1:
                    for part in parts:
                        if part.strip():
                            self.msgqueue.put((part, datainfo))
                    return
            
            self.msgqueue.put((msg, datainfo))
    
    def put_high_priority_msg(self, msg: str, datainfo: dict = {}):
        """添加高优先级消息"""
        if len(msg) > 0:
            if len(msg) > 100:
                sentences = re.split(r'([。！？.!?])', msg)
                parts = []
                for i in range(0, len(sentences) - 1, 2):
                    sentence = sentences[i]
                    punctuation = sentences[i + 1] if i + 1 < len(sentences) else ''
                    if sentence.strip():
                        parts.append(sentence.strip() + punctuation)
                
                if len(parts) > 1:
                    for part in parts:
                        if part.strip():
                            self.high_priority_queue.put((part, datainfo))
                    return
            
            self.high_priority_queue.put((msg, datainfo))
    
    def put_image_description(self, msg: str, datainfo: dict = {}):
        """添加图像描述消息，具有最高优先级，会立即中断当前播放"""
        if len(msg) > 0:
            logger.info(f"收到图像描述消息，长度：{len(msg)}字")
            # 图像描述通常是短句，不需要分割
            self.image_description_queue.put((msg, datainfo))
    
    def render(self, quit_event):
        """启动 TTS 处理线程"""
        process_thread = Thread(target=self.process_tts, args=(quit_event,))
        process_thread.start()
    
    def process_tts(self, quit_event):
        """处理 TTS 消息队列"""
        # 标记是否正在播放预生成音频，用于支持中断后恢复
        was_playing_pre_gen = False
        
        while not quit_event.is_set():
            try:
                msg = None
                
                # ★★★★★ 优先级 1: 用户问答（最高优先级，随时中断任何播放）
                if not self.high_priority_queue.empty():
                    msg = self.high_priority_queue.get_nowait()
                    logger.info("★★★★★ 处理用户问答（最高优先级） ★★★★★")
                    self.state = State.RUNNING
                    
                    # 播放用户问答
                    if msg and len(msg) >= 2 and isinstance(msg[0], str) and msg[0].strip():
                        self.txt_to_audio(msg)
                    
                    # 用户问答播放完成后，如果有预生成音频，继续播放
                    if was_playing_pre_gen:
                        logger.info("用户问答播放完成，恢复播放预生成音频")
                    continue
                
                # ★★★ 优先级 2: Image_Analysis 新音频（次高优先级，可中断预生成音频）
                check_start_time = time.time()  # 【时间戳】开始检查的时间
                newest_audio = self._get_newest_image_analysis_audio()
                if newest_audio is not None:
                    file_path, file_timestamp, filename = newest_audio
                    detect_time = time.time()  # 【时间戳】检测到文件的时间
                    
                    # 【新增】等待文件写入完成
                    # 检查文件大小是否稳定（500ms内没有变化）
                    logger.info(f"⏳ 等待文件写入完成...")
                    wait_start = time.time()
                    last_size = 0
                    stable_count = 0
                    max_wait = 10  # 最多等待10秒
                    
                    while (time.time() - wait_start) < max_wait:
                        try:
                            current_size = os.path.getsize(file_path)
                            if current_size == last_size and current_size > 0:
                                stable_count += 1
                                if stable_count >= 3:  # 连续3次大小不变，认为文件写入完成
                                    wait_end = time.time()
                                    logger.info(f"✅ 文件写入完成，等待耗时：{(wait_end - wait_start)*1000:.0f}ms，文件大小：{current_size} bytes")
                                    break
                            else:
                                stable_count = 0
                                last_size = current_size
                            time.sleep(0.15)  # 每150ms检查一次
                        except:
                            time.sleep(0.15)
                    
                    time_since_creation = detect_time - file_timestamp  # 【计算】文件生成到检测到的时间差
                    
                    logger.info("★★★ 发现新的图片分析音频，立即播放 ★★★")
                    logger.info(f"   文件名：{filename}")
                    logger.info(f"   文件创建时间（从文件名）：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(file_timestamp))}")
                    logger.info(f"   检测到文件时间：{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(detect_time))}.{int(detect_time % 1 * 1000):03d}")
                    logger.info(f"   ⏱️ 文件生成到检测耗时：{time_since_creation:.3f}秒")
                    logger.info(f"   检查开始时间：{time.strftime('%H:%M:%S', time.localtime(check_start_time))}.{int(check_start_time % 1 * 1000):03d}")
                    logger.info(f"   ⏱️ 检查耗时：{(detect_time - check_start_time)*1000:.1f}毫秒")
                    
                    was_playing_pre_gen = True  # 标记之前正在播放预生成音频
                    play_start_time = time.time()  # 【时间戳】开始播放准备的时间
                    self._play_image_analysis_audio()
                    play_end_time = time.time()  # 【时间戳】播放完成的时间
                    
                    logger.info("=" * 80)
                    logger.info("⏱️⏱️⏱️ 【图片分析音频时间统计】 ⏱️⏱️⏱️")
                    logger.info(f"   文件创建时间：{time.strftime('%H:%M:%S', time.localtime(file_timestamp))}")
                    logger.info(f"   开始播放时间：{time.strftime('%H:%M:%S', time.localtime(play_start_time))}.{int(play_start_time % 1 * 1000):03d}")
                    logger.info(f"   播放完成时间：{time.strftime('%H:%M:%S', time.localtime(play_end_time))}.{int(play_end_time % 1 * 1000):03d}")
                    logger.info(f"   📊 从文件生成到开始播放：{play_start_time - file_timestamp:.3f}秒")
                    logger.info(f"   📊 播放处理耗时（读取+转换）：{play_start_time - detect_time:.3f}秒")
                    logger.info(f"   📊 实际播放时长：{play_end_time - play_start_time:.3f}秒")
                    logger.info(f"   📊 总耗时（生成到播放完）：{play_end_time - file_timestamp:.3f}秒")
                    logger.info("=" * 80)
                    
                    logger.info("图片描述播放完成，将继续播放预生成音频")
                    continue
                
                # ★ 优先级 3: 循环播放预生成音频（基础优先级，默认状态）
                pre_gen_stream = self._get_next_pre_gen_audio()
                if pre_gen_stream is not None:
                    logger.info(f"★ 循环播放预生成音频 #{self.pre_gen_index}")
                    eventpoint = {'status': 'start', 'text': '预生成介绍'}
                    was_playing_pre_gen = True
                    self._play_audio_stream(pre_gen_stream, "[预生成介绍]", eventpoint)
                    continue
                
                # 没有预生成音频时，重置标记并等待普通消息
                was_playing_pre_gen = False
                try:
                    msg = self.msgqueue.get(block=True, timeout=0.5)
                except queue.Empty:
                    continue
                
                if msg and len(msg) >= 2 and isinstance(msg[0], str) and not msg[0].strip():
                    continue
                
                self.current_msg = msg
                self.current_msg_progress = 0
                
                # 处理普通消息
                if self.state == State.RUNNING:
                    self.txt_to_audio(msg)
                
                self.current_msg = None
                self.current_msg_progress = 0
                
            except Exception as e:
                logger.error(f"process_tts 错误：{e}")
                continue
        
        logger.info('qwen3tts thread stop')
    
    def _play_audio_stream(self, stream: np.ndarray, text: str, textevent: dict) -> bool:
        """播放音频流，返回是否完整播放（带速度控制）"""
        streamlen = stream.shape[0]
        idx = 0
        total_chunks = streamlen // self.chunk
        chunk_count = 0
        
        logger.info(f"   📦 音频流总帧数：{streamlen}, chunk 大小：{self.chunk}, 预计发送 {total_chunks} 个 chunk")
        logger.info(f"   ⏱️ 预计播放时长：{total_chunks * 0.02:.2f}秒")
        
        start_time = time.time()
        expected_frame_time = 0.0  # 期望的帧时间戳
        frame_duration = 0.02  # 20ms per frame
        
        while streamlen >= self.chunk and self.state == State.RUNNING:
            if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                logger.info("   ⚠️ 播放过程中发现高优先级消息，中断播放")
                return False
            
            eventpoint = {}
            streamlen -= self.chunk
            chunk_count += 1
            if idx == 0:
                eventpoint = {'status': 'start', 'text': text}
                eventpoint.update(**textevent)
                logger.info(f"   ▶️ 开始播放第 {chunk_count}/{total_chunks} 个 chunk (START)")
            elif streamlen < self.chunk:
                eventpoint = {'status': 'end', 'text': text}
                eventpoint.update(**textevent)
                logger.info(f"   ◀️ 开始播放第 {chunk_count}/{total_chunks} 个 chunk (END)")
            else:
                # 中间的 chunk，每 10 个记录一次
                if chunk_count % 10 == 1:
                    logger.info(f"   ➡️ 播放第 {chunk_count}/{total_chunks} 个 chunk")
            
            if self.state != State.RUNNING:
                logger.info(f"   ⚠️ 状态不是 RUNNING，停止播放")
                return False
            
            if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                logger.info("   ⚠️ 发送音频帧前发现高优先级消息，中断播放")
                return False
            
            # 发送音频帧
            self.parent.put_audio_frame(stream[idx:idx + self.chunk], eventpoint)
            idx += self.chunk
            
            # 速度控制：等待到正确的播放时间点
            expected_frame_time += frame_duration
            elapsed = time.time() - start_time
            delay = expected_frame_time - elapsed
            
            if delay > 0.005:  # 如果延迟超过 5ms，则等待
                # 计算需要等待的时间
                sleep_time = min(delay, 0.04)  # 最多等待 40ms
                time.sleep(sleep_time)
            elif delay < -0.1:  # 如果落后超过 100ms，跳过一些帧追赶
                logger.warning(f"   ⚠️ 播放落后 {abs(delay):.3f}s，跳过一些帧以追赶进度")
                # 不重置 expected_frame_time，让下一帧继续按正常节奏发送
        
        actual_duration = time.time() - start_time
        logger.info(f"   ✅ 所有 {chunk_count} 个 chunk 发送完成，实际耗时：{actual_duration:.2f}s (预期：{chunk_count * 0.02:.2f}s)")
        return True
    
    def txt_to_audio(self, msg: tuple[str, dict]):
        """将文本转换为音频（优先使用预生成音频）"""
        text, textevent = msg
        t = time.time()
        
        if self.state != State.RUNNING:
            return
        
        if not text.strip():
            return
        
        self.reset_interrupt_flag()
        
        if not self.high_priority_queue.empty():
            logger.info("发现高优先级消息，跳过当前普通消息处理")
            return
        
        try:
            # 首先尝试使用预生成音频
            pre_gen_stream = self._get_pre_generated_audio(text)
            
            if pre_gen_stream is not None:
                # 使用预生成音频
                logger.info(f"使用预生成音频: {text[:30]}...")
                
                if self.state != State.RUNNING:
                    return
                
                if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                    return
                
                self._play_audio_stream(pre_gen_stream, text, textevent)
                logger.info(f'-------预生成音频播放完成，耗时:{time.time()-t:.4f}s')
                return
            
            # 没有预生成音频，调用 API 生成
            logger.info(f"调用 API 生成音频: {text[:30]}...")
            
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            
            task = loop.create_task(self.__generate_audio(text))
            
            while not task.done():
                if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                    task.cancel()
                    logger.info("TTS 请求被高优先级消息中断")
                    break
                loop.run_until_complete(asyncio.sleep(0.05))
            
            try:
                loop.run_until_complete(task)
            except asyncio.CancelledError:
                logger.info("TTS 请求被中断")
                self.input_stream.seek(0)
                self.input_stream.truncate()
                return
            
            logger.info(f'-------qwen3 tts time:{time.time()-t:.4f}s')
            
            if self.input_stream.getbuffer().nbytes <= 0:
                logger.error('qwen3 tts err!!!!!')
                return
            
            self.input_stream.seek(0)
            stream = self.__create_bytes_stream(self.input_stream)
            
            if self.state != State.RUNNING:
                self.input_stream.seek(0)
                self.input_stream.truncate()
                return
            
            if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                self.input_stream.seek(0)
                self.input_stream.truncate()
                return
            
            self._play_audio_stream(stream, text, textevent)
            
            self.input_stream.seek(0)
            self.input_stream.truncate()
            
        except Exception as e:
            logger.error(f"qwen3tts txt_to_audio error: {e}")
    
    async def __generate_audio(self, text: str):
        """使用 VoxCPM2 本地模型生成音频"""
        try:
            import torch
            import numpy as np
            
            logger.info(f"🎤 VoxCPM2 开始生成音频: {text[:50]}...")
            
            # 检查模型是否加载成功
            if self.voxcpm2_model is None:
                raise Exception("VoxCPM2 模型未加载")
            
            # 检查参考音频是否存在
            if not os.path.exists(self.ref_audio_path):
                raise Exception(f"参考音频文件不存在：{self.ref_audio_path}")
            
            # 使用 VoxCPM2 生成音频（纯克隆模式）
            generate_start = time.time()
            wav = self.voxcpm2_model.generate(
                text=text,
                reference_wav_path=self.ref_audio_path,
                retry_badcase=False,  # 禁用 Badcase 重试，提升速度
                inference_timesteps=self.inference_timesteps,  # 推理步数
                cfg_value=self.cfg_value  # CFG 值
            )
            generate_time = time.time() - generate_start
            
            logger.info(f"✅ VoxCPM2 音频生成完成，耗时：{generate_time:.2f}秒")
            logger.info(f"   音频采样点数：{wav.shape[0]}")
            
            # VoxCPM2 输出是 48kHz，需要重采样到 16kHz
            if len(wav) > 0:
                # 重采样到 16kHz
                resample_start = time.time()
                wav_16k = resampy.resample(x=wav, sr_orig=48000, sr_new=self.sample_rate)
                resample_time = time.time() - resample_start
                logger.info(f"   重采样到 16kHz 完成，耗时：{resample_time:.2f}秒")
                
                # 转换为 float32
                wav_16k = wav_16k.astype(np.float32)
                
                # 写入输入流
                # 先将音频保存为 WAV 格式，然后读取到 BytesIO
                import io
                wav_bytes = io.BytesIO()
                sf.write(wav_bytes, wav_16k, self.sample_rate, format='WAV')
                wav_bytes.seek(0)
                
                # 读取到 input_stream
                wav_data, _ = sf.read(wav_bytes, dtype='float32')
                self.input_stream.write(wav_data.tobytes())
                
                logger.info(f"   音频数据写入 input_stream，大小：{self.input_stream.getbuffer().nbytes} bytes")
            else:
                logger.error("❌ VoxCPM2 生成的音频为空")
                raise Exception("生成的音频为空")
            
        except Exception as e:
            logger.error(f"❌ VoxCPM2 音频生成失败：{e}")
            raise

    def __create_bytes_stream(self, byte_stream):
        """将字节流转换为音频流"""
        stream, sample_rate = sf.read(byte_stream)
        logger.info(f'[INFO]tts audio stream {sample_rate}: {stream.shape}')
        stream = stream.astype(np.float32)

        if stream.ndim > 1:
            logger.info(f'[WARN] audio has {stream.shape[1]} channels, only use the first.')
            stream = stream[:, 0]
    
        if sample_rate != self.sample_rate and stream.shape[0] > 0:
            logger.info(f'[WARN] audio sample rate is {sample_rate}, resampling into {self.sample_rate}.')
            stream = resampy.resample(x=stream, sr_orig=sample_rate, sr_new=self.sample_rate)

        return stream