# -*- coding: utf-8 -*- # @Time : 2024/5/15 # @Author : 魏志亮 import multiprocessing import pandas as pd from etl.common.BaseDataTrans import BaseDataTrans from etl.wind_power.min_sec.ReadAndSaveTmp import ReadAndSaveTmp from etl.wind_power.min_sec.StatisticsAndSaveFile import StatisticsAndSaveFile from etl.wind_power.min_sec.TransParam import TransParam from service.plt_service import update_trans_status_success, update_trans_status_error from service.trans_service import batch_statistics, get_min_sec_conf from utils.conf.read_conf import read_conf from utils.df_utils.util import get_time_space from utils.file.trans_methods import read_excel_files, read_file_to_df from utils.log.trans_log import trans_print class MinSecTrans(BaseDataTrans): def __init__(self, data: dict = None, save_db=True, step=0, end=4): super(MinSecTrans, self).__init__(data, save_db, step, end) self.statistics_map = multiprocessing.Manager().dict() self.trans_param = self.get_trans_param() self.trans_param.wind_col_trans = self.wind_col_trans def get_filed_conf(self): return get_min_sec_conf(self.field_code, self.read_type) def get_trans_param(self): conf_map = self.get_filed_conf() if conf_map is None or type(conf_map) == tuple or len(conf_map.keys()) == 0: message = f"未找到{self.batch_no}的{self.read_type}配置" trans_print(message) update_trans_status_error(self.batch_no, self.read_type, message, self.save_db) else: resolve_col_prefix = read_conf(conf_map, 'resolve_col_prefix') wind_name_exec = read_conf(conf_map, 'wind_name_exec', None) is_vertical_table = read_conf(conf_map, 'is_vertical_table', False) merge_columns = read_conf(conf_map, 'merge_columns', False) vertical_cols = read_conf(conf_map, 'vertical_read_cols', '').split(',') index_cols = read_conf(conf_map, 'vertical_index_cols', '').split(',') vertical_key = read_conf(conf_map, 'vertical_col_key') vertical_value = read_conf(conf_map, 'vertical_col_value') need_valid_cols = not merge_columns boolean_sec_to_min = read_conf(conf_map, 'boolean_sec_to_min', 0) boolean_sec_to_min = int(boolean_sec_to_min) == 1 # self.boolean_sec_to_min = int(data['boolean_sec_to_min']) == 1 if 'boolean_sec_to_min' in data.keys() else False cols_trans_all = dict() trans_cols = ['wind_turbine_number', 'time_stamp', 'active_power', 'rotor_speed', 'generator_speed', 'wind_velocity', 'pitch_angle_blade_1', 'pitch_angle_blade_2', 'pitch_angle_blade_3', 'cabin_position', 'true_wind_direction', 'yaw_error1', 'set_value_of_active_power', 'gearbox_oil_temperature', 'generatordrive_end_bearing_temperature', 'generatornon_drive_end_bearing_temperature', 'wind_turbine_status', 'wind_turbine_status2', 'cabin_temperature', 'twisted_cable_angle', 'front_back_vibration_of_the_cabin', 'side_to_side_vibration_of_the_cabin', 'actual_torque', 'given_torque', 'clockwise_yaw_count', 'counterclockwise_yaw_count', 'unusable', 'power_curve_available', 'required_gearbox_speed', 'inverter_speed_master_control', 'outside_cabin_temperature', 'main_bearing_temperature', 'gearbox_high_speed_shaft_bearing_temperature', 'gearboxmedium_speed_shaftbearing_temperature', 'gearbox_low_speed_shaft_bearing_temperature', 'generator_winding1_temperature', 'generator_winding2_temperature', 'generator_winding3_temperature', 'turbulence_intensity', 'param1', 'param2', 'param3', 'param4', 'param5', 'param6', 'param7', 'param8', 'param9', 'param10'] for col in trans_cols: cols_trans_all[col] = read_conf(conf_map, col, '') return TransParam(read_type=self.read_type, read_path=self.read_path, cols_tran=cols_trans_all, wind_name_exec=wind_name_exec, is_vertical_table=is_vertical_table, vertical_cols=vertical_cols, vertical_key=vertical_key, vertical_value=vertical_value, index_cols=index_cols, merge_columns=merge_columns, resolve_col_prefix=resolve_col_prefix, need_valid_cols=need_valid_cols, boolean_sec_to_min=boolean_sec_to_min) # 第三步 读取 并 保存到临时文件 def read_and_save_tmp_file(self): read_and_save_tmp = ReadAndSaveTmp(self.pathsAndTable, self.trans_param) read_and_save_tmp.run() # 第四步 统计 并 保存到正式文件 def statistics_and_save_to_file(self): # 保存到正式文件 statistics_and_save_file = StatisticsAndSaveFile(self.pathsAndTable, self.trans_param, self.statistics_map, self.rated_power_and_cutout_speed_map) statistics_and_save_file.run() # 最后更新执行程度 def update_exec_progress(self): if self.end >= 4: all_files = read_excel_files(self.pathsAndTable.get_save_path()) if self.step <= 3: update_trans_status_success(self.batch_no, self.trans_param.read_type, len(all_files), self.statistics_map['time_granularity'], self.statistics_map['min_date'], self.statistics_map['max_date'], self.statistics_map['total_count'], self.save_db) else: df = read_file_to_df(all_files[0], read_cols=['time_stamp']) df['time_stamp'] = pd.to_datetime(df['time_stamp']) time_granularity = get_time_space(df, 'time_stamp') batch_data = batch_statistics("_".join([self.batch_no, self.trans_param.read_type])) if batch_data is not None: update_trans_status_success(self.batch_no, self.trans_param.read_type, len(read_excel_files(self.pathsAndTable.get_save_path())), time_granularity, batch_data['min_date'], batch_data['max_date'], batch_data['total_count'], self.save_db) else: update_trans_status_success(self.batch_no, self.trans_param.read_type, len(read_excel_files(self.pathsAndTable.get_save_path())), time_granularity, None, None, None, self.save_db)