from datetime import datetime from multiprocessing import Pool import json import os import time import traceback import warnings from sqlalchemy import text, delete, and_, or_, update import pandas as pd import traceback import sta_stacs from ZlwlAlgosCommon.utils.ProUtils import * from ZlwlAlgosCommon.service.iotp.IotpAlgoService import IotpAlgoService from ZlwlAlgosCommon.service.iotp.Beans import DataField from ZlwlAlgosCommon.orm.models import * #...................................电池包电芯安全诊断函数...................................................................................................................... def diag_cal(process_num, process):# # 环境变量配置（通过环境变量确定当前程序运行在开发、测试、生产环境） # 根据sn 获取对应的参数 #Hbase algo_list = ['soc_chracter'] # 本调度所包含的算法名列表。 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') 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 = 'CL3282A' 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']) df_algo_pack_param = {k: eval(v) if isinstance(v, str) else v for k, v in df_algo_pack_param.items()} folder_path = r"/data/common/benz" # 使用os.walk()函数遍历文件夹A及其子文件夹 try: for root, dirs, files in os.walk(folder_path): # 遍历每个子文件夹 for dir_name in dirs[5*(process_num):5*(process_num+1)]:#读取各个sn # 获取子文件夹的完整路径 dir_path = os.path.join(root, dir_name) # print(f'Reading files from folder: {dir_path}') itemsn = dir_path.split('/')[-1] # 遍历当前子文件夹中的所有文件 df_chrg_chrac = pd.DataFrame() for file_name in os.listdir(dir_path):#读取各sn下不同时间的数据 # if (itemsn == 'LY9F49BC5MALBZ879') and (file_name == '2022-10-15-00-00-00_2022-10-22-00-00-00.zippkl'): try: time_st = time.time() # 获取文件的完整路径 file_path = os.path.join(dir_path, file_name) loggers['soc_chracter'].info(f'开始执行算法{file_path}') df_data = pd.read_pickle(file_path, compression='zip') if len(df_data) > 30: df_data['mileage'].replace(0, pd.np.nan, inplace = True) df_data['mileage'].fillna(method = 'bfill', inplace = True) df_data['mileage'].fillna(method = 'ffill', inplace = True) df_out, df_table, cellvolt_name, celltemp_name = iotp_service.datacleaning(df_algo_pack_param,df_data)#进行数据清洗 if len(df_out) > 30: Diagsts_temp = sta_stacs.cell_statistic(df_algo_pack_param, df_out)#计算内阻 df_chrgrt_add = Diagsts_temp.rest_sta() if not df_chrgrt_add.empty: df_chrg_chrac = df_chrg_chrac.append(df_chrgrt_add) df_chrg_chrac.drop_duplicates(subset = ['sn','time_st'], keep = 'first', inplace = True) df_chrg_chrac.to_csv('/home/liuzhongxiao/zlwl-algos/USER/liuzhongxiao/SOC_pre/chracter/' + itemsn + '充电表单.csv',index=False,encoding='GB18030') loggers['soc_chracter'].info(f'算法运行完成{pack_code}，算法耗时{time.time()-time_st}') except Exception as e: loggers['soc_chracter'].error('算法运行出错') loggers['soc_chracter'].error(str(e)) loggers['soc_chracter'].error(traceback.format_exc()) except Exception as e: loggers['soc_chracter'].error('算法运行出错') loggers['soc_chracter'].error(str(e)) loggers['soc_chracter'].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", '10')) # 默认为1个进程 pool = Pool(processes = int(processes)) logger_main.info("开始分配子进程") for i in range(int(processes)): pool.apply_async(diag_cal, (i, 10,))# pool.close() logger_main.info("进程分配结束，堵塞主进程") pool.join() #log信息配置 #读取fault_code=C599的当前故障 # df_chrgrt = pd.read_csv(r'C:\Users\zldc\project\User\lzx\hz-application-algo\USER\lzx\状态统计\充电过程统计\充电表单-单车.csv' ,encoding='GB18030') # df_dschrg = pd.read_csv(r'C:\Users\zldc\project\User\lzx\hz-application-algo\USER\lzx\状态统计\充放电温度压差统计\放电表单\用车表单-单车.csv' ,encoding='GB18030') #定时任务....................................................................................................................................................................... # diag_cal()