检测到麦克风有语音输入就开始录音自动转写，设置唤醒热词。

一个简单的使用脚本

upload successful

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
import asyncio
import websockets
import json
import pyaudio
import numpy as np
import time


async def test_funasr():
    """测试FunASR服务器的简单脚本"""

    # 配置参数
    server_url = "ws://ip:端口号"
    sample_rate = 16000
    chunk_duration = 0.3  # 秒
    chunk_size = int(sample_rate * chunk_duration)

    print(f"连接到服务器: {server_url}")
    print(f"采样率: {sample_rate}Hz")
    print(f"块大小: {chunk_size} samples")

    try:
        # 连接服务器
        async with websockets.connect(server_url, ping_interval=None) as websocket:
            print("✅ 连接成功")

            # 默认配置 - 可直接修改以下参数
            default_mode = "2pass"  # 可选: online, 2pass, offline

            # 根据模式设置参数
            is_speaking = default_mode != "offline"

            config = {
                "mode": default_mode,
                "wav_name": "server",
                "wav_format": "pcm",
                "is_speaking": is_speaking,
                "chunk_size": [5, 10, 5],
                "chunk_interval": 10,
                "itn": True,
                "hotwords": '{"小翼小翼": 40, "你好小翼": 40 , "退下": 40}'
            }

            print(f"\n使用默认模式: {default_mode}")
            print("\n发送的配置参数:")
            import json
            print(json.dumps(config, indent=2, ensure_ascii=False))
            await websocket.send(json.dumps(config))

            # 初始化音频
            p = pyaudio.PyAudio()

            # 跳过音频设备列表打印

            # 打开默认音频输入
            stream = p.open(
                format=pyaudio.paInt16,
                channels=1,
                rate=sample_rate,
                input=True,
                frames_per_buffer=chunk_size
            )

            print(f"\n开始录音 ({config['mode']} 模式)...")
            print("请说话（持续录音模式）...\n")


            chunk_count = 0
            silent_count = 0
            speaking = False
            vad_threshold = 0.01  # 语音活动检测阈值
            min_speech_duration = 5  # 最小语音持续块数
            max_silent_duration = 10  # 最大静默持续块数

            # 持续录音
            try:
                while True:
                    # 读取音频
                    audio_data = stream.read(chunk_size, exception_on_overflow=False)

                    # 计算音频能量
                    audio_array = np.frombuffer(audio_data, dtype=np.int16)
                    energy = np.mean(np.abs(audio_array)) / 32768.0

                    # 语音活动检测
                    is_voice = energy > vad_threshold

                    # 显示音量指示
                    bars = int(energy * 40)
                    print(f"\r音量: [{'█' * bars}{' ' * (40 - bars)}] 能量: {energy:.4f} 语音: {'✓' if is_voice else '✗'} 块: {chunk_count}", end="")

                    # 基于语音活动控制发送
                    if is_voice:
                        # 检测到语音
                        await websocket.send(audio_data)
                        chunk_count += 1
                        silent_count = 0
                        if not speaking:
                            speaking = True
                            print("\n🔊 检测到语音，开始发送数据")
                    else:
                        # 检测到静默
                        silent_count += 1
                        if speaking and silent_count > max_silent_duration:
                            # 足够长的静默，认为语音结束
                            speaking = False
                            print("\n🔇 检测到静默，结束语音段")
                            # 发送结束标志获取结果
                            end_msg = json.dumps({"is_speaking": False})
                            await websocket.send(end_msg)
                            # 尝试接收结果
                            try:
                                result = await asyncio.wait_for(websocket.recv(), timeout=2.0)
                                try:
                                    data = json.loads(result)
                                    text = data.get('text', '无')
                                    if text and text.strip():
                                        print(f"\n完整文本: {text}")
                                    else:
                                        print("\n🔍 无有效文本")
                                except:
                                    pass
                            except asyncio.TimeoutError:
                                pass
                            # 重新发送开始标志继续录音
                            start_msg = json.dumps({"is_speaking": True})
                            await websocket.send(start_msg)
                            chunk_count = 0
                    # 定期发送保持连接
                    if chunk_count % 50 == 0 and chunk_count > 0:
                        # 发送一个小的音频包保持连接
                        await websocket.send(audio_data[:100])
            except KeyboardInterrupt:
                # 清理
                stream.stop_stream()
                stream.close()
                p.terminate()
    except Exception as e:
        print(f"\n❌ 错误: {e}")
        import traceback
        traceback.print_exc()


# 运行测试
if __name__ == "__main__":
    asyncio.run(test_funasr())