ax_speech_translate_demo_qwen_api_realtime.py

# import subprocess
import tempfile
import os
# import json
# import shutil
import time
import librosa
import torch
import argparse
import soundfile as sf
# from pathlib import Path
import cn2an
import requests
import re
import numpy as np
import onnxruntime as ort
import axengine as axe
import threading
import queue
from collections import deque

# 导入SenseVoice相关模块
from model import SinusoidalPositionEncoder
from utils.ax_model_bin import AX_SenseVoiceSmall
from utils.ax_vad_bin import AX_Fsmn_vad 
from utils.vad_utils import merge_vad
from funasr.tokenizer.sentencepiece_tokenizer import SentencepiecesTokenizer

# 导入MeloTTS相关模块
from libmelotts.python.split_utils import split_sentence
from libmelotts.python.text import cleaned_text_to_sequence
from libmelotts.python.text.cleaner import clean_text
from libmelotts.python.symbols import LANG_TO_SYMBOL_MAP

# 配置参数
# tts 参数
TTS_MODEL_DIR = "libmelotts/models"
TTS_MODEL_FILES = {
    "g": "g-zh_mix_en.bin",
    "encoder": "encoder-zh.onnx",
    "decoder": "decoder-zh.axmodel"
}

# Qwen大模型翻译API参数
QWEN_API_URL = "http://10.126.29.158:8000"  # API服务地址

# TTS辅助函数
def intersperse(lst, item):
    result = [item] * (len(lst) * 2 + 1)
    result[1::2] = lst
    return result

# def get_text_for_tts_infer(text, language_str, symbol_to_id=None):
#     norm_text, phone, tone, word2ph = clean_text(text, language_str)
#     phone, tone, language = cleaned_text_to_sequence(phone, tone, language_str, symbol_to_id)

#     phone = intersperse(phone, 0)
#     tone = intersperse(tone, 0)
#     language = intersperse(language, 0)

#     phone = np.array(phone, dtype=np.int32)
#     tone = np.array(tone, dtype=np.int32)
#     language = np.array(language, dtype=np.int32)
#     word2ph = np.array(word2ph, dtype=np.int32) * 2
#     word2ph[0] += 1

#     return phone, tone, language, norm_text, word2ph

# 处理字符无法不识别
def get_text_for_tts_infer(text, language_str, symbol_to_id=None):
    """修复版音素处理：确保所有数组长度一致"""
    try:
        norm_text, phone, tone, word2ph = clean_text(text, language_str)
        
        # 特殊音素直接映射为空字符串
        phone_mapping = {
            'ɛ': '', 'æ': '', 'ʌ': '', 'ʊ': '', 'ɔ': '', 'ɪ': '', 'ɝ': '', 'ɚ': '', 'ɑ': '',
            'ʒ': '', 'θ': '', 'ð': '', 'ŋ': '', 'ʃ': '', 'ʧ': '', 'ʤ': '', 'ː': '', 'ˈ': '',
            'ˌ': '', 'ʰ': '', 'ʲ': '', 'ʷ': '', 'ʔ': '', 'ɾ': '', 'ɹ': '', 'ɫ': '', 'ɡ': '',
        }
        
        # 同步处理 phone 和 tone，确保它们长度一致
        processed_phone = []
        processed_tone = []
        removed_symbols = set()
        
        for p, t in zip(phone, tone):
            if p in phone_mapping:
                # 特殊音素直接删除，同时删除对应的 tone
                removed_symbols.add(p)
            elif p in symbol_to_id:
                # 正常音素保留，同时保留对应的 tone
                processed_phone.append(p)
                processed_tone.append(t)
            else:
                # 其他未知音素也删除
                removed_symbols.add(p)
        
        # 记录被删除的音素
        if removed_symbols:
            print(f"[音素过滤] 删除了 {len(removed_symbols)} 个特殊音素: {sorted(removed_symbols)}")
            print(f"[音素过滤] 处理后音素序列长度: {len(processed_phone)}")
            print(f"[音素过滤] 处理后音调序列长度: {len(processed_tone)}")
        
        # 如果没有有效音素，使用默认音素，
        if not processed_phone:
            print("[警告] 没有有效音素，使用默认中文音素")
            processed_phone = ['ni', 'hao']
            processed_tone = ['1', '3']
            word2ph = [1, 1]
        
        # 确保 word2ph 的长度与处理后的音素序列匹配
        if len(processed_phone) != len(phone):
            print(f"[警告] 音素序列长度变化: {len(phone)} -> {len(processed_phone)}")
            # 简单处理：重新计算 word2ph
            word2ph = [1] * len(processed_phone)
        
        phone, tone, language = cleaned_text_to_sequence(processed_phone, processed_tone, language_str, symbol_to_id)

        phone = intersperse(phone, 0)
        tone = intersperse(tone, 0)
        language = intersperse(language, 0)

        phone = np.array(phone, dtype=np.int32)
        tone = np.array(tone, dtype=np.int32)
        language = np.array(language, dtype=np.int32)
        word2ph = np.array(word2ph, dtype=np.int32) * 2
        word2ph[0] += 1

        return phone, tone, language, norm_text, word2ph
        
    except Exception as e:
        print(f"[错误] 文本处理失败: {e}")
        import traceback
        traceback.print_exc()
        raise e

