AirTerminalDevice/apps/DHU/train.py

import os
from datetime import datetime
from pathlib import Path
from pprint import pprint

import pandas as pd

from ...model.DHU.DHU_AB import DHU_AB
from ...model.DHU.SDHU_AB import SDHU_AB
from ...model.Room.room import RoomDewPredictor
from .config_reader import ConfigReader
from ...tools.data_loader import DataLoader

NOW             = datetime.now().replace(second=0,microsecond=0)
PATH            = os.path.dirname(os.path.realpath(__file__)).replace('\\','/')
MODEL_FUNC_PATH = f'{PATH}/model_func.py'
MODEL_FILE_PATH = f'./model.pkl'

def train(*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':{
            'Data': {},
            'Fit' : {},
            'Save': {},
            'Plot': {}
        }
    }
    
    for each_eaup_name in config_reader.all_equp_names:
        equp_type = config_reader.get_equp_info(each_eaup_name,key='设备类型',info_type='str')
        print(f'{each_eaup_name}开始训练，设备类型为{equp_type}')
        
        # 获取数据
        try:
            equp_data = load_data(
                each_eaup_name=each_eaup_name,each_equp_type=equp_type,config_reader=config_reader,
                config_reader_path=config_reader_path,data_URL=data_URL
            )
        except Exception as E:
            ALL_RESULT['EXCEPTION']['Data'][each_eaup_name] = E
            raise E
            continue
        
        # 训练模型 
        try:
            equp_model,equp_data_clean = train_equp_model(
                each_eaup_name=each_eaup_name,each_equp_type=equp_type,equp_data=equp_data,
                config_reader=config_reader,config_reader_path=config_reader_path)
            train_room_model(
                each_eaup_name=each_eaup_name,each_equp_type=equp_type,equp_data=equp_data,
                config_reader=config_reader,config_reader_path=config_reader_path
            )
        except Exception as E:
            ALL_RESULT['EXCEPTION']['Fit'][each_eaup_name] = E
            print(f'{each_eaup_name}模型训练异常 {E}')
            continue
        
        # 保存可视化结果
        if config_reader.get_app_info(each_eaup_name,'模型训练','保存可视化结果','bool') and equp_model is not None:
            try:
                save_train_info(
                    equp_model=equp_model,equp_data=equp_data_clean,
                    config_reader_path=config_reader_path,each_eaup_name=each_eaup_name)
            except Exception as E:
                ALL_RESULT['EXCEPTION']['Plot'][each_eaup_name] = E
                pass
        
        # 模型迭代
        if not config_reader.get_app_info(each_eaup_name,'模型训练','迭代模型','bool') and equp_model is not None:
            continue
        try:
            monitor_point = config_reader.point.loc[lambda dt:dt.类型=='B']
            model_update_info = {}
            for i in range(len(monitor_point)):
                name    = monitor_point.loc[:,'编号'].iat[i]
                name_cn = monitor_point.loc[:,'名称'].iat[i]
                MAE     = monitor_point.loc[:,'指标MAE'].iat[i]
                model_update_info[name] = {
                    'point_id'   : name,
                    'point_name' : name_cn,
                    'point_class': name,
                    'thre_mae'   : MAE,
                    'thre_mape'  : 1,
                    'thre_days'  : 7
                }
            equp_model.save_to_platform(
                version_id      = datetime.now().strftime('%Y%m'),
                model_id        = config_reader.get_equp_info(each_eaup_name,'模型编号','str'),
                update_method   = 'update',
                model_info      = model_update_info,
                MODEL_FILE_PATH = MODEL_FILE_PATH,
                MODEL_FUNC_PATH = MODEL_FUNC_PATH,
            )
        except Exception as E:
            ALL_RESULT['EXCEPTION']['Save'][each_eaup_name] = E
            continue
    
    pprint(ALL_RESULT)


def save_data(dir,file:str,data:pd.DataFrame):
    Path(dir).mkdir(parents=True,exist_ok=True)
    if file.endswith('.csv'):
        data.to_csv(os.path.join(dir,file),index=True)
    elif file.endswith('.pkl'):
        data.to_pickle(os.path.join(dir,file))
    else:
        raise Exception('file type error')

