AirTerminalDevice/_opt/algorithm/constrains/execute.py

import re 
import numpy as np
import pandas as pd

def execute_constrains(data:pd.DataFrame,cst:str) -> np.ndarray:
    
    #######################################################################################
    # 以下函数用于execute_constrains中使用                                                 #
    #######################################################################################
    
    def threshold_min(*data,threshold):
        # 原数据中，所有大于threshold的数的最小值
        data     = np.column_stack(data)
        data_rm  = np.where(data<=threshold,np.nan,data)
        data_min = np.nanmin(data_rm,axis=1)
        data_min = np.where(np.isnan(data_min),threshold,data_min)
        return data_min
    
    def max(*data):
        return np.max(data,axis=0)

    #######################################################################################
    #######################################################################################
    
    mod_vars = re.findall(r'{(\w+)}',cst)
    if len(mod_vars) > 0:
        for var in mod_vars:
            cst = cst.replace('{'+var+'}',f'data.{var}')
        
        try:
            cv = eval(cst)
        except Exception as E:
            print(f'执行该约束时有误：{cst}，以下为输入的数据')
            with pd.option_context('display.max_columns', None):
                print(data)
            raise E
        
        cv = np.array(cv)
    else:
        cv = data.eval(cst).to_numpy()
        
    cv = cv.reshape(-1,1)
    
    return cv 

if __name__ == '__main__':
    res1 = execute_constrains(
        pd.DataFrame({'x':[1,2,3],'y':[4,5,6]}),
        cst='abs(x-y)'
    )
    print(res1)