hz-application-ml/LIB/MIDDLE/CellStateEstimation/Uniform/V1_0_0/CBMSBatUniform.py

import pandas as pd
import numpy as np
import datetime
import bisect
import matplotlib.pyplot as plt
from LIB.MIDDLE.CellStateEstimation.Common import BatParam

class BatUniform:
    def __init__(self,sn,celltype,df_bms):  #参数初始化

        self.sn=sn
        self.celltype=celltype
        self.param=BatParam.BatParam(celltype)
        self.df_bms=df_bms
        self.packcrnt=df_bms['总电流[A]']*self.param.PackCrntDec
        self.packvolt=df_bms['总电压[V]']
        self.bms_soc=df_bms['SOC[%]']
        self.bmstime= pd.to_datetime(df_bms['时间戳'], format='%Y-%m-%d %H:%M:%S')

    def batuniform(self):
        if self.celltype==1 or self.celltype==2 or self.celltype==3 or self.celltype==4:
            df_res=self._ncm_uniform()
            return df_res
                     
        elif self.celltype==99:
            df_res=self._lfp_uniform()
            return df_res
        
        else:
            return pd.DataFrame()
    
    #定义滑动滤波函数........................................................................................................................................
    def _np_move_avg(self,a, n, mode="same"): 
        return (np.convolve(a, np.ones((n,)) / n, mode=mode))
    
    #寻找当前行数据的最小温度值................................................................................................................................
    def _celltemp_weight(self,num):   
        celltemp = []
        for j in range(1, self.param.CellTempNums+1):
            s = str(j)
            celltemp.append(self.df_bms.loc[num,'单体温度' + s])
        celltemp.remove(min(celltemp))
        self.celltemp=celltemp
        if self.celltype==99:
            if min(celltemp)>=20:
                self.tempweight=1
                self.StandardStandingTime=2400
            elif min(celltemp)>=10:
                self.tempweight=0.6
                self.StandardStandingTime=3600
            elif min(celltemp)>=5:
                self.tempweight=0.
                self.StandardStandingTime=3600
            else:
                self.tempweight=0.1
                self.StandardStandingTime=7200
        else:
            if min(celltemp)>=20:
                self.tempweight=1
                self.StandardStandingTime=1800
            elif min(celltemp)>=10:
                self.tempweight=0.8
                self.StandardStandingTime=2400
            elif min(celltemp)>=5:
                self.tempweight=0.6
                self.StandardStandingTime=3600
            else:
                self.tempweight=0.2
                self.StandardStandingTime=7200
        
    #获取当前行所有电压数据............................................................................................................................
    def _cellvolt_get(self,num):
        cellvolt=[]
        for j in range(1, self.param.CellVoltNums+1): 
            s = str(j)
            cellvolt.append(self.df_bms.loc[num,'单体电压' + s]/1000)
        return(cellvolt)

    #寻找DVDQ的峰值点，并返回..........................................................................................................................
    def _dvdq_peak(self, time, soc, cellvolt, packcrnt):
        cellvolt = self._np_move_avg(cellvolt, 3, mode="same")
        Soc = 0
        Ah = 0
        Volt = cellvolt[0]
        DV_Volt = []
        DQ_Ah = []
        DVDQ = []
        time1 = []
        soc1 = []
        soc2 = []
        xvolt=[]

        for m in range(1, len(time)):
            Step = (time[m] - time[m - 1]).total_seconds()
            Soc = Soc - packcrnt[m] * Step * 100 / (3600 * self.param.Capacity)
            Ah = Ah - packcrnt[m] * Step / 3600
            if (cellvolt[m]-Volt)>0.0019 and Ah>0:
                DQ_Ah.append(Ah)
                DV_Volt.append(cellvolt[m]-Volt)
                DVDQ.append((DV_Volt[-1])/DQ_Ah[-1])
                xvolt.append(cellvolt[m])
                Volt=cellvolt[m]
                Ah = 0
                soc1.append(Soc)
                time1.append(time[m])
                soc2.append(soc[m])

        #切片，去除前后10min的数据
        df_Data1 = pd.DataFrame({'time': time1,
                                'SOC': soc2,
                                'DVDQ': DVDQ,
                                'AhSoc': soc1,
                                'DQ_Ah':DQ_Ah,
                                'DV_Volt':DV_Volt,
                                'XVOLT':xvolt})
        start_time=df_Data1.loc[0,'time']
        start_time=start_time+datetime.timedelta(seconds=900)
        end_time=df_Data1.loc[len(time1)-1,'time']
        end_time=end_time-datetime.timedelta(seconds=1200)
        if soc2[0]<36:
            df_Data1=df_Data1[(df_Data1['SOC']>40) & (df_Data1['SOC']<80)]
        else:
            df_Data1=df_Data1[(df_Data1['time']>start_time) & (df_Data1['SOC']<80)]
        df_Data1=df_Data1[(df_Data1['XVOLT']>self.param.PeakVoltLowLmt) & (df_Data1['XVOLT']<self.param.PeakVoltUpLmt)]

        # print(packcrnt[int(len(time)/2)], min(self.celltemp))
        # ax1 = plt.subplot(3, 1, 1)
        # plt.plot(df_Data1['XVOLT'],df_Data1['DVDQ'],'g*-')
        # plt.xlabel('Volt/V')
        # plt.ylabel('DV/DQ')
        # plt.legend()
        # ax1 = plt.subplot(3, 1, 2)
        # plt.plot(df_Data1['SOC'],df_Data1['XVOLT'],'y*-')
        # plt.xlabel('SOC/%')
        # plt.ylabel('Volt/V')
        # plt.legend()
        # ax1 = plt.subplot(3, 1, 3)
        # plt.plot(df_Data1['SOC'], df_Data1['DVDQ'], 'r*-')
        # plt.xlabel('SOC/%')
        # plt.ylabel('DV/DQ')
        # plt.legend()
        # plt.show()

        if len(df_Data1)>2:     #寻找峰值点，且峰值点个数>2
            PeakIndex = df_Data1['DVDQ'].idxmax()
            df_Data2 = df_Data1[(df_Data1['SOC'] > (df_Data1['SOC'][PeakIndex] - 0.5)) & (df_Data1['SOC'] < (df_Data1['SOC'][PeakIndex] + 0.5))]
            if len(df_Data2) > 2 and df_Data1.loc[PeakIndex,'XVOLT']<self.param.PeakVoltUpLmt-0.019:
                return df_Data1['AhSoc'][PeakIndex]
            else:
                df_Data1 = df_Data1.drop([PeakIndex])
                PeakIndex = df_Data1['DVDQ'].idxmax()
                df_Data2 = df_Data1[(df_Data1['SOC'] > (df_Data1['SOC'][PeakIndex] - 0.5)) & (df_Data1['SOC'] < (df_Data1['SOC'][PeakIndex] + 0.5))]
                if len(df_Data2) > 2 and df_Data1.loc[PeakIndex,'XVOLT']<self.param.PeakVoltUpLmt-0.019:
                    return df_Data1['AhSoc'][PeakIndex]
                else:
                    return 0
        else:
            return 0
 
    #三元电池一致性计算.................................................................................................................................
    def _ncm_uniform(self):
        column_name=['time','sn','cellsoc_diff','cellvolt_diff','cellmin_num','cellmax_num']
        df_res=pd.DataFrame(columns=column_name)
        standingtime=0

        for i in range(1,len(self.df_bms)-2):

            if abs(self.packcrnt[i]) < 0.1 and abs(self.packcrnt[i-1]) < 0.1 and abs(self.packcrnt[i+1]) < 0.1:     #电流为0
                delttime=(self.bmstime[i]-self.bmstime[i-1]).total_seconds()
                standingtime=standingtime+delttime
                self._celltemp_weight(i)     #获取不同温度对应的静置时间

                if standingtime>self.StandardStandingTime:      #静置时间满足要求
                    if abs(self.packcrnt[i+2]) >= 0.1:
                        standingtime=0                    
                        cellvolt_now=self._cellvolt_get(i)     #获取当前行电压数据
                        cellvolt_min=min(cellvolt_now)
                        cellvolt_max=max(cellvolt_now)
                        if 3<cellvolt_min<4.5 and 3<cellvolt_max<4.5:
                            cellmin_num=cellvolt_now.index(cellvolt_min)+1
                            cellmax_num=cellvolt_now.index(cellvolt_max)+1
                            cellsoc_min=np.interp(cellvolt_min,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellsoc_max=np.interp(cellvolt_max,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellvolt_diff=(cellvolt_max-cellvolt_min)*1000
                            cellsoc_diff=cellsoc_max-cellsoc_min
                            cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                            cellvolt_diff=eval(format(cellvolt_diff,'.0f'))
                            df_res.loc[len(df_res)]=[self.bmstime[i], self.sn, cellsoc_diff, cellvolt_diff, cellmin_num, cellmax_num]
                    elif standingtime>3600*12:
                        standingtime=0                    
                        cellvolt_now=self._cellvolt_get(i)     #获取当前行电压数据
                        cellvolt_min=min(cellvolt_now)
                        cellvolt_max=max(cellvolt_now)
                        if 3<cellvolt_min<4.5 and 3<cellvolt_max<4.5:
                            cellmin_num=cellvolt_now.index(cellvolt_min)+1
                            cellmax_num=cellvolt_now.index(cellvolt_max)+1
                            cellsoc_min=np.interp(cellvolt_min,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellsoc_max=np.interp(cellvolt_max,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellvolt_diff=(cellvolt_max-cellvolt_min)*1000
                            cellsoc_diff=cellsoc_max-cellsoc_min
                            cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                            cellvolt_diff=eval(format(cellvolt_diff,'.0f'))
                            df_res.loc[len(df_res)]=[self.bmstime[i], self.sn, cellsoc_diff, cellvolt_diff, cellmin_num, cellmax_num]
                    elif i>=len(self.df_bms)-4:
                        standingtime=0
                        cellvolt_now=self._cellvolt_get(i)     #获取当前行电压数据
                        cellvolt_min=min(cellvolt_now)
                        cellvolt_max=max(cellvolt_now)
                        if 3<cellvolt_min<4.5 and 3<cellvolt_max<4.5:
                            cellmin_num=cellvolt_now.index(cellvolt_min)+1
                            cellmax_num=cellvolt_now.index(cellvolt_max)+1
                            cellsoc_min=np.interp(cellvolt_min,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellsoc_max=np.interp(cellvolt_max,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellvolt_diff=(cellvolt_max-cellvolt_min)*1000
                            cellsoc_diff=cellsoc_max-cellsoc_min
                            cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                            cellvolt_diff=eval(format(cellvolt_diff,'.0f'))
                            df_res.loc[len(df_res)]=[self.bmstime[i], self.sn, cellsoc_diff, cellvolt_diff, cellmin_num, cellmax_num]
                        break
                    else:
                        continue
                else:
                    continue
            else:
                standingtime=0
                continue

        if df_res.empty:    #返回计算结果
            return pd.DataFrame()
        else:
            return df_res

    #磷酸铁锂电池一致性计算.........................................................................................................................
    def _lfp_uniform(self):
        column_name=['time','sn','cellsoc_diff','cellvolt_diff','cellmin_num','cellmax_num']
        df_res=pd.DataFrame(columns=column_name)
        standingtime=0
        chrg_start=[]
        chrg_end=[]
        charging=0

        for i in range(3,len(self.df_bms)-3):

            #静置电压法计算电芯一致性
            if abs(self.packcrnt[i]) < 0.1 and abs(self.packcrnt[i-1]) < 0.1 and abs(self.packcrnt[i+1]) < 0.1:     #电流为0
                delttime=(self.bmstime[i]-self.bmstime[i-1]).total_seconds()
                standingtime=standingtime+delttime
                self._celltemp_weight(i)     #获取不同温度对应的静置时间

                if standingtime>self.StandardStandingTime:      #静置时间满足要求
                    cellvolt_now=self._cellvolt_get(i)     #获取当前行电压数据
                    cellvolt_min=min(cellvolt_now)
                    cellvolt_max=max(cellvolt_now)
                    if abs(self.packcrnt[i+2]) >= 0.1:     
                        standingtime=0  
                        if 2 < cellvolt_max < self.param.OcvInflexionBelow-0.002:                 
                            cellmin_num=cellvolt_now.index(cellvolt_min)+1
                            cellmax_num=cellvolt_now.index(cellvolt_max)+1
                            cellsoc_min=np.interp(cellvolt_min,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellsoc_max=np.interp(cellvolt_max,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellvolt_diff=(cellvolt_max-cellvolt_min)*1000
                            cellsoc_diff=cellsoc_max-cellsoc_min
                            cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                            cellvolt_diff=eval(format(cellvolt_diff,'.0f'))
                            df_res.loc[len(df_res)]=[self.bmstime[i], self.sn, cellsoc_diff, cellvolt_diff, cellmin_num, cellmax_num]
                    elif i>=len(self.df_bms)-4:
                        standingtime=0
                        if 2 < cellvolt_max < self.param.OcvInflexionBelow-0.002:
                            cellmin_num=cellvolt_now.index(cellvolt_min)+1
                            cellmax_num=cellvolt_now.index(cellvolt_max)+1
                            cellsoc_min=np.interp(cellvolt_min,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellsoc_max=np.interp(cellvolt_max,self.param.LookTab_OCV,self.param.LookTab_SOC)
                            cellvolt_diff=(cellvolt_max-cellvolt_min)*1000
                            cellsoc_diff=cellsoc_max-cellsoc_min
                            cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                            cellvolt_diff=eval(format(cellvolt_diff,'.0f'))
                            df_res.loc[len(df_res)]=[self.bmstime[i], self.sn, cellsoc_diff, cellvolt_diff, cellmin_num, cellmax_num]
                    else:
                        pass
                else:
                    pass
            else:
                standingtime=0
                pass   

            #获取DVDQ算法所需数据——开始............................................................................................................
            if charging==0: #判断充电开始
                if self.packcrnt[i]<=-1 and self.packcrnt[i+1]<=-1 and self.packcrnt[i+2]<=-1 and self.bms_soc[i]<40:     #充电开始
                    charging=1
                    if len(chrg_start)>len(chrg_end):
                        chrg_start[-1]=i
                    else:
                        chrg_start.append(i)
                else:
                    pass

            else: #充电中
                if (self.bmstime[i+1]-self.bmstime[i]).total_seconds()>180:  #如果充电过程中时间间隔>180s，则舍弃该次充电
                    chrg_start.remove(chrg_start[-1])
                    charging=0
                    continue
                elif self.packcrnt[i]<=-1 and self.packcrnt[i+1]<=-1 and  self.packcrnt[i+2]>-1:  #判断电流波动时刻
                    cellvolt_now=self._cellvolt_get(i+1)
                    if max(cellvolt_now)>self.param.CellFullChrgVolt:   #电压>满充电压
                        chrg_end.append(i+1)
                        charging=0
                        continue
                    else:
                        pass
                elif self.packcrnt[i+1]>-0.1 and self.packcrnt[i+2]>-0.1:   #判断充电结束
                    charging=0
                    if len(chrg_start)>len(chrg_end):
                        if self.bms_soc[i]>90:
                            chrg_end.append(i)
                        else:
                            chrg_start.remove(chrg_start[-1])
                            continue
                    else:
                        continue
                elif i==len(self.packcrnt)-4 and self.packcrnt[i+1]<-1 and self.packcrnt[i+2]<-1:
                    charging=0
                    if len(chrg_start)>len(chrg_end):
                        if self.bms_soc[i]>90:   #soc>90
                            chrg_end.append(i)
                            continue
                        else:
                            chrg_start.remove(chrg_start[-1])
                            continue
                    else:
                        continue
                else:
                    continue   

        if chrg_end:    #DVDQ方法计算soc差
            peaksoc_list=[]
            for i in range(len(chrg_end)):
                peaksoc_list = []
                self._celltemp_weight(chrg_start[i])
                if min(self.celltemp)>10:
                    for j in range(1, self.param.CellVoltNums + 1):
                        s = str(j)
                        cellvolt = self.df_bms['单体电压' + s]/1000
                        cellvolt = list(cellvolt[chrg_start[i]:chrg_end[i]])
                        time = list(self.bmstime[chrg_start[i]:chrg_end[i]])
                        packcrnt = list(self.packcrnt[chrg_start[i]:chrg_end[i]])
                        soc = list(self.bms_soc[chrg_start[i]:chrg_end[i]])
                        peaksoc = self._dvdq_peak(time, soc, cellvolt, packcrnt)
                        if peaksoc>1:
                            peaksoc_list.append(peaksoc)    #计算到达峰值点的累计Soc
                        else:
                            pass
                    if len(peaksoc_list)>self.param.CellVoltNums/2:
                        peaksoc_max=max(peaksoc_list)
                        peaksoc_min=min(peaksoc_list)
                        peaksoc_maxnum=peaksoc_list.index(peaksoc_min)+1
                        peaksoc_minnum=peaksoc_list.index(peaksoc_max)+1
                        cellsoc_diff=peaksoc_max-peaksoc_min
                        cellsoc_diff=eval(format(cellsoc_diff,'.1f'))
                        df_res.loc[len(df_res)]=[self.bmstime[chrg_start[i]], self.sn, cellsoc_diff, 0, peaksoc_minnum, peaksoc_maxnum]
                    else:
                        pass
                    plt.show()
                else:
                    pass

        if df_res.empty:
            return pd.DataFrame()
        else:
            df_res.sort_values(by='time', ascending=True, inplace=True)
            return df_res