algorithm_platform/post_processing/cdq_coe_gen.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# @FileName  :cdq_coe_gen.py
# @Time      :2025/3/28 16:20
# @Author    :David
# @Company: shenyang JY
import os, requests, json, time, traceback
import pandas as pd
import numpy as np
from common.database_dml_koi import get_data_from_mongo
from pymongo import MongoClient
from flask import Flask,request,jsonify, g
from datetime import datetime
# from common.logs import Log
# logger = Log('post-processing').logger
from logging import getLogger
logger = getLogger('xx')
current_path = os.path.dirname(__file__)
API_URL = "http://ds2:18080/accuracyAndBiasByJSON"
app = Flask('cdq_coe_gen——service')


@app.before_request
def update_config():
    # ------------ 整理参数，整合请求参数 ------------
    g.coe = {}


def iterate_coe(pre_data, point, col_power, col_pre, coe):
    """
    更新16个点系数
    """
    T = 'T' + str(point + 1)
    best_acc, best_score1, best_coe_m, best_coe_n = 0, 0, 0, 0
    best_score, best_acc1, best_score_m, best_score_n = 999, 0, 0, 0
    req_his_fix = prepare_request_body(pre_data, col_power, 'his_fix')
    req_dq = prepare_request_body(pre_data, col_power, col_pre)

    his_fix_acc, his_fix_score = calculate_acc(API_URL, req_his_fix)
    dq_acc, dq_score = calculate_acc(API_URL, req_dq)
    for i in range(5, 210):
        for j in range(5, 210):
            pre_data["new"] = round(i / 170 * pre_data[col_pre] + j / 170 * pre_data['his_fix'], 3)
            req_new = prepare_request_body(pre_data, col_power, 'new')
            acc, acc_score = calculate_acc(API_URL, req_new)

            if acc > best_acc:
                best_acc = acc
                best_score1 = acc_score
                best_coe_m = i / 170
                best_coe_n = j / 170
            if acc_score < best_score:
                best_score = acc_score
                best_acc1 = acc
                best_score_m = i / 170
                best_score_n = j / 170

    pre_data["coe-acc"] = round(best_coe_m * pre_data[col_pre] + best_coe_n * pre_data['his_fix'], 3)
    pre_data["coe-ass"] = round(best_score_m * pre_data[col_pre] + best_score_n * pre_data['his_fix'], 3)
    logger.info(
        "1.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(
            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc, his_fix_acc))
    logger.info(
        "2.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(
            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score1, his_fix_score))
    logger.info(
        "3.过去{} - {}的短期的准确率：{:.4f}，自动确认系数后，{} 超短期的准确率：{:.4f}，历史功率：{:.4f}".format(
            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_acc, T, best_acc1, his_fix_acc))
    logger.info(
        "4.过去{} - {}的短期的考核分：{:.4f}，自动确认系数后，{} 超短期的考核分：{:.4f}，历史功率：{:.4f}".format(
            pre_data['C_TIME'][0], pre_data['C_TIME'].iloc[-1], dq_score, T, best_score, his_fix_score))

    coe[T]['score_m'] = round(best_score_m, 3)
    coe[T]['score_n'] = round(best_score_n, 3)
    coe[T]['acc_m'] = round(best_coe_m, 3)
    coe[T]['acc_n'] = round(best_coe_n, 3)
    logger.info("系数轮询后，最终调整的系数为：{}".format(coe))

def prepare_request_body(df, col_power, col_pre):
    """
    准备请求体，动态保留MongoDB中的所有字段
    """
    data = df.copy()
    # 转换时间格式为字符串
    if 'dateTime' in data.columns and isinstance(data['dateTime'].iloc[0], datetime):
        data['dateTime'] = data['dateTime'].dt.strftime('%Y-%m-%d %H:%M:%S')
    data['model'] = col_pre
    # 排除不需要的字段（如果有）
    exclude_fields = ['_id']  # 通常排除MongoDB的默认_id字段

    # 获取所有字段名（排除不需要的字段）
    available_fields = [col for col in data.columns if col not in exclude_fields]

    # 转换为记录列表（保留所有字段）
    data = data[available_fields].to_dict('records')

    # 构造请求体（固定部分+动态数据部分）
    request_body = {
        "stationCode": "J00600",
        "realPowerColumn": col_power,
        "ablePowerColumn": col_power,
        "predictPowerColumn": col_pre,
        "inStalledCapacityName": 153,
        "computTypeEnum": "E2",
        "computMeasEnum": "E2",
        "openCapacityName": 153,
        "onGridEnergy": 0,
        "price": 0,
        "fault": -99,
        "colTime": "dateTime",  #时间列名（可选，要与上面'dateTime一致'）
        # "computPowersEnum": "E4"  # 计算功率类型（可选）
        "data": data  # MongoDB数据
    }

    return request_body

