ty_zgh_data/run_data.py

import json
import multiprocessing
import os
import shutil
import warnings

from service import trans_service
from service.plt_service import get_wind_info, get_all_wind_by_company_code
from tool.ClassIdentifier import ClassIdentifier
from utils.common_utils import read_file, get_all_files

warnings.filterwarnings('ignore')

from service.trans_service import *


# def generate_warn_falut_code_maps(file_path):
#     df = pd.read_excel(file_path)
#     df['主控ID'] = df['主控ID'].str.strip()
#     df['状态编码'] = df['状态编码'].astype(int)
#     df['SC中文描述'] = df['SC中文描述'].str.strip()
#     df['告警等级'] = df['告警等级'].str.strip()
#     df['告警等级'] = df['告警等级'].fillna("告警信息")
#
#     result_map = dict()
#     for index, row in df.iterrows():
#         controller_id = row['主控ID'].strip()
#         status_code = int(row['状态编码'])
#         cn_des = row['SC中文描述']
#         level = row['告警等级']
#         if controller_id in result_map.keys():
#             result_map[controller_id][status_code] = (cn_des, level)
#         else:
#             result_map[controller_id] = {status_code: (cn_des, level)}
#
#     return result_map
#
#
# def generate_mc_version_maps(file_path):
#     df = pd.read_excel(file_path)
#
#     mc_version_maps = dict()
#     for _, data in df.iterrows():
#         changzhan = data['场站标准化编号'].strip()
#         wind_no = str(data['风机号'])
#         mc_versioin = data['主控版本'].strip()
#         if changzhan in mc_version_maps.keys():
#             mc_version_maps[changzhan][wind_no] = mc_versioin
#         else:
#             mc_version_maps[changzhan] = {wind_no: mc_versioin}
#
#     return mc_version_maps


def scada_read_and_save_db(file_path, wind_factory_map: dict):
    try:
        df = read_file(file_path)
        if 'wind_farm_code' in df.columns:
            del df['wind_farm_code']
        wind_no = os.path.basename(file_path).split('_')[0]

        df['wind_turbine_name'] = df['wind_turbine_name'].astype('str')
        wind_col_trans, rated_power_and_cutout_speed_map = get_wind_info(wind_no)
        df['wind_turbine_number'] = df['wind_turbine_name'].map(wind_col_trans).fillna(df['wind_turbine_name'])
        wind_turbine_number = df['wind_turbine_number'].unique()[0]

        df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce")
        df.dropna(subset=['time_stamp'], inplace=True)
        df.sort_values(by='time_stamp', inplace=True)
        logger.info(f"有功功率前10个 :{df.head(10)['active_power'].values}")
        power_df = df[df['active_power'] > 0]
        logger.info(f"{wind_turbine_number} 功率大于0的数量:{power_df.shape}")
        power = power_df.sample(int(power_df.shape[0] / 100))['active_power'].median()
        logger.info(f"{wind_turbine_number} 有功功率，中位数:{power}")

        rated_power_and_cutout_speed_tuple = rated_power_and_cutout_speed_map.get(str(wind_turbine_number), None)
        if rated_power_and_cutout_speed_tuple is None:
            rated_power_and_cutout_speed_tuple = (None, None)

        sec_df_origin = df.copy()
        class_identifiler = ClassIdentifier(wind_turbine_number=wind_turbine_number, origin_df=sec_df_origin,
                                            rated_power=rated_power_and_cutout_speed_tuple[0],
                                            cut_out_speed=rated_power_and_cutout_speed_tuple[1])
        sec_df = class_identifiler.run()
        if not sec_df.empty:
            sec_df['year'] = sec_df['time_stamp'].dt.year
            sec_df['month'] = sec_df['time_stamp'].dt.month
            sec_df['day'] = sec_df['time_stamp'].dt.day
            sec_df['time_stamp'] = sec_df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
            sec_df['year_month'] = sec_df[['year', 'month']].apply(lambda x: str(x['year']) + str(x['month']).zfill(2),
                                                                   axis=1)

        min_df_origin = df[df['time_stamp'].dt.minute % 10 == 0]
        class_identifiler = ClassIdentifier(wind_turbine_number=wind_turbine_number, origin_df=min_df_origin,
                                            rated_power=rated_power_and_cutout_speed_tuple[0],
                                            cut_out_speed=rated_power_and_cutout_speed_tuple[1])
        min_df = class_identifiler.run()
        if not min_df.empty:
            min_df['year'] = min_df['time_stamp'].dt.year
            min_df['month'] = min_df['time_stamp'].dt.month
            min_df['day'] = min_df['time_stamp'].dt.day
            min_df['time_stamp'] = min_df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d %H:%M:%S'))
            min_df['year_month'] = min_df[['year', 'month']].apply(lambda x: str(x['year']) + str(x['month']).zfill(2),
                                                                   axis=1)
        save_df_to_db(f'{wind_no}_second', sec_df)
        save_df_to_db(f'{wind_no}_minute', min_df)

        update_day_count(sec_df, wind_no, wind_factory_map[wind_no], 'second')
        update_day_count(min_df, wind_no, wind_factory_map[wind_no], 'minute')
        os.remove(file_path)
    except:
        logger.error(f"{file_path}转化失败")
        logger.error(traceback.format_exc(), exc_info=True)
        error_path = file_path.replace('scada', 'scada_error')
        os.makedirs(os.path.dirname(error_path), exist_ok=True)
        shutil.move(file_path, error_path)
    finally:
        logger.info(f'scada/{os.path.basename(file_path)}执行结束')


