zlwl-algos-new/USER/liuzhongxiao/SOC_Fct/main.py

import datetime, time
from multiprocessing import Pool
import json
import os
import traceback
import warnings
from sqlalchemy import text, delete, and_, or_, update
import pandas as pd
import traceback
import SOC_pre_vltVV
import OCV_SOC

from ZlwlAlgosCommon.utils.ProUtils import *
from ZlwlAlgosCommon.service.iotp.IotpAlgoService import IotpAlgoService
from ZlwlAlgosCommon.service.iotp.Beans import DataField
from ZlwlAlgosCommon.orm.models import *

import sys
from sklearn.datasets import load_iris
import joblib
import json

#...................................电池包电芯安全诊断函数......................................................................................................................
def diag_cal(process_num):#, process

    # 环境变量配置（通过环境变量确定当前程序运行在开发、测试、生产环境）
    # 根据sn 获取对应的参数
    #Hbase
    algo_list = ['soc_pre'] # 本调度所包含的算法名列表。
    process = 10
    loggers = sysUtils.get_loggers(algo_list, log_base_path, process_num) # 为每个算法分配一个logger
    logger_main.info(f"process-{process_num}: 配置中间件")
    hbase_params = sysUtils.get_cf_param('hbase-datafactory')
    iotp_service = IotpAlgoService(hbase_params=hbase_params)
    
    sql = f"select sn, imei, pack_model,organ_code, device_cell_type from t_device"
    mysql_kw_conn = mysql_kw_engine.connect()
    df_t_device = pd.read_sql(sql, mysql_kw_conn)
    sql = f"select algo_id, pack_code, param, param_ai from algo_adjustable_param"
    df_algo_adjustable_param = pd.read_sql(sql, mysql_kw_conn)
    sql = f"select pack_code, param from algo_pack_param"
    pack_code = 'JX18020'
    df_algo_pack_param = pd.read_sql(sql, mysql_kw_conn)
    adjustable_param = df_algo_adjustable_param[df_algo_adjustable_param['pack_code']==pack_code].drop(['pack_code'], axis=1)
    adjustable_param = adjustable_param.to_dict("records")
    pack_param = df_algo_pack_param[df_algo_pack_param['pack_code']==pack_code].drop(['pack_code'], axis=1)
    pack_param = pack_param.to_dict("records")
    df_algo_pack_param = json.loads(pack_param[0]['param'].replace(' ', ''))
    df_algo_pack_param = {k: eval(v) if isinstance(v, str) else v for k, v in df_algo_pack_param.items()}
    # snlst = list(df_t_device[df_t_device['pack_model'] == pack_code]['sn'])[0:1]#[3*(process_num):3*(process_num+1)]
    pk_sn = list(df_t_device[(df_t_device['pack_model'] == pack_code)]['sn'])
    snlst_ort = [item for item in pk_sn if 'S23620' in item]#
    snlst = snlst_ort[3*(process_num):3*(process_num+1)]
    def getDateList(start_date, end_date):
        date_list = []
        start_date = datetime.datetime.strptime(start_date, '%Y-%m-%d')
        end_date = datetime.datetime.strptime(end_date, '%Y-%m-%d')# %H:%M:%S
        date_list.append(start_date.strftime('%Y-%m-%d'))
        while start_date < end_date:
            start_date += datetime.timedelta(days=1)
            date_list.append(start_date.strftime('%Y-%m-%d'))
        return date_list
    # dnn_predict = joblib.load('/home/liuzhongxiao/zlwl-algos/USER/liuzhongxiao/SOC_Fct/Dnnmodel.pkl')
    # input_norm_X = joblib.load('/home/liuzhongxiao/zlwl-algos/USER/liuzhongxiao/SOC_Fct/normorlized_x')
    # input_norm_Y = joblib.load('/home/liuzhongxiao/zlwl-algos/USER/liuzhongxiao/SOC_Fct/normorlized_y')
    # 使用os.walk()函数遍历文件夹A及其子文件夹
    # snlst = ['PJXCLL532S234M030']#'PJXCLL532S234M007', 'PJXCLL532S234M010', 'PJXCLL532S234M004', 'PJXCLL532S234M030', 'PJXCLL532S234M018', , 'PJXCLL532S234M014', 'PJXCLL532S234M016'
    try:
        for itemsn in snlst:
            df_soc_pre = pd.DataFrame()
            try:
                time_st = time.time()
                # 获取文件的完整路径
                itemsn = 'PJXCLL128N234P002'#, 'PJXCLL532S234M012', 'PJXCLL532S234M015'
                loggers['soc_pre'].info(f'开始执行算法{itemsn}')
                data_lst = getDateList("2023-06-29","2023-06-30")
                for item_date in range(0, len(data_lst)-1):#
                    try:
                        st_time = '2023-06-28 10:30:00'#data_lst[item_date] + ' 00:00:00'
                        et = '2023-06-29 01:00:00'#data_lst[item_date + 1] + ' 09:00:00'
                        # st_time = (datetime.datetime.strptime(st, '%Y-%m-%d %H:%M:%S') - datetime.timedelta(hours=3)).strftime('%Y-%m-%d %H:%M:%S')
                        columns = [DataField.sn, DataField.time,DataField.pack_crnt, DataField.pack_volt, DataField.cell_voltage_count, DataField.cell_temp_count,
                                DataField.cell_voltage, DataField.cell_temp, DataField.pack_soc, DataField.other_temp_value, DataField.latitude, 
                                DataField.longitude, DataField.mileage, DataField.accum_chg_ah, DataField.accum_dschg_ah, DataField.accum_chg_wh,
                                DataField.accum_dschg_wh]
                        df_data_all = iotp_service.get_data(sn_list=[itemsn], columns=columns, start_time=st_time, end_time=et)
                        loggers['soc_pre'].info(f'sn-{itemsn},起始-{st_time},结束-{et},获取到{len(df_data_all)}条数据,取数耗时:{time.time()-time_st}')
                        if len(df_data_all) > 30:
                            df_data_all['mileage'].replace(0, pd.np.nan, inplace=True)
                            df_data_all['mileage'] = df_data_all['mileage'].fillna(method='backfill')
                            df_data_all['mileage'] = df_data_all['mileage'].fillna(method='ffill')
                            df_data_all.rename(columns={"mileage":"odo"}, inplace=True)
                            # df_data_gps=iotp_service.gps_datacleaning(df_data_all)
                            df_out, df_table, cellvolt_name, celltemp_name = iotp_service.datacleaning(df_algo_pack_param,df_data_all)#进行数据清洗
                            if (not df_out.empty):# and (not df_data_gps.empty)
                                # df_out['time'] = df_out['time'].dt.strftime('%Y-%m-%d %H:%M:%S')
                                # df_merge_gab = pd.merge(df_out, df_data_gps,on=['sn','time','datatype'],how='outer')
                                # df_merge_gab_fill = df_merge_gab.groupby("sn").fillna(method='backfill')
                                # df_merge_gab_fill =  pd.concat([df_merge_gab[['sn']], df_merge_gab_fill], axis=1)
                                # df_merge_gab_fill = df_merge_gab.groupby("sn").fillna(method='ffill')
                                # df_merge_gab_fill =  pd.concat([df_merge_gab[['sn']], df_merge_gab_fill], axis=1)
                                # df_merge = df_merge_gab_fill.sort_values(["sn","time"],ascending = [True, True])
                                df_merge_cal = df_out[df_out['datatype']==12]
                                if not df_merge_cal.empty:
                                    chrg_chracter = SOC_pre_vltVV.cell_chracter(df_merge_cal, df_algo_pack_param, cellvolt_name, celltemp_name)#计算内阻
                                    soc_pre = chrg_chracter.states_cal()
                                    OCV_chrac = OCV_SOC.cell_statistic(df_algo_pack_param, df_merge_cal)
                                    OCV_soc_rlt = OCV_chrac.soc_cal_est()
                                    df_soc_rlt = pd.concat([soc_pre, OCV_soc_rlt])
                                    if not df_soc_rlt.empty:
                                        df_soc_pre = df_soc_pre.append(df_soc_rlt)
                    except Exception as e:
                        loggers['soc_pre'].error('算法运行出错')
                        loggers['soc_pre'].error(str(e))
                        loggers['soc_pre'].error(traceback.format_exc())
                if not df_soc_pre.empty:
                    df_soc_pre.drop_duplicates(subset = ['sn','time'], keep = 'last', inplace = True)
                    df_soc_pre.to_csv('/home/liuzhongxiao/zlwl-algos/USER/liuzhongxiao/SOC_Fct/SOC_rlt/' + itemsn + 'SOC估计结果VV.csv',index=False,encoding='GB18030')
                loggers['soc_pre'].info(f'算法运行完成{itemsn}，算法耗时{time.time()-time_st}')
            except Exception as e:
                loggers['soc_pre'].error('算法运行出错')
                loggers['soc_pre'].error(str(e))
                loggers['soc_pre'].error(traceback.format_exc())
           
    except Exception as e:
        loggers['soc_pre'].error('算法运行出错')
        loggers['soc_pre'].error(str(e))
        loggers['soc_pre'].error(traceback.format_exc())
        
