zlwl-algos-new/USER/shouxueqi/hz_datacleaning/hz_data_clean.py

import os
import pandas as pd
import numpy as np





def clean_dead_value(series, num_dead_thresh):
    slide_list = [series.index[0]]
    slide_list_all = []
    for i in range(series.index[0],series.index[-1]):
        j = i + 1
        diff = series[j] - series[i]
        if diff == 0:
            slide_list.append(j)
        else:
            slide_list.clear()
            slide_list.append(j)
        if len(slide_list) >= num_dead_thresh:
            target_list = slide_list.copy()
            slide_list_all.append(target_list)
    index= []  # 将找到的满足条件的index合并
    for i in range(len(slide_list_all) - 1):
        if set(slide_list_all[i]) < set(slide_list_all[i + 1]):
            index.append(i)
    m = {i: element for i, element in enumerate(slide_list_all)}
    [m.pop(i) for i in index]
    return list(m.values())

def crnt_insert(df_data_temp):
    df_data_temp['time'] = pd.to_datetime(df_data_temp['time'])
    df_data_temp.sort_values(by = ['time'], inplace = True)
    interval = pd.Timedelta(minutes = 5)
    threshold = pd.Timedelta(minutes = 20)

    inserted_data = pd.DataFrame(columns=df_data_temp.columns)

    for i in range(1, len(df_data_temp)):
        current_time = df_data_temp.iloc[i]['time']
        prev_time = df_data_temp.iloc[i-1]['time']
        time_diff = current_time - prev_time
        
        if time_diff >= threshold:
            num_intervals = int(time_diff / interval) - 1
            new_times = pd.date_range(start=prev_time + interval, end=current_time - interval, freq=interval)
            new_rows = pd.DataFrame({col: np.nan for col in df_data_temp.columns}, index=range(len(new_times)))
            new_rows['time'] = new_times
            inserted_data = inserted_data.append(new_rows)

    df_data_temp = pd.concat([df_data_temp, inserted_data], ignore_index=True).sort_values('time')
    # df_data_temp['PackCrnt'] = df_data_temp['PackCrnt'].interpolate()
    df_data_temp['packcrnt'].fillna(0.2, inplace = True)
    df_data_temp.drop_duplicates(subset=['time'], keep = 'first', inplace = True)
    df_data_temp.sort_values(by = ['time'], inplace = True)
    df_data_temp.reset_index(drop = True, inplace = True)
    return df_data_temp