def generate_warn_fault_data(df):
    df['time_stamp'] = pd.to_datetime(df['time_stamp'])
    # 识别成功状态
    success_codes = {0, 309001}
    df['Is_Success'] = df['fault_code'].isin(success_codes)

    # 找到故障开始和结束时间
    result = []
    i = 0
    n = len(df)

    while i < n:
        current_status = df.iloc[i]['fault_code']
        current_time = df.iloc[i]['time_stamp']

        # 如果是故障状态（非成功状态）
        if current_status not in success_codes:
            # 记录故障开始
            fault_start = current_time
            fault_status = current_status

            # 跳过连续的相同故障状态
            while i < n and df.iloc[i]['fault_code'] == fault_status:
                i += 1

            # 找到下一个成功状态的时间
            success_time = None
            fault_duration = None

            # 寻找下一个成功状态
            for j in range(i, n):
                if df.iloc[j]['fault_code'] in success_codes:
                    success_time = df.iloc[j]['time_stamp']
                    fault_duration = int((success_time - fault_start).total_seconds())
                    break

            # 如果有成功时间，计算故障时长
            if success_time:
                # 添加到结果
                result.append([
                    df.iloc[0]['wind_turbine_name'],  # ID
                    fault_start,  # 故障开始时间
                    fault_status,  # 故障状态码
                    success_time,  # 成功时间
                    fault_duration  # 故障时长
                ])
            else:
                # 如果没有成功时间（故障持续到最后）
                result.append([
                    df.iloc[0]['wind_turbine_name'],
                    fault_start,
                    fault_status,
                    None,
                    None
                ])
        else:
            # 如果是成功状态，继续下一行
            i += 1

    # 创建结果DataFrame
    result_df = pd.DataFrame(result, columns=['wind_turbine_name', 'begin_time', 'fault_code', 'end_time', 'time_diff'])
    result_df['fault_code'] = result_df['fault_code'].astype(int)
    return result_df


