2. 音频数据采集与预处理流程
任务目标
本案例实现一个完整的音频采集与预处理流程,用于边缘AI语音唤醒场景。流程包含:从麦克风采集PCM音频数据、分帧、加窗、短时能量计算与静音段丢弃,最终输出适合后续特征提取(如MFCC)的预处理音频帧序列。代码可在PC或树莓派等Linux环境直接运行。
环境依赖
- Python 3.8+
- NumPy
- PyAudio(需系统安装PortAudio:
sudo apt install portaudio19-dev)
完整实现
import numpy as np
import pyaudio
import time
# ---------- 配置参数 ----------
SAMPLE_RATE = 16000 # 16kHz,语音唤醒常用采样率
CHUNK = 1024 # 每次读取的采样点数
FORMAT = pyaudio.paInt16 # 16bit PCM
CHANNELS = 1 # 单声道
FRAME_LEN = 400 # 帧长 25ms (400/16000)
FRAME_SHIFT = 160 # 帧移 10ms (160/16000)
ENERGY_THRESHOLD = 300.0 # 静音判定阈值(需根据环境调整)
MIN_SILENCE_FRAMES = 5 # 连续静音帧数超过该值则停止采集
# ---------- 预处理核心函数 ----------
def preprocess_frame(samples):
"""对单帧数据执行:加汉明窗 + 计算短时能量"""
window = np.hamming(len(samples))
framed = samples * window
energy = np.sum(framed ** 2) / len(samples)
return framed, energy
def collect_voice_utterance(max_duration=3.0):
"""
从麦克风采集一段语音,自动去除首尾静音。
返回:有效语音帧列表(每帧为加窗后的float数组)
"""
p = pyaudio.PyAudio()
stream = p.open(format=FORMAT,
channels=CHANNELS,
rate=SAMPLE_RATE,
input=True,
frames_per_buffer=CHUNK)
buffer = [] # 原始采样点缓冲
voice_frames = [] # 有效语音帧
silence_count = 0
voice_started = False
start_time = time.time()
try:
while time.time() - start_time < max_duration:
# 读取一块数据并转为float
raw = stream.read(CHUNK, exception_on_overflow=False)
data = np.frombuffer(raw, dtype=np.int16).astype(np.float32)
buffer.extend(data)
# 从缓冲中按帧长取出完整帧
while len(buffer) >= FRAME_LEN:
frame = np.array(buffer[:FRAME_LEN])
del buffer[:FRAME_SHIFT] # 帧移滑动
framed, energy = preprocess_frame(frame)
if energy > ENERGY_THRESHOLD:
voice_frames.append(framed)
silence_count = 0
voice_started = True
else:
if voice_started:
silence_count += 1
# 尾部静音超过阈值则结束
if silence_count >= MIN_SILENCE_FRAMES:
return voice_frames
# 未开始语音前的静音帧直接丢弃
finally:
stream.stop_stream()
stream.close()
p.terminate()
return voice_frames
# ---------- 主程序入口 ----------
if __name__ == "__main__":
print("请说话(自动检测语音起止)...")
frames = collect_voice_utterance()
if len(frames) == 0:
print("未检测到有效语音")
else:
# 拼接所有帧得到完整语音信号(用于验证)
full_signal = np.concatenate(frames)
total_time = len(full_signal) / SAMPLE_RATE
print(f"采集到 {len(frames)} 帧,语音时长约 {total_time:.2f} 秒")
print(f"每帧长度: {len(frames[0])} 采样点")
print("预处理完成,可送入特征提取模块")
运行与验证
- 安装依赖:
pip install numpy pyaudio - 运行程序:
python audio_preprocess.py - 验证步骤:
- 程序启动后,对着麦克风说“你好小智”等唤醒词,停顿1秒后程序自动停止;
- 观察输出:应显示采集帧数(通常20~60帧)和语音时长(0.5~2秒);
- 若环境安静时误触发,调高
ENERGY_THRESHOLD;若说话被截断,调低该值; - 若输出“未检测到有效语音”,检查麦克风权限或增大说话音量。
判定通过标准:程序能自动检测一段连续语音的起止边界,输出帧数大于10且每帧长度为400,说明采集、分帧、加窗和静音丢弃全链路工作正常。
关键说明
- 帧长25ms、帧移10ms是语音唤醒的常用配置,兼顾时域分辨率和计算开销;
- 能量阈值需按实际环境噪声调整,本代码默认值适用于较安静室内;
- 输出帧为加窗后的float数组,可直接输入MFCC或滤波器组特征提取模块;
- 若需在嵌入式设备(如ESP32)运行,可将PyAudio替换为I2S驱动,核心分帧算法不变。