Pywayne Dsp

Pywayne Dsp

数字信号处理工具集，提供滤波器、峰值检测、去趋势、曲线相似度等信号处理功能。

Quick Start

from pywayne.dsp import butter_bandpass_filter, peak_det, SignalDetrend

# Butterworth 低通滤波
filtered = butter_bandpass_filter(signal, order=3, lo=0.5, hi=40, fs=250)

# 峰值检测
peaks, valleys = peak_det(signal, delta=0.5)

# 信号去趋势
detrender = SignalDetrend(method='linear')
detrended = detrender(raw_signal)

Filtering - 滤波器

butter_bandpass_filter

巴特沃斯带通滤波器。

from pywayne.dsp import butter_bandpass_filter

# 带通滤波
filtered = butter_bandpass_filter(
    signal=raw_signal,
    order=4,
    lo=1,
    hi=50,
    fs=250,
    btype='bandpass'
)

参数说明：

ButterworthFilter

基于 NumPy 的巴特沃斯滤波器类，同时支持传递函数 ba 和数值更稳定的二阶节级联 sos。 BA/SOS 执行核心均为 Direct Form II Transposed，对齐 SciPy 的 lfilter/sosfilt 和 filtfilt/sosfiltfilt。

from pywayne.dsp import ButterworthFilter

# 方式 1：通过参数设计 BA
bf = ButterworthFilter.from_params(order=4, fs=200, btype='bandpass', cutoff=(1, 50))
y, zf = bf.lfilter(signal)

# 方式 2：通过系数构造
bf2 = ButterworthFilter.from_ba(b, a)
y, zf = bf2.lfilter(signal)

# 方式 3：SOS（高阶、窄带或接近 0/Nyquist 时推荐）
bf_sos = ButterworthFilter.from_params(
    order=8,
    fs=200,
    btype='bandpass',
    cutoff=(0.1, 10),
    output='sos',
)

# 也可直接使用 scipy.signal.butter(..., output='sos') 的系数
bf_sos2 = ButterworthFilter.from_sos(sos)

# 零相位滤波（前向-后向）
y = bf.filtfilt(signal)

# 流式处理：首段稳态初始化，后续传递 zf
state = bf_sos.zi() * chunks[0][0]
outputs = []
for chunk in chunks:
    y, state = bf_sos.lfilter(chunk, zi=state)
    outputs.append(y)

# 去趋势
detrended = ButterworthFilter.detrend(signal, method='linear')

参数设计方法：

ButterworthFilter.from_params(order, fs, btype, cutoff, cache_zi=True, output='ba')
ButterworthFilter.from_ba(b, a, cache_zi=True)
ButterworthFilter.from_sos(sos, cache_zi=True)
ButterworthFilter.lfilter_zi(b, a)
ButterworthFilter.sosfilt_zi(sos)

参数说明：

实例方法：

SOS 模式的 lfilter 接受 SciPy 形状 (n_sections, 2) 或等价扁平 zi，返回的 zf 始终是 (n_sections, 2)。在线滤波首段建议使用 zi() * x[0]，后续必须持续传递 前一段的 zf。filtfilt 不可用于实时流式处理，且有限长度信号仍需检查边界效应。

Peak Detection - 峰值检测

peak_det

峰值检测函数，基于 MATLAB peakdet 转换。

from pywayne.dsp import peak_det

max_peaks, min_peaks = peak_det(signal, delta=0.5)

参数说明：

返回值：(maxima_indices, minima_indices) - 峰值和谷值的索引位置

find_extremum_in_sliding_window

在滑动窗口中查找极值。

from pywayne.dsp import find_extremum_in_sliding_window

extrema = find_extremum_in_sliding_window(data, k=50)

参数说明：

返回值：[minima, maxima] - 含局部极值的列表

FindSlidingWindowExtremum

滑动窗口极值查找器类，用于实时数据流。

from pywayne.dsp import FindSlidingWindowExtremum

detector = FindSlidingWindowExtremum(win=100, find_max=True)

# 应用新值
for sample in stream:
    current_peak = detector.apply(sample)
    # 处理 current_peak

方法：

Detrending - 信号去趋势

SignalDetrend

信号去趋势处理器，支持多种去趋势算法。

from pywayne.dsp import SignalDetrend

# 去除线性趋势
detrender = SignalDetrend(method='linear')
detrended = detrender(signal)

# 去除均值趋势
detrender = SignalDetrend(method='mean')
detrended = detrender(signal)

# LOESS 去趋势
detrender = SignalDetrend(method='loess', span=0.3)
detrended = detrender(signal)

方法：

去趋势方法：

Curve Similarity - 曲线相似度

CurveSimilarity

曲线相似度计算，支持动态时间规整（DTW）。

from pywayne.dsp import CurveSimilarity

cs = CurveSimilarity()
distance = cs.dtw(curve1, curve2, mode='global')

# 可选后端：auto（默认）、python、numba
fast_distance = cs.dtw(curve1, curve2, backend='numba')

方法：

Numba 是可选性能依赖，可通过 pip install "pywayne[performance]" 安装。首次调用会进行 JIT 编译， 重复调用或较长曲线收益更明显。Python 和 Numba 后端保持相同的评分语义。

Other Tools - 其他工具

OneEuroFilter

一欧罗滤波器，用于平滑信号并减少延迟。

from pywayne.dsp import OneEuroFilter

# 初始化
euro_filter = OneEuroFilter(te=0.02, mincutoff=1.0, beta=0.007, dcutoff=1.0)

# 应用滤波
smooth_value = euro_filter.apply(new_measurement, te=0.02)

参数说明：

WelfordStd

使用 Welford 算法进行在线标准差计算。

from pywayne.dsp import WelfordStd

std_calculator = WelfordStd(win=50)

for sample in data_stream:
    current_std = std_calculator.apply(sample)
    # 使用 current_std 进行判断

方法：

应用场景