pdaaphm/algorithm/03特征提取/封装/特征提取.py

# 1.基本统计特征
from flask import Flask, request, jsonify
import pandas as pd
from scipy.stats import skew, kurtosis
import numpy as np
import os
import requests

app = Flask(__name__)

def ae_feature(raw, fs):
    mean = np.mean(raw)
    std_dev = np.std(raw)
    max_value = np.max(raw)
    min_value = np.min(raw)
    variance = np.var(raw)
    skewness = skew(raw)
    kurt = kurtosis(raw)
    range=np.max(raw) - np.min(raw)
    rms = np.sum(raw ** 2) / len(raw)
    ae_feature = pd.DataFrame()
    ae_feature['mean'] = [mean]
    ae_feature['std_dev'] = [std_dev]
    ae_feature['max_value'] = [max_value]
    ae_feature['min_value'] = [min_value]
    ae_feature['variance'] = [variance]
    ae_feature['skewness'] = [skewness]
    ae_feature['kurt'] = [kurt]
    ae_feature['range'] = [range]
    ae_feature['rms'] = [rms]
    return ae_feature

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

        file_url = data['file_url']

        # 使用requests获取文件内容
        response = requests.get(file_url)
        if response.status_code != 200:
            return jsonify({'code': 500, 'msg': 'Failed to retrieve file from URL'})

        # 读取数据集
        df = pd.read_csv(pd.compat.StringIO(response.text), header=None)

        fs = 300  # 设置提取特征值每组的数据量
        all_features_df = pd.DataFrame()
        for i in range(int(df.shape[1])):
            for j in range(0, len(df), fs):
                end_index = min(j + fs, len(df))
                data_test = np.array(df.iloc[j:end_index, i])
                sa = ae_feature(data_test, fs)
                all_features_df = pd.concat([all_features_df, sa], ignore_index=True)

        # 将结果保存为CSV文件
        result_file_path = os.path.join('/tmp', '特征值文件-统计.csv.csv')
        all_features_df.to_csv(result_file_path, index=False)
        # 构建返回数据，只返回文件路径
        return jsonify({
            'code': 200,
            'msg': '文件处理完成',
            'file_path': result_file_path
        })

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

if __name__ == '__main__':
    app.run(debug=True, port=8081, host='0.0.0.0')



## 2.时间域特征

from flask import Flask, request, jsonify
import pandas as pd
import numpy as np
from scipy import fftpack
import requests

app = Flask(__name__)

def time_features(raw, fs):
    n = len(raw)
    fft_raw = abs(fftpack.fft(raw, n) * 2 / n)
    main_freq_index = np.where(fft_raw[0:math.floor(n / 2)] == max(fft_raw[0:math.floor(n / 2)]))
    f = np.linspace(0, fs, n) * fs / n
    peak_freq = f[main_freq_index[0][0]]  # 峰值频率，单位Hz

    xm = np.mean(raw)  # 均值
    xstd = np.std(raw)  # 标准差
    kur = ((np.sum((raw - xm) ** 4)) / len(raw)) / (xstd ** 4)  # 峭度
    cal_ave_amp = np.sum(np.abs(raw)) / n
    power_2 = np.power(raw, 2)
    sum_power = np.sum(power_2)
    cal_rms = np.sqrt(sum_power / n)
    cal_form = cal_rms / cal_ave_amp  # 波峰因子，均方根/平均幅值
    cal_peak = np.max(np.abs(raw))  # 峰值
    cal_crest = cal_peak / cal_rms  # 峰值因子

    feature_df = pd.DataFrame()
    feature_df['peak_freq_khz'] = [peak_freq / 1000]  # 转换为kHz
    feature_df['均值v'] = [xm]
    feature_df['标准差'] = [xstd]
    feature_df['峭度'] = [kur]
    feature_df['平均幅值'] = [cal_ave_amp]
    feature_df['均方根'] = [cal_rms]
    feature_df['波峰因子'] = [cal_form]
    feature_df['峰值'] = [cal_peak]
    feature_df['峰值因子'] = [cal_crest]

    return feature_df

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

        file_url = data['file_url']

        # 使用requests获取文件内容
        response = requests.get(file_url)
        if response.status_code != 200:
            return jsonify({'code': 500, 'msg': 'Failed to retrieve file from URL'})
        # 读取数据集
        df = pd.read_csv(pd.compat.StringIO(response.text), header=None)

        fs = 300  # 每组数据量
        all_features_df = pd.DataFrame()
        for i in range(int(df.shape[1])):
            for j in range(0, len(df.iloc[:, i]), fs):
                end_index = min(j + fs, len(df.iloc[:, i]))
                data_test = np.array(df.iloc[j:end_index, i])
                sa = time_features(data_test, fs)
                all_features_df = pd.concat([all_features_df, sa], ignore_index=True)

        result_file_path = '/tmp/特征值文件-时域.csv'  # 使用相对路径
        all_features_df.to_csv(result_file_path, index=False)
        # 构建返回数据，只返回文件路径
        return jsonify({
            'code': 200,
            'msg': '文件处理完成',
            'file_path': result_file_path
        })

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

