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("预处理完成,可送入特征提取模块")

运行与验证

  1. 安装依赖:pip install numpy pyaudio
  2. 运行程序:python audio_preprocess.py
  3. 验证步骤:
    • 程序启动后,对着麦克风说“你好小智”等唤醒词,停顿1秒后程序自动停止;
    • 观察输出:应显示采集帧数(通常20~60帧)和语音时长(0.5~2秒);
    • 若环境安静时误触发,调高ENERGY_THRESHOLD;若说话被截断,调低该值;
    • 若输出“未检测到有效语音”,检查麦克风权限或增大说话音量。

判定通过标准:程序能自动检测一段连续语音的起止边界,输出帧数大于10且每帧长度为400,说明采集、分帧、加窗和静音丢弃全链路工作正常。

关键说明

  • 帧长25ms、帧移10ms是语音唤醒的常用配置,兼顾时域分辨率和计算开销;
  • 能量阈值需按实际环境噪声调整,本代码默认值适用于较安静室内;
  • 输出帧为加窗后的float数组,可直接输入MFCC或滤波器组特征提取模块;
  • 若需在嵌入式设备(如ESP32)运行,可将PyAudio替换为I2S驱动,核心分帧算法不变。