import datetime import gc import re from multiprocessing import Pool import json import logging import logging.handlers import os import time import traceback import warnings from sqlalchemy import text, delete, and_, or_, update import pandas as pd 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 invoke_algo1(logger, mysql_algo_conn, mysql_algo_Session, start_time, df_data): pass def invoke_algo2(logger, mysql_algo_conn, mysql_algo_Session, start_time, df_data): pass def main(process_num): # 程序不能停止 while(True): warnings.filterwarnings("ignore") try: # 调用算法前的准备工作 kafka_topic_key = 'topic_task_min_10' kafka_groupid_key = 'group_id_task_min_10' algo_list = ['FaultWarning', 'other'] # 本调度所包含的算法名列表。 loggers = sysUtils.get_loggers(algo_list, log_base_path, process_num) # 为每个算法分配一个logger logger_main.info(f"process-{process_num}: 配置中间件") # mysql mysql_algo_params = sysUtils.get_cf_param('mysql-algo') mysqlUtils = MysqlUtils() mysql_algo_engine, mysql_algo_Session= mysqlUtils.get_mysql_engine(mysql_algo_params) mysql_algo_conn = mysql_algo_engine.connect() # redis redis_params = sysUtils.get_cf_param('redis') redisUtils = RedisUtils() redis_conn = redisUtils.get_redis_conncect(redis_params) # hbase hbase_params = sysUtils.get_cf_param('hbase') iotp_service = IotpAlgoService(hbase_params=hbase_params) # kafka kafka_params = sysUtils.get_cf_param('kafka') kafkaUtils = KafkaUtils() kafka_consumer = kafkaUtils.get_kafka_consumer(kafka_params, kafka_topic_key, kafka_groupid_key, client_id=kafka_topic_key) logger_main.info(f"process-{process_num}: 获取算法参数及电池参数") except Exception as e: logger_main.error(f'process-{process_num}: {e}') logger_main.error(f'process-{process_num}: {traceback.format_exc()}') # 开始准备调度 try: logger_main.info(f"process-{process_num}: 监听topic {kafka_params[kafka_topic_key]}等待kafka 调度") param_update_timer = time.time() for message in kafka_consumer: try: logger_main.info(f'process-{process_num}: 收到调度 {message.value}') if not mysql_algo_conn.closed: mysql_algo_conn.close() mysql_algo_conn = mysql_algo_engine.connect() # 从连接池中获取一个myslq连接 schedule_params = json.loads(message.value) if (schedule_params is None) or (schedule_params ==''): logger_main.info(f'process-{process_num}: {message.value} kafka数据异常，跳过本次运算') continue # kafka 调度参数解析 df_snlist = pd.DataFrame(schedule_params['snlist']) df_algo_adjustable_param = pd.DataFrame([(d['algo_id'], d['param'],d['param_ai']) for d in schedule_params['adjustable_param']], columns=['algo_id', 'param','param_ai']) df_algo_pack_param = json.loads(schedule_params['pack_param'][0]['param']) df_algo_list = pd.DataFrame(schedule_params['algo_list']) start_time = schedule_params['start_time'] end_time = schedule_params['end_time'] pack_code = schedule_params['pack_code'] cell_type = schedule_params['cell_type'] sn_list=df_snlist['sn'].tolist() # 取数 logger_main.info(f"process-{process_num}: 开始取数") columns = [DataField.error_level, DataField.error_code, DataField.pack_crnt, DataField.pack_volt, DataField.bms_sta, DataField.cell_voltage_count, DataField.cell_temp_count, DataField.cell_voltage, DataField.cell_temp, DataField.pack_soc, DataField.other_temp_value, DataField.cell_balance, DataField.pack_soh, DataField.charge_sta] df_data = iotp_service.get_data(sn_list=sn_list, columns=columns, start_time=start_time, end_time=end_time) logger_main.info(f"process-{process_num}: {str(sn_list)}获取到{str(len(df_data))}条数据") except Exception as e: logger_main.error(f'process-{process_num}: {pack_code}运行出错') logger_main.error(f'process-{process_num}: {e}') logger_main.error(f'process-{process_num}: {traceback.format_exc()}') try: # 数据清洗 if len(df_data) == 0: logger_main.info(f"process-{process_num}: 无数据跳过本次运算") continue df_data,df_table,cellvolt_name,celltemp_name=iotp_service.data_clean(df_data,df_algo_pack_param)#进行数据清洗 if len(df_data) == 0: logger_main.info(f"process-{process_num}: 数据清洗完成, 无有效数据，跳过本次运算") continue else: logger_main.info(f"process-{process_num}: {pack_code}, time_type:{df_data.loc[0, 'time']} ~ {df_data.iloc[-1]['time']}, 数据清洗完成") except Exception as e: logger_main.error(f"process-{process_num}:{pack_code}数据清洗出错") logger_main.error(f"process-{process_num}:{e}") logger_main.error(f"process-{process_num}:{traceback.format_exc()}") # 算法调用 try: invoke_algo1(loggers, mysql_algo_conn, mysql_algo_Session, start_time, df_data) except Exception as e: loggers['FaultWarning'].error('{}运行出错'.format(pack_code)) loggers['FaultWarning'].error(str(e)) loggers['FaultWarning'].error(traceback.format_exc()) # 第二个算法调用 try: invoke_algo2(loggers, mysql_algo_conn, mysql_algo_Session, start_time, df_data) except Exception as e: loggers['FaultWarning'].error('{}运行出错'.format(pack_code)) loggers['FaultWarning'].error(str(e)) loggers['FaultWarning'].error(traceback.format_exc()) except Exception as e: logger_main.error(f'process-{process_num}: {pack_code}运行出错') logger_main.error(f'process-{process_num}: {e}') logger_main.error(f'process-{process_num}: {traceback.format_exc()}') finally: iotp_service.close() if __name__ == '__main__': while(True): try: # 配置量 cur_env = 'dev' # 设置运行环境 app_path = "/home/wangliming/project/zlwl-algos/" # 设置app绝对路径 log_base_path = f"{os.path.dirname(os.path.abspath(__file__))}/log" # 设置日志路径 app_name = "task_second_1" # 应用名 sysUtils = SysUtils(cur_env, app_path) 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(main, (i, )) pool.close() logger_main.info("进程分配结束，堵塞主进程") pool.join() except Exception as e: logger_main.error(str(e)) logger_main.error(traceback.format_exc()) finally: handlers = logger_main.handlers.copy() for h in handlers: logger_main.removeHandler(h) pool.terminate()