AirTerminalDevice/tools/optimizer.py

import os
import sys
import time 
import warnings
from contextlib import contextmanager

import pandas as pd

from ..model._base._base_device import BaseDevice
from .._opt.algorithm.sim_config import simulate_config as sim_opt_config
from .._opt.algorithm.main import main as main_opt
from .._opt.boundary.sim_config import simulate_config as sim_config_bound
from .._opt.boundary.main import main as main_bound
from .._opt.algorithm.model.model import SystemModel

def optimizer(
    model           : BaseDevice,
    opt_var_boundary: dict,
    opt_var_value   : pd.DataFrame,
    oth_var_value   : pd.DataFrame,
    target          : str,
    target_min      : bool,
    constrains      : list,
    other_kwargs    : dict = None,
    logging         : bool = True
):
    if other_kwargs is None:
        other_kwargs = {}
    
    var_name_opt  = []
    boundary_info = {}
    for var_name,var_info in opt_var_boundary.items():
        var_name_opt.append(var_name)
        boundary_info[var_name] = main_bound(
            None,None,None,
            opt_var_value.iloc[[0],:].loc[:,[var_name]],
            config = sim_config_bound(
                opt_var    = [var_name],
                syn_opt    = False,
                var_type   = True,
                lb_static  = var_info['lb'],
                ub_static  = var_info['ub'],
                var_precis = var_info.get('precis',1)
            )
        )
    var_name_oth = []
    oth_var_info = {}
    for var_name in oth_var_value.columns:
        var_name_oth.append(var_name)
        oth_var_info[var_name] = oth_var_value.iloc[[0],:].loc[:,[var_name]]

    opt_config = sim_opt_config(
        target        = target,
        dir_min       = target_min,
        var_opt       = var_name_opt,
        var_sys       = var_name_oth,
        diag_model    = False,
        algorithm     = 'soea_DE_best_1_L_templet',
        NIND          = other_kwargs.get('NIND',1000),
        MAXGEN        = other_kwargs.get('MAXGEN',100),
        constrains    = constrains,
        allow_neg_opt = False,
    )
    
    with suppress_output(show=logging):
        opt_output  = main_opt(
            *list(boundary_info.values()),
            *list(oth_var_info.values()),
            system_model = AirSystem(model=model),
            config       = opt_config
        )
    return {
        'opt_summary': opt_output[:3],
        'opt_var'    : opt_output[3:],
    }

class AirSystem(SystemModel):
    def __init__(self,model):
        super().__init__()
        self.model = model
    
    def predict(self,data:pd.DataFrame) -> pd.DataFrame:
        time_start   = time.time()
        self.PENALTY = {}
        self.index   = data.index
        with warnings.catch_warnings():
            warnings.simplefilter("ignore",category=RuntimeWarning)
            sys_out = self.model.predict_system(data)
        time_end  = time.time()
        past_time = round(time_end-time_start,2)
        self.PAST_TIME.append(past_time)
        # print(f'第{len(self.PAST_TIME)}次调用系统模型，本次调用时长为:{past_time}秒 \n')    
        return sys_out

@contextmanager
def suppress_output(show=True):
    """
    上下文管理器，根据参数控制是否显示输出
    
    参数:
        show (bool): 如果为True则显示输出，False则隐藏输出
    """
    if not show:
        original_stdout = sys.stdout
        sys.stdout = open(os.devnull, 'w')
    
    try:
        yield
    finally:
        if not show:
            sys.stdout = original_stdout