def warn_fault_read_and_save_db(file_path, wind_factory_map: dict, mc_vesion_map: dict, warn_falut_code_map: dict):
    try:
        wind_farm_code = os.path.basename(file_path).split('_')[0]
        wind_turbine_name = os.path.basename(file_path).split('_')[1].replace('.csv', '')
        df = read_file(file_path)
        col_map = {'wind_turbine_id': 'wind_turbine_name', 'timestamp': 'time_stamp', 'warn_fault_code': 'fault_code'}
        df.rename(columns=col_map, inplace=True)

        df['fault_code'] = df['fault_code'].astype(int, errors='ignore')
        df = df.dropna(subset=['fault_code'])

        df['wind_turbine_name'] = df['wind_turbine_name'].astype('str')
        df['wind_turbine_number'] = df['wind_turbine_name'].map(wind_factory_map).fillna(df['wind_turbine_name'])

        df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce")
        df.dropna(subset=['time_stamp'], inplace=True)
        df.sort_values(by='time_stamp', inplace=True)
        df.reset_index(drop=True, inplace=True)

        first_code = df.loc[0, 'fault_code']
        data_time = df.loc[0, 'time_stamp']
        if first_code in [0, 309001]:
            trans_service.update_warn_fault_exist_data_with_db(wind_farm_code, wind_turbine_name, data_time)

        wind_col_trans, rated_power_and_cutout_speed_map = get_wind_info(wind_farm_code)

        save_df = generate_warn_fault_data(df)
        save_df['wind_farm_code'] = wind_farm_code
        save_df['wind_farm_name'] = wind_factory_map[wind_farm_code]
        save_df['wind_turbine_name'] = save_df['wind_turbine_name'].astype('str')
        save_df['wind_turbine_number'] = save_df['wind_turbine_name'].map(wind_col_trans)
        mc_vesion = mc_vesion_map[wind_farm_code][wind_turbine_name]
        save_df['mc_version'] = mc_vesion

        def get_fault_dict(x, index):
            if x['mc_version'] not in warn_falut_code_map.keys():
                return None

            if x['fault_code'] not in warn_falut_code_map[x['mc_version']].keys():
                return None
            return warn_falut_code_map[x['mc_version']][x['fault_code']][index]

        save_df['fault_detail'] = save_df[['mc_version', 'fault_code']].apply(get_fault_dict, args=(0,), axis=1)
        save_df['fault_level'] = save_df[['mc_version', 'fault_code']].apply(get_fault_dict, args=(1,), axis=1)

        select_cols = ['wind_turbine_number', 'wind_turbine_name', 'mc_version', 'begin_time', 'end_time', 'time_diff',
                       'fault_code', 'fault_detail', 'fault_level']
        warn_df = save_df[save_df['fault_level'] != '故障'][select_cols]
        fault_df = save_df[save_df['fault_level'] == '故障'][select_cols]
        warn_df.sort_values(by='begin_time', inplace=True)
        fault_df.sort_values(by='begin_time', inplace=True)

        if not warn_df.empty:
            trans_service.save_df_to_db(f'{wind_farm_code}_warn', warn_df)
            warn_max_date = warn_df['begin_time'].max()
            add_date_str = warn_max_date.strftime('%Y-%m-%d')
            warn_last_date_str = warn_max_date.strftime('%Y-%m-%d %H:%M:%S')
            # wind_farm_code, wind_farm_name, add_date, trans_type, count, latest_data_time
            trans_service.update_wind_farm_day_count(wind_farm_code, wind_factory_map.get(wind_farm_code, ''),
                                                     add_date_str,
                                                     'warn', warn_df.shape[0], warn_last_date_str, 1)

        if not fault_df.empty:
            trans_service.save_df_to_db(f'{wind_farm_code}_fault', fault_df)
            fault_max_date = fault_df['begin_time'].max()
            add_date_str = fault_max_date.strftime('%Y-%m-%d')
            fault_last_date_str = fault_max_date.strftime('%Y-%m-%d %H:%M:%S')
            trans_service.update_wind_farm_day_count(wind_farm_code, wind_factory_map.get(wind_farm_code, ''),
                                                     add_date_str,
                                                     'fault', fault_df.shape[0], fault_last_date_str, 1)

        os.remove(file_path)
    except:
        logger.error(f"{file_path}转化失败")
        logger.error(traceback.format_exc(), exc_info=True)
        error_path = file_path.replace('warn_fault', 'warn_fault_error')
        os.makedirs(os.path.dirname(error_path), exist_ok=True)
        shutil.move(file_path, error_path)
    finally:
        logger.info(f'warn_fault/{os.path.basename(file_path)}执行结束')


def update_day_count(df_data: pd.DataFrame, wind_farm_code: str, wind_name, data_type: str):
    df = df_data[['wind_turbine_number', 'time_stamp']].copy()
    df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce")
    df['time_stamp'] = df['time_stamp'].apply(lambda x: x.strftime('%Y-%m-%d'))
    res_df = df.groupby('time_stamp')['wind_turbine_number'].count().reset_index()
    res_df.rename(columns={'time_stamp': 'add_date', 'wind_turbine_number': 'count'}, inplace=True)
    res_df['latest_data_time'] = res_df['add_date']
    res_df['sync_status'] = 1
    res_df['wind_farm_code'] = wind_farm_code
    res_df['wind_farm_name'] = wind_name
    res_df['type'] = data_type
    for index, data in res_df.iterrows():
        update_wind_farm_day_count(data['wind_farm_code'], data['wind_farm_name'], data['add_date'], data['type'],
                                   data['count'], data['latest_data_time'], sync_status=1)


if __name__ == '__main__':
    scada_dir = r'/home/trans/data/scada'
    warn_fault_dir = r'/home/trans/data/warn_fault'
    scada_files = get_all_files(scada_dir)
    warn_fault_files = get_all_files(warn_fault_dir)

    wind_factory_map = get_all_wind_by_company_code('COM00002')

    with open('conf/mc_vesion.json', 'r') as f:
        mc_vesion_map = json.load(f)

    with open('conf/warn_fault_mc_code.json', 'r') as f:
        warn_falut_code_map = json.load(f)

    with multiprocessing.Pool(4) as pool:
        pool.starmap(scada_read_and_save_db, [(i, wind_factory_map) for i in scada_files])

    with multiprocessing.Pool(4) as pool:
        pool.starmap(warn_fault_read_and_save_db,
                     [(i, wind_factory_map, mc_vesion_map, warn_falut_code_map) for i in warn_fault_files])