energy-data-trans/service/trans_service.py

# -*- coding: utf-8 -*-
# @Time    : 2024/6/7
# @Author  : 魏志亮
import os
import traceback

import pandas as pd

from utils.db.ConnectMysql import ConnectMysql
from utils.file.trans_methods import split_array
from utils.log.trans_log import trans_print

trans = ConnectMysql("trans")


def get_min_sec_conf(field_code, trans_type) -> dict:
    query_sql = "SELECT * FROM trans_conf where wind_code = %s and type = %s and status = 1"
    res = trans.execute(query_sql, (field_code, trans_type))
    print(res)
    if type(res) == tuple:
        return None
    return res[0]


def get_min_sec_conf_test(field_code, trans_type) -> dict:
    query_sql = "SELECT * FROM trans_conf where wind_name = %s and type = %s and status = 1"
    res = trans.execute(query_sql, (field_code, trans_type))
    print(res)
    if type(res) == tuple:
        return None
    return res[0]


def get_fault_warn_conf(field_code, trans_type) -> dict:
    types = list()
    if trans_type == 'fault':
        types.append(1)
    elif trans_type == 'warn':
        types.append(2)
    else:
        trans_print(f"未找到{trans_type}告警/故障的配置")
        raise ValueError(f"未找到{trans_type}告警/故障的配置")

    types.append(3)

    query_sql = "SELECT * FROM warn_fault_conf where wind_code = %s and type in %s and status = 1"
    res = trans.execute(query_sql, (field_code, types))
    print(res)
    if type(res) == tuple:
        return None
    return res[0]


def get_wave_conf(field_code) -> dict:
    query_sql = "SELECT * FROM wave_conf where wind_code = %s and status = 1"
    res = trans.execute(query_sql, (field_code))
    print(res)
    if type(res) == tuple:
        return None
    return res[0]


