AirTerminalDevice/_opt/algorithm/opt_alg.py

import numpy as np
import pandas as pd 
import geatpy as ea 

from ._utils.config_info import ConfigInfo
from .opt_obj import Opt

All_Algorithm = {
    'soea_DE_rand_1_L_templet':{
        'templet': ea.soea_DE_rand_1_L_templet,
        'param'  : {'F':0.6,'Cr':0.85,'L':15}
    },
    'soea_DE_currentToBest_1_L_templet':{
        'templet': ea.soea_DE_currentToBest_1_L_templet,
        'param'  : {'F':0.6,'Cr':0.85,'L':15}
    },
    'soea_DE_best_1_L_templet':{
        'templet': ea.soea_DE_best_1_L_templet,
        'param'  : {'F':0.5,'Cr':0.8,'L':15}
    },
}

def main(opt_obj:Opt,allow_neg_opt:bool,config=None) -> list:
    
    config_info = ConfigInfo(config)
    
    # 组件配置参数检查
    config_info.check_property_exist({'NIND':'种群数量','MAXGEN':'迭代次数'})
    
    # 种群参数
    Encoding   = 'RI'
    NIND       = config['NIND']
    Field      = ea.crtfld(
        Encoding   = Encoding,
        varTypes   = opt_obj.varTypes,
        ranges     = opt_obj.ranges,
        borders    = opt_obj.borders,
        precisions = opt_obj.var_precis
    )
    population = ea.Population(Encoding,Field,NIND)

    # 算法参数
    templete = All_Algorithm[config['algorithm']]['templet']
    param    = All_Algorithm[config['algorithm']]['param']
    
    Algorithm         = templete(opt_obj,population)
    Algorithm.MAXGEN  = config['MAXGEN']       # 复杂问题需增加代数。
    Algorithm.F       = param['F']             # 0.5~0.8 增大 F 增强全局搜索，但可能降低收敛速度。
    Algorithm.Cr      = param['Cr']            # 0.8~1.0 高 Cr 适合强相关变量，低 Cr 增强多样性。
    Algorithm.L       = param['L']             # 5~20    变量相关性越强，L 应越大。高维问题可设为变量维度的 10%~20%。
    Algorithm.logTras = int(config['MAXGEN']/10)
    Algorithm.verbose = False
    Algorithm.drawing = 0

    [bestindi,pop] = Algorithm.run()
    
    # 检查是否有可行解
    #TODO 设定没有可行解时取当前解
    if len(bestindi) == 0:
        print_cv(pop.CV,opt_obj.constrains,opt_obj.cur_sys_constrains_cfl)
        raise Exception('没有可行解')
    
    opt_system_x            = get_opt_system_x(opt_obj,bestindi)                # 完整的系统模型输入
    opt_var                 = opt_system_x.loc[:,opt_obj.name_opt_vars]         # 优化变量
    sys_before_after_output = get_sys_output_before_after(opt_obj,opt_system_x) # 优化前后系统模型输出
    opt_var_before_after    = get_opt_var_before_after(opt_obj,opt_var)         # 优化前后系统模型输入（优化变量）
    
    # 打印优化后的结果对比
    alg_log = print_opt_result(Algorithm,bestindi,opt_obj,sys_before_after_output,opt_var_before_after)
    
    # 检查本次优化是否为负优化
    check_neg_opt(opt_obj,sys_before_after_output,allow_neg_opt)
    
    # 组件输出并校验
    process_list = [alg_log,opt_var_before_after,sys_before_after_output]
    result_list  = process_list + config_info.split_df(method='id',df=opt_var,by_group='优化变量')
    
    return result_list
    

def print_opt_result(
    Algorithm,
    bestindi,
    opt_obj                : Opt,
    sys_before_after_output: pd.DataFrame,
    opt_var_before_after   : pd.DataFrame
) -> pd.DataFrame:
    
    print('+'*20+' 优化结果 '+'+'*20)
    
    print('\n算法迭代过程：')
    with pd.option_context('display.max_rows', None):
        with pd.option_context('display.max_columns', None):
            alg_log = pd.DataFrame(Algorithm.log).round(2)
            print(alg_log)
    
        print(f'\n评价次数 : {Algorithm.evalsNum}')
        print(f'所用时间 : {round(Algorithm.passTime,2)}秒')
        print(f'最优的目标函数值（{opt_obj.opt_target}） : {round(bestindi.ObjV[0][0],4)}')
    
        # print('\n最后一次迭代的系统模型输出：')
        # print(opt_obj.data_y.round(4).drop_duplicates().T)
        
        print('\n优化前后控制变量的对比：')
        print(opt_var_before_after)
    
        print('\n优化前后系统模型输出的结果对比：')
        # 优化后系统模型的结果不唯一，取离优化目标最接近的结果
        print(sys_before_after_output)
        
        print('\n系统模型的初始输入违背约束条件的情况：')
        print_sys_init_constrains(opt_obj.cur_sys_constrains_cfl)
    
    print(' \n系统模型调用时长统计：')
    opt_obj.system_model.summary_past_time()
    
    print('\n'+'+'*50)
    
    return alg_log

