Digital_Human/voxcpm2_api_tts.py

"""
VoxCPM2 API 调用模式 TTS
通过 HTTP API 调用独立的 VoxCPM2 模型服务，避免在数字人进程中加载模型
"""
from __future__ import annotations
import time
import os
import re
import hashlib
import numpy as np
import soundfile as sf
import resampy
import requests
from io import BytesIO
from queue import Queue
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 VoxCPM2APITTS:
    """通过 API 调用 VoxCPM2 语音合成服务"""
    
    def __init__(self, opt, parent: BaseReal):
        self.opt = opt
        self.parent = parent
        
        # API 配置（中性模式：降低 CFG，不使用 prompt）
        self.api_url = getattr(opt, 'VOXCPM2_API_URL', 'http://localhost:6003')
        self.ref_audio_path = getattr(opt, 'VOXCPM2_REF_WAV', 'voice_output.wav')
        self.ref_text = getattr(opt, 'VOXCPM2_REF_TEXT', '你好，买水果，卖水果，新鲜的水果。')
        # 中性模式：降低 CFG 值，减少情绪模仿
        self.cfg_value = getattr(opt, 'CFG_VALUE', 1.5)  # 从 2.0 降低到 1.5
        self.inference_timesteps = getattr(opt, 'INFERENCE_TIMESTEPS', 10)
        
        # 音频参数
        self.fps = opt.fps
        self.sample_rate = 16000
        self.chunk = self.sample_rate // self.fps  # 320 samples per chunk (20ms)
        
        # 消息队列
        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()
        
        # 预生成音频配置（已禁用，使用实时克隆）
        self.pre_gen_dir = "./data/pre_generated_tts"
        self.audio_map = {}  # 不再加载预生成音频
        self.use_pre_gen = False  # 禁用预生成音频
        
        # Image_Analysis 音频目录配置
        self.image_analysis_dir = "/mnt/nvme1data/Digital_Human/Image_Analysis/wav/wav"
        self.played_audio_files = set()
        
        # 实时克隆音频保存目录
        self.kelong_dir = "/mnt/nvme1data/Digital_Human/Image_Analysis/knowledge_kelong"
        os.makedirs(self.kelong_dir, exist_ok=True)
        logger.info(f"📁 实时克隆音频保存目录：{self.kelong_dir}")
        
        # 移除预生成音频循环播放配置
        # self.pre_gen_index = 0  # 不再需要
        # self.is_playing_pre_gen_loop = False  # 不再需要
        
        # 检查 API 服务
        self._check_api_service()
        
        logger.info(f"VoxCPM2 API TTS 初始化完成（实时克隆模式）:")
        logger.info(f"  API 地址：{self.api_url}")
        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"  预生成音频：已禁用（使用实时克隆）")
        logger.info(f"  克隆音频保存：{self.kelong_dir}")
        logger.info(f"  Image_Analysis 目录：{self.image_analysis_dir}")
    
    def _check_api_service(self):
        """检查 API 服务是否可用"""
        try:
            logger.info(f"🔍 检查 VoxCPM2 API 服务：{self.api_url}/health")
            resp = requests.get(f"{self.api_url}/health", timeout=5)
            if resp.status_code == 200:
                health = resp.json()
                logger.info(f"✅ VoxCPM2 API 服务正常")
                logger.info(f"   模型：{health.get('model', 'Unknown')}")
                logger.info(f"   设备：{health.get('device', 'Unknown')}")
                logger.info(f"   采样率：{health.get('sample_rate', 'Unknown')} Hz")
            else:
                raise Exception(f"API 服务异常：HTTP {resp.status_code}")
        except requests.exceptions.ConnectionError:
            logger.error(f"❌ 无法连接到 VoxCPM2 API 服务：{self.api_url}")
            logger.error("请先启动 API 服务：")
            logger.error("  cd /mnt/nvme1data/model && python voxcpm2_api.py")
            raise
        except Exception as e:
            logger.error(f"❌ VoxCPM2 API 服务检查失败：{e}")
            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
    
    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:
        """加载预生成音频映射表"""
        import json
        import os
        
        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"""
        import hashlib
        import os
        
        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]
            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):
        """获取最新的未播放的图片分析音频文件"""
        import re
        import time
        
        if not os.path.exists(self.image_analysis_dir):
            return None
            
        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)
                    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]
                
            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 generate_audio_via_api(self, text: str, save_to_kelong: bool = True) -> np.ndarray:
        """通过 API 生成音频，返回 16kHz numpy 数组
        
        Args:
            text: 要转换的文本
            save_to_kelong: 是否保存到 knowledge_kelong 目录
        """
        try:
            logger.info(f"🎤 调用 VoxCPM2 API 生成音频（中性模式）: {text[:30]}...")
            
            # 中性模式配置：只使用 reference_wav_path，不使用 prompt
            payload = {
                "text": text,
                "reference_wav_path": self.ref_audio_path,  # 音色参考
                # 不使用 prompt_wav_path 和 prompt_text（避免情绪模仿）
                "cfg_value": self.cfg_value,  # 降低 CFG，减少情绪
                "inference_timesteps": self.inference_timesteps
            }
            
            start_time = time.time()
            resp = requests.post(
                f"{self.api_url}/v1/tts/generate_audio",
                json=payload,
                timeout=120
            )
            
            if resp.status_code != 200:
                raise Exception(f"API 返回错误：HTTP {resp.status_code} - {resp.text}")
            
            api_sample_rate = int(resp.headers.get('X-Sample-Rate', 48000))
            generate_time = time.time() - start_time
            
            logger.info(f"✅ API 音频生成完成，耗时：{generate_time:.2f}秒")
            logger.info(f"   API 返回采样率：{api_sample_rate} Hz")
            
            wav_data = BytesIO(resp.content)
            stream, sr = sf.read(wav_data, dtype='float32')
            
            if stream.ndim > 1:
                stream = stream[:, 0]
            
            # 保存原始 48kHz 音频到 knowledge_kelong 目录
            if save_to_kelong:
                try:
                    timestamp = int(time.time())
                    # 使用文本前 20 个字符作为文件名（去除特殊字符）
                    safe_text = re.sub(r'[\\/*?:"<>|]', '', text[:20])
                    filename = f"{timestamp}_{safe_text}.wav"
                    kelong_path = os.path.join(self.kelong_dir, filename)
                    
                    # 保存 48kHz 原始音频（重采样之前）
                    sf.write(kelong_path, stream, sr, format='WAV')
                    logger.info(f"💾 已保存克隆音频：{kelong_path}")
                    logger.info(f"   采样率：{sr} Hz，文件大小：{os.path.getsize(kelong_path)} bytes")
                except Exception as e:
                    logger.error(f"保存克隆音频失败：{e}")
            
            # 重采样到 16kHz（数字人播放需要）
            if sr != self.sample_rate:
                logger.info(f"   重采样：{sr} Hz → {self.sample_rate} Hz")
                resample_start = time.time()
                # 使用高质量重采样滤波器，避免滋啦声和高频噪声
                stream = resampy.resample(
                    x=stream, 
                    sr_orig=sr, 
                    sr_new=self.sample_rate,
                    filter='kaiser_best'  # 使用 Kaiser 最佳滤波器
                )
                logger.info(f"   重采样完成，耗时：{time.time()-resample_start:.2f}秒")
            
            # 添加淡入淡出处理，避免首尾爆音和咔嗒声
            fade_duration = 0.015  # 15ms 淡入淡出
            fade_samples = int(fade_duration * self.sample_rate)
            if len(stream) > 2 * fade_samples:
                # 创建淡入淡出曲线（使用余弦曲线更平滑）
                fade_in = (1.0 - np.cos(np.linspace(0, np.pi, fade_samples))) / 2.0
                fade_out = (1.0 - np.cos(np.linspace(np.pi, 0, fade_samples))) / 2.0
                
                # 应用淡入淡出
                stream[:fade_samples] *= fade_in
                stream[-fade_samples:] *= fade_out
                logger.info(f"   已应用淡入淡出处理（{fade_duration*1000:.0f}ms）")
            
            logger.info(f"   音频采样点数：{stream.shape[0]}")
            logger.info(f"   音频时长：{stream.shape[0] / self.sample_rate:.2f}s")
            
            return stream
            
        except requests.exceptions.Timeout:
            logger.error("❌ API 请求超时（120秒）")
            raise Exception("语音生成超时")
        except Exception as e:
            logger.error(f"❌ API 调用失败：{e}")
            raise
    
    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
        
        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:
                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:
                sleep_time = min(delay, 0.04)
                time.sleep(sleep_time)
            elif delay < -0.1:
                logger.warning(f"   ⚠️ 播放落后 {abs(delay):.3f}s，跳过一些帧以追赶进度")
        
        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 put_msg_txt(self, msg: str, datainfo: dict = {}):
        """将文本消息放入队列"""
        if len(msg) > 0:
            self.msgqueue.put((msg, datainfo))
    
    def put_high_priority_msg(self, msg: str, datainfo: dict = {}):
        """添加高优先级消息"""
        if len(msg) > 0:
            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()
                    
                    logger.info("⏳ 等待文件写入完成...")
                    wait_start = time.time()
                    last_size = 0
                    stable_count = 0
                    max_wait = 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:
                                    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)
                        except:
                            time.sleep(0.15)
                    
                    time_since_creation = detect_time - file_timestamp
                    
                    logger.info("★★★ 发现新的图片分析音频，立即播放 ★★★")
                    logger.info(f"   文件名：{filename}")
                    logger.info(f"   ⏱️ 文件生成到检测耗时：{time_since_creation:.3f}秒")
                    
                    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"   📊 从文件生成到开始播放：{play_start_time - file_timestamp:.3f}秒")
                    logger.info(f"   📊 实际播放时长：{play_end_time - play_start_time:.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('voxcpm2_api_tts thread stop')
    
    def txt_to_audio(self, msg: tuple[str, dict]):
        """将文本转换为音频（通过 API）"""
        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]}...")
            
            stream = self.generate_audio_via_api(text)
            
            if self.state != State.RUNNING:
                return
            
            if self.interrupt_flag.is_set() or not self.high_priority_queue.empty():
                return
            
            self._play_audio_stream(stream, text, textevent)
            
        except Exception as e:
            logger.error(f"VoxCPM2 API TTS 失败：{e}")
    
    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:
            logger.info(f"🎵 播放图片分析音频：{filename}")
            logger.info(f"   音频文件路径：{file_path}")
                
            stream, sample_rate = sf.read(file_path, dtype='float32')
                
            logger.info(f"   音频原始信息：sample_rate={sample_rate}, shape={stream.shape}")
                
            if stream.ndim > 1:
                stream = stream[:, 0]
                
            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"   处理后音频信息：shape={stream.shape}, chunks={stream.shape[0]//self.chunk}")
                
            eventpoint = {'status': 'start', 'text': '图片描述'}
            completed = self._play_audio_stream(stream, f"[图片描述] {filename}", eventpoint)
                
            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