energy-data-trans/app.py

# -*- coding: utf-8 -*-
# @Time    : 2024/5/17
# @Author  : 魏志亮
import datetime

import pandas as pd

from base.TranseParam import TranseParam
from base.WindFarms import WindFarms
from utils.conf.read_conf import read_yaml_file, read_param_from_config
from utils.db.trans_mysql import get_exec_data, get_trans_conf
from utils.log.trans_log import trans_print, logger, init_log


def init_params() -> list[WindFarms]:
    df = get_exec_data()
    print(df)
    exec_subjects = list()

    path = r'D:\transdata\test\虹梯官风电场-山西-大唐\收资数据'
    changshang_name = '虹梯官风电场'
    # read_type = 'minute'
    read_type = 'second'
    batch_no = 'hongtiguan_' + str(datetime.datetime.now().strftime('%Y%m%d'))

    df = pd.DataFrame(columns=['batch_no', 'transfer_type', 'transfer_file_addr', 'field_code', 'field_name'],
                      data=[
                          [batch_no, read_type, path, 'A', changshang_name]])

    if df is None:
        trans_print("当前任务正在执行")

    if df.empty:
        trans_print("当前没有需要执行的任务")
    else:
        for batch_no, transfer_type, transfer_file_addr, field_code, field_name in zip(df['batch_no'],
                                                                                       df['transfer_type'],
                                                                                       df['transfer_file_addr'],
                                                                                       df['field_code'],
                                                                                       df['field_name']):
            init_log(batch_no, field_name, transfer_type)

            conf_df = get_trans_conf(field_name, transfer_type)
            if conf_df is None:
                trans_print(f"未找到{field_name}的{transfer_type}配置")
                continue

            conf = conf_df.iloc[0].to_dict()

            resolve_col_prefix = read_param_from_config(conf, 'resolve_col_prefix')
            wind_name_exec = read_param_from_config(conf, 'wind_name_exec', None)
            wind_full_name = read_param_from_config(conf, 'wind_full_name')
            is_vertical_table = read_param_from_config(conf, 'is_vertical_table', False)
            merge_columns = read_param_from_config(conf, 'merge_columns', False)

            vertical_cols = read_param_from_config(conf, 'vertical_read_cols', '').split(',')
            index_cols = read_param_from_config(conf, 'vertical_index_cols', '').split(',')
            vertical_key = read_param_from_config(conf, 'vertical_col_key')
            vertical_value = read_param_from_config(conf, 'vertical_col_value')
            need_valid_cols = not merge_columns

            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_param_from_config(conf, col, '')

            trans_subject = WindFarms(field_name, batch_no=batch_no, field_code=field_code,
                                      wind_full_name=wind_full_name)

            params = TranseParam(read_type=transfer_type, read_path=transfer_file_addr,
                                 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)

            trans_subject.set_trans_param(params)
            exec_subjects.append(trans_subject)
    return exec_subjects


if __name__ == '__main__':
    exec_subjects = init_params()
    for exec_subject in exec_subjects:
        try:
            exec_subject.run()
        except Exception as e:
            print(e)
            logger.exception(e)