from typing import Union import numpy as np import pandas as pd from .opt_obj import Opt from .opt_alg import main as opt_alg_main from .constrains.build import main as build_all_constrains from ._utils.config_info import ConfigInfo from .model.diagnosis import ModelDiag def main(*args, system_model=None, config=None): config_info = ConfigInfo(config=config) group_data = config_info.get_io_group_info(io='in',type='data',data=args) group_point = config_info.get_io_group_info(io='in',type='point_id') check_var_data(group_data) # 校验数据的维度 # 组件输入一：模型变量 group_opt_var = group_data.get('优化变量') group_sys_var = group_data.get('系统变量') group_oth_var = group_data.get('其他变量') group_opt_col = group_point.get('优化变量') group_sys_col = group_point.get('系统变量') group_oth_col = group_point.get('其他变量') # 组件输入二：外部变量 group_cst_var = group_data.get('外部变量') group_cst_col = group_point.get('外部变量') # 组件输入三：动态配置项 group_dyn_cfg = group_data.get('动态配置',[None]) allow_neg_opt = group_dyn_cfg[0] # 是否需要校验负优化 # 初始化边界条件 boundary,boundary_insert = get_boundary(group_opt_var) lb,ub,var_type,var_precis,collection_map,var_map = init_boundary(boundary) opt_var = get_opt_var(group_opt_var,group_opt_col) sys_var = get_sys_var(group_sys_var,group_sys_col) oth_var = get_oth_var(group_oth_var,group_oth_col,index=opt_var.index) cst_var = get_cst_var(group_cst_var,group_cst_col,index=opt_var.index) # 初始化可行性约束，动态可行性约束需要提前解析 constrains = build_all_constrains(config) # 系统模型 if system_model is None: from .model.model import main as get_system_model system_model = get_system_model(config) # 优化对象 opt_object = Opt( lb = lb, ub = ub, varTypes = var_type, var_precis = var_precis, maxormins = 1 if config['dir_min'] is True else -1, collection_map = collection_map, var_map = var_map, boundary_insert = boundary_insert, opt_var = opt_var, sys_var = sys_var, system_model = system_model, oth_var = oth_var, cst_var = cst_var, opt_target = config['target'], constrains = constrains ) # 系统模型诊断 diag = config.get('diag_model',True) if diag == True: model_diag = ModelDiag( model = system_model, data = opt_object.current_system_x, c_var = opt_object.name_opt_vars_c, d_var = 1, ) model_diag.summary_diag() # 优化算法 # 是否允许负优化，组件的配置或输入中任意一处允许负优化，则不检查负优化 allow_neg_opt_config = config_info.get_property('allow_neg_opt',default=False) allow_neg_opt = any([allow_neg_opt,allow_neg_opt_config]) result_list = opt_alg_main(opt_obj = opt_object, allow_neg_opt = allow_neg_opt, config =config) # 输出中增加系统模型的初始输出（对应系统模型的初始输入） output_groups = config_info.get_io_group_info(io='out',type='point_id') result_list = result_append_init_sys_y(result_list,output_groups,opt_object) return result_list def check_var_data(group_data) -> None: for name,data in group_data.items(): if not isinstance(data,pd.DataFrame): continue if data.shape[0] == 1: raise Exception(f'{name}的数据行数不等于1，不满足要求，数据为{data}') def get_opt_var(group_opt_var:list,group_opt_col:list) -> pd.DataFrame: opt_var = [] for bound_info in group_opt_var: opt_var.append(bound_info['cur_val']) opt_var = pd.concat(opt_var,axis=1) opt_var.columns = group_opt_col return opt_var def get_sys_var(group_sys_var:list,group_sys_col:list) -> Union[pd.DataFrame,None]: if group_sys_var is None or len(group_sys_var)==0: print('【警告】未获取到"系统变量"，确保实际不存在该变量 ') sys_var = None else: group_sys_var = [df.iloc[:,[0]] for df in group_sys_var] sys_var = pd.concat(group_sys_var,axis=1) sys_var.columns = group_sys_col return sys_var def get_oth_var(group_oth_var:list,group_oth_col:list,index) -> Union[pd.DataFrame,None]: oth_var_exist = group_oth_col is not None and len(group_oth_var) > 0 if oth_var_exist: is_oth_var_df = all([isinstance(df,pd.DataFrame) for df in group_oth_var]) if not oth_var_exist: # print('【警告】未获取到"其他变量"，确保实际不存在该变量 ') oth_var = None elif is_oth_var_df and oth_var_exist: group_oth_var = [df.iloc[:,[0]] for df in group_oth_var] oth_var = pd.concat(group_oth_var,axis=1) oth_var.columns = group_oth_col elif (not is_oth_var_df) and oth_var_exist: print('【警告】未获取到有效的"其他变量"，用0值代替所有该类变量 ') oth_var = pd.DataFrame( data = np.zeros_like(group_oth_col,dtype='float').reshape(1,-1), columns = group_oth_col, index = index ) return oth_var def get_cst_var(group_cst_var:list,group_cst_col:list,index) -> Union[pd.DataFrame,None]: group_cst_var = None if (group_cst_var is None) or (len(group_cst_var)==0) else group_cst_var group_cst_col = None if (group_cst_col is None) or (len(group_cst_col)==0) else group_cst_col if group_cst_var is None or len(group_cst_var) == 0: # print('【警告】未获取到"外部变量"，确保实际不存在该变量 ') return None new_group_cst_var = [] # 校验数据类型，并重置Index（对齐）和Columns（便于引用） for var in group_cst_var: if not isinstance(var,pd.DataFrame): raise Exception(f'外部变量中存在错误的数据类型{type(var)}({var})') if var.shape[0] != 1: raise Exception(f'外部变量的数据形状有误{var.shape}(var)') var.index = index new_group_cst_var.append(var.iloc[:,[0]]) cst_var = pd.concat(new_group_cst_var,axis=1) cst_var.columns = group_cst_col return cst_var def get_boundary(group_opt_var:list) -> dict: boundary = {} # {'id1':boundary_info1 , 'id2':boundary_info2} boundary_insert = {} # {0:0, 1:2, 3:1} 0位置插入0条，1位置插入2条，3位置插入1条 print('-'*20+'边界条件'+'-'*20) pos_group_id = -1 for idx,bound_info in enumerate(group_opt_var): print('-'*20+ str(idx) +'-'*20) for k,v in bound_info.items(): v = v.iat[0,0] if isinstance(v,pd.DataFrame) else v print(f'{k} : {v} \n') if bound_info['id'] not in boundary.keys(): pos_group_id += 1 # 多个优化变量属于一个组内时（相同id），所有边界的信息以该组的id变量为准 boundary[bound_info['id']] = bound_info # boundary 仅保留唯一的id boundary_insert[pos_group_id] = 0 # 插入的位置：插入的数量 else: boundary_insert[pos_group_id] += 1 return boundary,boundary_insert def init_boundary(all_boundary) -> tuple: # 生成边界 lb = [] # 所有类型边界 ub = [] # 所有类型边界 var_type = [] # 所有类型边界 var_precis = [] # 区间类边界 collection_map = {} # 集合类边界/映射类边界 {'u1_name':{0:u1_a, 1:u1_b}} var_map = {} # 映射类边界 {'u1_name':{0:x1_a, 1:x1_b}} for boundary_id,boundary_info in all_boundary.items(): boundary_type = boundary_info.get('boundary_type') var_name = boundary_info.get('var_name') # 区间类型的边界 if boundary_type == 'interval': lb.append(boundary_info.get('lb')) ub.append(boundary_info.get('ub')) var_type.append(boundary_info.get('var_type')) var_precis.append(boundary_info.get('var_precis')) # 集合类型的边界 elif boundary_type == 'collection': # 序号对应的数值也应该从小到大排序 collection = boundary_info.get('collection') lb.append(0) ub.append(len(collection)-1) var_type.append(1) var_precis.append(0) collection_map[var_name] = dict(enumerate(collection)) # 映射类型边界 elif boundary_type == 'map': collection = boundary_info.get('collection') lb.append(0) ub.append(len(collection)-1) var_type.append(1) var_precis.append(0) collection_map[var_name] = dict(enumerate(collection)) var_map[var_name] = boundary_info.get('var_map') else: raise ValueError('边界类型有误') return lb,ub,var_type,var_precis,collection_map,var_map def result_append_init_sys_y(result:list,output_groups:dict,opt_obj:Opt): if '初始输出' not in output_groups.keys(): return result if len(output_groups['初始输出']) == 0: return result init_sys_y = [] for y_col in output_groups['初始输出']: #TODO 报错时提示 sys_y = opt_obj.current_system_y.loc[:,[y_col]] init_sys_y.append(sys_y) #TODO 放在最后可能不太合适，最好能指定 final_result = result + init_sys_y return final_result