177 lines
5.5 KiB
Python
177 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试sounddevice麦克风录音功能
|
|
用于验证新的麦克风输入实现是否正常工作
|
|
"""
|
|
|
|
import numpy as np
|
|
import sounddevice as sd
|
|
import time
|
|
import threading
|
|
import queue
|
|
import sys
|
|
|
|
def test_microphone():
|
|
"""测试麦克风录音"""
|
|
print("=== SoundDevice麦克风录音测试 ===")
|
|
|
|
# 1. 检查音频输入设备
|
|
print("\n1. 检查音频输入设备...")
|
|
try:
|
|
devices = sd.query_devices()
|
|
input_devices = [dev for dev in devices if dev['max_input_channels'] > 0]
|
|
|
|
print(f"找到 {len(input_devices)} 个输入设备:")
|
|
for i, dev in enumerate(input_devices):
|
|
print(f" [{i}] {dev['name']} (输入通道: {dev['max_input_channels']})")
|
|
|
|
if not input_devices:
|
|
print("错误: 没有找到可用的音频输入设备")
|
|
return False
|
|
|
|
# 查找默认输入设备
|
|
default_input = sd.default.device[0] if isinstance(sd.default.device, tuple) else sd.default.device
|
|
print(f"默认输入设备: {default_input}")
|
|
|
|
except Exception as e:
|
|
print(f"音频设备检查失败: {e}")
|
|
return False
|
|
|
|
# 2. 测试录音5秒
|
|
print("\n2. 测试录音5秒...")
|
|
try:
|
|
sample_rate = 16000
|
|
channels = 1
|
|
duration = 5
|
|
chunk_size = 3200
|
|
|
|
print(f"录音参数: 采样率={sample_rate}Hz, 通道={channels}, 时长={duration}秒")
|
|
print("开始录音，请说话...")
|
|
|
|
# 创建音频队列
|
|
audio_queue = queue.Queue()
|
|
recording = True
|
|
|
|
def audio_callback(indata, frames, time_info, status):
|
|
"""音频数据回调"""
|
|
if status:
|
|
print(f"音频流状态: {status}")
|
|
if recording:
|
|
audio_queue.put(indata.copy())
|
|
|
|
# 创建输入流
|
|
with sd.InputStream(
|
|
samplerate=sample_rate,
|
|
channels=channels,
|
|
dtype='int16',
|
|
blocksize=chunk_size,
|
|
callback=audio_callback
|
|
) as stream:
|
|
# 录音指定时长
|
|
start_time = time.time()
|
|
audio_data = []
|
|
|
|
while time.time() - start_time < duration:
|
|
try:
|
|
data = audio_queue.get(timeout=1.0)
|
|
audio_data.append(data)
|
|
except queue.Empty:
|
|
print("警告: 音频队列为空")
|
|
break
|
|
|
|
print(f"录音完成，共收集到 {len(audio_data)} 个音频块")
|
|
|
|
# 3. 播放录制的音频
|
|
if audio_data:
|
|
print("\n3. 播放录制的音频...")
|
|
|
|
# 合并音频数据
|
|
recorded_audio = np.concatenate(audio_data, axis=0)
|
|
print(f"录制音频形状: {recorded_audio.shape}")
|
|
|
|
# 播放
|
|
print("开始播放录制的音频...")
|
|
sd.play(recorded_audio, sample_rate)
|
|
sd.wait()
|
|
print("✓ 音频播放完成")
|
|
|
|
# 保存音频文件
|
|
print("\n4. 保存音频文件...")
|
|
try:
|
|
from scipy.io import wavfile
|
|
wavfile.write('test_recording.wav', sample_rate, recorded_audio)
|
|
print("✓ 音频已保存为 test_recording.wav")
|
|
except ImportError:
|
|
print("提示: 安装scipy可保存WAV文件: pip install scipy")
|
|
|
|
else:
|
|
print("警告: 没有录制到音频数据")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"录音测试失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
def test_stream_reading():
|
|
"""测试流式读取"""
|
|
print("\n5. 测试流式读取...")
|
|
|
|
try:
|
|
sample_rate = 16000
|
|
channels = 1
|
|
chunk_size = 3200
|
|
|
|
# 创建输入流
|
|
with sd.InputStream(
|
|
samplerate=sample_rate,
|
|
channels=channels,
|
|
dtype='int16',
|
|
blocksize=chunk_size
|
|
) as stream:
|
|
print("开始流式读取测试...")
|
|
|
|
# 读取10个数据块
|
|
for i in range(10):
|
|
audio_data = stream.read(chunk_size)
|
|
print(f"读取第 {i+1} 块数据: 形状={audio_data.shape}, 类型={audio_data.dtype}")
|
|
|
|
# 转换为字节数据
|
|
byte_data = audio_data.tobytes()
|
|
print(f"字节数据长度: {len(byte_data)} 字节")
|
|
|
|
time.sleep(0.1) # 模拟实际处理间隔
|
|
|
|
print("✓ 流式读取测试完成")
|
|
|
|
except Exception as e:
|
|
print(f"流式读取测试失败: {e}")
|
|
return False
|
|
|
|
return True
|
|
|
|
if __name__ == "__main__":
|
|
print("SoundDevice麦克风录音功能测试")
|
|
print("=" * 50)
|
|
|
|
success = True
|
|
|
|
# 测试麦克风
|
|
if not test_microphone():
|
|
success = False
|
|
|
|
# 测试流式读取
|
|
if not test_stream_reading():
|
|
success = False
|
|
|
|
print("\n" + "=" * 50)
|
|
if success:
|
|
print("✓ 所有麦克风测试通过")
|
|
print("树莓派应该可以正常录音了！")
|
|
else:
|
|
print("✗ 部分测试失败，需要检查音频设备和权限")
|
|
print("请确保:")
|
|
print("1. 麦克风已正确连接")
|
|
print("2. 用户有音频设备访问权限")
|
|
print("3. 没有其他程序占用音频设备") |