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


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))


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
            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()
            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))


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.update_time
    LIMIT 1
    """
    data = trans.execute(query_next_exec_sql)
    if type(data) == tuple:
        return None
    return data[0]


def get_data_by_id(id):
    query_exec_sql = f"""
    SELECT
        t.*,a.field_name,b.batch_name
    FROM
        data_transfer t INNER JOIN wind_field a on t.field_code = a.field_code
        inner join wind_field_batch b on t.id = b.id
    WHERE
         t.trans_sys_status in (-1,1,2) and t.transfer_status = 2 and t.id = '{id}'
    AND t.read_dir != ''
    """

    data = trans.execute(query_exec_sql)
    if type(data) == tuple:
        return None
    return data[0]


def create_wave_table(table_name, 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` varchar(50) DEFAULT NULL COMMENT '采样频率',
          `mesure_point_name` varchar(100) DEFAULT NULL COMMENT '测点名称',
          `type` int(11) DEFAULT '-1' COMMENT '-1:不存在 0:角度 1:速度 2:加速度 3:位移,默认 -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
        """
        trans.execute(exec_sql)