platform/validate.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# time: 2023/11/21 13:33
# file: validate.py
# author: David
# company: shenyang JY
import random
from cache.limited_power_curve import LimitPower
import pandas as pd
import numpy as np
np.random.seed(42)
import pickle, os
from cache.data_cleaning import key_field_row_cleaning, cleaning
current_path = os.path.dirname(__file__)

def readVar(path):
    if os.path.exists(path):
        with open(path, "rb") as file:
            var = pickle.load(file)
        return var
    else:
        return None

class ValidationError(Exception):
    def __init__(self, message):
        self.msg = message


class Validation(object):
    glob_rp = readVar(current_path + '/var/glob_rp.pickle')

    def __init__(self, log, args):
        self.logger = log
        self.args = args
        self.opt = args.parse_args_and_yaml()
        self.status = 1
        self.lp = LimitPower(log, args, None)

    @classmethod
    def set_lp(cls):
        cls.glob_rp = readVar(current_path + '/var/glob_rp.pickle')

    def validate_nwp(self, nwp):
        nwp_cs = [x for x in nwp.columns.to_list() if x in self.opt.nwp_columns]
        if 'C_TIME' in nwp_cs:
            nwp_cs.pop(nwp_cs.index('C_TIME'))
        nwp_filter = key_field_row_cleaning(nwp, nwp_cs, self.logger)
        if len(nwp_filter) < 8:
            self.status = 2
            raise ValidationError("NWP数据缺失")
        nwp_filter['C_TIME'] = pd.to_datetime(nwp_filter['C_TIME'])
        return nwp_filter

    def validate_his_data(self, history_rp, env, history_dq):
        if len(history_rp) <= 1 and len(env) <= 1:
            self.status = 3
            raise ValidationError("鉴权失败-历史数据时刻不满足USTA_HISTORY个数")
        if len(history_dq) < self.opt.Model["his_points"]:
            self.status = 2
            raise ValidationError("history_dq缺失")
        his = pd.merge(history_rp, history_dq, on='C_TIME')
        his = pd.merge(env, his, on='C_TIME')
        if self.opt.usable_power['clean_power_which'] in range(1, 3):
            his_clean = key_field_row_cleaning(his, [self.opt.usable_power['env']], self.logger)
        else:
            his_clean = his
        his_clean = his_clean.replace(-99, np.nan)
        points = self.opt.Model['his_points']
        if len(his_clean) < points * 0.5:
            self.status = 2
            raise ValidationError("历史数据缺失，实际长度：{}".format(len(his_clean)))
        elif len(his_clean) < self.opt.Model["his_points"] or his_clean.isnull().any().any():
            his_clean.drop(columns=['C_FP_VALUE'], inplace=True)
            his = pd.merge(history_dq, his_clean, on='C_TIME', how='left')
            his.set_index('C_TIME', inplace=True)
            his = his.interpolate(method='linear')
            his = his.fillna(method='ffill')
            his = his.fillna(method='bfill')
            his.reset_index(drop=False, inplace=True)
            his['LIMIT_STATUS'] = np.where((his['LIMIT_STATUS'] != 0) & (his['LIMIT_STATUS'] != 1), 1, his['LIMIT_STATUS'])
            return his
        else:
            return his_clean

    def validate_authentic(self, nwp, his):
        # 验证当前天气是否经过鉴权，第一个点一定要是当天的
        predict_dt = nwp.iloc[0, :]["C_TIME"]
        history_dt = his.iloc[-1, :]["C_TIME"]
        if predict_dt - history_dt > pd.Timedelta(hours=4):
            self.status = 3
            raise ValidationError("鉴权失败-环境数据结束时间早于预测开始时间4小时")

    def validate_power(self, his):
        self.lp.mapping_relation(self.glob_rp)
        history_rp_ws = his.copy()
        new_rp = []
        for index, row in history_rp_ws.iterrows():
            zfs = row[self.opt.usable_power["env"]]
            rp = row['C_REAL_VALUE']
            able_rp = row['C_ABLE_VALUE']
            status = row['LIMIT_STATUS']  # 限电是1，不限电是0
            # 以下三种情况用可用功率替代：
            # 1. api_able_power可用功率参数置为True
            # 2. 取消模型融合，解析接口数据填充
            # 3. 用样板机法或单独信号法清洗限电
            if self.opt.usable_power['api_able_power'] is True or self.opt.Model['fusion'] is False or self.opt.usable_power['clean_power_which'] not in range(1, 3):
                new_rp.append(able_rp)
            elif not self.lp.filter_unlimited_power(zfs, rp) or status:  # 任意判断限电
                new_rp.append(able_rp)
            else:
                new_rp.append(rp)  # 全部为不限电
        history_rp_ws["NEW_RP"] = new_rp
        return history_rp_ws[['C_TIME', 'NEW_RP']]