import pandas as pd import numpy as np import datetime import bisect import matplotlib.pyplot as plt import BatParam class BatUniform: def __init__(self,sn,celltype,df_bms,df_volt,df_temp,df_accum): #参数初始化 self.sn=sn self.celltype=celltype CellVoltNums=int(df_volt.loc[5,'单体电池总数']) if CellVoltNums==110: self.celltype=99 else: self.celltype==100 self.param=BatParam.BatParam(self.celltype) self.df_volt=df_volt self.df_temp=df_temp self.packcrnt=(df_volt['可充电储能装置电流(A)'].astype('float'))*self.param.PackCrntDec self.packvolt=df_volt['可充电储能装置电压(V)'].astype('float') self.bms_soc=df_bms['SOC'] # self.bms_soh=df_volt['SOH[%]'] self.bmstime= pd.to_datetime(df_volt['上报时间'], format='%Y-%m-%d %H:%M:%S') self.param.CellVoltNums=CellVoltNums self.param.CellTempNums=int(df_temp.loc[5,'可充电储能温度探针个数']) def batuniform(self): if self.celltype==1 or self.celltype==2 or self.celltype==3 or self.celltype==4 or self.celltype==100: 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_temp.loc[num, s+'.0']) celltemp.remove(min(celltemp)) self.celltemp=celltemp if self.celltype==99: if min(celltemp)>=20: self.tempweight=1 self.StandardStandingTime=1800 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_volt.loc[num,s+'.0']) 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']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'] (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.StandardStandingTime: #静置时间满足要求 if abs(self.packcrnt[i+1]) >= 0.1: standingtime=0 cellvolt_now=self._cellvolt_get(i) #获取当前行电压数据 cellvolt_last=self._cellvolt_get(i-1) cellvolt_min=min(cellvolt_now) cellvolt_max=max(cellvolt_now) cellvolt_maxlast=max(cellvolt_last) if 33600*12: standingtime=0 cellvolt_now=self._cellvolt_get(i) #获取当前行电压数据 cellvolt_last=self._cellvolt_get(i-1) cellvolt_min=min(cellvolt_now) cellvolt_max=max(cellvolt_now) cellvolt_maxlast=max(cellvolt_last) if 3=len(self.df_volt)-4: standingtime=0 cellvolt_now=self._cellvolt_get(i) #获取当前行电压数据 cellvolt_last=self._cellvolt_get(i-1) cellvolt_min=min(cellvolt_now) cellvolt_max=max(cellvolt_now) cellvolt_maxlast=max(cellvolt_last) if 3self.StandardStandingTime: #静置时间满足要求 cellvolt_now=self._cellvolt_get(i) #获取当前行电压数据 cellvolt_last=self._cellvolt_get(i-1) cellvolt_min=min(cellvolt_now) cellvolt_max=max(cellvolt_now) cellvolt_maxlast=max(cellvolt_last) if abs(self.packcrnt[i+1]) >= 0.1 and abs(cellvolt_maxlast-cellvolt_max)<0.003: standingtime=0 if 2 < cellvolt_max < self.param.OcvInflexionBelow-0.02: 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_volt)-4: standingtime=0 if 2 < cellvolt_max < self.param.OcvInflexionBelow-0.02: 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 float(self.bms_soc[i].strip('%'))<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 or (self.packcrnt[i]>self.param.Capacity/3 and self.packcrnt[i+1]>self.param.Capacity/3): #如果充电过程中时间间隔>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 float(self.bms_soc[i].strip('%'))>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 float(self.bms_soc[i].strip('%'))>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_volt[s+'.0'] # 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