3. 数据预处理

任务目标

本案例完成EdgeImpulse数据预处理中最基础的两步：对原始音频样本进行裁剪（去除首尾静音/无用片段）与重采样（统一采样率）。你将得到一个可直接运行的Python脚本，它读取一个WAV文件，输出裁剪并重采样后的新WAV文件，并打印处理前后的关键参数用于验证。

环境与依赖

需要Python 3.8+，并安装以下库（建议使用虚拟环境）：

pip install numpy scipy soundfile

完整实现代码

将以下代码保存为 preprocess_audio.py：

import numpy as np
import soundfile as sf
from scipy import signal

def load_audio(file_path):
    """读取音频，返回数据(归一化到[-1,1])和采样率"""
    data, sr = sf.read(file_path, dtype='float32')
    # 若为立体声，取单声道（平均）
    if data.ndim > 1:
        data = np.mean(data, axis=1)
    return data, sr

def trim_silence(data, sr, top_db=30, frame_ms=20, hop_ms=10):
    """
    基于能量裁剪首尾静音。
    逻辑：按帧计算RMS能量，找到能量超过阈值(相对最大值)的首帧和末帧。
    """
    frame_len = int(sr * frame_ms / 1000)
    hop_len = int(sr * hop_ms / 1000)
    if len(data) < frame_len:
        return data  # 太短则不裁剪

    # 计算每帧RMS
    n_frames = 1 + (len(data) - frame_len) // hop_len
    rms = np.zeros(n_frames)
    for i in range(n_frames):
        frame = data[i*hop_len : i*hop_len + frame_len]
        rms[i] = np.sqrt(np.mean(frame**2) + 1e-12)

    if rms.max() == 0:
        return data  # 全静音，不裁剪

    # 阈值 = 最大值 / (10^(top_db/20))
    threshold = rms.max() / (10 ** (top_db / 20))
    voiced_idx = np.where(rms > threshold)[0]
    if len(voiced_idx) == 0:
        return data

    start_frame = voiced_idx[0]
    end_frame = voiced_idx[-1]
    # 转换为样本索引，并向外扩展hop_len的一半作为平滑余量
    start_sample = max(0, start_frame * hop_len - hop_len // 2)
    end_sample = min(len(data), (end_frame + 1) * hop_len + hop_len // 2)
    return data[start_sample:end_sample]

def resample_audio(data, orig_sr, target_sr):
    """使用scipy的多相重采样将数据重采样到target_sr"""
    if orig_sr == target_sr:
        return data
    # 计算重采样后的长度
    new_len = int(len(data) * target_sr / orig_sr)
    # scipy.signal.resample使用FFT方法，适合短音频；polyphase更省内存
    # 这里使用resample_poly，更稳定
    gcd = np.gcd(orig_sr, target_sr)
    up = target_sr // gcd
    down = orig_sr // gcd
    resampled = signal.resample_poly(data, up, down)
    # 修正长度（resample_poly可能多1个样本）
    return resampled[:new_len]

def process_file(input_path, output_path, target_sr=16000):
    """主处理流程：读取 -> 裁剪 -> 重采样 -> 保存"""
    print(f"处理文件: {input_path}")
    data, orig_sr = load_audio(input_path)
    print(f"原始采样率: {orig_sr} Hz, 样本数: {len(data)}, 时长: {len(data)/orig_sr:.2f}s")

    # 1. 裁剪静音
    trimmed = trim_silence(data, orig_sr)
    print(f"裁剪后样本数: {len(trimmed)}, 时长: {len(trimmed)/orig_sr:.2f}s")

    # 2. 重采样
    resampled = resample_audio(trimmed, orig_sr, target_sr)
    print(f"重采样后采样率: {target_sr} Hz, 样本数: {len(resampled)}, 时长: {len(resampled)/target_sr:.2f}s")

    # 3. 保存（使用PCM_16格式）
    sf.write(output_path, resampled, target_sr, subtype='PCM_16')
    print(f"已保存到: {output_path}")
    return len(resampled)

if __name__ == "__main__":
    # 使用示例：处理当前目录下的 input.wav，输出到 output.wav
    process_file("input.wav", "output.wav", target_sr=16000)

运行与验证

运行方式：将你的原始音频命名为 input.wav 放在脚本同目录，然后执行：

python preprocess_audio.py

验证方法与通过条件：

1. 程序正常退出，无报错信息。
2. 观察控制台输出：裁剪后的时长应小于或等于原始时长（若原始首尾无静音则相等）；重采样后的采样率应显示为16000 Hz。
3. 检查输出文件：用任意音频工具（如Audacity）打开 output.wav，确认：
   - 文件属性中采样率为16000 Hz；
   - 波形首尾不再有长段平直的静音区（若原始音频本身无静音则波形基本不变）。
4. 预期输出示例（具体数值取决于输入文件）：
   原始采样率: 44100 Hz, 样本数: 220500, 时长: 5.00s
裁剪后样本数: 180000, 时长: 4.08s
重采样后采样率: 16000 Hz, 样本数: 65306, 时长: 4.08s

常见问题

• 输入为立体声：脚本自动取左右声道平均转为单声道，这是EdgeImpulse默认推荐做法。
• 裁剪阈值：top_db=30 表示低于峰值30dB的帧视为静音。若发现裁剪过多/过少，可调整该参数（减小则裁剪更少，增大则裁剪更多）。
• 重采样质量：使用 resample_poly 可避免FFT重采样在长音频上的边界伪影，适合语音/环境音预处理。