import pandas as pd import numpy as np from pymongo import MongoClient import requests import json from datetime import datetime from flask import Flask, request import time import logging import traceback app = Flask('evaluation_accuracy——service') url = 'http://49.4.78.194:17160/apiCalculate/calculate' ''' 准确率接口使用手顺： ①入口方法为 calculate_acc ② 按照参数传传值 data含有C_TIME时间、realValue实际功率、ableValue可用功率（没有数值用实际功率替代）、forecastAbleValue预测功率 opt为包含场站必要信息的字典，字段为：cap装机容量 province省份 formulaType公式类型 electricType电站类型 stationCode场站编码 具体介绍参考接口文档 ③公式计算分为按天和按点两种，指定好opt.formulaType，即可设置公式类型，再在求取的每天或每个点的结果上进行平均，结果返回 ''' def wrap_json(df, opt): """ 包装json :param df: 列名为 C_TIME realValue ableValue forecastAbleValue的DataFrame :param opt: 参数字典 :return: json列表 """ d = opt['formulaType'].split('_')[0] jata, dfs = [], [] if d == 'POINT': df['time'] = df['C_TIME'].apply(datetime_to_timestamp) for i, row in df.iterrows(): dfs.append(row.to_frame().T) elif d == 'DAY': df = df.copy() # df['time'] = df['C_TIME'].apply(datetime_to_timestamp) int(round(time.mktime(df['C_TIME'].timetuple()))*1000) # df.loc[:, 'time'] = df['C_TIME'].apply(datetime_to_timestamp) df['time'] = df['C_TIME'].apply(lambda x: datetime_to_timestamp(x)) # df['time'] = df.apply(lambda row: datetime_to_timestamp(row['C_TIME']), axis=1) # df['C_TIME'] = df['C_TIME'].dt.strftime('%y%m%d') # 转换成年月日 df.loc[:, 'C_TIME'] = df['C_TIME'].dt.strftime('%y%m%d') for i, group in df.groupby('C_TIME'): dfs.append(group) outter_dict = {"electricCapacity": str(opt['cap']), "province": opt['province'], "formulaType": opt['formulaType'], "electricType":opt['electricType'], "stationCode": opt['stationCode']} timestamp = int(time.mktime(datetime.now().timetuple()) * 1000 + datetime.now().microsecond / 1000.0) inner_dict = {"genTime": str(timestamp)+"L", "capacity": str(opt['cap']), "openCapacity": str(opt['cap'])} for df in dfs: calculationInfoList = df.iloc[:, 1:].to_json(orient='records') outter_dict['calculationInfoList'] = [dict(calculation, **inner_dict) for calculation in eval(calculationInfoList)] jata.append(json.dumps(outter_dict)) return jata def send_reqest(url, jata): """ 发送请求 :param url: 请求地址 :param jata: Json数据 :return: 准确率 """ headers = { 'content-type': 'application/json;charset=UTF-8', "Authorization": "dXNlcjoxMjM0NTY=" } acc, number = 0, 0 for i in range(len(jata)): res = requests.post(url, headers=headers, data=jata[i]) if res.json()['code'] == '500': print("没通过考核标准", end=' ') continue number += 1 acc += float(res.json()['data'][:-1]) if number != 0: acc /= number else: print("无法迭代计算准确率平均值，分母为0") return acc def calculate_acc(data, opt): """ 准确率调用接口计算 :param data: 列名为 C_TIME realValue ableValue forecastAbleValue的DataFrame :param opt: 参数字段 :return: 计算结果 """ jata = wrap_json(data, opt) acc = send_reqest(url=url, jata=jata) return acc def datetime_to_timestamp(dt): return int(round(time.mktime(dt.timetuple()))*1000) def get_data_from_mongo(args): mongodb_connection,mongodb_database,mongodb_read_table = "mongodb://root:sdhjfREWFWEF23e@192.168.1.43:30000/",args['mongodb_database'],args['mongodb_read_table'] client = MongoClient(mongodb_connection) # 选择数据库（如果数据库不存在，MongoDB 会自动创建） db = client[mongodb_database] collection = db[mongodb_read_table] # 集合名称 data_from_db = collection.find() # 这会返回一个游标（cursor） # 将游标转换为列表，并创建 pandas DataFrame df = pd.DataFrame(list(data_from_db)) client.close() return df def insert_data_into_mongo(res_df,args): mongodb_connection,mongodb_database,mongodb_write_table = "mongodb://root:sdhjfREWFWEF23e@192.168.1.43:30000/",args['mongodb_database'],args['mongodb_write_table'] client = MongoClient(mongodb_connection) db = client[mongodb_database] if mongodb_write_table in db.list_collection_names(): db[mongodb_write_table].drop() print(f"Collection '{mongodb_write_table} already exist, deleted successfully!") collection = db[mongodb_write_table] # 集合名称 # 将 DataFrame 转为字典格式 data_dict = res_df.to_dict("records") # 每一行作为一个字典 # 插入到 MongoDB collection.insert_many(data_dict) print("data inserted successfully!") # def compute_accuracy(df,args): # col_time,col_rp,col_pp,formulaType = args['col_time'],args['col_rp'],args['col_pp'],args['formulaType'].split('_')[0] # dates = [] # accuracy = [] # df = df[(~np.isnan(df[col_rp]))&(~np.isnan(df[col_pp]))] # df = df[[col_time,col_rp,col_pp]].rename(columns={col_time:'C_TIME',col_rp:'realValue',col_pp:'forecastAbleValue'}) # df['ableValue'] = df['realValue'] # df['C_TIME'] = df['C_TIME'].apply(lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S")) # if formulaType=='DAY': # df['C_DATE'] = df['C_TIME'].apply(lambda x: x.strftime("%Y-%m-%d")) # days_list = df['C_DATE'].unique().tolist() # for day in days_list: # df_tmp = df[df['C_DATE'] == day] # dates.append(day) # accuracy.append(calculate_acc(df_tmp, args)) # else: # points = df['C_TIME'].unique().tolist() # for point in points: # df_tmp = df[df['C_TIME'] == point] # dates.append(point) # accuracy.append(calculate_acc(df_tmp, args)) # print("accuray compute successfully!") # return pd.DataFrame({'date':dates,'accuracy':accuracy}) # 定义 RMSE 和 MAE 计算函数 def rmse(y_true, y_pred): return np.sqrt(np.mean((y_true - y_pred) ** 2)) def mae(y_true, y_pred): return np.mean(np.abs(y_true - y_pred)) def compute_accuracy(df,args): col_time,col_rp,col_pp = args['col_time'],args['col_rp'],args['col_pp'] df[col_time] = df[col_time].apply(lambda x:pd.to_datetime(x).strftime("%Y-%m-%d")) # 按日期分组并计算 RMSE 和 MAE results = df.groupby(col_time).apply( lambda group: pd.Series({ "RMSE": rmse(group[col_rp], group[col_pp]), "MAE": mae(group[col_rp], group[col_pp]) }) ).reset_index() return results @app.route('/evaluation_accuracy', methods=['POST']) def evaluation_accuracy(): # 获取程序开始时间 start_time = time.time() result = {} success = 0 print("Program starts execution!") try: args = request.values.to_dict() print('args',args) logger.info(args) power_df = get_data_from_mongo(args) acc_result = compute_accuracy(power_df,args) insert_data_into_mongo(acc_result,args) success = 1 except Exception as e: my_exception = traceback.format_exc() my_exception.replace("\n","\t") result['msg'] = my_exception end_time = time.time() result['success'] = success result['args'] = args result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time)) result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time)) print("Program execution ends!") return result if __name__=="__main__": print("Program starts execution!") logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("evaluation_accuracy log") from waitress import serve serve(app, host="0.0.0.0", port=10091) print("server start!")