import multiprocessing import traceback from os import path import numpy as np import pandas as pd from etl.common.PathsAndTable import PathsAndTable from etl.wind_power.min_sec import TransParam from etl.wind_power.min_sec.ClassIdentifier import ClassIdentifier from service.trans_conf_service import update_trans_transfer_progress from utils.conf.read_conf import read_conf from utils.df_utils.util import get_time_space from utils.file.trans_methods import create_file_path, read_excel_files, read_file_to_df, split_array from utils.log.trans_log import trans_print from utils.systeminfo.sysinfo import use_files_get_max_cpu_count exec("import math") class StatisticsAndSaveTmpFormalFile(object): def __init__(self, paths_and_table: PathsAndTable, trans_param: TransParam, statistics_map, rated_power_and_cutout_speed_map): self.paths_and_table = paths_and_table self.trans_param = trans_param self.statistics_map = statistics_map self.lock = multiprocessing.Manager().Lock() self.rated_power_and_cutout_speed_map = rated_power_and_cutout_speed_map def set_statistics_data(self, df): if not df.empty: df['time_stamp'] = pd.to_datetime(df['time_stamp']) min_date = df['time_stamp'].min() max_date = df['time_stamp'].max() with self.lock: if 'min_date' in self.statistics_map.keys(): if self.statistics_map['min_date'] > min_date: self.statistics_map['min_date'] = min_date else: self.statistics_map['min_date'] = min_date if 'max_date' in self.statistics_map.keys(): if self.statistics_map['max_date'] < max_date: self.statistics_map['max_date'] = max_date else: self.statistics_map['max_date'] = max_date if 'total_count' in self.statistics_map.keys(): self.statistics_map['total_count'] = self.statistics_map['total_count'] + df.shape[0] else: self.statistics_map['total_count'] = df.shape[0] if 'time_granularity' not in self.statistics_map.keys(): self.statistics_map['time_granularity'] = get_time_space(df, 'time_stamp') def save_to_csv(self, filename): df = read_file_to_df(filename) if self.trans_param.is_vertical_table: df = df.pivot_table(index=['time_stamp', 'wind_turbine_number'], columns=self.trans_param.vertical_key, values=self.trans_param.vertical_value, aggfunc='max') # 重置索引以得到普通的列 df.reset_index(inplace=True) # 转化风机名称 origin_wind_name = str(df['wind_turbine_number'].values[0]) df['wind_turbine_number'] = df['wind_turbine_number'].astype('str') # df['wind_turbine_name'] = df['wind_turbine_number'] df['wind_turbine_number'] = df['wind_turbine_number'].map( self.trans_param.wind_col_trans).fillna(df['wind_turbine_number']) wind_col_name = str(df['wind_turbine_number'].values[0]) not_double_cols = ['wind_turbine_number', 'wind_turbine_name', 'time_stamp', 'param6', 'param7', 'param8', 'param9', 'param10'] # 删除 有功功率 和 风速均为空的情况 df.dropna(subset=['active_power', 'wind_velocity'], how='all', inplace=True) trans_print(origin_wind_name, wind_col_name, "删除有功功率和风速均为空的情况后:", df.shape) df.replace(np.nan, -999999999, inplace=True) number_cols = df.select_dtypes(include=['number']).columns.tolist() for col in df.columns: if col not in not_double_cols and col not in number_cols: if not df[col].isnull().all(): df[col] = pd.to_numeric(df[col], errors='coerce') # 删除包含NaN的行（即那些列A转换失败的行） df = df.dropna(subset=[col]) trans_print(origin_wind_name, wind_col_name, "删除非数值列名:", col) df.replace(-999999999, np.nan, inplace=True) df.drop_duplicates(['wind_turbine_number', 'time_stamp'], keep='first', inplace=True) df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce") df.dropna(subset=['time_stamp'], inplace=True) df.sort_values(by='time_stamp', inplace=True) df = df[[i for i in self.trans_param.cols_tran.keys() if i in df.columns]] # 如果秒级有可能合并到分钟级 # TODO add 秒转分钟 if self.trans_param.boolean_sec_to_min: df['time_stamp'] = df['time_stamp'].apply(lambda x: x + pd.Timedelta(minutes=(10 - x.minute % 10) % 10)) df['time_stamp'] = df['time_stamp'].dt.floor('10T') df = df.groupby(['wind_turbine_number', 'time_stamp']).mean().reset_index() trans_print('有功功率前10个', df.head(10)['active_power'].values) power_df = df[df['active_power'] > 0] trans_print(origin_wind_name, wind_col_name, "功率大于0的数量:", power_df.shape) power = power_df.sample(int(power_df.shape[0] / 100))['active_power'].median() del power_df trans_print(origin_wind_name, wind_col_name, '有功功率，中位数', power) if power > 100000: df['active_power'] = df['active_power'] / 1000 ## 做数据检测前,羡强行处理有功功率 # df = df[df['active_power'] < 50000] rated_power_and_cutout_speed_tuple = read_conf(self.rated_power_and_cutout_speed_map, str(wind_col_name)) if rated_power_and_cutout_speed_tuple is None: rated_power_and_cutout_speed_tuple = (None, None) # 如果有需要处理的,先进行代码处理,在进行打标签 # exec_code = get_trans_exec_code(self.paths_and_table.exec_id, self.paths_and_table.read_type) # if exec_code: # if 'import ' in exec_code: # raise Exception("执行代码不支持导入包") # exec(exec_code) class_identifiler = ClassIdentifier(wind_turbine_number=origin_wind_name, origin_df=df, rated_power=rated_power_and_cutout_speed_tuple[0], cut_out_speed=rated_power_and_cutout_speed_tuple[1]) df = class_identifiler.run() df['year'] = df['time_stamp'].dt.year df['month'] = df['time_stamp'].dt.month df['day'] = df['time_stamp'].dt.day df['time_stamp'] = df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S')) df['wind_turbine_name'] = str(origin_wind_name) df['year_month'] = df[['year', 'month']].apply(lambda x: str(x['year']) + str(x['month']).zfill(2), axis=1) cols = df.columns if self.paths_and_table.read_type == 'second': type_col = 'year_month' else: type_col = 'year' date_strs = df[type_col].unique().tolist() for date_str in date_strs: save_path = path.join(self.paths_and_table.get_tmp_formal_path(), str(date_str), str(origin_wind_name) + '.csv') create_file_path(save_path, is_file_path=True) now_df = df[df[type_col] == date_str][cols] if self.paths_and_table.save_zip: save_path = save_path + '.gz' now_df.to_csv(save_path, compression='gzip', index=False, encoding='utf-8') else: now_df.to_csv(save_path, index=False, encoding='utf-8') del now_df self.set_statistics_data(df) del df trans_print("保存" + str(wind_col_name) + "成功") def mutiprocessing_to_save_file(self): # 开始保存到正式文件 all_tmp_files = read_excel_files(self.paths_and_table.get_read_tmp_path()) # split_count = self.pathsAndTable.multi_pool_count split_count = use_files_get_max_cpu_count(all_tmp_files) all_arrays = split_array(all_tmp_files, split_count) try: for index, arr in enumerate(all_arrays): with multiprocessing.Pool(split_count) as pool: pool.starmap(self.save_to_csv, [(i,) for i in arr]) update_trans_transfer_progress(self.paths_and_table.id, round(50 + 15 * (index + 1) / len(all_arrays), 2), self.paths_and_table.save_db) except Exception as e: trans_print(traceback.format_exc()) message = "保存文件错误,系统返回错误:" + str(e) raise ValueError(message) def run(self): self.mutiprocessing_to_save_file() update_trans_transfer_progress(self.paths_and_table.id, 65, self.paths_and_table.save_db)