AirTerminalDevice/apps/DHU/optimize.py

import os
from datetime import datetime,timedelta
from pathlib import Path
from pprint import pprint
from typing import Union

import pandas as pd

from ...model.DHU.DHU_AB import DHU_AB
from ...model.DHU.SDHU_AB import SDHU_AB
from .config_reader import ConfigReader
from ...tools.data_loader import DataLoader
from ..._utils.data_summary import print_dataframe,summary_dataframe


def optimize(*inputs,config=None):
    config = {} if config is None else config
    
    if '__LOCAL' in config.keys():
        config_reader_path = config['__LOCAL']
        data_URL           = config['__URL']
    else:
        config_reader_path = '/mnt/workflow_data'
        data_URL           = 'http://basedataportal-svc:8080/data/getpointsdata'
        
    config_reader = ConfigReader(path=f'{config_reader_path}/DHU配置.xlsx')
    
    ALL_RESULT = {
        'EXCEPTION':{
            'State'    : {},
            'Mod'      : {},
            'Data_ATD' : {},
            'Opt'      : {},
            'Push'     : {}
        },
        'STATUS':{
            'Mode_Steady': [],
            'Mode_Low'   : [],
            'Mode_Off'   : []
        }
    }
    ALL_OPT_RES = []
    
    for each_eaup_name,each_eaup_name_short in zip(
        config_reader.all_equp_names,
        config_reader.all_equp_names_short
    ):
        # 加载模型
        # 加载数据
        # 运行判断
        #   稳态判断：房间露点设定值与反馈值是否接近
        #   模型调整
        #   模型判断：模型精度是否满足要求
        #   模式判断：
        #       1. 基于当前露点优化模式：基于房间露点设定值减偏差
        #       2. 快速提升送风露点模式：约束送风露点保持不变
        
        if not config_reader.get_app_info(each_eaup_name,'实时优化','开始实时优化','bool'):
            continue
        each_equp_type = config_reader.get_equp_info(each_eaup_name,key='设备类型',info_type='str')
        
        global NOW
        NOW = config_reader.get_app_info(each_eaup_name,'实时优化','运行时间','datetime').replace(second=0,microsecond=0)
        print(f'{each_eaup_name}开始实时优化，设备类型为{each_equp_type}，运行时间为{NOW}')
        
        try:
            dhu_State = load_dhu_State(
                each_eaup_name     = each_eaup_name,
                config_reader      = config_reader,
                config_reader_path = config_reader_path,
                data_URL           = data_URL,
                NOW                = NOW
            )
            print(f'{each_eaup_name}设备状态为{dhu_State}')
            if dhu_State < 1:
                print('设备处于非运行状态，跳过')
                ALL_RESULT['STATUS']['Mode_Off'].append(each_eaup_name)
                continue
        except Exception as e:
            ALL_RESULT['EXCEPTION']['State'][each_eaup_name] = e
            raise e
            continue
        
        # 加载模型
        try:
            MODEL = load_model(
                each_eaup_name     = each_eaup_name,
                each_equp_type     = each_equp_type,
                config_reader      = config_reader,
                config_reader_path = config_reader_path
            )
        except Exception as e:
            ALL_RESULT['EXCEPTION']['Mod'][each_eaup_name] = e
            continue
        
        try:
            data_dhu = load_data_dhu(
                each_eaup_name     = each_eaup_name,
                each_equp_type     = each_equp_type,
                config_reader      = config_reader,
                config_reader_path = config_reader_path,
                data_URL           = data_URL
            )
        except Exception as e:
            ALL_RESULT['EXCEPTION']['Data_ATD'][each_eaup_name] = e
            continue
        
        try:
            ALL_RESULT['STATUS']['Mode_Steady'].append(each_eaup_name)
            
            opt_result = optimize_dhu(
                MODEL            = MODEL,
                data_cur         = data_dhu,
                config_reader    = config_reader,
                each_eaup_name   = each_eaup_name,
                each_equp_type   = each_equp_type
            )
            ALL_OPT_RES.append(opt_result['opt_summary'][1].assign(equp_name=each_eaup_name))
        except Exception as e:
            ALL_RESULT['EXCEPTION']['Opt'][each_eaup_name] = e
            raise e
            continue
            
        try:
            push_result(
                each_eaup_name       = each_eaup_name,
                each_eaup_name_short = each_eaup_name_short,
                each_equp_type       = each_equp_type,
                config_reader        = config_reader,
                opt_result           = opt_result,
                data_cur             = data_dhu
            )
        except Exception as e:
            ALL_RESULT['EXCEPTION']['Push'][each_eaup_name] = e
            continue

    pprint(ALL_RESULT)
    print_dataframe(pd.concat(ALL_OPT_RES,axis=0),'优化结果')
    


