energy-data-trans/etl/wind_power/min_sec/MinSecTrans.py

# -*- coding: utf-8 -*-
# @Time    : 2024/5/15
# @Author  : 魏志亮
import multiprocessing
import os.path

from etl.common.BaseDataTrans import BaseDataTrans
from etl.common.CombineAndSaveFormalFile import CombineAndSaveFormalFile
from etl.wind_power.min_sec.ReadAndSaveTmp import ReadAndSaveTmp
from etl.wind_power.min_sec.StatisticsAndSaveTmpFormalFile import StatisticsAndSaveTmpFormalFile
from etl.wind_power.min_sec.TransParam import TransParam
from service.trans_conf_service import update_trans_status_success, update_trans_status_error
from service.trans_service import get_min_sec_conf
from utils.conf.read_conf import read_conf
from utils.log.trans_log import trans_print


class MinSecTrans(BaseDataTrans):

    def __init__(self, data: dict = None, save_db=True, yaml_config=None, step=0, end=6):
        super(MinSecTrans, self).__init__(data, save_db, yaml_config, 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.wind_farm_code, self.transfer_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.id}的{self.transfer_type}配置"
            trans_print(message)
            update_trans_status_error(self.id, 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.transfer_type, read_path=self.read_dir,
                              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_tmp_formal_file(self):
        # 保存到正式文件
        statistics_and_save_tmp_formal_file = StatisticsAndSaveTmpFormalFile(self.pathsAndTable, self.trans_param,
                                                                             self.statistics_map,
                                                                             self.rated_power_and_cutout_speed_map)
        statistics_and_save_tmp_formal_file.run()

    def combine_and_save_formal_file(self):
        combine_and_save_formal_file = CombineAndSaveFormalFile(self.pathsAndTable)
        self.update_files = combine_and_save_formal_file.run()

    # 最后更新执行程度
    def update_exec_progress(self):
        all_files = set([os.path.basename(i) for i in self.update_files])
        update_trans_status_success(self.id, 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)