# -*- coding: utf-8 -*- # @Time : 2025/1/9 # @Author : 魏志亮 from datetime import datetime from service.common_connect import trans from utils.log.trans_log import info def update_timeout_trans_data(): sql = """ UPDATE data_transfer SET trans_sys_status = 2,err_info='运行超时失败',transfer_status=2 WHERE TIMESTAMPDIFF(HOUR, transfer_start_time, NOW()) > 24 AND trans_sys_status = 0 """ trans.execute(sql) def update_trans_status_running(id, save_db=True): if save_db: exec_sql = """ update data_transfer set transfer_status = 0,trans_sys_status = 0 ,transfer_start_time = now(),err_info='', engine_count =0,time_granularity=0,transfer_finish_time=null,transfer_progress=0, data_min_time= null,data_max_time= null,transfer_data_count=null where id = %s """ trans.execute(exec_sql, id) def update_archive_success(id, archive_path, save_db=True): if save_db: exec_sql = """ update data_transfer set transfer_progress=70,archive_path = %s where id = %s """ trans.execute(exec_sql, (archive_path, id)) def update_trans_status_error(id, message="", save_db=True): if save_db: exec_sql = """ update data_transfer set transfer_status = 2,trans_sys_status=2 ,err_info= %s,transfer_finish_time=now() where id = %s """ message = message if len(message) <= 200 else message[0:200] trans.execute(exec_sql, (message, id)) info("执行失败:", message) def update_trans_status_success(id, wind_count=0, time_granularity=0, min_date=datetime.now(), max_date=datetime.now(), total_count=0, save_db=True): if save_db: if min_date is not None: exec_sql = """ update data_transfer set transfer_status = 1,trans_sys_status = 1,transfer_progress=100,err_info = '', engine_count =%s,time_granularity=%s,transfer_finish_time=now(), data_min_time= %s,data_max_time= %s,transfer_data_count=%s,is_high_priority=0 where id = %s """ trans.execute(exec_sql, (wind_count, time_granularity, min_date, max_date, total_count, id)) else: exec_sql = """ update data_transfer set transfer_status = 1,trans_sys_status = 1,transfer_progress = 100,err_info = '', engine_count =%s,time_granularity=%s,transfer_finish_time=now(),is_high_priority=0 where id = %s """ trans.execute(exec_sql, (wind_count, time_granularity, id)) def update_trans_transfer_progress(id, transfer_progress=0, save_db=True): print(id, transfer_progress) if save_db: exec_sql = """ update data_transfer set transfer_progress =%s where id = %s """ trans.execute(exec_sql, (int(transfer_progress), id)) info('当前进度:', transfer_progress) def get_now_running_count(): query_running_sql = """ select count(1) as count from data_transfer where trans_sys_status = 0 """ data = trans.execute(query_running_sql) now_count = int(data[0]['count']) return now_count # 获取执行的数据 def get_batch_exec_data() -> dict: query_next_exec_sql = """ SELECT * FROM data_transfer t WHERE t.trans_sys_status in (-1,1,2) and t.transfer_status = -1 AND t.read_dir != '' ORDER BY t.is_high_priority desc, t.update_time LIMIT 1 """ data = trans.execute(query_next_exec_sql) if type(data) == tuple: return None return data[0] def create_wave_table(table_name, wind_farm_name: str = '', save_db=True): # if save_db: # exec_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 '时间', # `rotational_speed` float DEFAULT NULL COMMENT '转速', # `sampling_frequency` int DEFAULT NULL COMMENT '采样频率', # `mesure_point_name` varchar(100) DEFAULT NULL COMMENT '测点名称', # `origin_type` int(11) DEFAULT '-1' COMMENT '厂商原始的类型,默认 -1', # `type` int(11) DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移 4:包络,默认 -1', # `mesure_data` longtext COMMENT '测点数据', # PRIMARY KEY (`id`), # KEY `wind_turbine_number` (`wind_turbine_number`), # KEY `time_stamp` (`time_stamp`), # KEY `mesure_point_name` (`mesure_point_name`) # ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='{wind_farm_name}' # """ # trans.execute(exec_sql) if save_db: exec_sql = f""" CREATE TABLE `{table_name}` ( `id` int NOT NULL AUTO_INCREMENT COMMENT '主键', `wind_turbine_number` varchar(20) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '风机编号', `wind_turbine_name` varchar(20) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '原始风机编号', `time_stamp` datetime DEFAULT NULL COMMENT '时间', `eu_spectrum` varchar(10) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '频谱单位', `detection_type` int DEFAULT NULL COMMENT '频谱的检测类型 0:None,1:Peak,2:Peak-to-peak,3:RMS', `window_type` int DEFAULT NULL COMMENT '窗函数类型:0:Hanning,1:Uniform,2:Flat Top', `start_frequency` int DEFAULT NULL COMMENT '频谱的起始频率(Hz)', `end_frequency` int DEFAULT NULL COMMENT '频谱的结束频率(Hz)', `samples` int DEFAULT NULL COMMENT '采样点数(时域波形)或谱线数(频谱)', `rotational_speed` float DEFAULT NULL COMMENT '转速', `sampling_frequency` int DEFAULT NULL COMMENT '采样频率', `mesure_point_name` varchar(100) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '测点名称', `type` int DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移 4:包络,默认 -1', `mesure_data_time` longtext CHARACTER SET utf8mb4 COMMENT '时域数据', `mesure_data_frenquency` longtext CHARACTER SET utf8mb4 COMMENT '频谱数据', `mesure_data_env` longtext CHARACTER SET utf8mb4 COMMENT '包络数据', `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `status_1` int DEFAULT NULL COMMENT '轴承诊断 0:正常 1:预警 2:危险', `status_2` int DEFAULT NULL COMMENT '齿轮诊断 0:正常 1:预警 2:危险', `status_3` int DEFAULT NULL COMMENT '不对中诊断 0:正常 1:预警 2:危险', `status_4` int DEFAULT NULL COMMENT '不平衡诊断 0:正常 1:预警 2:危险', `status_5` int DEFAULT NULL COMMENT '松动诊断 0:正常 1:预警 2:危险', `diag_flag` int DEFAULT NULL COMMENT '诊断标识 0：未诊断 1：已诊断', `retry_times` int(11) DEFAULT '0' COMMENT '尝试计算次数', PRIMARY KEY (`id`) USING BTREE, KEY `wind_turbine_number` (`wind_turbine_number`) USING BTREE, KEY `time_stamp` (`time_stamp`) USING BTREE, KEY `mesure_point_name` (`mesure_point_name`) USING BTREE ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='{wind_farm_name}' """ trans.execute(exec_sql) def get_trans_exec_code(wind_code, trans_type): exec_sql = f"select exec_code from batch_exec_code where wind_code = '{wind_code}' and type = '{trans_type}' and status = 1" data = trans.execute(exec_sql) if type(data) == tuple: return None return data[0]['exec_code']