pdaaphm/algorithm/06去噪/封装/去噪.py

#开发时间：2024/8/28 10:56
#开发时间：2024/6/12 9:36
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib import rcParams
from flask import Flask, request, jsonify
from threading import Lock

# 移动平均法
def moving_average(signal, window_size):
    return np.convolve(signal, np.ones(window_size) / window_size, mode='same')

app = Flask(__name__)
lock = Lock()

@app.route('/moving_average', methods=['POST'])
def set_file_address1():
    try:
        # 检查请求体中是否包含文件地址
        data = request.json
        if 'file_url' not in data:
            return jsonify({'code': 400, 'msg': 'No file URL provided'})

        file_url = request.json.get('file_url')
        result_file_path = request.json.get('result_file_path')
        result_img_path = request.json.get('result_img_path')

        # 读取数据集
        noisy_signal = pd.read_csv('{0}'.format(file_url))

        plt.rcParams['xtick.direction'] = 'in'
        plt.rcParams['ytick.direction'] = 'in'
        plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
        plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

        config = {
            "font.family": 'serif',
            "font.size": 20,
            "mathtext.fontset": 'stix',
            "font.serif": ['Times New Roman'],  # 宋体
            'axes.unicode_minus': False  # 处理负号
        }

        rcParams.update(config)

        # 检查是否包含时间列
        if 'Time' in noisy_signal.columns:
            time_column = noisy_signal['Time'].values.reshape(-1, 1)
            other_columns = noisy_signal.drop(columns='Time')
        else:
            time_column = None
            other_columns = noisy_signal

        # 应用去噪方法
        window_size = 20  # 窗口宽度
        denoised_signals = pd.DataFrame()

        for column in other_columns.columns:
            denoised_signals[column] = moving_average(other_columns[column].values, window_size)

        # 如果有时间列，将其添加回去
        if time_column is not None:
            denoised_signals.insert(0, 'time', time_column)

        # 保存去噪后的信号到CSV文件
        denoised_signals.to_csv(result_file_path, index=False)

        # 绘图
        plt.figure(figsize=(12, 10))

        # 绘制原始信号和去噪后信号（这里只绘制第一列作为示例）
        plt.subplot(2, 1, 1)
        plt.plot(noisy_signal.iloc[:, 1], label='Noisy Signal')
        plt.legend()

        plt.subplot(2, 1, 2)
        plt.plot(denoised_signals.iloc[:, 1], label='Moving Average')
        plt.legend()
        plt.tight_layout()
        plt.savefig(result_img_path)
        plt.close()  # 关闭图形以释放资源
        return jsonify({
            'code': 200,
            'msg': 'moving_average_denoise completed'

        })
    except Exception as e:
        return jsonify({
            'code': 500,
            'msg': str(e)
        })

# 傅里叶变换去噪
from scipy.fft import fft, ifft

def fft_denoise(signal, cutoff_freq, fs):
    signal_fft = fft(signal)
    frequencies = np.fft.fftfreq(len(signal), 1 / fs)
    signal_fft[np.abs(frequencies) > cutoff_freq] = 0
    return ifft(signal_fft).real

@app.route('/fft_denoise', methods=['POST'])
def set_file_address2():
    try:
        # 检查请求体中是否包含文件地址
        data = request.json
        if 'file_url' not in data:
            return jsonify({'code': 400, 'msg': 'No file URL provided'})

        file_url = request.json.get('file_url')
        result_file_path = request.json.get('result_file_path')
        result_img_path = request.json.get('result_img_path')

        # 读取数据集
        noisy_signal = pd.read_csv('{0}'.format(file_url))

        plt.rcParams['xtick.direction'] = 'in'
        plt.rcParams['ytick.direction'] = 'in'
        plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
        plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

        config = {
            "font.family": 'serif',
            "font.size": 20,
            "mathtext.fontset": 'stix',
            "font.serif": ['Times New Roman'],  # 宋体
            'axes.unicode_minus': False  # 处理负号
        }

        rcParams.update(config)

        # 读取信号
        fs = 1000
        # 检查是否包含时间列
        if 'Time' in noisy_signal.columns:
            time_column = noisy_signal['Time'].values.reshape(-1, 1)
            other_columns = noisy_signal.drop(columns='Time')
        else:
            time_column = None
            other_columns = noisy_signal

        # 应用去噪方法
        cutoff_freq = 10  # 截止频率
        denoised_signals = pd.DataFrame()

        for column in other_columns.columns:
            denoised_signals[column] = fft_denoise(other_columns[column].values, cutoff_freq, fs)

        # 如果有时间列，将其添加回去
        if time_column is not None:
            denoised_signals.insert(0, 'Time', time_column)

        # 保存去噪后的信号到CSV文件
        denoised_signals.to_csv(result_file_path, index=False)

        # 绘图
        plt.figure(figsize=(12, 10))

        # 绘制原始信号和去噪后信号（这里只绘制第一列作为示例）
        plt.subplot(2, 1, 1)
        plt.plot(noisy_signal.iloc[:, 1], label='Noisy Signal')
        plt.legend()

        plt.subplot(2, 1, 2)
        plt.plot(denoised_signals.iloc[:, 1], label='FFT Denoised')
        plt.legend()

        plt.tight_layout()
        plt.savefig(result_img_path)
        plt.close()  # 关闭图形以释放资源
        return jsonify({
            'code': 200,
            'msg': 'fft_denoise completed'

        })
    except Exception as e:
        return jsonify({
            'code': 500,
            'msg': str(e)
        })