def split_data(df_data2):
    df_data_ori = df_data2.copy()
    df_data_ori.reset_index(drop = True, inplace = True)
    df_data = crnt_insert(df_data_ori)
    df_data['crnt_flg'] = 0
    df_data.loc[df_data['packcrnt'] > 0.05*102, 'crnt_flg'] = 1
    df_data.loc[df_data['packcrnt'] < -0.05*102, 'crnt_flg'] = -1
    df_data['sts_flg'] = 2
    df_sts_chrg = pd.DataFrame(columns=list(df_data.columns))
    df_crnt_flg = df_data['crnt_flg']
    num_dead_thresh = 15
    indexs_to_delelte = clean_dead_value(df_crnt_flg, num_dead_thresh)#获得连续数据所在的行
    rest_num = len(indexs_to_delelte)
    if rest_num > 0:#仅有一个连续数据时
        for splice_item in range(0, rest_num):#rest_num
            df_data_temp = df_data.iloc[indexs_to_delelte[splice_item][0]:(indexs_to_delelte[splice_item][-1]+1)]#获得电流连续数据
            df_data_temp.reset_index(drop = True, inplace = True)
            # cal_ah_temp = cal_ah(df_data_temp)
            delta_soc = df_data_temp['packsoc'].iloc[-1] - df_data_temp['packsoc'].iloc[0]
            df_time_temp = pd.to_datetime(df_data_temp['time'])
            delta_time = (df_time_temp.iloc[-1] - df_time_temp.iloc[0])/pd.Timedelta(1, 'hours')
            if all(df_data_temp['crnt_flg'] == 0):#静置判断
                if delta_time > 0.17:
                    df_data_temp['sts_flg'] = 0
                    df_sts_chrg = df_sts_chrg.append(df_data_temp)
                    df_sts_chrg.reset_index(drop = True, inplace = True)
            elif all(df_data_temp['crnt_flg'] == -1) and (delta_soc > 2):#充电判断0.1*self.capty
                df_data_temp['sts_flg'] = 1
                
                df_time_temp = pd.to_datetime(df_data_temp['time'])
                df_soc_temp = df_data_temp['packsoc']
                delta_soc = round((df_soc_temp.iloc[-1] - df_soc_temp.iloc[0]), 3)
                delta_time = (df_time_temp.iloc[-1] - df_time_temp.iloc[0])/pd.Timedelta(1, 'hours')
                if delta_time >= 0.03:
                    rate = round(delta_soc/(100*delta_time), 3)
                else:
                    rate = 0
                if abs(rate) > 0.3:
                    status = 'offboard_charge'
                else:
                    status = 'onboard_charge'
                if df_soc_temp.iloc[-1] >= 99:
                    ful_chrg_flg = 1
                else:
                    ful_chrg_flg = 0                    
                df_data_temp['charge_method']= status
                df_data_temp['full_charge_flg']= ful_chrg_flg

                
                df_sts_chrg = df_sts_chrg.append(df_data_temp)
                df_sts_chrg.reset_index(drop = True, inplace = True)
            # elif all(df_data_temp['crnt_flg'] == 1) and (cal_ah_temp > 0.1*self.capty):#充电判断
            #     df_data_temp['sts_flg'] = 1
            #     df_sts_chrg = df_sts_chrg.append(df_data_temp)
            #     df_sts_chrg.reset_index(drop = True, inplace = True)
    df_dschrg = pd.concat([df_data, df_sts_chrg, df_sts_chrg]).drop_duplicates(subset = ['time', 'packsoc'], keep = False)
    data_temp = pd.concat([df_dschrg, df_sts_chrg])
    data_temp.sort_values(by = 'time', inplace = True)
    data_temp.reset_index(drop = True, inplace = True)
    mask = (data_temp['sts_flg'] == 2) & (data_temp['sts_flg'].shift(13) == 0) & (data_temp['sts_flg'].shift(-13) == 0)
    data_temp.loc[mask, 'sts_flg'] = 0#静置中的单点行车修改为静置
    df_chrg_temp = pd.DataFrame()
    chrgr_splice_num = []
    if len(df_sts_chrg) > 30:
        #---------------------------------充电过程判断--------------------------------------
        df_chrg_temp = data_temp.loc[(data_temp['sts_flg'] == 1)]#获取电池充电段
        if not df_chrg_temp.empty:
            df_chrg_temp.reset_index(inplace = True, drop = True)
            chrgr_time = pd.to_datetime(df_chrg_temp['time'])
            delta_time = (np.diff(chrgr_time)/pd.Timedelta(1, 'min'))#
            pos = np.where(delta_time > 30)
            splice_num = []
            if len(pos[0]) >= 1:
                pos_ful_tem = np.insert(pos, 0, 0)
                pos_len = len(pos_ful_tem)
                data_len = len(chrgr_time)
                pos_ful = np.insert(pos_ful_tem, pos_len, data_len-1)
                for item in range(0,len(pos_ful)-1):
                    splice_num.extend(item*np.ones(pos_ful[item +1]-pos_ful[item]))
                splice_num = np.insert(splice_num, 0, 0)
            else:
                splice_num = np.zeros(len(chrgr_time))
                pos_ful = np.array([0])
            if len(splice_num) > 0:
                df_chrg_temp['chrgr_rest'] = splice_num
                chrgr_splice_num = np.unique(df_chrg_temp['chrgr_rest'])#判断有几段充电数据
    if not data_temp.empty:
        stat_delta_flg = np.diff(data_temp['sts_flg'])#计算时间差的分钟数
        delta_time = np.diff(pd.to_datetime(data_temp['time']))/pd.Timedelta(1, 'min')
        stat_pos = np.where((stat_delta_flg != 0) | (delta_time > 40))#充电数据分段,大于10min时，认为是两个充电过程
        sts_splice_num = []
        if len(stat_pos[0]) >= 1:
            pos_ful_tem = np.insert(stat_pos, 0, 0)
            pos_len = len(pos_ful_tem)
            data_len = len(data_temp)
            pos_ful = np.insert(pos_ful_tem, pos_len, data_len-1)
            for item in range(0,len(pos_ful)-1):
                sts_splice_num.extend(item*np.ones(pos_ful[item +1]-pos_ful[item]))
            sts_splice_num = np.insert(sts_splice_num, 0, 0)
        else:
            sts_splice_num = np.zeros(len(data_temp))
            pos_ful = np.array([0])
        if len(sts_splice_num) > 0:
            data_temp['state_bms'] = sts_splice_num
            state_splice_num = np.unique(data_temp['state_bms'])#判断有几段充电数据
    return data_temp, df_chrg_temp, chrgr_splice_num, state_splice_num

