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')