def audio_numpy_concat(segment_data_list, sr, speed=1.):
    audio_segments = []
    for segment_data in segment_data_list:
        audio_segments += segment_data.reshape(-1).tolist()
        audio_segments += [0] * int((sr * 0.05) / speed)
    audio_segments = np.array(audio_segments).astype(np.float32)
    return audio_segments

def merge_sub_audio(sub_audio_list, pad_size, audio_len):
    if pad_size > 0:
        for i in range(len(sub_audio_list) - 1):
            sub_audio_list[i][-pad_size:] += sub_audio_list[i+1][:pad_size]
            sub_audio_list[i][-pad_size:] /= 2
            if i > 0:
                sub_audio_list[i] = sub_audio_list[i][pad_size:]

    sub_audio = np.concatenate(sub_audio_list, axis=-1)
    return sub_audio[:audio_len]

def calc_word2pronoun(word2ph, pronoun_lens):
    indice = [0]
    for ph in word2ph[:-1]:
        indice.append(indice[-1] + ph)
    word2pronoun = []
    for i, ph in zip(indice, word2ph):
        word2pronoun.append(np.sum(pronoun_lens[i : i + ph]))
    return word2pronoun

def generate_slices(word2pronoun, dec_len):
    pn_start, pn_end = 0, 0
    zp_start, zp_end = 0, 0
    zp_len = 0
    pn_slices = []
    zp_slices = []
    while pn_end < len(word2pronoun):
        if pn_end - pn_start > 2 and np.sum(word2pronoun[pn_end - 2 : pn_end + 1]) <= dec_len:
            zp_len = np.sum(word2pronoun[pn_end - 2 : pn_end])
            zp_start = zp_end - zp_len
            pn_start = pn_end - 2
        else:
            zp_len = 0
            zp_start = zp_end
            pn_start = pn_end
            
        while pn_end < len(word2pronoun) and zp_len + word2pronoun[pn_end] <= dec_len:
            zp_len += word2pronoun[pn_end]
            pn_end += 1
        zp_end = zp_start + zp_len
        pn_slices.append(slice(pn_start, pn_end))
        zp_slices.append(slice(zp_start, zp_end))
    return pn_slices, zp_slices

# 确认中英文
def lang_detect_with_regex(text):
    text_without_digits = re.sub(r'\d+', '', text)
    
    if not text_without_digits:
        return 'unknown'
    
    if re.search(r'[\u4e00-\u9fff]', text_without_digits):
        return 'chinese'
    else:
        if re.search(r'[a-zA-Z]', text_without_digits):
            return 'english'
        else:
            return 'unknown'

