zlwl-algos-new/OFFLINEALGO/lib/algos.py

import pandas as pd
import time
import datetime
import traceback

from ZlwlAlgos.PERIODIC.task_min_5.group_1.safetyalarm.V_1_0_0.CBMSSafetyAlarm import SafetyAlarm
from ZlwlAlgos.PERIODIC.task_min_10.group_1.FaultWarning.V_1_0_0 import CBMSBatDiag, CBMSBatDiag_SOCCrntIsnV1, CBMSBatDiag_TempV4
from ZlwlAlgos.PERIODIC.task_hour_1.group_2.safety_score.V1_0_0 import Safety_Score
from ZlwlAlgos.PERIODIC.task_hour_6.group_1.intershort.V_1_0_0.CBMSBatInterShort import BatInterShort
from ZlwlAlgos.PERIODIC.task_hour_6.group_1.uniform.V_1_0_0.CBMSBatUniform import CellUniform
from ZlwlAlgos.PERIODIC.task_hour_6.group_1.safetywarning.V_1_0_0.CBMSSafetyWarning import SafetyWarning

from ZlwlAlgos.PERIODIC.task_day_1.group_1.socdiag.V_1_0_0.SOCBatDiag import SocDiag
from ZlwlAlgos.PERIODIC.task_day_1.group_1.LowSocAlarm.V1_0_0.low_soc_alarm import Low_soc_alarm 
from ZlwlAlgos.PERIODIC.task_day_1.group_1.SorCal.V_1_0_0.sorcal import sor_est  

from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_status as ds  ##充电状态标准化程序
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_split  as dt ##分段函数程序
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_drive_stat as ddt ##行驶数据按行驶段汇总统计
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_charge_stat as dct ##充电数据按充电段汇总
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_stand_stat as dst ##静置数据按静置段汇总
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import data_drive_stat_period as ddtp ##行驶数据按充电周期汇总统计
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import trans_day  as trd  ##解决跨天的问题
from ZlwlAlgos.PERIODIC.task_day_1.group_1.DataSplit.V_1_0_0  import stand_status as ss

from ZlwlAlgos.PERIODIC.task_day_7.group_1.soh.V_1_0_0.CBMSBatSoh import BatSoh 

 
import lib.glo as glo