def creat_min_sec_table(table_name, win_names, read_type):
    create_sql = f"""
    CREATE TABLE
    IF NOT EXISTS `{table_name}` (
        `wind_turbine_number` VARCHAR (20) DEFAULT NULL COMMENT '风机编号',
        `wind_turbine_name` VARCHAR(20) DEFAULT NULL COMMENT '风机原始名称',
        `time_stamp` datetime NOT NULL COMMENT '时间戳',
        `active_power` DOUBLE DEFAULT NULL COMMENT '有功功率',
        `rotor_speed` DOUBLE DEFAULT NULL COMMENT '风轮转速',
        `generator_speed` DOUBLE DEFAULT NULL COMMENT '发电机转速',
        `wind_velocity` DOUBLE DEFAULT NULL COMMENT '风速',
        `pitch_angle_blade_1` DOUBLE DEFAULT NULL COMMENT '桨距角1',
        `pitch_angle_blade_2` DOUBLE DEFAULT NULL COMMENT '桨距角2',
        `pitch_angle_blade_3` DOUBLE DEFAULT NULL COMMENT '桨距角3',
        `cabin_position` DOUBLE DEFAULT NULL COMMENT '机舱位置',
        `true_wind_direction` DOUBLE DEFAULT NULL COMMENT '绝对风向',
        `yaw_error1` DOUBLE DEFAULT NULL COMMENT '对风角度',
        `set_value_of_active_power` DOUBLE DEFAULT NULL COMMENT '有功功率设定值',
        `gearbox_oil_temperature` DOUBLE DEFAULT NULL COMMENT '齿轮箱油温',
        `generatordrive_end_bearing_temperature` DOUBLE DEFAULT NULL COMMENT '发电机驱动端轴承温度',
        `generatornon_drive_end_bearing_temperature` DOUBLE DEFAULT NULL COMMENT '发电机非驱动端轴承温度',
        `cabin_temperature` DOUBLE DEFAULT NULL COMMENT '机舱内温度',
        `twisted_cable_angle` DOUBLE DEFAULT NULL COMMENT '扭缆角度',
        `front_back_vibration_of_the_cabin` DOUBLE DEFAULT NULL COMMENT '机舱前后振动',
        `side_to_side_vibration_of_the_cabin` DOUBLE DEFAULT NULL COMMENT '机舱左右振动',
        `actual_torque` DOUBLE DEFAULT NULL COMMENT '实际力矩',
        `given_torque` DOUBLE DEFAULT NULL COMMENT '给定力矩',
        `clockwise_yaw_count` DOUBLE DEFAULT NULL COMMENT '顺时针偏航次数',
        `counterclockwise_yaw_count` DOUBLE DEFAULT NULL COMMENT '逆时针偏航次数',
        `unusable` DOUBLE DEFAULT NULL COMMENT '不可利用',
        `power_curve_available` DOUBLE DEFAULT NULL COMMENT '功率曲线可用',
        `required_gearbox_speed` DOUBLE DEFAULT NULL COMMENT '齿轮箱转速',
        `inverter_speed_master_control` DOUBLE DEFAULT NULL COMMENT '变频器转速(主控)',
        `outside_cabin_temperature` DOUBLE DEFAULT NULL COMMENT '环境温度',
        `main_bearing_temperature` DOUBLE DEFAULT NULL COMMENT '主轴承轴承温度',
        `gearbox_high_speed_shaft_bearing_temperature` DOUBLE DEFAULT NULL COMMENT '齿轮箱高速轴轴承温度',
        `gearboxmedium_speed_shaftbearing_temperature` DOUBLE DEFAULT NULL COMMENT '齿轮箱中速轴轴承温度',
        `gearbox_low_speed_shaft_bearing_temperature` DOUBLE DEFAULT NULL COMMENT '齿轮箱低速轴轴承温度',
        `generator_winding1_temperature` DOUBLE DEFAULT NULL COMMENT '发电机绕组1温度',
        `generator_winding2_temperature` DOUBLE DEFAULT NULL COMMENT '发电机绕组2温度',
        `generator_winding3_temperature` DOUBLE DEFAULT NULL COMMENT '发电机绕组3温度',
        `wind_turbine_status` DOUBLE DEFAULT NULL COMMENT '风机状态1',
        `wind_turbine_status2` DOUBLE DEFAULT NULL COMMENT '风机状态2',
        `turbulence_intensity` DOUBLE DEFAULT NULL COMMENT '湍流强度',
        `lab` int DEFAULT NULL COMMENT '-1:停机 0:好点  1:欠发功率点；2:超发功率点；3:额定风速以上的超发功率点 4: 限电',
        `year` INT (4) DEFAULT NULL COMMENT '年',
        `month` INT (2) DEFAULT NULL COMMENT '月',
        `day` INT (2) DEFAULT NULL COMMENT '日',
        `param1` DOUBLE DEFAULT NULL COMMENT '预留1',
        `param2` DOUBLE DEFAULT NULL COMMENT '预留2',
        `param3` DOUBLE DEFAULT NULL COMMENT '预留3',
        `param4` DOUBLE DEFAULT NULL COMMENT '预留4',
        `param5` DOUBLE DEFAULT NULL COMMENT '预留5',
        `param6` VARCHAR (20) DEFAULT NULL COMMENT '预留6',
        `param7` VARCHAR (20) DEFAULT NULL COMMENT '预留7',
        `param8` VARCHAR (20) DEFAULT NULL COMMENT '预留8',
        `param9` VARCHAR (20) DEFAULT NULL COMMENT '预留9',
        `param10` VARCHAR (20) DEFAULT NULL COMMENT '预留10',
         KEY `time_stamp` (`time_stamp`),
         KEY `wind_turbine_number` (`wind_turbine_number`)
    ) ENGINE = myisam DEFAULT CHARSET = utf8mb4
    """

    if read_type == 'second' and win_names and len(win_names) > 1:

        create_sql = create_sql + f" PARTITION BY LIST COLUMNS(`wind_turbine_number`) ("
        partition_strs = list()
        for wind_name in win_names:
            partition_strs.append(f" PARTITION p{wind_name} VALUES IN('{wind_name}')")

        create_sql = create_sql + ",".join(partition_strs) + ")"

    trans.execute(create_sql)


def rename_table(table_name, renamed_table_name, save_db=True):
    if save_db:
        rename_sql = f"RENAME TABLE {table_name} TO {renamed_table_name}"
        try:
            trans.execute(rename_sql)
        except:
            trans_print(traceback.format_exc())


def drop_table(table_name, save_db=True):
    if save_db:
        rename_sql = f"drop TABLE `{table_name}`"
        try:
            trans.execute(rename_sql)
        except:
            trans_print(traceback.format_exc())


def clear_table(table_name, save_db=True):
    if save_db:
        rename_sql = f"truncate TABLE `{table_name}`"
        try:
            trans.execute(rename_sql)
        except:
            trans_print(traceback.format_exc())


def save_file_to_db(table_name: str, file: str, batch_count=100000):
    base_name = os.path.basename(file)
    try:
        for i, df in enumerate(pd.read_csv(file, chunksize=batch_count)):
            # df.to_sql(table_name, engine, if_exists='append', index=False)
            trans.execute_df_save(df, table_name)
            count = (i + 1) * batch_count
            trans_print(base_name, f"Chunk {count} written to MySQL.")
    except Exception as e:
        trans_print(traceback.format_exc())
        message = base_name + str(e)
        raise Exception(message)


