📚 课程目录第3章峰值检测算法实现

3. 峰值检测算法实现

任务说明

本案例要求设计并实现一个基于阈值的峰值检测算法，从三轴加速度计采集的步行信号中提取步态峰值（即每一步落地时产生的冲击波峰）。输入为一段连续的加速度采样序列，输出为检测到的峰值位置索引及对应的幅值，用于后续步数统计。

算法设计

步态加速度信号具有准周期性，每一步对应一个明显的波峰。本算法采用以下策略：

信号预处理：计算三轴加速度的合成幅值 acc_mag = sqrt(x² + y² + z²)，消除传感器朝向影响；
滑动均值滤波：对合成幅值做窗口为5的移动平均，抑制高频噪声；
动态阈值：以信号均值为基准，加上固定偏移量作为检测阈值（threshold = mean + offset），offset取0.8倍标准差；
峰值判定：候选点需满足：幅值大于阈值、且大于左右各2个相邻点（局部极大值）；
最小间隔约束：相邻有效峰值间隔不小于10个采样点（约0.2s@50Hz），避免双峰误检。

代码实现

import numpy as np

def peak_detection(acc_data, fs=50.0, offset_factor=0.8, min_interval=10):
    """
    基于阈值的步态峰值检测
    参数:
        acc_data: (N,3) 数组，列为 [ax, ay, az]，单位 g
        fs: 采样率 Hz
        offset_factor: 阈值偏移系数（×标准差）
        min_interval: 最小峰值间隔（采样点数）
    返回:
        peaks_idx: 峰值索引列表
        peaks_val: 对应幅值列表
    """
    # 1. 合成幅值
    mag = np.sqrt(np.sum(np.square(acc_data), axis=1))

    # 2. 滑动均值滤波（窗口=5）
    win = 5
    kernel = np.ones(win) / win
    mag_filt = np.convolve(mag, kernel, mode='same')

    # 3. 动态阈值
    mean_val = np.mean(mag_filt)
    std_val = np.std(mag_filt)
    threshold = mean_val + offset_factor * std_val

    # 4. 峰值检测
    peaks_idx = []
    n = len(mag_filt)
    k = 2  # 左右比较半径

    for i in range(k, n - k):
        if mag_filt[i] <= threshold:
            continue
        # 局部极大值判断
        if mag_filt[i] > mag_filt[i-k] and mag_filt[i] > mag_filt[i+k]:
            # 检查与上一个峰值的间隔
            if len(peaks_idx) == 0 or (i - peaks_idx[-1]) >= min_interval:
                peaks_idx.append(i)

    peaks_val = [mag_filt[i] for i in peaks_idx]
    return peaks_idx, peaks_val

# ========== 样例数据生成与测试 ==========
if __name__ == "__main__":
    # 生成模拟步行信号：50Hz采样，时长10秒，步频约2Hz
    fs = 50.0
    t = np.arange(0, 10, 1/fs)
    n = len(t)

    # 模拟三轴加速度（步行时z轴为主，叠加噪声）
    step_freq = 2.0  # Hz
    gravity = 1.0    # g
    step_amp = 0.4   # 步态波动幅值

    # 构造周期性步态信号 + 随机噪声
    phase = 2 * np.pi * step_freq * t
    ax = 0.05 * np.sin(phase) + np.random.normal(0, 0.02, n)
    ay = 0.03 * np.cos(phase) + np.random.normal(0, 0.02, n)
    az = gravity + step_amp * np.sin(phase) + np.random.normal(0, 0.03, n)
    acc_data = np.column_stack([ax, ay, az])

    # 执行峰值检测
    peaks_idx, peaks_val = peak_detection(acc_data, fs=fs)

    # 输出结果
    print(f"采样点数: {n}, 采样时长: {n/fs:.1f}s")
    print(f"检测到峰值数: {len(peaks_idx)}")
    print(f"预期步数(2Hz×10s): {int(step_freq*10)}")
    print(f"峰值索引(前10个): {peaks_idx[:10]}")
    print(f"峰值幅值(前10个): {[round(v,3) for v in peaks_val[:10]]}")

落地与验证

运行方式：将代码保存为 peak_detection.py，在装有NumPy的Python环境中执行 python peak_detection.py。代码内置了模拟步行信号生成器，无需外部数据文件。

验证方法：观察输出峰值数量是否接近预期步数（模拟信号为2Hz×10s=20步）。由于随机噪声影响，检测结果允许±2步偏差。若峰值数在18~22之间，且峰值索引间隔均匀（约25个采样点），则算法实现正确。

判定标准：程序无报错运行完成；输出峰值数在18~22范围内；相邻峰值间隔在20~30个采样点之间（对应0.4~0.6s步态周期）。满足以上条件即判定通过。