def datacleaning(df):
    if not df.empty:
        # df['Time'] = pd.to_datetime(list(df['Time']), utc=True, unit='ms').tz_convert('Asia/Shanghai')
        df=df.replace('', np.nan)
        df.dropna(axis=0,subset = [ "time", "cellvoltage", "celltemp", "packcrnt"], inplace=True)
        df['time'] = pd.to_datetime(df['time'], unit='s') #+ pd.Timedelta(hours=8)
        df.drop(df.index[(df['packvoltage'] > 1000) | abs(df['packcrnt'] > 999) | (df['packsoc'] > 100) ], inplace=True)
        df.drop(df.index[(df['packvoltage'] < 0.001) & (abs(df['packcrnt']) < 0.001) ], inplace=True)
        if not df.empty:
            df.sort_values(by="time", inplace=True)
            df.drop_duplicates(subset="time", inplace=True)
            df.reset_index(drop=True, inplace=True)
            #num_lst = df.loc[:,'CellVoltTotalCount'].value_counts()   #统计电芯数量这一列每个元素出现的个数
            CellVoltNums = 96 #num_lst.idxmax()  # 找出电芯数量出现最多的次数
            #num_lst = df.loc[:,'CellTempTotalCount'].value_counts()   #统计电芯数量这一列每个元素出现的个数
            CellTempNums = 16#num_lst.idxmax()  # 找出电芯数量出现最多的次数
            cellvolt_name=['cellvoltage'+str(x) for x in range(1, CellVoltNums+1)]
            celltemp_name=['celltemp'+str(x) for x in range(1, CellTempNums+1)]
            df_volt=pd.DataFrame([x[0].split(",") for x in np.array(df[['cellvoltage']])]).iloc[:,list(range(CellVoltNums))]
            df_volt.columns = cellvolt_name
            df_volt=df_volt.astype('float')
            cellvoltmax = df_volt.max(axis=1)
            cellvoltmin = df_volt.min(axis=1)
            df_volt[['cellvoltmax','cellvoltmin']] = pd.concat([cellvoltmax,cellvoltmin], axis=1)
            df_temp=pd.DataFrame([x[0].split(",") for x in np.array(df[['celltemp']])]).iloc[:,list(range(CellTempNums))]
            df_temp.columns = celltemp_name
            df_temp=df_temp.astype('float')
            celltempmax = df_temp.max(axis=1)
            celltempmin = df_temp.min(axis=1)
            df_temp[['celltempmax','celltempmin']] = pd.concat([celltempmax,celltempmin], axis=1)
            
            df=pd.concat([df,df_volt,df_temp],axis=1)
            df.drop(df.index[(df['cellvoltmin'] <0.1)], inplace=True)
            df.reset_index(inplace=True, drop=True)
            # df_table = df.drop_duplicates(subset=['SN'], keep='first', ignore_index=True)
            # df_table = df_table.set_index('SN')
        else:
            df = pd.DataFrame()
            #df_table = pd.DataFrame()
            cellvolt_name = []
            celltemp_name = []
        return df,  cellvolt_name, celltemp_name #df_table,
    else:
        return pd.DataFrame(), [], [] # pd.DataFrame(),
 




def get_all_files_in_folder(folder_path):
    all_files = []
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            file_path = os.path.join(root, file)
            all_files.append(file_path)
    return all_files



def concat_csv_files(csv_files):
    dfs = []
    for file in csv_files:
        df = pd.read_csv(file)
        dfs.append(df)
    return pd.concat(dfs, ignore_index=True)



sn_list=['LUZAGAAA9LA012791','LUZAGAAA2LA014379','LUZAGAAA1MA020272','LUZAGAAA0MA020540','LUZAGAAA3MA033346','LUZAGAAA0MA035149', 'LUZAGAAA7MA043572','LUZAGAAA0MA070726','LUZAGAAA9NA020991','LUZAGAAA9NA022448']
for i in range (0,len(sn_list)):
    
    folder_path = "/data_highspeed/common/hz/lifecycle/{}.csv".format(sn_list[i])  # 替换为实际的文件夹路径
    
    out_put_folder_path="{}_charge.feather".format(sn_list[i])   #/data/common/hz/lifecycle/LUZAGAAAXKA008957/
    # file_list = get_all_files_in_folder(folder_path)
    # if len(file_list) > 0:
    #     concatenated_df = concat_csv_files(file_list)
    print('读取第'+str(i)+'个电池'+sn_list[i])
    concatenated_df=pd.read_csv(folder_path)




    if concatenated_df is not None:
        print('清洗第'+str(i)+'个电池'+sn_list[i])
        df_cleaned,cellvolt_name,celltemp_name=datacleaning(concatenated_df)
        print('分段第'+str(i)+'个电池'+sn_list[i])
        data_temp, df_chrg_splited, chrgr_splice_num, state_splice_num=split_data(df_cleaned)
        print('写入第'+str(i)+'个电池'+sn_list[i])
        df_chrg_splited.to_feather(out_put_folder_path)
        print("Finished.")
    else:
        print("No CSV files found in the folder.")