def load_model(
    each_eaup_name,
    each_equp_type,
    config_reader:ConfigReader,
    config_reader_path,
    use_adj_name = True
):
    adjust_equp_name = config_reader.get_equp_info(each_eaup_name,'替代模型','str')
    if (adjust_equp_name in config_reader.all_equp_names_total) and use_adj_name:
        print(f'{each_eaup_name}使用替代模型{adjust_equp_name}')
        each_eaup_name = adjust_equp_name
    else:
        pass
    
    if each_equp_type in ['DHU_A','DHU_B']:
        MODEL = DHU_AB
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        MODEL = SDHU_AB
    else:
        raise NotImplementedError
    
    if config_reader.get_app_info(
        each_eaup_name,
        app_type  = '实时优化',
        key       = '使用临时模型',
        info_type = 'bool'
    ):
        # 从文件夹中获取临时模型
        MODEL = MODEL.load(path=f'{config_reader_path}/model/{each_eaup_name}.pkl')
    else:
        MODEL = MODEL.load_from_platform(
            source   = 'id',
            model_id = config_reader.get_equp_info(
                equp_name = each_eaup_name,
                key       = '模型编号',
                info_type = 'str'
            )
        )
    return MODEL

def load_dhu_State(
    each_eaup_name,
    config_reader,
    config_reader_path,
    data_URL,
    NOW,
):
    data_last = (
        DataLoader(
            path          = f'{config_reader_path}/data/optimize/data_cur/',
            start_time    = NOW - timedelta(minutes=1),
            end_time      = NOW,
            print_process = False
        )
        .download_equp_data(
            equp_name   = each_eaup_name,
            point       = {'State':config_reader.get_equp_point(each_eaup_name,equp_class=['A'])['State']},
            url         = data_URL,
            clean_cache = True
        )
        .get_equp_data(
            equp_name = each_eaup_name,
        )
    )
    return data_last.iat[-1,0]

def load_data_room(
    each_eaup_name,
    each_equp_type,
    config_reader,
    config_reader_path,
    data_URL
):
    room_steady_len = config_reader.get_app_info(each_eaup_name,'实时优化','房间稳态判断时长','float')
    room_data_loader = DataLoader(
        path          = f'{config_reader_path}/data/optimize/data_cur/',
        start_time    = NOW - timedelta(minutes=room_steady_len),
        end_time      = NOW,
        print_process = False
    )
    room_data_loader.download_equp_data(
        equp_name   = each_eaup_name,
        point       = config_reader.get_equp_point(each_eaup_name,equp_class=['C']),
        url         = data_URL,
        clean_cache = True
    )
    room_data       = room_data_loader.get_equp_data(each_eaup_name)
    summary_dataframe(room_data,f'{each_eaup_name}房间数据')
    room_Dew_SP_adj = config_reader.get_app_info(each_eaup_name,'实时优化','房间露点设定值偏差','float')
    room_Dew_SP     = room_data.room_DSP.mean() + room_Dew_SP_adj
    room_Dew_PV     = room_data.room_DPV.mean()
    room_Dew_diff_dwlim = config_reader.get_app_info(each_eaup_name,'实时优化','房间露点过低阈值','float')
    
    # 模式判断
    is_room_Dew_steady = (room_Dew_PV < (room_Dew_SP + 0.5)) and (room_Dew_PV > (room_Dew_SP - 0.5))
    is_room_Dew_low    = room_Dew_PV < (room_Dew_SP - room_Dew_diff_dwlim)
    
    data_room = {
        'is_room_Dew_steady': is_room_Dew_steady,
        'is_room_Dew_low'   : is_room_Dew_low,
        'room_Dew_SP'       : room_Dew_SP,
        'room_Dew_PV'       : room_Dew_PV
    }
    return data_room

def load_data_dhu(
    each_eaup_name,
    each_equp_type,
    config_reader,
    config_reader_path,
    data_URL,
):
    dhu_steady_len   = config_reader.get_app_info(each_eaup_name,'实时优化','除湿机工况均值时长','float')
    data_input_point = config_reader.get_equp_point(each_eaup_name,equp_class=['A','B'])
    data_last = (
        DataLoader(
            path          = f'{config_reader_path}/data/optimize/data_cur/',
            start_time    = NOW - timedelta(minutes=dhu_steady_len),
            end_time      = NOW,
            print_process = False
        )
        .download_equp_data(
            equp_name   = each_eaup_name,
            point       = data_input_point,
            url         = data_URL,
            clean_cache = True
        )
        .get_equp_data(
            equp_name = each_eaup_name,
        )
    )
    data_cur = data_last.mean(axis=0).to_frame().T
    summary_dataframe(data_last,f'{each_eaup_name}除湿机数据')
    return data_cur