def save_df_to_db(table_name: str, df: pd.DataFrame(), batch_count=100000):
    split_dfs = [df.iloc[i:i + batch_count] for i in range(0, len(df), batch_count)]
    try:
        for i, split_df in enumerate(split_dfs):
            trans.execute_df_save(split_df, table_name)
            count = (i + 1) * batch_count
            trans_print(f"Chunk {count} written to MySQL.")
    except Exception as e:
        trans_print(traceback.format_exc())
        raise Exception(str(e))


def batch_statistics(table_name):
    query_sql = f"select count(1) as total_count ,min(t.time_stamp) as min_date ,max(t.time_stamp) as max_date from `{table_name}` t "
    try:
        res = trans.execute(query_sql)
        return res[0]
    except:
        trans_print(traceback.format_exc())
        return None


def create_warn_fault_table(table_name):
    sql = f"""
    CREATE TABLE `{table_name}` (
      `wind_turbine_number` varchar(20) DEFAULT NULL COMMENT '风机编号',
      `wind_turbine_name` varchar(20) DEFAULT NULL COMMENT '原始风机编号',
      `begin_time` datetime DEFAULT NULL COMMENT '开始时间',
      `end_time` datetime DEFAULT NULL COMMENT '结束时间',
      `time_diff` int DEFAULT NULL COMMENT '处理耗时,单位秒',
      `fault_id` varchar(20) DEFAULT NULL COMMENT '报警或者故障ID',
      `fault_code` varchar(50) DEFAULT NULL COMMENT '报警或者故障CODE',
      `fault_detail` varchar(255) DEFAULT NULL COMMENT '错误描述',
      `fault_level` varchar(20) DEFAULT NULL COMMENT '报警等级',
      `fault_type` varchar(20) DEFAULT NULL COMMENT '报警类型',
      `stop_status` varchar(20) DEFAULT NULL COMMENT '刹车状态',
      KEY `wind_turbine_number` (`wind_turbine_number`),
      KEY `begin_time` (`begin_time`),
      KEY `end_time` (`end_time`)
    ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4
    """

    trans.execute(sql)


def get_or_create_wave_table(table_name):
    create_table = False
    query_sql = f"select 1 from `{table_name}` limit 1"
    try:
        trans.execute(query_sql)
    except:
        create_table = True

    if create_table:
        sql = f"""
        CREATE TABLE `{table_name}` (
          `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
          `wind_turbine_number` varchar(20) DEFAULT NULL COMMENT '风机编号',
          `wind_turbine_name` varchar(20) DEFAULT NULL COMMENT '原始风机编号',
          `time_stamp` datetime DEFAULT NULL COMMENT '时间',
          `sampling_frequency` varchar(50) DEFAULT NULL COMMENT '分析频率',
          `mesure_point_name` varchar(100) DEFAULT NULL COMMENT '测点名称',
          `mesure_data` mediumtext COMMENT '测点数据',
          PRIMARY KEY (`id`),
          KEY `wind_turbine_number` (`wind_turbine_number`),
          KEY `time_stamp` (`time_stamp`),
          KEY `mesure_point_name` (`mesure_point_name`)
        ) ENGINE=MyISAM DEFAULT CHARSET=utf8mb4
        """

        trans.execute(sql)


def get_wave_data(table_name, min_data, max_data):
    query_sql = f"""
    select  id,wind_turbine_number,wind_turbine_name,time_stamp,sampling_frequency,mesure_point_name from `{table_name}` where time_stamp >= '{min_data}' and time_stamp <= '{max_data}'
    """
    return trans.read_sql_to_df(query_sql)


def delete_exist_wave_data(table_name, ids):
    all_arrays = split_array(ids, 1000)
    for array in all_arrays:
        ids_str = ",".join(['%s'] * len(array))
        delete_sql = f"delete from `{table_name}` where id in ({ids_str})"
        trans.execute(delete_sql, array)


if __name__ == '__main__':
    # path_prix = r"/data/download/collection_data/2完成/招远风电场-山东-大唐/清理数据/WOF063100040-WOB00013/second"
    # files = ["WOG00030.csv", "WOG00034.csv"]
    # for path in files:
    #     save_file_to_db("WOF063100040-WOB00013_second", path_prix + os.sep + path, batch_count=100000)

    # sql = """
    # SELECT wind_turbine_number, time_stamp, wind_velocity, active_power
    #                            FROM `WOF085500002-WOB000001_second`
    #                            WHERE  time_stamp >= '2024-02-17 00:00:00' AND time_stamp <= '2024-05-14 00:00:00' AND lab = 0
    # """
    #
    # begin = datetime.datetime.now()
    # df = trans.read_sql_to_df(sql)
    # end = datetime.datetime.now()
    # print(df.shape)
    # print(df.info())
    # print("Time used:", (end - begin).seconds)
    # get_fault_warn_conf("test", "fault")

    delete_exist_wave_data('SKF001_wave', [1, 2, 3])