ASR

Note

在展馆等多人声环境中，建议通过物理方式（如定向麦克风）或音频预处理（如降噪、波束成形）减少环境干扰后再送入ASR服务

from funasr import AutoModel

model_dir = r"C:\Users\wjn\yx\hello-xiaoxin\models\asr\SenseVoiceSmall"
wav_path="test.wav"
wav_path = r"C:\Users\wjn\github\xiaozhi-esp32-server\main\xiaozhi-server\models\SenseVoiceSmall\example\zh.mp3"
model = AutoModel(
    model=model_dir,
    vad_model="fsmn-vad",
    vad_kwargs={"max_single_segment_time": 30000},
    device="cpu",
)
res = model.generate(input=[wav_path], cache={}, batch_size_s=0)
text = res[0]["text"]
print(text)

asr & tts

import os
import sys
import pyttsx3
from funasr import AutoModel

# ================= 配置区域 =================
# 1. ASR 模型路径 (SenseVoiceSmall)
# 请确保该路径下包含 model.pt, config.yaml, tokens.json 等文件
ASR_MODEL_DIR = r"C:\Users\wjn\yx\hello-xiaoxin\models\asr\SenseVoiceSmall"

# 2. 测试音频路径
# 优先使用你提供的 mp3 路径，如果不存在则 fallback 到 test.wav
AUDIO_FILE = r"C:\Users\wjn\github\xiaozhi-esp32-server\main\xiaozhi-server\models\SenseVoiceSmall\example\zh.mp3"
if not os.path.exists(AUDIO_FILE):
    AUDIO_FILE = "test.wav"

# 3. TTS 配置
TTS_RATE = 150  # 语速 (默认约 200，调低更清晰)
TTS_VOLUME = 1.0  # 音量 (0.0 - 1.0)
TTS_VOICE_ID = None  # 留空则使用默认，可设置为特定中文语音 ID


# ===========================================

def init_tts():
    """初始化 TTS 引擎并配置中文声音"""
    print("🔊 正在初始化 TTS 引擎...")
    engine = pyttsx3.init()

    # 配置语速和音量
    engine.setProperty('rate', TTS_RATE)
    engine.setProperty('volume', TTS_VOLUME)

    # 尝试自动选择中文声音
    voices = engine.getProperty('voices')
    selected_voice = None

    # 优先寻找中文女声 (通常包含 'zh' 或 'xiaoxiao' 等关键字)
    for voice in voices:
        if 'zh' in voice.id.lower() or 'chinese' in voice.name.lower():
            selected_voice = voice.id
            print(f"✅ 找到中文语音: {voice.name}")
            break

    if selected_voice:
        engine.setProperty('voice', selected_voice)
    else:
        print("⚠️ 未找到专用中文语音，使用系统默认语音。")
        # 可选：打印所有可用声音以便调试
        # for v in voices: print(v.name)

    return engine


def speak(engine, text):
    """执行朗读"""
    if not text:
        print("⚠️ 没有文本可朗读。")
        return

    print(f"🗣️  正在朗读: {text}")
    engine.say(text)
    engine.runAndWait()  # 阻塞直到朗读完成


def run_asr(model, audio_path):
    """执行语音识别"""
    if not os.path.exists(audio_path):
        raise FileNotFoundError(f"❌ 音频文件不存在: {audio_path}")

    print(f"🎤 正在识别音频: {os.path.basename(audio_path)} ...")

    try:
        # FunASR 的 generate 方法
        # cache={} 用于清空上下文，确保每次都是独立识别
        res = model.generate(input=[audio_path], cache={}, batch_size_s=0)

        if res and len(res) > 0:
            text = res[0].get("text", "")
            # SenseVoice 有时会返回带标点或不带标点的结果，视模型训练而定
            return text.strip()
        else:
            return ""
    except Exception as e:
        print(f"❌ ASR 识别出错: {e}")
        return None


def main():
    # 1. 初始化 TTS
    tts_engine = init_tts()

    # 2. 加载 ASR 模型
    print(f"🧠 正在加载 ASR 模型 (SenseVoiceSmall)... 路径: {ASR_MODEL_DIR}")
    print("(首次加载可能需要几分钟，请耐心等待...)")

    try:
        asr_model = AutoModel(
            model=ASR_MODEL_DIR,
            vad_model="fsmn-vad",  # 语音活动检测，自动切除静音
            vad_kwargs={"max_single_segment_time": 30000},  # 最大切片长度 30s
            device="cpu",  # 强制使用 CPU (如果没有 GPU)
            # disable_update=True       # 可选：禁止检查模型更新
        )
        print("✅ ASR 模型加载成功！")
    except Exception as e:
        print(f"❌ 模型加载失败: {e}")
        print("💡 请检查路径是否正确，以及是否安装了 torchaudio/torch。")
        sys.exit(1)

    # 3. 执行流程：听 -> 说
    recognized_text = run_asr(asr_model, AUDIO_FILE)

    if recognized_text is None:
        speak(tts_engine, "抱歉，识别过程中出错了。")
    elif recognized_text == "":
        speak(tts_engine, "我没听清，请再说一遍。")
    else:
        print(f"✨ 识别结果: {recognized_text}")
        # 这里可以加入简单的逻辑回复，或者直接复述
        reply = f"我听到了：{recognized_text}"
        speak(tts_engine, reply)


if __name__ == "__main__":
    main()