def calculate_acc(api_url, request_body):
    """
    调用API接口
    """
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }
    try:
        response = requests.post(
            api_url,
            data=json.dumps(request_body, ensure_ascii=False),
            headers=headers
        )
        result = response.json()
        if response.status_code == 200:
            acc = np.average([res['accuracy'] for res in result])
            # ass = np.average([res['accuracyAssessment'] for res in result])
            print("111111111")
            return acc, 0
        else:
            logger.info(f"失败：{result['status']},{result['error']}")
            print(f"失败：{result['status']},{result['error']}")
            print("22222222")
    except requests.exceptions.RequestException as e:
        print(f"API调用失败: {e}")
        print("333333333")
        return None

def history_error(data, col_power, col_pre):
    data['error'] =  data[col_power] - data[col_pre]
    data['error'] = data['error'].round(2)
    data.reset_index(drop=True, inplace=True)
    # 用前面5个点的平均error，和象心力相加
    numbers = len(data) - 5
    datas = [data.iloc[x: x+5, :].reset_index(drop=True) for x in range(0, numbers)]
    data_error = [np.mean(d.iloc[0:5, -1]) for d in datas]
    pad_data_error = np.pad(data_error, (5, 0), mode='constant', constant_values=0)
    data['his_fix'] = data[col_pre] + pad_data_error
    data = data.iloc[5:, :].reset_index(drop=True)
    data.loc[data[col_pre] <= 0, ['his_fix']] = 0
    data['dateTime'] = pd.to_datetime(data['dateTime'])
    data = data.loc[:, ['dateTime', col_power, col_pre, 'his_fix']]
    # data.to_csv('J01080原始数据.csv', index=False)
    return data

@app.route('/cdq_coe_gen', methods=['POST'])
def get_station_cdq_coe():
    # 获取程序开始时间
    start_time = time.time()
    result = {}
    success = 0
    args = {}
    coe = g.coe
    try:
        args = request.values.to_dict()
        logger.info(args)
        data = get_data_from_mongo(args).sort_values(by='dateTime', ascending=True)
        pre_data = history_error(data, col_power='realPower', col_pre='dq')
        for point in range(0, 16, 1):
            iterate_coe(pre_data, point, 'realPower', 'dq', coe)
        success = 1
    except Exception as e:
        my_exception = traceback.format_exc()
        my_exception.replace("\n", "\t")
        result['msg'] = my_exception
        logger.info("调系数出错：{}".format(my_exception))
    end_time = time.time()
    result['success'] = success
    result['args'] = args
    result['coe'] = coe
    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))
    return result

if __name__ == "__main__":
    # args = {
    #     'mongodb_database': 'ldw_ftp',
    #     'mongodb_read_table': 'j00600',
    #     # 'timeBegin': '2025-01-01 00:00:00',
    #     # 'timeEnd': '2025-01-03 23:45:00'
    # }
    # data = get_data_from_mongo(args).sort_values(by='dateTime', ascending=True)
    # pre_data = history_error(data, col_power='realPower', col_pre='dq')
    # for point in range(0, 16, 1):
    #     iterate_coe(pre_data, point, 'realPower', 'dq')
    # run_code = 0
    print("Program starts execution!")
    from waitress import serve

    serve(
        app,
        host="0.0.0.0",
        port=10123,
        threads=8,  # 指定线程数（默认4，根据硬件调整）
        channel_timeout=600  # 连接超时时间（秒）
    )