def optimize_dhu(
    MODEL           : Union[DHU_AB,SDHU_AB],
    data_cur        : pd.DataFrame,
    config_reader   : ConfigReader,
    each_eaup_name  : str,
    each_equp_type  : str
):
    # 模型精度判断
    predict  = MODEL.predict_system(input_data=data_cur)
    TVP_data = (
        predict.T.set_axis(['pred'],axis=1)
        .join(
            data_cur.T.set_axis(['real'],axis=1),
            how = 'left'
        )
        .dropna(axis=0)
        .assign(diff=lambda x: x['pred'] - x['real'])
    )
    print(TVP_data)
    
    # 实时优化
    
    if each_equp_type in ['DHU_A','DHU_B']:
        opt_res = MODEL.optimize(
            cur_input_data=data_cur,
            wheel_1_TinR=(
                config_reader.get_equp_info(each_eaup_name,key='前再生温度上限',info_type='float'),
                config_reader.get_equp_info(each_eaup_name,key='前再生温度下限',info_type='float')
            ),
            wheel_2_TinR=(
                config_reader.get_equp_info(each_eaup_name,key='后再生温度上限',info_type='float'),
                config_reader.get_equp_info(each_eaup_name,key='后再生温度下限',info_type='float')
            ),
            fan_2_Hz   = None,
            constrains = ['coil_3_DoutA-[coil_3_DoutA]<0'],
            logging    = False,
            target     = 'summary_waste',
            target_min = True
        )
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        opt_res = MODEL.optimize(
            cur_input_data = data_cur,
            wheel_1_TinR   = (
                config_reader.get_equp_info(each_eaup_name,key='前再生温度上限',info_type='float'),
                config_reader.get_equp_info(each_eaup_name,key='前再生温度下限',info_type='float')
            ),
            fan_2_Hz   = (30,45),
            constrains = ['wheel_1_DoutP-[wheel_1_DoutP]<0'],
            logging    = False,
            target     = 'summary_waste',
            target_min = True
        )
    opt_summary = opt_res['opt_summary']
    opt_var     = opt_res['opt_var']
    print_dataframe(opt_summary,'优化结果')
    print_dataframe(opt_var,'优化变量')
    return {
        'opt_summary': opt_summary,
        'opt_var'    : opt_var,
    }

def push_result(
    each_eaup_name,
    each_eaup_name_short,
    each_equp_type,
    config_reader:ConfigReader,
    opt_result:dict,
    data_cur:pd.DataFrame
):
    equp_point = config_reader.get_equp_point(each_eaup_name,equp_class=['D'])
    if each_equp_type in ['DHU_A','DHU_B']:
        rcmd = [
            {
                'name'      : '前再生加热温度',
                'rcmd_value': opt_result['opt_var'][0].iat[0,0],
                'children'  : [
                    {
                        'name'      : '前再生加热温度',
                        'rcmd_value': opt_result['opt_var'][0].round(1).iat[0,0],
                        'curr_value': data_cur.loc[:,['wheel_1_TinR']].round(1).iat[0,0],
                        'point_id'  : f'{each_eaup_name}_{equp_point["wheel_1_TinR_AISP"]}'
                    }
                ]
            },
            {
                'name'      : '后再生加热温度',
                'rcmd_value': opt_result['opt_var'][1].iat[0,0],
                'children'  : [
                    {
                        'name'      : '后再生加热温度',
                        'rcmd_value': opt_result['opt_var'][1].round(1).iat[0,0],
                        'curr_value': data_cur.loc[:,['wheel_2_TinR']].round(1).iat[0,0],
                        'point_id'  : f'{each_eaup_name}_{equp_point["wheel_2_TinR_AISP"]}'
                    }
                ]
            },
        ]
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        rcmd = [
            {
                'name'      : '再生加热温度',
                'rcmd_value': opt_result['opt_var'][0].iat[0,0],
                'children'  : [
                    {
                        'name'      : '再生加热温度',
                        'rcmd_value': opt_result['opt_var'][0].round(1).iat[0,0],
                        'curr_value': data_cur.loc[:,['wheel_1_TinR']].round(1).iat[0,0],
                        'point_id'  : f'{each_eaup_name}_{equp_point["wheel_1_TinR_AISP"]}'
                    }
                ]
            },
            {
                'name'      : '排风机频率',
                'rcmd_value': opt_result['opt_var'][1].iat[0,0],
                'children'  : [
                    {
                        'name'      : '排风机频率',
                        'rcmd_value': opt_result['opt_var'][1].round(1).iat[0,0],
                        'curr_value': data_cur.loc[:,['fan_2_Hz']].round(1).iat[0,0],
                        'point_id'  : f'{each_eaup_name}_{equp_point["fan_2_Hz_AISP"]}'
                    }
                ]
            },
        ]
    else:
        raise Exception('MODEL_TYPE_ERROR')
    
    if config_reader.get_app_info(each_eaup_name,'实时优化','推送策略','bool'):
        add_ai_rcmd_operation(
            code           = each_eaup_name_short,
            job_id         = os.environ.get('JOB_ID', None),
            rcmd           = rcmd,
            custom_details = []
        )
    else:
        pprint(rcmd)
        
        
def add_ai_rcmd_operation(code,job_id,rcmd,custom_details):
    from workflowlib import requests
    json_info = {
        "job_id": job_id,
        "code"  : code,
        "data"  : {
            "rcmd"          : rcmd,
            "custom_details": custom_details
        },
    }
    print('ai_rcmd',json_info)
    try:
        # 下发分步指令
        url      = 'http://m2-backend-svc:8000/api/ai/ai_rcmd_operation/add_ai_rcmd_operation_v3'
        r        = requests.post(url,json=json_info,headers={'Content-Type': "application/json", "Connection": "close"})
        jsonResp = r.json()
        print(f'下发指令 jsonResp',jsonResp)
    except Exception as e:
        print(e)
    return