#...............................................主函数起定时作用.......................................................................................................................
if __name__ == "__main__":
    cur_env = 'dev' # 设置运行环境
    app_path = "/home/liuzhongxiao/zlwl-algos/" # 设置相对路径
    log_base_path = f"{os.path.dirname(os.path.abspath(__file__))}/log" # 设置日志路径
    app_name = "schedule" # 应用名, 建议与topic的后缀相同
    
    sysUtils = SysUtils(cur_env, app_path)
    
    mysqlUtils = MysqlUtils()
    mysql_iotp_params = sysUtils.get_cf_param('mysql-iotp')
    mysql_iotp_engine, mysql_iotp_Session= mysqlUtils.get_mysql_engine(mysql_iotp_params)
    
    
    mysql_kw_params = sysUtils.get_cf_param('mysql-algo')
    mysql_kw_engine, mysql_kw_Session= mysqlUtils.get_mysql_engine(mysql_kw_params)
    logger_main = sysUtils.get_logger(app_name, log_base_path)
    logger_main.info(f"本次主进程号: {os.getpid()}")
    # 读取配置文件 (该部分请不要修改)
    processes = int(sysUtils.env_params.get("PROCESS_NUM_PER_NODE", '1')) # 默认为1个进程
    pool = Pool(processes = int(processes))
    logger_main.info("开始分配子进程")
    for i in range(int(processes)):
        pool.apply_async(diag_cal, (i, ))#5,10,  
    pool.close()
    logger_main.info("进程分配结束，堵塞主进程")
    pool.join()