energy-data-trans/service/trans_conf_service.py

# -*- 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']