Python怎么实现EMD算法

• 筛选（Sifting）

1. 求极值点 通过Find Peaks算法获取信号序列的全部 极大值和 极小值
2. 拟合包络曲线 通过信号序列的 极大值和 极小值组，经过 三次样条插值法获得两条光滑的波峰/波谷拟合曲线，即信号的 上包络线与 下包络线
3. 均值包络线 将两条极值曲线平均获得 平均包络线
4. 中间信号 原始信号减均值包络线，得到 中间信号
5. 判断本征模函数（IMF） IMF需要符合两个条件：1）在整个数据段内，极值点的个数和过零点的个数必须相等或相差最多不能超过一个。2）在任意时刻，由局部极大值点形成的上包络线和由局部极小值点形成的下包络线的平均值为零，即上、下包络线相对于时间轴局部对称。

下面利用公式来说明上面的分析过程。

EMD算法步骤

任何复杂的信号均可视为多个不同的固有模态函数叠加之和，任何模态函数可以是线性的或非线性的，并且任意两个模态之间都是相互独立的。在这个假设 基础上，复杂信号的EMD分解步骤如下：
步骤1：
寻找信号 全部极值点，通过三次样条曲线将局部极大值点连成上包络线，将局部极小值点连成下包络线。上、下包络线包含所有的数据点。

步骤2：
由上包络和下包络线的平均值 ，得出

若满足IMF的条件，则可认为是的第一个IMF分量。

步骤3：
若不符合IMF条件，则将作为原始数据，重复步骤1、步骤2，得到上、下包络的均值，通过计算是否适合IMF分量的必备条件，若不满足，重复如上两步次，直到满足前提下得到。第1个IMF表示如下:

步骤4：
将从信号中分离得到：

将作为原始信号重复上述三个步骤，循环次，得到第二个IMF分量直到第个IMF分量 ，则会得出：

步骤5：
当变成单调函数后，剩余的成为残余分量。所有IMF分量和残余分量之和为原始信号：

from scipy.signal import argrelextrema
"""通过Scipy的argrelextrema函数获取信号序列的极值点"""# 构建100个随机数data = np.random.random(100)# 获取极大值max_peaks = argrelextrema(data, np.greater)#获取极小值min_peaks = argrelextrema(data, np.less)
# 绘制极值点图像plt.figure(figsize = (18,6))plt.plot(data)plt.scatter(max_peaks, data[max_peaks], c='r', label='Max Peaks')plt.scatter(min_peaks, data[min_peaks], c='b', label='Max Peaks')plt.legend()plt.xlabel('time (s)')plt.ylabel('Amplitude')plt.title("Find Peaks")

from scipy.signal import argrelextrema
#进行样条差值import scipy.interpolate as spi
data = np.random.random(100)-0.5index = list(range(len(data)))
# 获取极值点max_peaks = list(argrelextrema(data, np.greater)[0])min_peaks = list(argrelextrema(data, np.less)[0])
# 将极值点拟合为曲线ipo3_max = spi.splrep(max_peaks, data[max_peaks],k=3) #样本点导入，生成参数iy3_max = spi.splev(index, ipo3_max) #根据观测点和样条参数，生成插值
ipo3_min = spi.splrep(min_peaks, data[min_peaks],k=3) #样本点导入，生成参数iy3_min = spi.splev(index, ipo3_min) #根据观测点和样条参数，生成插值
# 计算平均包络线iy3_mean = (iy3_max+iy3_min)/2
# 绘制图像plt.figure(figsize = (18,6))plt.plot(data, label='Original')plt.plot(iy3_max, label='Maximun Peaks')plt.plot(iy3_min, label='Minimun Peaks')plt.plot(iy3_mean, label='Mean')plt.legend()plt.xlabel('time (s)')plt.ylabel('microvolts (uV)')plt.title("Cubic Spline Interpolation")

def sifting(data):    index = list(range(len(data)))
    max_peaks = list(argrelextrema(data, np.greater)[0])    min_peaks = list(argrelextrema(data, np.less)[0])
    ipo3_max = spi.splrep(max_peaks, data[max_peaks],k=3) #样本点导入，生成参数    iy3_max = spi.splev(index, ipo3_max) #根据观测点和样条参数，生成插值
    ipo3_min = spi.splrep(min_peaks, data[min_peaks],k=3) #样本点导入，生成参数    iy3_min = spi.splev(index, ipo3_min) #根据观测点和样条参数，生成插值
    iy3_mean = (iy3_max+iy3_min)/2    return data-iy3_mean

def hasPeaks(data):    max_peaks = list(argrelextrema(data, np.greater)[0])    min_peaks = list(argrelextrema(data, np.less)[0])
    if len(max_peaks)>3 and len(min_peaks)>3:        return True    else:        return False

# 判断IMFsdef isIMFs(data):    max_peaks = list(argrelextrema(data, np.greater)[0])    min_peaks = list(argrelextrema(data, np.less)[0])
    if min(data[max_peaks]) < 0 or max(data[min_peaks])>0:        return False    else:        return True

def getIMFs(data):    while(not isIMFs(data)):        data = sifting(data)    return data

# EMD函数def EMD(data):    IMFs = []    while hasPeaks(data):        data_imf = getIMFs(data)        data = data-data_imf        IMFs.append(data_imf)    return IMFs

# 绘制对比图data = np.random.random(1000)-0.5IMFs = EMD(data)n = len(IMFs)+1
# 原始信号plt.figure(figsize = (18,15))plt.subplot(n, 1, 1)plt.plot(data, label='Origin')plt.title("Origin ")
# 若干条IMFs曲线for i in range(0,len(IMFs)):    plt.subplot(n, 1, i+2)    plt.plot(IMFs[i])    plt.ylabel('Amplitude')    plt.title("IMFs "+str(i+1))
plt.legend()plt.xlabel('time (s)')plt.ylabel('Amplitude')

# 导入工具库import numpy as npfrom PyEMD import EMD, Visualisation

# 构建信号t = np.arange(0,1, 0.01)S = 2*np.sin(2*np.pi*15*t) +4*np.sin(2*np.pi*10*t)*np.sin(2*np.pi*t*0.1)+np.sin(2*np.pi*5*t)

# 提取imfs和剩余emd = EMD()emd.emd(S)imfs, res = emd.get_imfs_and_residue()# 绘制 IMFvis = Visualisation()vis.plot_imfs(imfs=imfs, residue=res, t=t, include_residue=True)# 绘制并显示所有提供的IMF的瞬时频率vis.plot_instant_freq(t, imfs=imfs)vis.show()