| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124 |
- # -*- 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)
|