def load_data(each_eaup_name,each_equp_type,config_reader,config_reader_path,data_URL):
    # 部分情况下设备不需要部分点位表中的点位
    rm_point_name = []
    if not config_reader.get_equp_info(each_eaup_name,'存在回风口','bool'):
        rm_point_name += ['mixed_1_TinM','mixed_1_DinM']
    if not config_reader.get_equp_info(each_eaup_name,'存在补风口','bool'):
        rm_point_name += ['mixed_2_TinM','mixed_2_DinM']
    
    # 获取历史数据
    data_loader = DataLoader(
        path          = f'{config_reader_path}/data/train/data_his/',
        start_time    = config_reader.get_app_info(each_eaup_name,app_type='模型训练',key='开始时间',info_type='datetime'),
        end_time      = config_reader.get_app_info(each_eaup_name,app_type='模型训练',key='结束时间',info_type='datetime'),
        print_process = config_reader.get_app_info(each_eaup_name,app_type='模型训练',key='打印取数日志',info_type='bool'),
    )
    data_loader.download_equp_data(
        equp_name     = each_eaup_name,
        point         = config_reader.get_equp_point(each_eaup_name,equp_class=['A','B','C']),
        url           = data_URL,
        clean_cache   = False,
        rm_point_name = rm_point_name
    )
    equp_data = data_loader.get_equp_data(each_eaup_name)
    save_data(f'{config_reader_path}/data/train/data_his_raw',f'{each_eaup_name}.pkl',equp_data)
    return equp_data

def train_equp_model(each_eaup_name,each_equp_type,equp_data,config_reader,config_reader_path):
    if each_equp_type in ['DHU_A','DHU_B']:
        equp_model = DHU_AB(
            DHU_type   = each_equp_type,
            exist_Fa_H = config_reader.get_equp_info(each_eaup_name,'存在回风口','bool'),
            exist_Fa_B = config_reader.get_equp_info(each_eaup_name,'存在补风口','bool'),
        )
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        equp_model = SDHU_AB(
            DHU_type   = each_equp_type,
            exist_Fa_H = config_reader.get_equp_info(each_eaup_name,'存在回风口','bool'),
            exist_Fa_H0= config_reader.get_equp_info(each_eaup_name,'存在回风口（前表冷后）','bool'),
        )
    else:
        raise NotImplementedError
    
    # 清洗数据
    Path(f'{config_reader_path}/data/train/clean_log/').mkdir(parents=True, exist_ok=True)
    equp_data_clean = equp_model.clean_data(
        data          = equp_data,
        data_type     = ['input','observed'],
        print_process = True,
        fill_zero     = False,
        save_log      = f'{config_reader_path}/data/train/clean_log/{each_eaup_name}.txt',
    )
    equp_data_clean = equp_data_clean.resample('15min').mean().dropna()
    save_data(f'{config_reader_path}/data/train/data_his_clean',f'{each_eaup_name}.pkl',equp_data_clean)
    
    if not config_reader.get_app_info(each_eaup_name,'模型训练','训练设备模型','bool'):
        return None,None
    
    if each_equp_type in ['DHU_A','DHU_B']:
        equp_model.fit(
            input_data    = equp_data_clean,
            observed_data = equp_data_clean,
            plot_TVP      = False,
            rw_FA_val     = config_reader.get_app_info(each_eaup_name,'模型训练','新风阀门开度参数','bool')
        )
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        equp_model:SDHU_AB
        equp_model.fit(
            input_data    = equp_data_clean,
            observed_data = equp_data_clean,
            plot_TVP      = False
        )
    else:
        raise NotImplementedError
    Path(f'{config_reader_path}/model').mkdir(parents=True, exist_ok=True)
    equp_model.save(f'{config_reader_path}/model/{each_eaup_name}.pkl')
    save_data(f'{config_reader_path}/data/train/data_TVP',f'{each_eaup_name}.csv',equp_model.TVP_data)
    save_data(f'{config_reader_path}/data/train/data_metric',f'{each_eaup_name}.csv',equp_model.TVP_metric.round(2))
    
    return equp_model,equp_data_clean