class AlgosInvoke():
    
    def __init__(self, logger, df_algo_param, df_table, cellvolt_name, celltemp_name, pack_code, df_snlist, df_algo_adjustable_param, df_algo_pack_param, cell_type):
        self.logger = logger
        self.df_algo_param = df_algo_param
        self.df_table = df_table
        self.cellvolt_name = cellvolt_name
        self.cellvolt_name = cellvolt_name
        self.celltemp_name = celltemp_name
        self.pack_code = pack_code
        self.df_snlist = df_snlist
        self.df_algo_adjustable_param = df_algo_adjustable_param
        self.df_algo_pack_param = df_algo_pack_param
        self.df_algo_pack_param_eval = {k: eval(v) if isinstance(v, str) else v for k, v in df_algo_pack_param.items()}

        self.cell_type = cell_type
        
    def _process_ing_done(self, df_all_fault_ing, df_all_fault_done, df_res_new, df_res_end, df_res_update=None, update_flag=False):
            # 处理新增
        df_all_fault_ing = pd.concat([df_all_fault_ing, df_res_new])
        df_all_fault_ing.reset_index(drop=True, inplace=True)
        # 处理update
        if update_flag:
            upd_list = df_res_update.to_dict("records")
            for u in upd_list:
                index1=df_all_fault_ing[(df_all_fault_ing['sn']==u['sn']) & (df_all_fault_ing['start_time']==u['start_time']) & (df_all_fault_ing['fault_code']==u['fault_code'])].index[0]
                df_all_fault_ing.drop(index1, inplace=True)
                
                index2=df_res_update[(df_res_update['sn']==u['sn']) & (df_res_update['start_time']==u['start_time']) & (df_res_update['fault_code']==u['fault_code'])].index[0]
                df_all_fault_ing = pd.concat([df_all_fault_ing, df_res_update.loc[index2:index2]])
                
                df_all_fault_ing.reset_index(drop=True, inplace=True)
        # 处理delete
        del_list = df_res_end.to_dict("records")
        for d in del_list:
            index1=df_all_fault_ing[(df_all_fault_ing['sn']==d['sn']) & (df_all_fault_ing['start_time']==d['start_time']) & (df_all_fault_ing['fault_code']==d['fault_code'])].index[0]
            df_all_fault_ing.drop(index1, inplace=True)
            
            index2=df_res_end[(df_res_end['sn']==d['sn']) & (df_res_end['start_time']==d['start_time']) & (df_res_end['fault_code']==d['fault_code'])].index[0]
            df_all_fault_done = pd.concat([df_all_fault_done, df_res_end.loc[index2:index2]])
            
            df_all_fault_ing.reset_index(drop=True, inplace=True)
            df_all_fault_done.reset_index(drop=True, inplace=True)
        return df_all_fault_ing, df_all_fault_done
    def safety_alarm(self, df_data, df_all_fault_ing, df_all_fault_done):
        try:
            algo_name = 'safety_alarm'
            self.logger.info(f"开始执行{algo_name}算法")
            # 变量定义
            df_bms_ram = pd.DataFrame(columns=['time', 'sn', 'packvolt', 'cellvolt', 'celltemp', 'packsoc','packcrnt'])
            df_alarm_ram = pd.DataFrame(columns=['time','sn','safetywarning1','safetywarning2'])
            df_diag_ram = pd.DataFrame(columns=['start_time','end_time','sn','imei','model','fault_level','fault_code','fault_info','fault_reason','fault_advice','fault_location','device_status','odo'])
            
            # 执行算法
            safety_alarm = SafetyAlarm(df_data, df_diag_ram, df_bms_ram, df_alarm_ram, self.df_algo_param, self.df_table, self.cellvolt_name, self.celltemp_name, self.pack_code, self.df_snlist)
            df_res_new, df_res_end, df_bms_ram, df_alarm_ram = safety_alarm.safety_alarm_diag()
            # 处理结果
            df_all_fault_ing, df_all_fault_done = self._process_ing_done(df_all_fault_ing, df_all_fault_done, df_res_new, df_res_end)
            self.logger.info(f"{algo_name}算法执行完成")
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_all_fault_ing, df_all_fault_done
        
    
    def fault_warning(self, df_data, df_all_fault_ing, df_all_fault_done):
        algo_name = 'fault_warning'
        self.logger.info(f"开始执行{algo_name}算法")
        #电芯电压温度故障诊断

        try:
            df_res_new_volt=pd.DataFrame()
            df_res_update_volt=pd.DataFrame()
            df_res_end_volt=pd.DataFrame() 
            FaultDiagVolt = CBMSBatDiag.BatDiagVolt(df_data, self.df_table, self.cellvolt_name, self.celltemp_name, self.cell_type)
            df_res_new_volt, df_res_update_volt, df_res_end_volt = FaultDiagVolt.diag(df_all_fault_ing, self.df_algo_adjustable_param, self.df_algo_pack_param, self.df_algo_param, self.df_snlist)#df_soh,                            
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())

        try:    
            df_res_new_crnt=pd.DataFrame()
            df_res_update_crnt=pd.DataFrame()
            df_res_end_crnt=pd.DataFrame()     
            FaultDiagTemp = CBMSBatDiag_SOCCrntIsnV1.BatDiag_SOCCrntIsn(df_data, self.df_table, self.cellvolt_name, self.celltemp_name)
            df_res_new_crnt, df_res_update_crnt, df_res_end_crnt = FaultDiagTemp.diag(df_all_fault_ing, self.df_algo_adjustable_param, self.df_algo_pack_param, self.df_algo_param, self.df_snlist)
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        try:
            df_res_new_temp=pd.DataFrame()
            df_res_update_temp=pd.DataFrame()
            df_res_end_Temp=pd.DataFrame()     
            FaultDiagTemp = CBMSBatDiag_TempV4.BatDiag(df_data, self.df_table, self.celltemp_name, self.cellvolt_name, self.pack_code)
            df_res_new_temp, df_res_update_temp, df_res_end_Temp = FaultDiagTemp.diag(df_all_fault_ing, self.df_algo_adjustable_param, self.df_algo_pack_param, self.df_algo_param, self.df_snlist)
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        try:
            df_res_new = pd.concat([df_res_new_volt, df_res_new_crnt, df_res_new_temp]) #, res1
            df_res_update=pd.concat([df_res_update_volt,df_res_update_crnt, df_res_update_temp]) #, res1
            df_res_end = pd.concat([df_res_end_volt,df_res_end_crnt, df_res_end_Temp]) #, res2
            df_res_new.reset_index(drop=True, inplace=True)
            df_res_update.reset_index(drop=True, inplace=True)
            df_res_end.reset_index(drop=True, inplace=True)
           
           # 处理结果
            df_all_fault_ing, df_all_fault_done = self._process_ing_done(df_all_fault_ing, df_all_fault_done, df_res_new, df_res_end, df_res_update, update_flag=True)
            self.logger.info(f"{algo_name}算法执行完成")
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_all_fault_ing, df_all_fault_done
    
    def safety_score(self, df_r_risk_model, df_r_risk_list, df_dept, df_allfltinfoing, df_allfltinfodone, organ_code):
        try:
            algo_name = 'safety_score'
            self.logger.info(f"开始执行{algo_name}算法")
            df_safety_score_h = pd.DataFrame()
            df_safety_score_d = pd.DataFrame()
            df_safety_core_hour = pd.DataFrame(columns=["create_time", "sn", "score_h", "risk_index"])
            SafetyScole=Safety_Score.SafetyScore(self.df_snlist, df_allfltinfoing, df_allfltinfodone, df_safety_core_hour, self.df_algo_param, df_r_risk_model,
                                                 df_r_risk_list, df_dept, organ_code)
            df_safety_score_h, df_safety_score_d=SafetyScole.safety_score() 
            self.logger.info(f"{algo_name}算法执行完成")
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_safety_score_h, df_safety_score_d
    
    def bat_intershort_cell_uniform_safety_warning(self, df_data, df_all_fault_ing, df_all_fault_done, df_soh):
        try:
            algo_name = 'bat_intershort_cell_uniform_safety_warning'
            self.logger.info(f"开始执行{algo_name}算法")
            df_all_short = pd.DataFrame()
            df_all_uniform = pd.DataFrame()
            BatShort = BatInterShort(self.cell_type, self.df_algo_pack_param, self.celltemp_name, self.cellvolt_name)
            BatUniform = CellUniform(self.cell_type, self.df_algo_pack_param, self.celltemp_name, self.cellvolt_name)     
            BatWarning = SafetyWarning(self.df_algo_adjustable_param, self.df_algo_param, self.cellvolt_name, self.pack_code, self.df_snlist)
            for sn, df in df_data.groupby('sn'):
                df_short, df_ram_res, df_ram_res1, df_ram_res2, df_ram_res3, df_lfp_ram, df_bal_ram=BatShort.intershort(
                    sn, df, df_soh, pd.DataFrame(columns=['sn', 'time', 'deltsoc', 'cellsoc']), pd.DataFrame(columns=['sn', 'time1', 'deltsoc1']), 
                    pd.DataFrame(columns=['sn', 'time2', 'delt_volt', 'packcrnt_avg', 'temp_avg']), pd.DataFrame(columns=['sn','time3','standingtime','standingtime1','standingtime2','standingtime3']), [],
                    pd.DataFrame(columns=['sn', 'time', 'bal_time', 'bal_time1', 'bal_time2']))
                df_uniform, df_ram_res3=BatUniform.batuniform(sn, df, df_ram_res3)
                df_new, df_update, df_end = BatWarning.warning_diag(sn, df, df_short, df_uniform, pd.DataFrame(columns=['start_time','end_time','sn','imei','model','fault_level','fault_code','fault_info','fault_reason','fault_advice','fault_location','device_status','odo']))
                df_all_short = pd.concat([df_all_short, df_short])
                df_all_uniform = pd.concat([df_all_uniform, df_uniform])
                # 处理结果
                df_all_fault_ing, df_all_fault_done = self._process_ing_done(df_all_fault_ing, df_all_fault_done, df_new, df_end, df_update, update_flag=True)
            self.logger.info(f"{algo_name}算法执行完成")
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_all_short, df_all_uniform, df_all_fault_ing, df_all_fault_done
    
    def soc_diag(self, df_data, df_all_fault_ing, df_all_fault_done):
        try:
            algo_name = 'soc_diag'
            self.logger.info(f"开始执行{algo_name}算法")
            df_diag_ram = df_all_fault_ing[df_all_fault_ing['sn'].isin(self.df_snlist['sn'].tolist())]
            period = 24*60 #算法周期min
            end_time = str(df_data['time'].tolist()[-1])
            soc_diag = SocDiag(self.cell_type, self.df_algo_pack_param_eval, self.df_algo_adjustable_param, self.df_algo_param, end_time, period, self.pack_code, self.df_snlist, df_data)
            df_res_new_C109, df_res_end_C109= soc_diag.soc_block(df_diag_ram)
            df_res_end_C107 = soc_diag.soc_jump()

            df_res_new_soc = df_res_new_C109
            df_res_end_soc = pd.concat([df_res_end_C107, df_res_end_C109])
            df_all_fault_ing, df_all_fault_done = self._process_ing_done(df_all_fault_ing, df_all_fault_done, df_res_new_soc, df_res_end_soc)
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_all_fault_ing, df_all_fault_done
    def low_soc(self, df_data, df_all_fault_ing, df_all_fault_done):
        try:
            algo_name = 'low_soc'
            self.logger.info(f"开始执行{algo_name}算法")
            df_diag_ram = df_all_fault_ing[df_all_fault_ing['sn'].isin(self.df_snlist['sn'].tolist())]
            low_soc_warning = Low_soc_alarm(df_data, self.cellvolt_name)
            df_res_new_lw_soc, df_res_update_lw_soc,df_res_end_lw_soc= low_soc_warning.diag(
                self.df_algo_pack_param_eval, self.df_algo_param, self.df_algo_adjustable_param, df_data, self.df_table, df_diag_ram, self.df_snlist)
            df_all_fault_ing, df_all_fault_done = self._process_ing_done(df_all_fault_ing, df_all_fault_done, df_res_new_lw_soc, df_res_end_lw_soc, df_res_update_lw_soc, True)

        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_all_fault_ing, df_all_fault_done
    def sor(self, df_data):
        try:
            algo_name = 'sor'
            self.logger.info(f"开始执行{algo_name}算法")
            Diagsor_temp = sor_est(df_data, self.df_algo_pack_param)#计算内阻
            df_sor_add = Diagsor_temp.sor_cal()
            df_sor_add.reset_index(drop = True, inplace = True)
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_sor_add
    # def li_plted(self, df_data):
    #     try:
    #         algo_name = 'li_plted'
    #         self.logger.info(f"开始执行{algo_name}算法")
    #         Diagsor_temp = sor_est(df_data, self.df_algo_pack_param)#计算内阻
    #         df_sor_add = Diagsor_temp.sor_cal()  
    #     except Exception as e:
    #         self.logger.error(str(e))
    #         self.logger.error(traceback.format_exc())      
    def data_split(self, df_merge):
        try:
            algo_name = 'data_split'
            self.logger.info(f"开始执行{algo_name}算法")
            df_merge=ds.data_status(df_merge,c_soc_dif_p=0.05,s_soc_dif_p=0,c_order_delta=1200,s_order_delta=300)
            ##基于各个状态码，进行分段，分段函数
            df_drive,df_charge,df_stand,df_data_split_rlt=dt.split(df_merge, self.celltemp_name, self.cellvolt_name, 
                    drive_interval_time_min=1200,charge_interval_time_min=1200,stand_interval_time_min=1200,single_num_min=3,drive_sts=3,charge_sts=[21,22],stand_sts=0)                    
            df_data_split_rlt.reset_index(drop = True, inplace = True)
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_data_split_rlt
    
    def soh(self, df_data):
        try:
            algo_name = 'soh'
            self.logger.info(f"开始执行{algo_name}算法")
            df_ram_soh_result = pd.DataFrame(columns=['time_st','time_sp','sn','method','soh','soh_real','cellsoh_diff','cellsoh','odo'])
            bat_soh = BatSoh(self.cell_type, df_data, df_ram_soh_result, self.df_algo_pack_param_eval, self.cellvolt_name, self.celltemp_name, self.pack_code)
            df_soh_res = bat_soh.batsoh(self.df_snlist)
            df_soh_res.index = list(range(len(df_soh_res)))
        except Exception as e:
            self.logger.error(str(e))
            self.logger.error(traceback.format_exc())
        return df_soh_res