if __name__ == '__main__':
    app.run(debug=True, port=8081, host='0.0.0.0')


## 3.频域特征

from flask import Flask, request, jsonify
import pandas as pd
from scipy.fft import fft, fftfreq
from scipy.stats import skew, kurtosis
import numpy as np
import requests

app = Flask(__name__)

def fe_feature(signal, fs):
    n = len(signal)
    freqs = fftfreq(n, d=1/fs)  # 生成频率序列
    signal_fft = fft(signal)  # 计算FFT
    signal_fft_shifted = np.fft.fftshift(signal_fft)  # 将FFT结果中心化

    # 计算频域特征
    total_energy = np.sum(np.abs(signal_fft)**2)
    total_power = total_energy / n
    rms = np.sqrt(total_power)
    freq_centroid = np.sum(np.abs(freqs) * np.abs(signal_fft_shifted)**2) / total_energy
    freq_bandwidth = np.sum(np.abs(freqs[1:]**2 * signal_fft_shifted[1:]**2) - np.abs(freqs[:-1]**2 * signal_fft_shifted[:-1]**2)) / (2 * np.sum(np.abs(signal_fft_shifted)**2))
    peak_freq = (np.argmax(np.abs(signal_fft_shifted)) + 1) * fs / n  # 峰值频率
    freq_skewness = skew(freqs[1:n//2] * (np.abs(signal_fft[1:n//2]) / np.max(np.abs(signal_fft[1:n//2]))))
    freq_kurtosis = kurtosis(freqs[1:n//2] * (np.abs(signal_fft[1:n//2]) / np.max(np.abs(signal_fft[1:n//2]))))

    feature_df = pd.DataFrame({
        '总能量': [total_energy],
        '总功率': [total_power],
        '均方根': [rms],
        '频率中心': [freq_centroid],
        '频率带宽': [freq_bandwidth],
        '峰值频率': [peak_freq],
        '频率偏度': [freq_skewness],
        '频率峰度': [freq_kurtosis]
    })

    return feature_df

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

        file_url = data['file_url']

        # 使用requests获取文件内容
        response = requests.get(file_url)
        if response.status_code != 200:
            return jsonify({'code': 500, 'msg': 'Failed to retrieve file from URL'})

        # 读取数据集
        df = pd.read_csv(pd.compat.StringIO(response.text), header=None)
        fs = 300  # 采样频率，根据实际情况调整
        all_features_df = pd.DataFrame()

        for i in range(int(df.shape[1])):
            for j in range(0, len(df.iloc[:, i]), fs):
                end_index = min(j + fs, len(df.iloc[:, i]))
                data_test = df.iloc[j:end_index, i].to_numpy()
                fa = fe_feature(data_test, fs)
                all_features_df = pd.concat([all_features_df, fa], ignore_index=True)

        result_file_path = '/tmp/特征值文件-频域.csv'  # 保存结果的路径
        all_features_df.to_csv(result_file_path, index=False, header=True)
        # 构建返回数据，只返回文件路径
        return jsonify({
            'code': 200,
            'msg': '文件处理完成',
            'file_path': result_file_path
        })

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

if __name__ == '__main__':
    app.run(debug=True, port=8081, host='0.0.0.0')

# #4.时频域特征

from flask import Flask, request, jsonify
import pandas as pd
from scipy.stats import skew, kurtosis
import numpy as np
from scipy.fft import fft, fftfreq
import requests

app = Flask(__name__)

def tf_features(raw, fs):
    # 计算快速傅里叶变换
    fft_values = fft(raw)
    fft_magnitude = np.abs(fft_values)

    # 计算频率向量
    n = len(raw)
    freq = fftfreq(n, 1 / fs)

    # 时域特征
    mean = np.mean(raw)
    std_dev = np.std(raw)
    max_value = np.max(raw)
    min_value = np.min(raw)
    variance = np.var(raw)
    skewness = skew(raw)
    kurt = kurtosis(raw)

    # 频域特征
    power_spectrum = np.square(fft_magnitude) / n
    mean_freq = np.mean(freq[power_spectrum > 0])
    std_freq = np.std(freq[power_spectrum > 0])
    freq_skewness = skew(freq[power_spectrum > 0])
    freq_kurt = kurtosis(freq[power_spectrum > 0])
    peak_freq_index = np.argmax(power_spectrum)
    peak_freq = freq[peak_freq_index]
    bandwidth = np.max(freq[power_spectrum > 0]) - np.min(freq[power_spectrum > 0])
    energy = np.sum(power_spectrum)

    feature_df = pd.DataFrame({
        '均值': [mean],
        '标准差': [std_dev],
        '最大值': [max_value],
        '最小值': [min_value],
        '方差': [variance],
        '偏度': [skewness],
        '峰度': [kurt],
        '频率均值': [mean_freq],
        '频率标准差': [std_freq],
        '频率偏度': [freq_skewness],
        '频率峰度': [freq_kurt],
        '峰值频率': [peak_freq],
        '带宽': [bandwidth],
        '能量': [energy]
    })

    return feature_df

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

        file_url = data['file_url']

        # 使用requests获取文件内容
        response = requests.get(file_url)
        if response.status_code != 200:
            return jsonify({'code': 500, 'msg': 'Failed to retrieve file from URL'})

        # 读取数据集
        df = pd.read_csv(pd.compat.StringIO(response.text), header=None)
        fs = 300  # 采样频率
        all_features_df = pd.DataFrame()

        for i in range(int(df.shape[1])):
            for j in range(0, len(df.iloc[:, i]), fs):
                end_index = min(j + fs, len(df.iloc[:, i]))
                data_test = df.iloc[j:end_index, i].to_numpy()
                fa = tf_features(data_test, fs)
                all_features_df = pd.concat([all_features_df, fa], ignore_index=True)

        result_file_path = '/tmp/特征值文件-时频域.csv'  # 保存结果的路径
        all_features_df.to_csv(result_file_path, index=False, header=True)
        # 构建返回数据，只返回文件路径
        return jsonify({
            'code': 200,
            'msg': '文件处理完成',
            'file_path': result_file_path
        })

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

if __name__ == '__main__':
    app.run(debug=True, port=8081, host='0.0.0.0')

##5.形态学特征提取
from flask import Flask, request, jsonify
import pandas as pd
from scipy.signal import find_peaks
import numpy as np
import requests

app = Flask(__name__)

def morphological_features(raw):
    nonzero_elements = np.count_nonzero(raw)
    mean_slope = np.mean(np.diff(raw) / np.diff(np.arange(len(raw))))

    peaks, _ = find_peaks(raw)
    num_peaks = len(peaks)
    troughs, _ = find_peaks(-raw)  # 寻找谷值
    num_troughs = len(troughs)

    # 假设峰值和谷值的宽度为相邻峰值或谷值之间的距离
    peak_widths = np.diff(np.sort(peaks)[1:] - np.sort(peaks)[:-1]) if num_peaks > 1 else np.array([])
    trough_widths = np.diff(np.sort(troughs)[1:] - np.sort(troughs)[:-1]) if num_troughs > 1 else np.array([])

    # 计算曲率和加速度变化需要一阶和二阶导数
    first_derivative = np.diff(raw) / np.diff(np.arange(len(raw)))
    second_derivative = np.diff(first_derivative) / np.diff(np.arange(len(raw) - 1))
    curvature = np.sum(second_derivative ** 2)  # 曲率的简单估计
    acceleration_change = np.sum(np.abs(np.diff(first_derivative)))  # 加速度变化的简单估计

    feature_df = pd.DataFrame({
        '非零元素数': [nonzero_elements],
        '平均斜率': [mean_slope],
        '峰值数量': [num_peaks],
        '谷值数量': [num_troughs],
        '平均峰宽': [np.mean(peak_widths)] if peak_widths.size else [0],
        '平均谷宽': [np.mean(trough_widths)] if trough_widths.size else [0],
        '曲率': [curvature],
        '加速度变化': [acceleration_change]
    })

    return feature_df

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

        file_url = data['file_url']

        # 使用requests获取文件内容
        response = requests.get(file_url)
        if response.status_code != 200:
            return jsonify({'code': 500, 'msg': 'Failed to retrieve file from URL'})

        # 读取数据集
        df = pd.read_csv(pd.compat.StringIO(response.text), header=None)
        all_features_df = pd.DataFrame()

        for i in range(int(df.shape[1])):
            # 假设数据是等间隔采样的，fs 不再作为函数参数
            data_test = df.iloc[:, i].to_numpy()
            fa = morphological_features(data_test)
            all_features_df = pd.concat([all_features_df, fa], ignore_index=True)

        result_file_path = '/tmp/特征值文件-形态学.csv'  # 保存结果的路径
        all_features_df.to_csv(result_file_path, index=False, header=True)
        # 构建返回数据，只返回文件路径
        return jsonify({
            'code': 200,
            'msg': '文件处理完成',
            'file_path': result_file_path
        })

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

if __name__ == '__main__':
    app.run(debug=True, port=8081, host='0.0.0.0')