def train_room_model(each_eaup_name,each_equp_type,equp_data,config_reader:ConfigReader,config_reader_path):
    if not config_reader.get_app_info(each_eaup_name,'模型训练','训练房间模型','bool'):
        return None
    
    N_fit = 24 * 60 * 3
    
    try:
        equp_model_path = f'{config_reader_path}/model/{each_eaup_name}.pkl'
        if each_equp_type in ['DHU_A','DHU_B']:
            equp_model = DHU_AB.load(equp_model_path)
            Dout       = equp_model.predict(equp_data.iloc[-N_fit:,:])['coil_3']['DoutA']
        elif each_equp_type in ['SDHU_A','SDHU_B']:
            equp_model = SDHU_AB.load(equp_model_path)
            Dout       = equp_model.predict(equp_data.iloc[-N_fit:,:])['wheel_1']['DoutP']
        else:
            raise NotImplementedError
    except Exception as E:
        Dout = None
        print(f'{each_eaup_name}设备模型加载失败，只训练基于实际送风露点的房间露点模型')
    
    # 实际送风露点
    if each_equp_type in ['DHU_A','DHU_B']:
        Dout_real  = equp_data.iloc[-N_fit:,:].loc[:,'wheel_2_DoutP'].values
    elif each_equp_type in ['SDHU_A','SDHU_B']:
        Dout_real  = equp_data.iloc[-N_fit:,:].loc[:,'coil_2_DoutA'].values
    else:
        raise NotImplementedError
    
    N_room = config_reader.get_equp_info(each_eaup_name,'房间数量','int')
    
    path_diffdata    = f'{config_reader_path}/plot/plot_room_diffdata/'
    path_lagcorr     = f'{config_reader_path}/plot/plot_room_lagcorr/'
    path_diffdata_bk = f'{config_reader_path}/plot/plot_room_diffdata_bk/'
    path_lagcorr_bk  = f'{config_reader_path}/plot/plot_room_lagcorr_bk/'
    Path(path_diffdata).mkdir(parents=True, exist_ok=True)
    Path(path_lagcorr).mkdir(parents=True, exist_ok=True)
    Path(path_diffdata_bk).mkdir(parents=True, exist_ok=True)
    Path(path_lagcorr_bk).mkdir(parents=True, exist_ok=True)
    
    for i in range(1,N_room+1):
        Droom = equp_data.iloc[-N_fit:,:].loc[:,f'room_{i}_Dpv'].values
        if Dout is not None:
            room_model = RoomDewPredictor().fit_Droom(Dout=Dout,Droom=Droom)
            room_model.save(f'{config_reader_path}/model/{each_eaup_name}_room_{i}_Dpv.pkl')
            room_model.plot_diffdata(Dout,Droom).save(filename=f'{path_diffdata}/{each_eaup_name}_room_{i}_Dpv.png')
            room_model.plot_diffdata_lagcorr(Dout,Droom).save(filename=f'{path_lagcorr}/{each_eaup_name}_room_{i}_Dpv.png')
        room_model_bk = RoomDewPredictor().fit_Droom(Dout=Dout_real,Droom=Droom)
        room_model_bk.save(f'{config_reader_path}/model/{each_eaup_name}_room_{i}_Dpv_bk.pkl')
        room_model_bk.plot_diffdata(Dout_real,Droom).save(filename=f'{path_diffdata_bk}/{each_eaup_name}_room_{i}_Dpv.png')
        room_model_bk.plot_diffdata_lagcorr(Dout_real,Droom).save(filename=f'{path_lagcorr_bk}/{each_eaup_name}_room_{i}_Dpv.png')
    
def save_train_info(equp_model,equp_data,config_reader_path,each_eaup_name):
    for plot_name,plot in equp_model.plot_check(equp_data).items():
        path = f'{config_reader_path}/plot/{plot_name}/'
        Path(path).mkdir(parents=True, exist_ok=True)
        plot.save(filename=f'{path}/{each_eaup_name}.png')
    
    path = f'{config_reader_path}/plot/TVP'
    Path(path).mkdir(parents=True, exist_ok=True)
    equp_model.plot_TVP(equp_model.TVP_data,save_path=f'{path}/{each_eaup_name}.png')