CGN_taiyuan_104/service/trans_service.py

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

import pandas as pd

from service.common_connect import trans
from utils.log.trans_log import logger


def create_tmp_table(table_name):
    create_sql = f"""
    CREATE TABLE `{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 '湍流强度'
  )
    """
    trans.execute(create_sql)


def boolean_table_exists(table_name):
    table_sql = f"""
    select count(1) as count from information_schema.tables where table_name = '{table_name}'
    """
    data = trans.execute(table_sql)[0]

    if int(data['count']) == 0:
        return False
    return True


def add_partition(table_name, pname, date_str):
    try:
        sql = f"""
        ALTER TABLE {table_name} REORGANIZE PARTITION pmax INTO (
            PARTITION {pname} VALUES LESS THAN ('{date_str}'),
            PARTITION pmax VALUES LESS THAN (MAXVALUE)
        );
        """
        trans.execute(sql)
        logger.info(f"添加{table_name}分区{pname}成功")
    except:
        logger.error(traceback.format_exc())


def delelet_partition(table_name, pmonth):
    pname = f'p{str(pmonth.year) + str(pmonth.month).zfill(2)}'

    exists_partition_sql = f"""
    SELECT count(1) from INFORMATION_SCHEMA.`PARTITIONS` t  where t.TABLE_NAME = '{table_name}' and t.PARTITION_NAME = '{pname}'
    """
    data = trans.execute(exists_partition_sql)[0]
    if data > 0:
        del_sql = f"""
        ALTER TABLE {table_name} DROP PARTITION {pname}
        """
        trans.execute(del_sql)
        logger.info(f"删除{table_name}分区{pname}成功")

    else:
        logger.info(f"删除{table_name}分区{pname}不存在")


def get_all_partitioned_tables() -> list:
    all_tables_sql = """
    SELECT t.TABLE_NAME FROM INFORMATION_SCHEMA.`TABLES` t where t.CREATE_OPTIONS = 'partitioned'
    """
    return trans.execute(all_tables_sql)


def save_df_to_db(table_name: str, df: pd.DataFrame(), batch_count=1000):
    try:
        if 'index' in df.columns:
            del df['index']
        trans.execute_df_save(df, table_name, batch_count)
    except Exception as e:
        logger.error(traceback.format_exc())
        raise Exception(str(e))


def load_data_local(table_name, df):
    columns_str = 'wind_turbine_number,wind_turbine_name,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,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,wind_turbine_status,wind_turbine_status2,turbulence_intensity,lab,year,month,day,year_month'
    cols = columns_str.split(',')
    print(cols)
    df = df[cols]
    # trans.execute_df_save(df, table_name, batch_count)
    trans.safe_load_data_local(df, table_name)


def drop_table(table_name):
    drop_sql = f"DROP TABLE `{table_name}`"
    try:
        trans.execute(drop_sql)
    except:
        logger.error(traceback.format_exc())


def get_yesterday_tables(yesterday):
    query_sql = f"""
    select * from wind_farm_day_count where add_date = '{yesterday}' and sync_status = 0
    """
    return trans.execute(query_sql)


def update_sync(id):
    update_sql = f"update wind_farm_day_count set sync_status = 1 where id = {id}"
    trans.execute(update_sql)


def update_wind_farm_day_count(wind_farm_code, wind_farm_name, add_date, trans_type, count, latest_data_time,
                               sync_status=0):
    select_sql = f"SELECT * from wind_farm_day_count  WHERE `wind_farm_code` = '{wind_farm_code}' and `add_date` = '{add_date}' and `type` = '{trans_type}' "
    result = trans.execute(select_sql)
    if result:
        id = result[0]['id']
        update_sql = f"update wind_farm_day_count set count = count + {count}, latest_data_time = '{latest_data_time}' where id = {id}"
        trans.execute(update_sql)
    else:
        insert_sql = f"""
        INSERT INTO `wind_farm_day_count` (
        `wind_farm_code`,
        `wind_farm_name`,
        `add_date`,
        `type`,
        `count`,
        `sync_status`,
        `del_status`,
        `latest_data_time`
    )
        VALUES
        (
            '{wind_farm_code}',
            '{wind_farm_name}',
            '{add_date}',
            '{trans_type}',
            '{count}',
            '{sync_status}',
            '0',
            '{latest_data_time}'
        )
        """
        trans.execute(insert_sql)


def get_sys_conf_by_key(type, param_key, default_value=None):
    sql = f"SELECT * from sys_conf t where t.type ='{type}'  and t.param_key = '{param_key}' and status = 1"
    datas = trans.execute(sql)
    if isinstance(datas, tuple):
        return default_value
    else:
        return datas[0]['param_value']


def get_sys_conf(type) -> dict:
    sql = f"SELECT * from sys_conf t where t.type ='{type}' and status = 1"
    datas = trans.execute(sql)
    if isinstance(datas, tuple):
        return {}
    else:
        result_dict = dict()
        for data in datas:
            result_dict[data['param_key']] = data['param_value']
        return result_dict


def read_data_from_table(table_name):
    df = pd.read_sql_table(con=trans.get_engine(), table_name=table_name)
    return df


def update_warn_fault_exist_data(changzhan, seq_no, end_time):
    updata_tables = [f'{changzhan}_warn', f'{changzhan}_fault']

    for table_name in updata_tables:
        update_sql = f"""
        update {table_name} set end_time = '{end_time}', time_diff = TIMESTAMPDIFF(SECOND, begin_time,'{end_time}') 
        where seq_no = {seq_no} and end_time is null
        """
        trans.execute(update_sql)
        logger.info(f"更新{changzhan}故障顺序号{seq_no}成功")


def update_expired_data(table_type, exists_date):
    update_expired_sql = f"""
    update wind_farm_day_count set del_status = 1 where type = '{table_type}' and add_date < '{exists_date}'
    """
    trans.execute(update_expired_sql)
    logger.info(f"删除类型{table_type},截止日期{exists_date}成功")


def exists_windno_seq_fault(changzhan):
    types = ['warn', 'fault']
    result_dict = dict(list())
    for type in types:
        result_dict[type] = list()
        query_sql = f"select * from {changzhan}_{type} where time_diff is null"
        fault_datas = trans.execute(query_sql)
        if isinstance(fault_datas, tuple):
            result_dict[type] = []
        else:
            for data in fault_datas:
                # t.wind_turbine_name,t.seq_no,t.fault_code
                result_dict[type].append(f"{data['wind_turbine_name']}_{data['seq_no']}_{data['fault_code']}")

    return result_dict


def update_warn_fault_update_time(date_str, max_time):
    sql = f"update wind_farm_day_count set update_time = '{max_time}' where add_date = '{date_str}' and type in ('warn','fault')"
    trans.execute(sql)
    logger.info(f"更新{date_str}的故障报警到当前时间")


if __name__ == '__main__':
    update_wind_farm_day_count('1', '1', '2025-04010', 100)