# -*- coding: utf-8 -*- # @Time : 2024/6/7 # @Author : 魏志亮 import os import traceback import pandas as pd from utils.db.ConnectMysql import ConnectMysql 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) 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")