123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363 |
- import datetime #from datetime
- from multiprocessing import Pool
- import json
- import os
- import time
- import traceback
- import warnings
- from keras.models import load_model
- import pickle
- 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 DataSplit.COMMON.service.iotp.IotpAlgoService import IotpAlgoService #test_code
- from ZlwlAlgosCommon.service.iotp.Beans import DataField
- from ZlwlAlgosCommon.orm.models import *
-
- from DataSplit.V_1_0_0 import data_status as ds ##充电状态标准化程序
- from DataSplit.V_1_0_0 import data_split as dt ##分段函数程序
- from DataSplit.V_1_0_0 import data_drive_stat as ddt ##行驶数据按行驶段汇总统计
- from DataSplit.V_1_0_0 import data_charge_stat as dct ##充电数据按充电段汇总
- from DataSplit.V_1_0_0 import data_stand_stat as dst ##静置数据按静置段汇总
- from DataSplit.V_1_0_0 import data_drive_stat_period as ddtp ##行驶数据按充电周期汇总统计
- from DataSplit.V_1_0_0 import trans_day as trd ##解决跨天的问题
- from DataSplit.V_1_0_0 import stand_status as ss
- def update_param(db_engine):
- # 从redis中获取参数,如果redis中获取不到,则去数据库中获取
- df_algo_adjustable_param = pd.read_sql("select * from algo_adjustable_param", db_engine)
- df_algo_list = pd.read_sql("select * from algo_list", db_engine)
- df_algo_pack_param = pd.read_sql("select* from algo_pack_param", db_engine)
- return df_algo_adjustable_param, df_algo_list, df_algo_pack_param
- def main():
- process_num=1
- # 程序不能停止
- while(True):
- try:
-
- warnings.filterwarnings("ignore")
- try:
- cleanUtils = CleanUtils()
- # 调用算法前的准备工作
- mysql_algo_conn = None
- mysql_algo_engine = None
- mysql_iotp_conn = None
- mysql_iotp_engine= None
- kafka_consumer = None
- rc= None
-
- algo_list = ['socdiag','low_soc_diag','Sor_Diag','Li_Plted','Data_Split'] # 本调度所包含的算法名列表。
-
- 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()
-
-
- mysql_iotp_data = sysUtils.get_cf_param('mysql-iotp')
- mysqlUtils = MysqlUtils()
- mysql_iotp_engine, mysql_iopt_Session= mysqlUtils.get_mysql_engine(mysql_iotp_data)
- mysql_iotp_conn = mysql_iotp_engine.connect()
-
-
- #Hbase
- hbase_params = sysUtils.get_cf_param('hbase-datafactory')# test_code
- iotp_service = IotpAlgoService(hbase_params=hbase_params)
- #redis
- redis_params = sysUtils.get_cf_param('redis')
- reidsUtils = RedisUtils()
- rc = reidsUtils.get_redis_conncect(redis_params)
- except Exception as e:
- logger_main.error(f'process-{process_num}: {e}')
- logger_main.error(f'process-{process_num}: {traceback.format_exc()}')
- cleanUtils.clean(mysql_algo_conn, mysql_algo_engine, mysql_iotp_conn, mysql_iotp_engine, kafka_consumer, rc)
-
-
- # 开始准备调度
- # for message in kafka_consumer:
- #KafkaConsumer.commit(self)
- try:
- logger_main.info(f'收到调度')
- if not mysql_algo_conn.closed:
- mysql_algo_conn.close()
- mysql_algo_conn = mysql_algo_engine.connect() # 从连接池中获取一个myslq连接
-
- df_algo_adjustable_param, df_algo_list, df_algo_pack_param=update_param(mysql_algo_conn)
- df_algo_adjustable_param[df_algo_adjustable_param['pack_code']=='JX19220']
- df_algo_pack_param=df_algo_pack_param[df_algo_pack_param['pack_code']=='JX19220']
- df_algo_pack_param=json.loads(df_algo_pack_param.iloc[0]['param'])
- pack_code='JX19220'
- # schedule_params = json.loads(message.value)
- # if (schedule_params is None) or (schedule_params ==''):
- # logger_main.info('{} kafka数据异常,跳过本次运算'.format(str(message.value)))
- # 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_pack_param = {k: eval(v) if isinstance(v, str) else v for k, v in df_algo_pack_param.items()}
- # df_algo_param = 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']
- 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()
- df_snpk_list = pd.read_sql("select sn, imei,pack_model,device_cell_type,scrap_status from t_device", mysql_algo_conn)
- df_snpk_list=df_snpk_list[df_snpk_list['scrap_status']<4]
- df_snpk_list=df_snpk_list.rename(columns={'pack_model':'pack_code'})
- df_snpk_list=df_snpk_list[df_snpk_list['sn'].str.contains('N234P')]
- df_sn_list=list(df_snpk_list['sn'])
- sn_list_total=df_sn_list
-
- sn_list=sn_list_total
-
- start_time = "2023-03-25 00:00:00"
- end_time = "2023-06-25 00:00:00"
- columns = [ DataField.time, DataField.sn, DataField.pack_crnt, DataField.pack_volt, DataField.pack_soc,
- DataField.cell_voltage_count, DataField.cell_temp_count, DataField.cell_voltage, DataField.cell_temp,
- DataField.other_temp_value, DataField.bms_sta, DataField.charge_sta,DataField.latitude,DataField.longitude,
- DataField.mileage, DataField.accum_chg_wh, DataField.accum_dschg_wh, DataField.accum_chg_ah,DataField.accum_dschg_ah,DataField.vin]
- for i in range(0,len(sn_list_total)):
- sn_list=[sn_list_total[i]]
- # 取数
- time_st = time.time()
- logger_main.info(f"process-{process_num}: 开始取数,{start_time} ~ {end_time}\n{str(sn_list)}")
- df_data = iotp_service.get_data(sn_list=sn_list, columns=columns, start_time=start_time, end_time=end_time)
- #df_data.to_excel('df_data_1.xlsx')
- logger_main.info(f'process-{process_num},获取到{len(df_data)}条数据,取数d耗时:{time.time()-time_st}')
- if len(df_data) == 0:
- logger_main.info(f"process-{process_num}: 数据清洗耗时{time.time()-time_st}, 无有效数据,跳过本次运算")
- continue
- # 将字符串转换成datetime对象
- str_date = start_time
- date_time =datetime.datetime.strptime(str_date, '%Y-%m-%d %H:%M:%S')
- # 将datetime对象减去6小时
- new_date_time = date_time - datetime.timedelta(hours=8)
- # 将datetime对象转换成字符串
- start_time_8h = new_date_time.strftime('%Y-%m-%d %H:%M:%S')
- df_data_8h = iotp_service.get_data(sn_list=sn_list, columns=columns, start_time=start_time_8h, end_time=start_time)
- logger_main.info(f'process-{process_num},获取到{len(df_data_8h)}条数据,取数耗时:{time.time()-time_st}')
- # 数据清洗
- try:
- time_st = time.time()
- logger_main.info(f'process-{process_num}数据清洗')
- if len(df_data_8h)==0:
- df_data_8h=pd.DataFrame(columns=['datatype','time','sn','pack_crnt','pack_volt','pack_soc',
- 'cell_voltage_count','cell_temp_count','cell_voltage','cell_temp','other_temp_value',
- 'bms_sta','charge_sta','latitude','longitude','mileage','accum_dschg_wh','accum_chg_wh','accum_chg_ah','accum_dschg_ah','vin'])
- #df_data_8h = df_data_8h.drop(['latitude','longitude','mileage','accum_chg_wh''accum_dschg_wh','accum_dschg_ah','accum_chg_ah','vin'], axis=1, errors='ignore')
- if len(df_data)==0:
- df_data=pd.DataFrame(columns=['datatype','time','sn','pack_crnt','pack_volt','pack_soc',
- 'cell_voltage_count','cell_temp_count','cell_voltage','cell_temp','other_temp_value',
- 'bms_sta','charge_sta','latitude','longitude','mileage','accum_dschg_wh','accum_chg_wh','accum_chg_ah','accum_dschg_ah','vin'])
- #df_data = df_data.drop(['latitude','longitude','mileage','accum_chg_wh','accum_dschg_wh','accum_dschg_ah','accum_chg_ah','vin'], axis=1, errors='ignore')
- df_data_gps=iotp_service.gps_datacleaning(df_data)
- df_data_gps_8h=iotp_service.gps_datacleaning(df_data_8h)
- df_data_accum=iotp_service.accum_datacleaning(df_data)
- df_data_accum_8h=iotp_service.accum_datacleaning(df_data_8h)
- df_data_vin=iotp_service.vin_datacleaning(df_data)
- df_data_vin_8h=iotp_service.vin_datacleaning(df_data_8h)
- df_data, df_table, cellvolt_name, celltemp_name = iotp_service.datacleaning(df_algo_pack_param,df_data)#进行数据清洗
- df_data_8h, df_table_t, cellvolt_name_t, celltemp_name_t = iotp_service.datacleaning(df_algo_pack_param,df_data_8h)#进行数据清洗
- #df_data.to_excel('df_data_2.xlsx')
- if len(df_data_gps_8h)==0:
- df_data_gps_8h=pd.DataFrame(columns=['sn','time','datatype','latitude','longitude','mileage'])
- if len(df_data_gps)==0:
- df_data_gps=pd.DataFrame(columns=['sn','time','datatype','latitude','longitude','mileage'])
- if len(df_data_accum_8h)==0:
- df_data_accum_8h=pd.DataFrame(columns=['sn','time','datatype','accum_chg_wh','accum_dschg_wh','accum_dschg_ah','accum_chg_ah'])
- if len(df_data_accum)==0:
- df_data_accum=pd.DataFrame(columns=['sn','time','datatype','accum_chg_wh','accum_dschg_wh','accum_dschg_ah','accum_chg_ah'])
- if len(df_data_vin_8h)==0:
- df_data_vin_8h=pd.DataFrame(columns=['sn','time','datatype','vin'])
- if len(df_data_vin)==0:
- df_data_vin=pd.DataFrame(columns=['sn','time','datatype','vin'])
- try:
- df_data_32h=pd.concat([df_data_8h,df_data])
- df_data_32h=df_data_32h.reset_index(drop=True)
- #df_data_32h.to_excel('data_32h.xlsx')
- if len(df_data_32h)==0:
- logger_main.info(f"process-{process_num}: 数据清洗耗时{time.time()-time_st}, 无有效数据,跳过本次运算")
- continue
- else:
- df_data_gps_32h=pd.concat([df_data_gps_8h,df_data_gps])
- df_data_gps_32h=df_data_gps_32h.reset_index(drop=True)
- if len(df_data_gps_32h)==0:
- df_data_gps_32h=pd.DataFrame(columns=['sn','time','datatype','latitude','longitude','mileage'])
-
- df_data_accum_32h=pd.concat([df_data_accum_8h,df_data_accum])
- df_data_accum_32h=df_data_accum_32h.reset_index(drop=True)
- if len(df_data_accum_32h)==0:
- df_data_accum_32h=pd.DataFrame(columns=['sn','time','datatype','accum_chg_wh','accum_dschg_wh','accum_dschg_ah','accum_chg_ah'])
- df_data_vin_32h=pd.concat([df_data_vin_8h,df_data_vin])
- df_data_vin_32h=df_data_vin_32h.reset_index(drop=True)
- if len(df_data_vin_32h)==0:
- df_data_vin_32h=pd.DataFrame(columns=['sn','time','datatype','vin'])
- ##先关联gps和accum
-
- df_merge_ga = pd.merge(df_data_gps_32h, df_data_accum_32h,on=['sn','time','datatype'],how='outer')
- df_merge_ga = pd.merge(df_merge_ga, df_data_vin_32h,on=['sn','time','datatype'],how='outer')
- df_merge_ga=df_merge_ga.sort_values(["sn","time"],ascending = [True, True])
- df_merge_ga_filled = df_merge_ga.groupby("sn").fillna(method='ffill')
- df_merge_ga_filled= pd.concat([df_merge_ga[['sn']], df_merge_ga_filled], axis=1)
- df_merge_ga_filled = df_merge_ga_filled.groupby("sn").fillna(method='bfill')
- df_merge_ga= pd.concat([df_merge_ga[['sn']], df_merge_ga_filled], axis=1)
- ##识别静置的情况,关联非静置数据用前值填充
- df_merge_ga['time'] = pd.to_datetime(df_merge_ga['time'], format='%Y-%m-%d %H:%M:%S')
- df_merge_bga = pd.merge(df_data_32h, df_merge_ga, on=['sn','time','datatype'],how='outer')
- #df_merge=ss.stand_status(df_merge_bga,s_order_delta=600,lon_delta=0.1)
- df_merge=df_merge_bga.sort_values(["sn","time"],ascending = [True, True])
- df_merge_filled = df_merge.groupby("sn").fillna(method='ffill')
- df_merge = pd.concat([df_merge[['sn']], df_merge_filled], axis=1)
- df_merge_filled = df_merge.groupby("sn").fillna(method='bfill')
- df_merge = pd.concat([df_merge[['sn']], df_merge_filled], axis=1)
- df_merge=df_merge.dropna(subset=['pack_crnt','latitude','longitude','vin'],axis=0,how='all')
- df_merge=df_merge[df_merge['datatype']==12]
- #df_merge.to_excel('merged.xlsx')
-
-
-
-
-
-
-
- except Exception as e:
- logger_main.error(f"process-{process_num}:32小时拼接出错")
- logger_main.error(f"process-{process_num}:{e}")
- logger_main.error(f"process-{process_num}:{traceback.format_exc()}")
- if len(df_data) == 0:
- logger_main.info(f"process-{process_num}: 数据清洗耗时{time.time()-time_st}, 无有效数据,跳过本次运算")
- continue
- else:
- logger_main.info(f"process-{process_num}: {pack_code}, time_type:{df_data.loc[0, 'time']} ~ {df_data.iloc[-1]['time']}, 数据清洗完成耗时{time.time()-time_st}")
- except Exception as e:
- logger_main.error(f"process-{process_num}:数据清洗出错")
- logger_main.error(f"process-{process_num}:{e}")
- logger_main.error(f"process-{process_num}:{traceback.format_exc()}")
- continue
- try:
- loggers['Data_Split'].info(f'开始执行算法{pack_code}, time:{start_time}~{end_time},\n sn_list:{sn_list}')
- df_merge=ds.data_status(df_merge,c_soc_dif_p=0.05,s_soc_dif_p=0,c_order_delta=1200,s_order_delta=300)
- ##基于各个状态码,进行分段,分段函数
- cell_type='L'
- df_drive,df_charge,df_stand,df_data_split_rlt,df_rank_abnormal=dt.split(df_merge,celltemp_name,cellvolt_name,drive_interval_time_min=1200,charge_interval_time_min=1200,stand_interval_time_min=1200,single_num_min=3,drive_sts=3,charge_sts=[21,22],stand_sts=0,cell_type=cell_type,capacity=456)
- #df_drive,df_charge,df_stand,df_data_split_rlt=dt.split(df_merge,celltemp_name,cellvolt_name,drive_interval_time_min=1200,charge_interval_time_min=1200,stand_interval_time_min=1200,single_num_min=3,drive_sts=3,charge_sts=[21,22],stand_sts=0,cell_type=cell_type,capacity=456)
- df_data_split_rlt.to_excel(f'{sn_list[0]}_NEW.xlsx')
- loggers['Data_Split'].info(f'算法运行完成{pack_code},算法耗时{time.time()-time_st}')
- except Exception as e:
- loggers['Data_Split'].error('算法运行出错')
- loggers['Data_Split'].error(str(e))
- loggers['Data_Split'].error(traceback.format_exc())
- df_drive=pd.DataFrame()
- df_charge=pd.DataFrame()
- df_stand=pd.DataFrame()
- df_data_split_rlt=pd.DataFrame()
- try:
- if not df_data_split_rlt.empty:
- time_record = time.time()
- df_data_split_rlt.reset_index(drop = True, inplace = True)
- #df_data_split_rlt.to_sql("algo_charge_info",con=mysql_algo_conn, if_exists="append",index=False)
- #write_mysql_time = write_mysql_time + time.time()-time_record
- logger_main.info(f'process-{process_num}数据分段统计入库{pack_code}完成')
- logger_main.info(f"process-{process_num}: 结果入库耗时:{time.time()-time_st}")
-
- except Exception as e:
- logger_main.error(f"process-{process_num}:数据分段结果入库出错")
- logger_main.error(f"process-{process_num}:{e}")
- logger_main.error(f"process-{process_num}:{traceback.format_exc()}")
- except Exception as e:
- logger_main.error(f"process-{process_num}:获取原始数据出错")
- logger_main.error(f"process-{process_num}:{e}")
- logger_main.error(f"process-{process_num}:{traceback.format_exc()}")
- continue
-
-
-
- #算法5_日数据分段
-
-
-
-
-
-
-
- except Exception as e:
- logger_main.error(f'process-{process_num}: {e}')
- logger_main.error(f'process-{process_num}: {traceback.format_exc()}')
- cleanUtils.clean(mysql_algo_conn, mysql_algo_engine, mysql_iotp_conn, mysql_iotp_engine, kafka_consumer, rc)
- if __name__ == '__main__':
- while(True):
- try:
- # 配置量
-
- cur_env = 'dev' # 设置运行环境
- app_path = "/home/shouxueqi/projects/zlwl-algos/" # 设置app绝对路径
- log_base_path = f"{os.path.dirname(os.path.abspath(__file__))}/log" # 设置日志路径
- app_name = "task_day_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("开始分配子进程")
- main()
- except Exception as e:
- print(str(e))
- print(traceback.format_exc())
- 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()
|