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'], } }