def print_cv(
    cv                    : np.ndarray,
    constrains            : list,
    cur_sys_constrains_cfl: dict
) -> None:
    
    #TODO 增加每一次迭代的CV结果
    
    # 最后一次迭代的CV
    print('每一个约束条件最后一次迭代的CV(大于0表明不满足约束条件):')
    last_cv = [
        cv.mean(axis=0),
        cv.max(axis=0),
        cv.min(axis=0),
        (cv<0).sum(axis=0),
        (cv<0).mean(axis=0),
    ]
    last_cv = pd.DataFrame(
        data=np.vstack(last_cv).T,
        columns=['Mean_CV','Max_CV','Min_CV','Count_CV<0','Pct_CV<0']
    )
    print(last_cv)
    
    #TODO 联合各个约束条件检查CV
    
    print('\n对应以下约束条件：')
    for idx,cst in enumerate(constrains):
        print(f'{idx} : {cst}<0')
    
    print('\n系统模型的初始输入违背约束条件的情况：')
    print_sys_init_constrains(cur_sys_constrains_cfl)

def get_opt_system_x(opt_obj:Opt,bestindi) -> pd.DataFrame:
    # 优化变量的优化结果
    data = opt_obj.insert_sync_var(bestindi.Phen.reshape(1,-1))
    opt_var = pd.DataFrame(
        data    = data,
        columns = opt_obj.name_opt_vars,
        index   = opt_obj.index
    )
    system_x = pd.concat([opt_var,opt_obj.sys_var,opt_obj.oth_var],axis=1)
    # 将优化结果中映射边界的对应的变量进行还原
    system_x = opt_obj.restore_data(system_x)
    return system_x

def get_sys_output_before_after(opt_obj:Opt,system_x):
    # 优化前后，系统模型输出的变化
    opt_system_y   = opt_obj.system_model.predict(system_x)
    
    sys_before_after_output = (
        pd.concat([
            opt_obj.current_system_y,
            opt_system_y
        ],axis=0)
        .transpose()
        .set_axis(['Before','After'],axis=1)
        .assign(
            Diff     = lambda dt:dt.After - dt.Before,
            Diff_pct = lambda dt:((dt.After - dt.Before)/dt.Before * 100).round(2).astype('str') + '%'
        )
        .round(2)
    )
    return sys_before_after_output

def get_opt_var_before_after(opt_obj,opt_var):
    # 优化前后，优化变量的变化
    result_opt = (
        pd.concat([opt_obj.opt_var,opt_var],axis=0)
        .reset_index(drop=True)
        .transpose()
        .set_axis(['Before','After'],axis=1)
        .assign(
            Diff     = lambda dt:dt.After - dt.Before,
            Diff_pct = lambda dt:((dt.After - dt.Before)/dt.Before * 100).round(2).astype('str') + '%'
        )
        .round(2)
    )
    return result_opt


def check_neg_opt(opt_obj,sys_before_after_output,allow_neg_opt):
    if allow_neg_opt is None:
        allow_neg_opt = False
    
    if allow_neg_opt is True:
        return None
    if not isinstance(allow_neg_opt,bool):
        raise Exception(f'判断是否需要校验负优化的输入不是布尔值({type(allow_neg_opt)})')
    
    # 检查本次优化是否为负优化
    raw_pred = sys_before_after_output.at[opt_obj.opt_target,'Before']
    opt_pred = sys_before_after_output.at[opt_obj.opt_target,'After']
    
    neg_min_opt = (opt_obj.maxormins[0] == 1) and (opt_pred > raw_pred)
    neg_max_opt = (opt_obj.maxormins[0] == -1) and (opt_pred < raw_pred)
    
    if neg_min_opt or neg_max_opt:
        raise Exception(f'出现负优化，当前值为{raw_pred}，优化后为{opt_pred}')
    
    return None

def print_sys_init_constrains(cur_sys_constrains_cfl:dict):
    # 系统模型的初始输入违背约束条件的情况
    for cst_type,var_message in cur_sys_constrains_cfl.items():
        # 约束类型（边界/可行性）
        print(cst_type.upper())
        
        if isinstance(var_message,dict):
            for var_name,message in var_message.items():
                print(var_name+' : '+message)
        
        if isinstance(var_message,list):
            for message in var_message:
                print(message)