Python中FFT波形的索引峰值

Question

我有从加速度计获得的采样数据，每个轴（“x”、“y”和“z”）上的加速度。

该数据存储为 Pandas DataFrame，每个轴都有一列。

这样，我得到了 FFT：

import pandas as pd
from scipy import fft
from typing import Optional

def fft_raw(df: pd.DataFrame) -> pd.DataFrame:
    """Calculates the raw Fast Fourier Transform of a DataFrame

    Parameters
    ----------
    df : pd.DataFrame
        DataFrame whose FFT should be calculated

    Returns
    -------
    pd.DataFrame
        Raw FFT
    """

    fft_raw = pd.DataFrame()
    for c in df:
        fft_raw = fft_raw.join(
            pd.DataFrame(fft.rfft(np.array(df[c])), columns=[c]), how="outer"
        )
    return fft_raw


def norm_fft(fft_raw: pd.DataFrame, length: Optional[int] = None) -> pd.DataFrame:
    """Normalizes a raw FFT

    Parameters
    ----------
    fft_raw : pd.DataFrame
        Raw FFT to be normalized
    length : int, optional
        How many sample points were used for calculating the raw FFT
        It uses the length of `fft_raw` by default

    Returns
    -------
    pd.DataFrame
        Normalized FFT
    """

    if length is None:
        length = len(fft_raw)

    return 2.0 * (fft_raw.abs() / length)

它最终会看起来像这样：

然后我想从这种 FFT 中提取峰值。

在上面给出的示例中，“z”轴的峰值约为：

  2.90 Hz at 0.15 g
 54.56 Hz at 0.90 g
106.22 Hz at 0.10 g

我考虑做的一件事是过滤掉所有幅度低于给定阈值的频率（例如，它可能在0.1 g 左右），这会给我一个只有峰值的波形。

这样做的问题是，我仍然会在峰值周围有很多采样点，因为它通常会“上升”到峰值的最大值，然后“下降”到几乎没有，这需要增加多个点，而不仅仅是一个。

然后我考虑尝试将波形“拆分”成代表单个峰值的点组，这样我就可以找到它们的最大值，但我不太确定这样做的有效方法。

我试图找到一个类似的问题，然后遇到了this one，但我无法让它工作，即使在将我的数据简化回 Numpy 数组之后。

所以我决定在这里询问是否有人知道一种有效的方法，最好是使用 Pandas 来获得 FFT 的峰值。

Answer 1

我找到了一种实现理想结果的方法，基于 this question，并使用 PeakUtils 库：

import pandas as pd
import peakutils


def find_peaks(
    df: pd.DataFrame, threshold: pd.Series, min_dist: int = 50
) -> Dict[str, pd.Series]:
    index = df.index
    df.reset_index(drop=True, inplace=True)
    all_peaks = dict()
    for c in df:
        if c in threshold:
            data = df[c]
            peaks = peakutils.indexes(
                data, thres=threshold[c], min_dist=min_dist, thres_abs=True
            )
            all_peaks[c] = pd.Series()
            for peak in peaks:
                peak_index = index[peak]
                val = data[peak]
                to_append = pd.Series([val], index=[peak_index])
                all_peaks[c] = all_peaks[c].append(to_append)

    return all_peaks


def find_fft_peaks(
    df: pd.DataFrame, threshold: pd.Series, dist_hz: float = 1
) -> Dict[str, pd.Series]:
    index = df.index
    index_interval = index[-1] - index[0]
    points_per_hz = len(index) / index_interval
    min_dist = int(points_per_hz * dist_hz)
    return find_peaks(df=df, threshold=threshold, min_dist=min_dist)

它似乎做得很好，虽然我仍然觉得它有点 hacky。

它可能不是超级高效或干净，但它现在可以完成这项工作。