Hang_KG/api/api.py

from flask import Flask, jsonify, request
import requests
from io import StringIO
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split
import json


app = Flask(__name__)


def detect_outlier(data):
    # 计算平均值和标准差
    mean = np.mean(data)
    std = np.std(data)
    # 设定上下界
    lower = mean - 1 * std
    upper = mean + 1 * std
    # 创建一个空列表，用于存储新列的数据
    result = []
    # 遍历原列的数据，判断是否为异常值
    for x in data:
        # 如果小于下界或大于上界，认为是异常值，标记为1
        if x < lower or x > upper:
            result.append(1)
        # 否则认为是正常值，标记为0
        else:
            result.append(0)
    # 返回新列的数据
    return result


# 异常值剔除
@app.route('/abnormal', methods=['POST'])
def process_data():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 200, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')

        # 对每一列的数据调用函数，生成新列
        for col in dataframe.columns:
            if col != '时间':    # 跳过 时间列
                dataframe[col + '_是否异常'] = detect_outlier(dataframe[col])  # 新列的名字为原列名加_是否异常

        # 返回处理后的数据
        result = {'status': 200, 'data': dataframe.to_dict(orient='records'), 'msg': 'Successfully'}
        print(jsonify(result))
        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


# 缺失值补全
@app.route('/completion', methods=['POST'])
def process_data1():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 500, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')

        # 用数值型列的平均值插补缺失值
        dataframe.fillna(dataframe.select_dtypes(include=np.number).mean().round(3), inplace=True)

        # 返回处理后的数据
        result = {'status': 200, 'data': dataframe.to_dict(orient='records')}
        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


def alarm_suppression(alarms, threshold=0.6):
    """
    输入：
      alarms: 报警序列，由 0 和 1 组成
    输出：
      is_true_alarms: 是否是真的报警序列，由 True 和 False 组成
    """

    # 定义虚警门限
    # threshold = 0.6

    # 遍历报警序列，如果报警值超过门限，则认为是真的报警
    alarms = alarms.apply(pd.to_numeric, errors='coerce')

    # 遍历报警序列的每一列，如果列中的值超过门限，则认为是真的报警
    # is_true_alarms = alarms.apply(lambda col: col > threshold, axis=0)
    # 遍历每一列，创建新的列以标记是否是真实的报警
    for column_name in alarms:
        new_column_name = f"{column_name}_是否虚警"
        alarms[new_column_name] = alarms[column_name] > threshold

    return alarms


def list_tran_json(list):
    str_json = json.dumps(list, ensure_ascii=False, indent=2)
    return str_json


# 虚警抑制
@app.route('/false_alarm', methods=['POST'])
def process_data2():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        # 虚警门限
        threshold = data.get('threshold')
        faultCode = data.get('faultCode')
        threshold = 0.3

        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 500, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data,  index_col=False, encoding='UTF-8')
        columns_to_remove = ['时间', '故障名称']
        excluded_columns = [col for col in columns_to_remove if col in dataframe.columns]
        excluded_columns_data = dataframe[excluded_columns]
        dataframe = dataframe.drop(columns=excluded_columns, errors='ignore')

        # 用数值型列的平均值插补缺失值
        is_true_alarms = alarm_suppression(dataframe, float(threshold))

        # 返回处理后的数据
        # result_df = pd.DataFrame(is_true_alarms, columns=dataframe.columns)

        result_df = pd.concat([is_true_alarms, excluded_columns_data], axis=1)

        # 返回处理后的数据
        # result = {'status': 200, 'data': result_df.to_dict(orient='records')}
        data_json = [{'故障编码': faultCode, '是否虚警': '是'}]
        result = {'status': 200, 'data': data_json}

        return jsonify(result)
    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


# 故障诊断
@app.route('/fault_diagnosis', methods=['POST'])
def process_data3():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 500, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')

        # 计算均值
        xm = np.mean(dataframe.iloc[:, 1])
        xstd = np.std(dataframe.iloc[:, 1])  # 标准差
        kur = ((sum((dataframe.iloc[:, 1] - xm) ** 4)) / len(dataframe.iloc[:, 1])) / (xstd ** 4)  # 峭度
        # 结果图片上传
        # // 文件上传 http://localhost:9090/als/resource/oss/upload
        # 文件上传API ,返回结果 {
        #     "code": 200,
        #     "msg": "操作成功",
        #     "data": {
        #         "ossId": "236309167646711808"
        #     }
        # }
        # python 调用 API 得到ossId 返回
        data1 = {"result" : "字符串结果", "ossId": 236309167646711808}
        result = {'status': 200, 'data': data1}
        # 最终返回结果
        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


