AirTerminalDevice/model/DHU/DHU_B.py

import numpy as np
import pandas as pd
import pymc as pm
import pytensor.tensor as pt

from .._base._base_device import BaseDevice
from ...components.coil_water import CoolingCoil2
from ...components.coil_steam import SteamCoil
from ...components.wheel2 import WheelS2V3
from ...components.wheel3 import WheelS3
from ...components.mixed import Mixed
from ..utils.fit_utils import (
    observe,record,reorder_posterior
)
from ...tools.optimizer import optimizer
from .DHU_AB import AirFlow_DHU_AB



class DHU_B(BaseDevice):
    
    model_input_data_columns = {
        'Tin_F'       : 'coil_1_ToutA',
        'Hin_F'       : 'coil_1_HoutA',
        'fan_1_Hz'    : 'fan_1_Hz',
        'fan_2_Hz'    : 'fan_2_Hz',
        'coil_1_TinW' : 'coil_1_TinW',
        'coil_2_TinW' : 'coil_2_TinW',
        'coil_3_TinW' : 'coil_3_TinW',
        'coil_1_Val'  : 'coil_1_Val',
        'coil_2_Val'  : 'coil_2_Val',
        'coil_3_Val'  : 'coil_3_Val',
        'wheel_1_TinR': 'wheel_1_TinR',
        'wheel_2_TinR': 'wheel_2_TinR',
        'mixed_1_TinM': 'mixed_1_TinM',
        'mixed_2_TinM': 'mixed_2_TinM',
        'mixed_1_HinM': 'mixed_1_HinM',
        'mixed_2_HinM': 'mixed_2_HinM',
    }
    model_observe_data_columns = {
        'mixed_1_ToutA'  : 'mixed_1_ToutA',
        'mixed_1_DoutA'  : 'mixed_1_DoutA',
        'coil_2_ToutA'   : 'coil_2_ToutA',
        'coil_2_DoutA'   : 'coil_2_DoutA',
        'wheel_2_ToutP'  : 'wheel_2_ToutP',
        'wheel_2_DoutP'  : 'wheel_2_DoutP',
        'wheel_2_ToutR'  : 'wheel_2_ToutR',
        'wheel_2_ToutC'  : 'wheel_2_ToutC', 
    }
    val_rw_adj_target = ('coil_2_DoutA','coil_3_DoutA')
    
    def __init__(
        self,
        wheel_1       = 'WheelS3',
        wheel_2       = 'WheelS3',
        coolingcoil_2 = 'CoolingCoil2',
        coolingcoil_3 = 'CoolingCoil2',
        heatingcoil_1 = 'SteamCoil',
        heatingcoil_2 = 'SteamCoil',
        mixed_1       = 'Mixed',
        mixed_2       = 'Mixed',
        **kwargs
    ) -> None:
        super().__init__()
        if 'components' in kwargs:
            self.components = kwargs['components']
        else:
            self.components = {
                'wheel_1'      : wheel_1,
                'wheel_2'      : wheel_2,
                'coil_2'       : coolingcoil_2,
                'coil_3'       : coolingcoil_3,
                'heatingcoil_1': heatingcoil_1,
                'heatingcoil_2': heatingcoil_2,
                'mixed_1'      : mixed_1,
                'mixed_2'      : mixed_2
            }
        self.record_load_info(components=self.components)
        self.components = {k:eval(v)(k) for k,v in self.components.items()}
        
    def fit(
        self,
        input_data   : pd.DataFrame,
        observed_data: pd.DataFrame,
        exist_Fa_H   : bool,
        exist_Fa_B   : bool,
        rw_FA_val    : bool = False,
        plot_TVP     : bool = True
    ):
        with pm.Model() as self.MODEL_PYMC:
            param_prior = {name:comp.prior() for name,comp in self.components.items()}
            param_prior['F_air'] = AirFlow_DHU_AB.prior(
                rw_FA_val  = rw_FA_val,
                N          = len(input_data),
                exist_Fa_H = exist_Fa_H,
                exist_Fa_B = exist_Fa_B
            )
            
            res = self.model(
                **{k:input_data.loc[:,v].values for k,v in self.model_input_data_columns.items()},
                engine       = 'pymc',
                components   = self.components,
                param        = param_prior
            )
            for std_name,name in self.model_observe_data_columns.items():
                if name not in observed_data.columns:
                    continue
                observed_data = observed_data.rename(columns={name:std_name})
            observe('mixed_1_ToutA',res['mixed_1']['ToutA'],observed=observed_data)
            observe('mixed_1_DoutA',res['mixed_1']['DoutA'],observed=observed_data)
            observe('coil_2_ToutA',res['coil_2']['ToutA'],observed=observed_data)
            observe('coil_2_DoutA',res['coil_2']['DoutA'],observed=observed_data)
            observe('wheel_2_ToutP',res['wheel_2']['ToutP'],observed=observed_data)
            observe('wheel_2_DoutP',res['wheel_2']['DoutP'],observed=observed_data)
            observe('wheel_2_ToutC',res['wheel_2']['ToutC'],observed=observed_data)
            observe('wheel_2_ToutR',res['wheel_2']['ToutR'],observed=observed_data)
            observe('wheel_1_ToutR',res['wheel_1']['ToutR'],observed=observed_data,sigma=5)
            
            record('steamcoil_1_Fs',res['steamcoil_1']['Fs'])
            record('steamcoil_2_Fs',res['steamcoil_2']['Fs'])
            
            record('mixed_2_ToutA',res['mixed_2']['ToutA'])
            record('wheel_1_FaP',res['wheel_1']['FP'])
            record('wheel_1_FaR',res['wheel_1']['FR'])
            record('wheel_1_FaC',res['wheel_1']['FC'])
            record('mixed_1_FaA',res['mixed_1']['FA'])
            record('mixed_1_FaM',res['mixed_1']['FM'])
            record('F_air_S',res['Fa']['Fa_S'])
            record('F_air_H',res['Fa']['Fa_H'])
            record('F_air_X',res['Fa']['Fa_X'])
            self.param_posterior = pm.find_MAP(maxeval=50000,include_transformed=False)
            
        self.record_model(
            model_name   = 'ATD',
            model        = reorder_posterior(param_prior,self.param_posterior),
            train_data   = {'x':np.array([1])},
            train_metric = {'R2':1,'MAE':1,'MAPE':1}
        )
        self.TVP_data,self.TVP_metric = get_fitted_result(self.param_posterior,observed_data,plot_TVP)
        return self
    
    def optimize(
        self,
        cur_input_data : pd.DataFrame,
        wheel_1_TinR_ub: float = 120,
        wheel_1_TinR_lb: float = 70,
        wheel_2_TinR_ub: float = 120,
        wheel_2_TinR_lb: float = 70,
        constrains     : list  = None,
        logging        : bool  = True
    ) -> list:
        constrains = [] if constrains is None else constrains
        cur_input_data = cur_input_data.iloc[[0],:]
        opt_var_boundary  = {
            'wheel_1_TinR':{'lb':wheel_1_TinR_lb,'ub':wheel_1_TinR_ub},
            'wheel_2_TinR':{'lb':wheel_2_TinR_lb,'ub':wheel_2_TinR_ub},
        }
        opt_var_value = cur_input_data.loc[:,list(opt_var_boundary.keys())]
        oth_var_value = (
            cur_input_data
            .loc[:,list(self.model_input_data_columns.values())]
            .drop(opt_var_value.columns,axis=1)
        )
        opt_res = optimizer(
            model            = self,
            opt_var_boundary = opt_var_boundary,
            opt_var_value    = opt_var_value,
            oth_var_value    = oth_var_value,
            target           = 'Fs',
            target_min       = True,
            constrains       = constrains,
            logging          = logging,
            other_kwargs     = {'NIND':2000,'MAXGEN':50}
        )
        return opt_res
    
    
    @classmethod
    def model(
        cls,
        Tin_F,        # 前表冷后温度
        Hin_F,        # 前表冷后湿度
        fan_1_Hz,     # 处理侧风机频率 
        fan_2_Hz,     # 再生侧风机频率
        coil_1_TinW,  # 前表冷进水温度
        coil_2_TinW,  # 中表冷进水温度
        coil_3_TinW,  # 后表冷进水温度
        coil_1_Val,   # 前表冷阀门开度
        coil_2_Val,   # 中表冷阀门开度
        coil_3_Val,   # 后表冷阀门开度
        wheel_1_TinR, # 前转轮再生侧温度
        wheel_2_TinR, # 后转轮再生侧温度
        mixed_1_TinM, # 回风温度（处理侧）
        mixed_1_HinM, # 回风湿度（处理侧）
        mixed_2_TinM, # 补风温度（再生侧）
        mixed_2_HinM, # 补风湿度（再生侧）
        engine    : str,
        components: dict,
        param     : dict,
    ) ->  dict:
        
        # 水的质量流量
        coil_2_FW = coil_2_Val / 100
        coil_3_FW = coil_3_Val / 100
        # 空气的质量流量
        air_flow = AirFlow_DHU_AB.model(fan_1_Hz=fan_1_Hz,fan_2_Hz=fan_2_Hz,param=param,type='DHU_B')
        
        # 前转轮
        wheel_1_res = components['wheel_1'].model(
            TinP   = Tin_F,
            HinP   = Hin_F,
            FP     = air_flow['wheel_1_FaP'],
            TinR   = wheel_1_TinR,
            HinR   = 0,
            FR     = air_flow['wheel_1_FaR'],
            engine = engine,
            param  = param,
        )
        
        # 处理侧混风（回风）
        mixed_1_res = components['mixed_1'].model(
            TinA   = wheel_1_res['ToutP'],
            HinA   = wheel_1_res['HoutP'],
            FA     = air_flow['mixed_1_FaA'],
            TinM   = mixed_1_TinM,
            HinM   = mixed_1_HinM,
            FM     = air_flow['mixed_1_FaM'],
            engine = engine
        )
        
        # 中表冷
        coil_2_res = components['coil_2'].model(
            TinA   = mixed_1_res['ToutA'],
            HinA   = mixed_1_res['HoutA'],
            FA     = air_flow['coil_2_FaA'],
            TinW   = coil_2_TinW,
            FW     = coil_2_FW,
            engine = engine,
            param  = param['coil_2']
        )
        
        # 后转轮
        wheel_2_res = components['wheel_2'].model(
            TinP   = coil_2_res['ToutA'],
            HinP   = coil_2_res['HoutA'],
            FP     = air_flow['wheel_2_FaP'],
            TinC   = mixed_1_res['ToutA'],
            HinC   = mixed_1_res['HoutA'],
            FC     = air_flow['wheel_2_FaC'],
            TinR   = wheel_2_TinR,
            HinR   = 0,
            FR     = air_flow['wheel_2_FaR'],
            engine = engine,
            param  = param['wheel_2'],
        )
        
        # 后表冷
        coil_3_res = components['coil_3'].model(
            TinA   = wheel_2_res['ToutP'],
            HinA   = wheel_2_res['HoutP'],
            FA     = air_flow['coil_3_FaA'],
            TinW   = coil_3_TinW,
            FW     = coil_3_FW,
            engine = engine,
            param  = param['coil_3']
        )
        
        
        # 后转轮湿度修正
        wheel_2_res_adj = components['wheel_2'].model(
            TinP   = coil_2_res['ToutA'],
            HinP   = coil_2_res['HoutA'],
            FP     = air_flow['wheel_2_FaP'],
            TinC   = mixed_1_res['ToutA'],
            HinC   = mixed_1_res['HoutA'],
            FC     = air_flow['wheel_2_FaC'],
            TinR   = wheel_2_TinR,
            HinR   = wheel_2_res['HoutC'],
            FR     = air_flow['wheel_2_FaR'],
            engine = engine,
            param  = param['wheel_2'],
        )
        
        # 再生侧混风（排风）
        mixed_2_res = components['mixed_2'].model(
            TinA   = wheel_2_res_adj['ToutR'],
            HinA   = wheel_2_res_adj['HoutR'],
            FA     = air_flow['mixed_2_FaA'],
            TinM   = mixed_2_TinM,
            HinM   = mixed_2_HinM,
            FM     = air_flow['mixed_2_FaM'],
            engine = engine
        )
        
        # 前转轮湿度修正
        wheel_1_res_adj = components['wheel_1'].model(
            TinP   = Tin_F,
            HinP   = Hin_F,
            FP     = air_flow['wheel_1_FaP'],
            TinR   = wheel_1_TinR,
            HinR   = mixed_2_res['HoutA'],
            FR     = air_flow['wheel_1_FaR'],
            engine = engine,
            param  = param,
        )
        
        # 前蒸气盘管
        steamcoil_1_res = components['steamcoil_1'].model(
            TinA   = mixed_2_res['ToutA'],
            ToutA  = wheel_1_TinR,
            FA     = air_flow['steamcoil_1_Fa'],
            param  = param['steamcoil_1'],
            engine = engine
        )
        
        # 后蒸气盘管
        steamcoil_2_res = components['steamcoil_2'].model(
            TinA   = wheel_2_res_adj['ToutC'],
            ToutA  = wheel_2_TinR,
            FA     = air_flow['steamcoil_2_Fa'],
            param  = param['steamcoil_2'],
            engine = engine
        )
        
        return {
            'coil_2'     : coil_2_res,
            'coil_3'     : coil_3_res,
            'wheel_1'    : wheel_1_res_adj,
            'wheel_2'    : wheel_2_res_adj,
            'mixed_1'    : mixed_1_res,
            'mixed_2'    : mixed_2_res,
            'steamcoil_1': steamcoil_1_res,
            'steamcoil_2': steamcoil_2_res,
            'Fa'         : air_flow,
            'summary'    : {
                'Fs':steamcoil_1_res['Fs'] + steamcoil_2_res['Fs'],
            }
        }