使用 UliEngineering 在 Python 中轻松计算和可视化 FFT

UliEngineering 是 Python 中的混合数据分析库 - 它提供的实用工具之一是易于使用的 FFT 计算包。与其他包不同,此库面向实际用例,允许你仅用一行代码完成 FFT!不需要数学知识。 首先,安装 UliEngineering

入门

首先,我们将生成一些测试数据。请参见此之前的文章了解如何生成正弦测试数据的更多详情:

generate_test_data.py
from UliEngineering.SignalProcessing.Simulation import *

# 生成测试数据:100 Hz + 400 Hz 音调
data = sine_wave(frequency=100.0, samplerate=1000, amplitude=1.) \
       + sine_wave(frequency=400.0, samplerate=1000, amplitude=0.5)

测试数据由 100 Hz 正弦波加上 400 Hz 正弦波(振幅为一半)组成。该信号以 1000 Hz 的采样率采样。

现在我们可以使用 matplotlib 计算和可视化 FFT:

compute_fft.py
# 计算 FFT。使用相同的采样率
# 注意:窗口默认为 "blackman"!
from UliEngineering.SignalProcessing.FFT import compute_fft
fft = compute_fft(data, samplerate=1e3)

# 绘图
from matplotlib import pyplot as plt
plt.style.use("ggplot")

plt.gcf().set_size_inches(10, 5) # 使用 (20, 10) 获取更大的图
plt.plot(fft.frequencies, fft.amplitudes)
plt.xlabel("Frequency")
plt.ylabel("Amplitude")

compute_fft(data, samplerate=1e3) 返回一个 FFT 对象,包含 frequenciesamplitudephase 等字段。它执行与输入大小相同的 FFT(即由于 data 是长度为 1000 的数组,FFT 大小将为 1000)。

fft.frequencies 是频率数组(以 Hz 为单位),对应于 fft.amplitudes 中的值。你也可以使用 fft.angles 获取以度为单位的相对角度,但这不在本博客文章中涵盖。

如上面显示的图所示,可以检测的最大频率始终为采样率的一半,即对于我们的采样率 $f_s = 1000,\text{Hz}$,它是 $500,\text{Hz}$。如果你想了解更多详情,请参见这个更以数学为中心的 FFT 解释

在内部,compute_fft() 执行此计算:

$$2 \cdot \frac{\text{abs}\left(\text{FFT}(\text{data} \cdot \text{Window})\right)}{\text{len(data)}}$$

选择频率范围

使用 UliEngineering API,选择 FFT 的频率范围非常简单:只需使用 fft[lowfreq:highfreq]。你可以使用 fft[lowfreq:] 选择从 lowfreq 开始的所有内容,或使用 fft[:highfreq] 选择直到 highfreq 的所有内容。

fft_frequency_range.py
from UliEngineering.SignalProcessing.FFT import compute_fft
fft = compute_fft(data, samplerate=1e3)

# 选择频率范围:50 到 200 Hz
fft = fft[50.0:200.0]

# 绘图
from matplotlib import pyplot as plt
plt.style.use("ggplot")

plt.gcf().set_size_inches(10, 5) # 使用 (20, 10) 获取更大的图
plt.plot(fft.frequencies, fft.amplitudes)
plt.xlabel("Frequency")
plt.ylabel("Amplitude")
plt.savefig("/ram/fft-frequency-range.svg")

Frequency window

提取特定频率的振幅和角度

通过使用 [frequency],即带有单个值的 getitem 操作符,你将获得一个 FFTPoint() 对象,包含给定频率的频率、振幅和相对角度。库会自动选择最近的 FFT 桶,因此即使你的 FFT 没有该特定频率的桶,你也会得到合理的结果。

fft_point.py
from UliEngineering.SignalProcessing.FFT import compute_fft
fft = compute_fft(data, samplerate=1e3)

# 显示特定频率的值
print(fft[30]) # FFTPoint(frequency=30.0, value=2.91e-08, angle=0.0)
print(fft[100]) # FFTPoint(frequency=100.0, value=0.419, angle=0.0)

长数据的短 FFT

使用 compute_fft(),如果我们有一个极长的数据数组,这意味着我们将计算一个极长的 FFT。在许多情况下,这是不可取的,你想计算固定大小的 FFT(通常是 2 的幂 FFT,例如 102420484096 等)。

让我们生成一些长测试数据并假设我们想在其上计算大小为 1024 的 FFT

simple_serial_fft.py
from UliEngineering.SignalProcessing.FFT import *
fft = simple_serial_fft_reduce(data, samplerate=1e3, fftsize=1024)
# fft.frequencies、fft.amplitudes 等

像上面那样绘制数据会产生

FFT plot of 100 Hz and 400 Hz sine waves using serial FFT reduce这看起来几乎与之前我们的 compute_fft() 图完全一样 - 正如我们所预期的。

simple_serial_fft_reduce() 为我们处理所有魔法和归一化,包括将数据分区为重叠块、添加 FFT 并正确归一化结果。

命名约定在此很重要:

并行化 FFT

如果你有庞大的数据集,你可以像 simple_serial_fft_reduce() 一样使用 simple_parallel_fft_reduce()

simple_parallel_fft.py
from UliEngineering.SignalProcessing.FFT import *
fft = simple_parallel_fft_reduce(data, samplerate=1e3, fftsize=1024)
# 使用 fft.frequencies、fft.amplitudes 等

但是在大多数情况下,你想手动初始化执行器以便稍后重用:

simple_parallel_fft_executor.py
from UliEngineering.SignalProcessing.FFT import *
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor() # 无参数 => 使用 num_cpus 线程
fft = simple_parallel_fft_reduce(data, samplerate=1e3, fftsize=1024, executor=executor)

我们可以使用 ThreadPoolExecutor(),因为 scipy.fftpackUliEngineering 用来做繁重数学计算的)解锁了 Python GIL。

注意由于需要执行大量内务任务,如果你的数据集太小以至于并行化无效,simple_parallel_fft_reduce()simple_serial_fft_reduce() 慢得多。我最初的建议是如果串行变体的总执行时间大于 $0.5s$,则考虑使用并行变体


Check out similar posts by category: Data Science, Mathematics, Python