# 卡尔曼滤波去噪
import numpy as np
import pandas as pd
from filterpy.kalman import KalmanFilter
import matplotlib.pyplot as plt
from matplotlib import rcParams
from flask import Flask, request, jsonify

def kalman_denoise(signal):
    kf = KalmanFilter(dim_x=2, dim_z=1)
    kf.x = np.array([0., 0.])
    kf.F = np.array([[1., 1.],
                     [0., 1.]])
    kf.H = np.array([[1., 0.]])
    kf.P *= 1000.
    kf.R = 5
    kf.Q = np.array([[1e-5, 0.],
                     [0., 1e-5]])

    filtered_signal = []
    for z in signal:
        kf.predict()
        kf.update(z)
        filtered_signal.append(kf.x[0])
    return np.array(filtered_signal)

@app.route('/kalman_denoise', methods=['POST'])
def set_file_address3():
    try:
        # 检查请求体中是否包含文件地址
        data = request.json
        if 'file_url' not in data:
            return jsonify({'code': 400, 'msg': 'No file URL provided'})

        file_url = request.json.get('file_url')
        result_file_path = request.json.get('result_file_path')
        result_img_path = request.json.get('result_img_path')

        # 读取数据集
        noisy_signal = pd.read_csv('{0}'.format(file_url))

        plt.rcParams['xtick.direction'] = 'in'
        plt.rcParams['ytick.direction'] = 'in'
        plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
        plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号

        config = {
            "font.family": 'serif',
            "font.size": 20,
            "mathtext.fontset": 'stix',
            "font.serif": ['Times New Roman'],  # 宋体
            'axes.unicode_minus': False  # 处理负号
        }

        rcParams.update(config)

        # 检查是否包含时间列
        if 'Time' in noisy_signal.columns:
            time_column = noisy_signal['Time'].values.reshape(-1, 1)
            other_columns = noisy_signal.drop(columns='Time')
        else:
            time_column = None
            other_columns = noisy_signal

        # 应用去噪方法
        denoised_signals = pd.DataFrame()

        for column in other_columns.columns:
            denoised_signals[column] = kalman_denoise(other_columns[column].values)

        # 如果有时间列，将其添加回去
        if time_column is not None:
            denoised_signals.insert(0, 'time', time_column)

        # 保存去噪后的信号到CSV文件
        denoised_signals.to_csv(result_file_path, index=False)

        # 绘图
        plt.figure(figsize=(12, 10))

        # 绘制原始信号和去噪后信号（这里只绘制第一列作为示例）
        plt.subplot(2, 1, 1)
        plt.plot(noisy_signal.iloc[:, 1], label='Noisy Signal')
        plt.legend()

        plt.subplot(2, 1, 2)
        plt.plot(denoised_signals.iloc[:, 1], label='Kalman Filtered')
        plt.legend()

        plt.tight_layout()
        plt.savefig(result_img_path)
        plt.close()  # 关闭图形以释放资源
        return jsonify({
            'code': 200,
            'msg': 'kalman_denoise completed'

        })
    except Exception as e:
        return jsonify({
            'code': 500,
            'msg': str(e)
        })


if __name__ == '__main__':
    app.run(port=10004, debug=True, use_reloader=False)