class QwenTranslationAPI:
    def __init__(self, api_url=QWEN_API_URL):
        self.api_url = api_url
        self.session_id = f"speech_translate_{int(time.time())}"
        self.last_reset_time = time.time()
        self.request_count = 0
        self.max_requests_before_reset = 10
        
    def reset_context(self):
        """重置API上下文"""
        try:
            reset_url = f"{self.api_url}/api/reset"
            response = requests.post(reset_url, json={}, timeout=5)
            if response.status_code == 200:
                print("[翻译API] ✓ 上下文重置成功")
                self.last_reset_time = time.time()
                self.request_count = 0
                return True
            else:
                print(f"[翻译API] ✗ 重置失败，状态码: {response.status_code}, 响应: {response.text}")
        except Exception as e:
            print(f"[翻译API] ✗ 重置上下文失败: {e}")
        return False

    def check_and_reset_if_needed(self):
        """检查是否需要重置上下文"""
        current_time = time.time()
        if (self.request_count >= 10 or 
            current_time - self.last_reset_time > 120):  # 2分钟
            print(f"[翻译API] 重置 (请求数: {self.request_count}, 时间: {current_time - self.last_reset_time:.1f}秒)")
            return self.reset_context()
        return True
        
    def translate(self, text_content, max_retries=3, timeout=120):
        if not text_content or text_content.strip() == "":
            return "输入文本为空"
        
        # 过滤太短的文本
        if len(text_content.strip()) < 3:
            return text_content
            
        if lang_detect_with_regex(text_content)=='chinese':
            prompt_f = "翻译成英文"
        else:
            prompt_f= "翻译成中文"

        prompt = f"{prompt_f}：{text_content}"
        print(f"[翻译API] 发送请求: {prompt}")
        
        # 检查是否需要重置
        self.check_and_reset_if_needed()
        
        for attempt in range(max_retries):
            try:
                generate_url = f"{self.api_url}/api/generate"
                payload = {
                    "prompt": prompt,
                    "temperature": 0.1,
                    "repetition_penalty": 1.0,
                    "top-p": 0.9,
                    "top-k": 40,
                    "max_new_tokens": 512
                }
                
                print(f"[翻译API] 开始生成请求 (尝试 {attempt + 1}/{max_retries})")
                response = requests.post(generate_url, json=payload, timeout=30)
                response.raise_for_status()
                print("[翻译API] 生成请求成功")
                
                result_url = f"{self.api_url}/api/generate_provider"
                start_time = time.time()
                full_translation = ""
                error_detected = False
                
                while time.time() - start_time < timeout:
                    try:
                        result_response = requests.get(result_url, timeout=10)
                        result_data = result_response.json()
                        
                        current_chunk = result_data.get("response", "")
                        
                        # 检查是否有错误
                        if "error:" in current_chunk.lower() or "setkvcache failed" in current_chunk.lower():
                            print(f"[翻译API] ✗ 检测到错误: {current_chunk}")
                            error_detected = True
                            print("[翻译API] 立即重置上下文...")
                            self.reset_context()
                            break
                        
                        full_translation += current_chunk
                        
                        if result_data.get("done", False):
                            if full_translation and len(full_translation.strip()) > 0:
                                self.request_count += 1
                                print(f"[翻译API] ✓ 翻译完成: {full_translation}")
                                return full_translation
                            else:
                                print(f"[翻译API] ✗ 翻译结果为空")
                                break
                        
                        time.sleep(0.05)
                        
                    except requests.exceptions.RequestException as e:
                        print(f"[翻译API] 轮询请求失败: {e}")
                        if time.time() - start_time > timeout:
                            break
                        time.sleep(0.5)
                        continue
                
                if error_detected:
                    if attempt < max_retries - 1:
                        wait_time = 1
                        print(f"[翻译API] 等待 {wait_time} 秒后重试...")
                        time.sleep(wait_time)
                        continue
                    else:
                        print("[翻译API] 达到最大重试次数，返回原文")
                        return text_content
                
                print(f"[翻译API] 轮询超时，尝试第 {attempt + 1} 次重试")
                
            except requests.exceptions.RequestException as e:
                print(f"[翻译API] 请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt
                    print(f"[翻译API] 等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)
                else:
                    return text_content
            except Exception as e:
                print(f"[翻译API] 翻译过程出错: {e}")
                if attempt < max_retries - 1:
                    time.sleep(1)
                    continue
                return text_content
        
        print("[翻译API] 翻译超时，返回原文")
        return text_content

class AudioResampler:
    """音频重采样器"""
    def __init__(self, target_sr=16000):
        self.target_sr = target_sr
        
    def resample_audio(self, audio_data, original_sr):
        """重采样音频到目标采样率，asr统一输入16000Hz"""
        if original_sr == self.target_sr:
            return audio_data
            
        print(f"[重采样] {original_sr}Hz -> {self.target_sr}Hz")
        return librosa.resample(y=audio_data, orig_sr=original_sr, target_sr=self.target_sr)
    
    def resample_chunk(self, audio_chunk, original_sr):
        """重采样音频块：长音频进行过冲采样后，音频块可以不做重采样"""
        if original_sr == self.target_sr:
            return audio_chunk
            
        if len(audio_chunk) < 1000:
            return self._linear_resample(audio_chunk, original_sr, self.target_sr)
        else:
            return librosa.resample(y=audio_chunk, orig_sr=original_sr, target_sr=self.target_sr)
    
    def _linear_resample(self, audio_chunk, original_sr, target_sr):
        """线性插值重采样"""
        ratio = target_sr / original_sr
        old_length = len(audio_chunk)
        new_length = int(old_length * ratio)
        
        old_indices = np.arange(old_length)
        new_indices = np.linspace(0, old_length - 1, new_length)
        
        resampled = np.interp(new_indices, old_indices, audio_chunk)
        return resampled

class StreamProcessor:
    """流式处理"""
    def __init__(self, pipeline, chunk_duration=7.0, overlap_duration=0.01, target_sr=16000):
        self.pipeline = pipeline
        self.chunk_duration = chunk_duration  # 增加4->7秒
        self.overlap_duration = overlap_duration  # 减少到0.1->0.01秒
        self.target_sr = target_sr
        self.chunk_samples = int(chunk_duration * target_sr)
        self.overlap_samples = int(overlap_duration * target_sr)
        self.audio_buffer = deque()
        self.result_queue = queue.Queue()
        self.is_running = False
        self.processing_thread = None
        self.resampler = AudioResampler(target_sr=target_sr)
        self.segment_counter = 0  # 音频段计数器
        self.processed_texts = set()  # 记录已处理的文本，避免重复
        
    def start_processing(self):
        """开始流式处理"""
        self.is_running = True
        self.processing_thread = threading.Thread(target=self._process_loop)
        self.processing_thread.daemon = True
        self.processing_thread.start()
        
    def stop_processing(self):
        """停止流式处理"""
        self.is_running = False
        if self.processing_thread:
            self.processing_thread.join(timeout=5)
            
    def add_audio_chunk(self, audio_chunk, original_sr=None):
        """添加音频块到缓冲区"""
        if original_sr and original_sr != self.target_sr:
            audio_chunk = self.resampler.resample_chunk(audio_chunk, original_sr)
        
        self.audio_buffer.append(audio_chunk)
        
    def get_next_result(self, timeout=1.0):
        """获取下一个处理结果"""
        try:
            return self.result_queue.get(timeout=timeout)
        except queue.Empty:
            return None
            
    def _process_loop(self):
        """处理循环"""
        accumulated_audio = np.array([], dtype=np.float32)
        last_asr_result = ""  # 记录上一次的ASR结果，防止重复处理
        
        while self.is_running:
            if len(self.audio_buffer) > 0:
                audio_chunk = self.audio_buffer.popleft()
                accumulated_audio = np.concatenate([accumulated_audio, audio_chunk])
                
                # 当积累的音频足够处理时
                if len(accumulated_audio) >= self.chunk_samples:
                    # 提取处理块（减少重叠）
                    process_chunk = accumulated_audio[:self.chunk_samples]
                    accumulated_audio = accumulated_audio[self.chunk_samples - self.overlap_samples:]
                    
                    try:
                        # 实时ASR识别
                        asr_result = self._stream_asr(process_chunk)
                        
                        # 过滤条件：
                        # # 1. 文本有效且足够长
                        # 2. 与上次结果不同（避免重复）
                        # 3. 不是已处理过的文本
                        if (asr_result and asr_result.strip() and 
                            # len(asr_result.strip()) >= 5 and
                            asr_result != last_asr_result and
                            asr_result not in self.processed_texts):
                            
                            print(f"[实时ASR] {asr_result}")
                            last_asr_result = asr_result
                            self.processed_texts.add(asr_result)
                            
                            # 实时翻译
                            try:
                                translation_result = self.pipeline.run_translation(asr_result)
                                
                                # 检查翻译结果是否有效
                                if (translation_result and 
                                    translation_result != asr_result and
                                    "翻译失败" not in translation_result and 
                                    "error:" not in translation_result.lower() and
                                    "输入文本为空" not in translation_result):
                                    
                                    print(f"[实时翻译] {translation_result}")
                                    
                                    # TTS合成
                                    try:
                                        self.segment_counter += 1
                                        tts_filename = f"stream_segment_{self.segment_counter:04d}.wav"
                                        tts_start_time = time.time()
                                        
                                        tts_path = self.pipeline.run_tts(
                                            translation_result, 
                                            self.pipeline.output_dir, 
                                            tts_filename
                                        )
                                        
                                        tts_time = time.time() - tts_start_time
                                        print(f"[实时TTS] 音频已保存: {tts_path} (耗时: {tts_time:.2f}秒)")
                                        
                                        # 将完整结果放入队列
                                        self.result_queue.put({
                                            'type': 'complete',
                                            'original': asr_result,
                                            'translated': translation_result,
                                            'audio_path': tts_path,
                                            'timestamp': time.time(),
                                            'segment_id': self.segment_counter
                                        })
                                        
                                    except Exception as tts_error:
                                        print(f"[实时TTS错误] {tts_error}")
                                        import traceback
                                        traceback.print_exc()
                                else:
                                    print(f"[实时翻译] 翻译结果无效，已跳过")
                                    
                            except Exception as translation_error:
                                print(f"[实时翻译错误] {translation_error}")
                        else:
                            if asr_result == last_asr_result:
                                print(f"[实时ASR] 重复内容已跳过: {asr_result}")
                                
                    except Exception as e:
                        print(f"[流式处理错误] {e}")
                        import traceback
                        traceback.print_exc()
                        
            time.sleep(0.01)
            
    def _stream_asr(self, audio_chunk):
        """流式ASR识别（带VAD）"""
        try:
            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
            # 步骤1: VAD检测 - 过滤静音段
            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
            res_vad = self.pipeline.model_vad(audio_chunk)[0]
            vad_segments = merge_vad(res_vad, 15 * 1000)
            
            # 如果没有检测到语音段，直接返回空
            if not vad_segments or len(vad_segments) == 0:
                print(f"[VAD] 未检测到语音活动，跳过此音频块")
                return ""
            
            print(f"[VAD] 检测到 {len(vad_segments)} 个语音段")
            
            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
            # 步骤2: 对检测到的语音段进行ASR识别
            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
            all_results = ""
            
            for i, segment in enumerate(vad_segments):
                segment_start, segment_end = segment
                start_sample = int(segment_start / 1000 * self.target_sr)
                end_sample = min(int(segment_end / 1000 * self.target_sr), len(audio_chunk))
                segment_audio = audio_chunk[start_sample:end_sample]
                
                # 跳过太短的片段，减少误识别（小于0.3秒）
                if len(segment_audio) < int(0.3 * self.target_sr):
                    continue
                
                # 写入临时文件
                with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file:
                    sf.write(temp_file.name, segment_audio, self.target_sr)
                    temp_filename = temp_file.name
                
                try:
                    # ASR识别
                    segment_result = self.pipeline.model_bin(
                        temp_filename, 
                        "auto",
                        True,
                        self.pipeline.position_encoding, 
                        tokenizer=self.pipeline.tokenizer,
                    )
                    
                    if segment_result and segment_result.strip():
                        all_results += segment_result + " "
                    
                    # 清理临时文件
                    os.unlink(temp_filename)
                    
                except Exception as e:
                    print(f"[ASR错误] 处理VAD段 {i} 时出错: {e}")
                    if os.path.exists(temp_filename):
                        os.unlink(temp_filename)
                    continue
            
            return all_results.strip()
                
        except Exception as e:
            print(f"[ASR错误] {e}")
            return ""

class SpeechTranslationPipeline:
    def __init__(self, 
                 tts_model_dir, tts_model_files,
                 asr_model_dir="ax_model", seq_len=132,
                 tts_dec_len=128, sample_rate=44100, tts_speed=0.8,
                 qwen_api_url=QWEN_API_URL, target_sr=16000,
                 output_dir="./output"):
        self.tts_model_dir = tts_model_dir
        self.tts_model_files = tts_model_files
        self.asr_model_dir = asr_model_dir
        self.seq_len = seq_len
        self.tts_dec_len = tts_dec_len
        self.sample_rate = sample_rate
        self.tts_speed = tts_speed
        self.qwen_api_url = qwen_api_url
        self.target_sr = target_sr
        self.output_dir = output_dir
        
        # 输出目录
        os.makedirs(self.output_dir, exist_ok=True)
        
        # 初始化音频重采样器
        self.resampler = AudioResampler(target_sr=target_sr)
        
        # 初始化ASR模型
        self._init_asr_models()
        
        # 初始化TTS模型
        self._init_tts_models()
        
        # 初始化翻译API
        self.translator = QwenTranslationAPI(api_url=qwen_api_url)
        
        # 初始化流式处理器
        self.stream_processor = StreamProcessor(self, target_sr=target_sr)
        
        # 验证所有必需文件存在
        self._validate_files()
        
        # 初始化时重置API上下文
        print("[初始化] 重置API上下文...")
        self.translator.reset_context()
    
    def _init_asr_models(self):
        """初始化语音识别相关模型"""
        print("Initializing SenseVoice models...")
        
        self.model_vad = AX_Fsmn_vad(self.asr_model_dir)
        
        self.embed = SinusoidalPositionEncoder()
        self.position_encoding = self.embed.get_position_encoding(
            torch.randn(1, self.seq_len, 560)).numpy()
        
        self.model_bin = AX_SenseVoiceSmall(self.asr_model_dir, seq_len=self.seq_len)
        
        tokenizer_path = os.path.join(self.asr_model_dir, "chn_jpn_yue_eng_ko_spectok.bpe.model")
        self.tokenizer = SentencepiecesTokenizer(bpemodel=tokenizer_path)
        
        print("SenseVoice models initialized successfully.")
    
    def _init_tts_models(self):
        """初始化TTS相关模型"""
        print("Initializing MeloTTS models...")
        init_start = time.time()
        
        enc_model = os.path.join(self.tts_model_dir, self.tts_model_files["encoder"])
        dec_model = os.path.join(self.tts_model_dir, self.tts_model_files["decoder"])
        
        model_load_start = time.time()
        self.sess_enc = ort.InferenceSession(enc_model, providers=["CPUExecutionProvider"], sess_options=ort.SessionOptions())
        self.sess_dec = axe.InferenceSession(dec_model)
        print(f"  Load encoder/decoder models: {(time.time() - model_load_start)*1000:.2f}ms")
        
        g_file = os.path.join(self.tts_model_dir, self.tts_model_files["g"])
        self.tts_g = np.fromfile(g_file, dtype=np.float32).reshape(1, 256, 1)
        
        self.tts_language = "ZH_MIX_EN"
        self.symbol_to_id = {s: i for i, s in enumerate(LANG_TO_SYMBOL_MAP[self.tts_language])}
        
        print("  Warming up TTS modules...")
        warmup_start = time.time()
        
        try:
            warmup_text_mix = "这是一个test测试。"
            _, _, _, _, _ = get_text_for_tts_infer(warmup_text_mix, self.tts_language, symbol_to_id=self.symbol_to_id)
            print(f"  Mixed ZH-EN warm-up: {(time.time() - warmup_start)*1000:.2f}ms")
        except Exception as e:
            print(f"  Warning: Mixed warm-up failed: {e}")
        
        total_init_time = (time.time() - init_start) * 1000
        print(f"MeloTTS models initialized successfully. Total init time: {total_init_time:.2f}ms")
    
    def _validate_files(self):
        """验证所有必需的文件都存在"""
        for key, filename in self.tts_model_files.items():
            filepath = os.path.join(self.tts_model_dir, filename)
            if not os.path.exists(filepath):
                raise FileNotFoundError(f"TTS模型文件不存在: {filepath}")
        
        try:
            response = requests.get(f"{self.qwen_api_url}/api/generate_provider", timeout=5)
            print("[API检查] 千问API服务连接正常")
        except:
            print("[API警告] 无法连接到千问API服务，请确保已启动API服务")

    def start_stream_processing(self):
        """开始流式处理"""
        self.stream_processor.start_processing()
        print("[流式处理] 已启动")
        
    def stop_stream_processing(self):
        """停止流式处理"""
        self.stream_processor.stop_processing()
        print("[流式处理] 已停止")
        
    def process_audio_stream(self, audio_chunk, original_sr=None):
        """处理音频流数据"""
        self.stream_processor.add_audio_chunk(audio_chunk, original_sr)
        
    def get_stream_results(self):
        """获取流式处理结果"""
        return self.stream_processor.get_next_result()

    def load_and_resample_audio(self, audio_file):
        """加载音频并重采样到目标采样率"""
        print(f"加载音频文件: {audio_file}")
        speech, original_sr = librosa.load(audio_file, sr=None)
        
        audio_duration = len(speech) / original_sr
        print(f"原始音频: {original_sr}Hz, 时长: {audio_duration:.2f}秒")
        
        if original_sr != self.target_sr:
            speech = self.resampler.resample_audio(speech, original_sr)
            print(f"重采样后: {self.target_sr}Hz, 时长: {len(speech)/self.target_sr:.2f}秒")
        
        return speech, self.target_sr
    
    def run_translation(self, text_content):
        """调用Qwen大模型API中英互译"""
        print("Starting translation via API...")
        translation_start_time = time.time()
        
        translate_content = self.translator.translate(text_content)
        
        translation_time_cost = time.time() - translation_start_time
        print(f"Translation processing time: {translation_time_cost:.2f} seconds")
        print(f"Translation Result: {translate_content}")
        
        return translate_content
    
    def run_tts(self, translate_content, output_dir, output_wav=None):
        """使用TTS模型合成语音"""
        output_path = os.path.join(output_dir, output_wav)
        
        try:
            if lang_detect_with_regex(translate_content) == "chinese":
                translate_content = cn2an.transform(translate_content, "an2cn")
            
            print(f"TTS synthesis for text: {translate_content}")
            
            sens = split_sentence(translate_content, language_str=self.tts_language)
            print(f"Text split into {len(sens)} sentences")
            
            audio_list = []
            
            for n, se in enumerate(sens):
                if self.tts_language in ['EN', 'ZH_MIX_EN']:
                    se = re.sub(r'([a-z])([A-Z])', r'\1 \2', se)
                
                print(f"Processing sentence[{n}]: {se}")
                
                phones, tones, lang_ids, norm_text, word2ph = get_text_for_tts_infer(
                    se, self.tts_language, symbol_to_id=self.symbol_to_id)
                
                encoder_start = time.time()
                z_p, pronoun_lens, audio_len = self.sess_enc.run(None, input_feed={
                    'phone': phones, 'g': self.tts_g,
                    'tone': tones, 'language': lang_ids, 
                    'noise_scale': np.array([0], dtype=np.float32),
                    'length_scale': np.array([1.0 / self.tts_speed], dtype=np.float32),
                    'noise_scale_w': np.array([0], dtype=np.float32),
                    'sdp_ratio': np.array([0], dtype=np.float32)})
                print(f"Encoder run time: {1000 * (time.time() - encoder_start):.2f}ms")
                
                word2pronoun = calc_word2pronoun(word2ph, pronoun_lens)
                pn_slices, zp_slices = generate_slices(word2pronoun, self.tts_dec_len)
                
                audio_len = audio_len[0]
                sub_audio_list = []
                
                for i, (ps, zs) in enumerate(zip(pn_slices, zp_slices)):
                    zp_slice = z_p[..., zs]
                    
                    sub_dec_len = zp_slice.shape[-1]
                    sub_audio_len = 512 * sub_dec_len
                    
                    if zp_slice.shape[-1] < self.tts_dec_len:
                        zp_slice = np.concatenate((zp_slice, np.zeros((*zp_slice.shape[:-1], self.tts_dec_len - zp_slice.shape[-1]), dtype=np.float32)), axis=-1)
                    
                    decoder_start = time.time()
                    audio = self.sess_dec.run(None, input_feed={"z_p": zp_slice, "g": self.tts_g})[0].flatten()
                    
                    audio_start = 0
                    if len(sub_audio_list) > 0:
                        if pn_slices[i - 1].stop > ps.start:
                            audio_start = 512 * word2pronoun[ps.start]
                    
                    audio_end = sub_audio_len
                    if i < len(pn_slices) - 1:
                        if ps.stop > pn_slices[i + 1].start:
                            audio_end = sub_audio_len - 512 * word2pronoun[ps.stop - 1]
                    
                    audio = audio[audio_start:audio_end]
                    print(f"Decode slice[{i}]: decoder run time {1000 * (time.time() - decoder_start):.2f}ms")
                    sub_audio_list.append(audio)
                
                sub_audio = merge_sub_audio(sub_audio_list, 0, audio_len)
                audio_list.append(sub_audio)
            
            audio = audio_numpy_concat(audio_list, sr=self.sample_rate, speed=self.tts_speed)
            
            sf.write(output_path, audio, self.sample_rate)
            print(f"TTS audio saved to {output_path}")
            
            return output_path
            
        except Exception as e:
            print(f"TTS synthesis failed: {e}")
            import traceback
            traceback.print_exc()
            raise e

    def process_long_audio_stream(self, audio_file, chunk_size=64000):
        """
        处理长音频文件的流式模拟
        chunk_size增加到64000（4秒 * 16000Hz），与StreamProcessor的chunk_duration匹配
        4秒有点短，改到7秒感觉更好点
        """
        print(f"[流式处理] 开始处理长音频: {audio_file}")
        
        # 加载并重采样音频
        speech, fs = self.load_and_resample_audio(audio_file)
        
        # 启动流式处理
        self.start_stream_processing()
        
        total_chunks = (len(speech) + chunk_size - 1) // chunk_size
        print(f"[流式处理] 音频总长度: {len(speech)/fs:.2f}秒, 分块数: {total_chunks}")
        
        # 收集所有结果
        all_results = []
        
        # 模拟流式输入
        chunk_count = 0
        for i in range(0, len(speech), chunk_size):
            chunk = speech[i:i+chunk_size]
            chunk_count += 1
            
            # 处理最后一块：如果不足chunk_size，填零补齐
            if len(chunk) < chunk_size:
                padding_size = chunk_size - len(chunk)
                chunk = np.concatenate([chunk, np.zeros(padding_size, dtype=np.float32)])
                print(f"\n[流式处理] 处理音频块 {chunk_count}/{total_chunks} (最后一块，已填零 {padding_size} 样本)")
            else:
                print(f"\n[流式处理] 处理音频块 {chunk_count}/{total_chunks}")
            
            self.process_audio_stream(chunk, fs)
            
            # 获取并显示实时结果
            result = self.get_stream_results()
            while result:
                print(f"\n{'='*70}")
                print(f"[实时结果 #{len(all_results) + 1}]")
                print(f"段落ID: {result['segment_id']}")
                print(f"原文: {result['original']}")
                print(f"翻译: {result['translated']}")
                print(f"音频: {result['audio_path']}")
                print(f"{'='*70}")
                all_results.append(result)
                result = self.get_stream_results()
                
            time.sleep(0.1)
        
        # 输出结果
        # print(f"\n[流式处理] 等待处理剩余音频块...")
        max_wait_time = 20  # 增加等待时间到20秒
        wait_start = time.time()
        
        while time.time() - wait_start < max_wait_time:
            result = self.get_stream_results()
            if result:
                print(f"\n{'='*70}")
                print(f"[实时结果 #{len(all_results) + 1}]")
                print(f"段落ID: {result['segment_id']}")
                print(f"原文: {result['original']}")
                print(f"翻译: {result['translated']}")
                print(f"音频: {result['audio_path']}")
                print(f"{'='*70}")
                all_results.append(result)
                wait_start = time.time()  # 重置等待时间
            else:
                time.sleep(0.2)
                
        # 停止流式处理
        self.stop_stream_processing()
        
        print(f"\n[流式处理] 完成！共处理 {len(all_results)} 个有效结果")
        return all_results

def main():
    parser = argparse.ArgumentParser(description="实时语音翻译pipeline")
    parser.add_argument("--audio_file", type=str, required=True, help="输入音频文件路径")
    parser.add_argument("--output_dir", type=str, default="./output", help="输出目录")
    parser.add_argument("--api_url", type=str, default=QWEN_API_URL, help="Qwen API服务器URL")
    parser.add_argument("--target_sr", type=int, default=16000, help="ASR目标采样率 (默认: 16000)")
    parser.add_argument("--chunk_duration", type=float, default=7.0, help="音频块时长(秒) (默认: 7.0)")
    parser.add_argument("--overlap_duration", type=float, default=0.01, help="重叠时长(秒) (默认: 0.1)")
    
    args = parser.parse_args()
    print("-------------------实时语音翻译pipeline-------------------\n")
    os.makedirs(args.output_dir, exist_ok=True)
    
    print(f"处理音频文件: {args.audio_file}")
    print(f"输出目录: {args.output_dir}")
    print(f"音频块时长: {args.chunk_duration}秒")
    print(f"重叠时长: {args.overlap_duration}秒\n")
    
    # 初始化Pipeline
    pipeline = SpeechTranslationPipeline(
        tts_model_dir=TTS_MODEL_DIR,
        tts_model_files=TTS_MODEL_FILES,
        asr_model_dir="ax_model",
        seq_len=132,
        tts_dec_len=128,
        sample_rate=44100,
        tts_speed=0.8,
        qwen_api_url=args.api_url,
        target_sr=args.target_sr,
        output_dir=args.output_dir
    )
    
    # # 可选：调整流式处理参数
    # if args.chunk_duration != 7.0 or args.overlap_duration != 0.01:
    #     pipeline.stream_processor.chunk_duration = args.chunk_duration
    #     pipeline.stream_processor.overlap_duration = args.overlap_duration
    #     pipeline.stream_processor.chunk_samples = int(args.chunk_duration * args.target_sr)
    #     pipeline.stream_processor.overlap_samples = int(args.overlap_duration * args.target_sr)
    #     print(f"[配置] 已更新流式处理参数: chunk_duration={args.chunk_duration}s, overlap_duration={args.overlap_duration}s\n")
    
    start_time = time.time()
    try:
        # 流式处理模式
        print("使用流式处理模式（包含TTS）...")
        print("="*70 + "\n")
        
        # 计算chunk_size以匹配chunk_duration
        chunk_size = int(args.chunk_duration * args.target_sr)
        results = pipeline.process_long_audio_stream(args.audio_file, chunk_size=chunk_size)
        
        print("\n" + "="*70)
        print("                   处理完成")
        print("="*70)
        print(f"\n  成功处理 {len(results)} 个有效翻译段落\n")
        
        # 显示所有结果
        if results:
            print("所有翻译结果：")
            print("-" * 70)
            for idx, result in enumerate(results, 1):
                print(f"\n【段落 {idx}】(ID: {result['segment_id']})")
                print(f"  原文: {result['original']}")
                print(f"  译文: {result['translated']}")
                print(f"  音频: {result['audio_path']}")
                print(f"  时间: {time.strftime('%H:%M:%S', time.localtime(result['timestamp']))}")
            print("-" * 70)
            
            # 保存结果到文件
            result_file = os.path.join(args.output_dir, "stream_results.txt")
            with open(result_file, 'w', encoding='utf-8') as f:
                f.write(f"流式翻译+TTS结果 - {args.audio_file}\n")
                f.write(f"处理时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"音频块时长: {args.chunk_duration}秒, 重叠时长: {args.overlap_duration}秒\n")
                f.write("="*70 + "\n\n")
                for idx, result in enumerate(results, 1):
                    f.write(f"【段落 {idx}】(ID: {result['segment_id']})\n")
                    f.write(f"原文: {result['original']}\n")
                    f.write(f"译文: {result['translated']}\n")
                    f.write(f"音频: {result['audio_path']}\n")
                    f.write(f"时间: {time.strftime('%H:%M:%S', time.localtime(result['timestamp']))}\n")
                    f.write("\n" + "-"*70 + "\n\n")
            print(f"\n✓ 结果已保存到: {result_file}")
            
            # 统计音频文件
            audio_files = [r['audio_path'] for r in results]
            print(f"\n  生成 {len(audio_files)} 个TTS音频文件:")
            for audio_file in audio_files:
                print(f"  - {audio_file}")
        else:
            print("\n  未获取到有效的翻译结果")
        
        print("="*70)
        
        # 总耗时
        total_time = time.time() - start_time
        print(f"\n总处理时间: {total_time:.2f} 秒")
        
    except Exception as e:
        print(f"Pipeline执行失败: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()