# 深度隔离
@app.route('/fault_isolation', methods=['POST'])
def process_data4():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 500, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')

        # 计算均值
        xm = np.mean(dataframe.iloc[:, 1])

        # 判断输出是否故障
        if xm > 0.1:
            result = {'status': 200, 'data': '发动机振动故障'}
        else:
            result = {'status': 200, 'data': '发动机振动正常'}

        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


# 故障预测
@app.route('/fault_prediction', methods=['POST'])
def process_data5():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 500, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')
        X_train, X_test, y_train, y_test = train_test_split(dataframe.iloc[:, :-1], dataframe.iloc[:, -1],
                                                            test_size=0.05, random_state=42)

        # 训练线性回归模型
        model = LinearRegression()
        model.fit(X_train, y_train)

        # 对测试集进行预测
        y_pred = model.predict(X_test)

        # 计算均方误差
        mse = mean_squared_error(y_test, y_pred)

        # 判断输出是否故障
        if np.mean(y_pred) > 2:
            result = {'status': 200, 'data': '发动机寿命后期'}
        else:
            result = {'status': 200, 'data': '发动机寿命前期'}

        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)


# 定义合理参数范围的字典，实际应用中需依据具体飞机型号及飞行手册设定
PARAMETER_RANGES = {
    "发动机排气温度": (500, 1200),  # 示例范围，单位：摄氏度
    "喘振值": (0.8, 1.2),  # 示例范围，具体范围需根据实际情况设定
    "时间": None,  # 时间一般不设定范围，但确保数据连续性
    "机匣加速": (-200, 200),  # 示例范围，单位：m/s^2
    "机匣振幅": (0, 0.1),  # 示例范围，单位：毫米
    "机匣频率": (50, 500),  # 示例范围，单位：Hz
    "燃油流量": (0, 10),  # 示例范围，单位：kg/s，具体根据发动机规格
    "风扇转子转速": (5000, 12000),  # 示例范围，单位：RPM
    "高压压气机转子转数": (8000, 15000),  # 示例范围，单位：RPM
}

def check_parameter_validity(value, range_info):
    """检查单个参数值是否在合理范围内"""
    if range_info is None or (value >= range_info[0] and value <= range_info[1]):
        return True
    return False


def calculate_testability(df_flight_params: pd.DataFrame) -> float:
    """
    根据飞参数据评估可测试性得分。
    参数:
    df_flight_params (pd.DataFrame): 包含飞参数据的DataFrame。

    返回:
    float: 可测试性得分，范围0-1，值越接近1表示数据越符合预期，可测试性越高。
    """
    total_rows = len(df_flight_params)
    valid_count = 0

    for param in PARAMETER_RANGES.keys():
        if param in df_flight_params.columns:
            # 检查该参数的每个值是否都在合理范围内
            valid_for_param = df_flight_params[param].apply(check_parameter_validity, args=(PARAMETER_RANGES[param],))
            valid_count += valid_for_param.sum()

    # 计算整体有效性百分比作为可测试性得分
    testability_score = valid_count / (total_rows * len(PARAMETER_RANGES)) if total_rows > 0 else 0

    return testability_score

# 测试性
@app.route('/testability', methods=['POST'])
def process_data6():
    try:
        # 从请求中获取数据
        data = request.get_json()
        # 从在线数据链接下载 CSV 数据
        data_url = data.get('url')
        response = requests.get(data_url)
        # 检查下载是否成功
        if response.status_code != 200:
            error_message = {'status': 200, 'message': 'Failed to download CSV data'}
            return jsonify(error_message)

        # 将 CSV 数据转换为 Pandas DataFrame
        response.encoding = 'UTF-8'
        csv_data = StringIO(response.text)
        dataframe = pd.read_csv(csv_data, index_col=False, encoding='UTF-8')
        df_flight_params = pd.DataFrame(dataframe)

        # 计算可测试性得分
        testability = calculate_testability(df_flight_params)
        res = f"根据飞参数据评估，可测试性得分为: {testability:.2f}"
        result = {'status': 200, 'data': res}

        return jsonify(result)

    except Exception as e:
        error_message = {'status': 500, 'message': str(e)}
        return jsonify(error_message)

app.debug = True
if __name__ == '__main__':
    # 启动 Flask 应用
    app.run(debug=True, port=8888, host='0.0.0.0')