dalian-university/wind-LSTM-CNN-v2.0-nwp+rp+环境/data_features.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# time: 2023/4/12 17:42
# file: data_features.py
# author: David
# company: shenyang JY
import pandas as pd
from sklearn.model_selection import train_test_split
import numpy as np
from data_utils import *


class data_features(object):
    def __init__(self, opt, mean, std):
        self.opt = opt
        self.time_step = self.opt.Model["time_step"]
        self.mean = mean
        self.std = std
        self.columns = list()
        self.columns_lstm = list()
        self.columns_cnn = list()

    def get_train_data(self, dfs):
        train_x, valid_x, train_y, valid_y = [], [], [], []
        self.opt.feature_columns = dfs[0].columns.tolist()
        self.opt.feature_columns.insert(0, 'C_TIME')
        self.opt.label_in_feature_index = (lambda x, y: [x.index(i) for i in y])(self.opt.feature_columns,
                                                                        self.opt.label_columns)  # 因为feature不一定从0开始
        self.opt.input_size = len(self.opt.feature_columns)
        for df in dfs:
            datax, datay = self.get_data_features(df)
            trainx_ = [[np.array(x[0]), np.array(x[1])] for x in datax]
            # trainx = np.array(datax)
            trainy = [y['C_REAL_VALUE'].values for y in datay]
            trainy = np.expand_dims(np.array(trainy), axis=-1)  # 在最后一维加一维度
            tx, vx, ty, vy = train_test_split(trainx_, trainy, test_size=self.opt.valid_data_rate,
                                                                  random_state=self.opt.Model["random_seed"],
                                                                  shuffle=self.opt.shuffle_train_data)  # 划分训练和验证集
            # 分裂 tx 和 vx
            train_x.extend(tx)
            valid_x.extend(vx)
            train_y.append(ty)
            valid_y.append(vy)
        # train_x = np.concatenate(train_x, axis=0)
        # valid_x = np.concatenate(valid_x, axis=0)
        train_y = np.concatenate(train_y, axis=0)
        valid_y = np.concatenate(valid_y, axis=0)

        train_x = self.norm_features(train_x)
        valid_x = self.norm_features(valid_x)
        train_y = self.norm_label(train_y)
        valid_y = self.norm_label(valid_y)

        cnn_x, cnn_x1 = [], []
        lstm_x, lstm_x1 = [], []
        for i in range(0, len(train_x)):
            cnn_x.append(train_x[i][0])
            lstm_x.append(train_x[i][1])
        train_x = [np.array(cnn_x), np.array(lstm_x)]
        for i in range(0, len(valid_x)):
            cnn_x1.append(valid_x[i][0])
            lstm_x1.append(valid_x[i][1])
        valid_x = [np.array(cnn_x1), np.array(lstm_x1)]
        return train_x, valid_x, train_y, valid_y

    def get_test_data(self, dfs):
        test_x, test_y, data_y = [], [], []
        self.opt.feature_columns = dfs[0].columns.tolist()
        self.opt.feature_columns.insert(0, 'C_TIME')
        self.opt.label_in_feature_index = (lambda x, y: [x.index(i) for i in y])(self.opt.feature_columns,
                                                                        self.opt.label_columns)  # 因为feature不一定从0开始
        self.opt.input_size = len(self.opt.feature_columns)
        for df in dfs:
            datax, datay = self.get_data_features(df)
            trainx_ = [[np.array(x[0]), np.array(x[1])] for x in datax]
            # trainx = np.array(datax)
            trainy = [y['C_REAL_VALUE'].values for y in datay]
            trainy = np.expand_dims(np.array(trainy), axis=-1)  # 在最后一维加一维度
            test_x.extend(trainx_)
            test_y.append(trainy)
            data_y.append(datay)

        test_y = np.concatenate(test_y, axis=0)

        test_x = self.norm_features(test_x)
        test_y = self.norm_label(test_y)

        cnn_x, lstm_x = [], []

        for i in range(0, len(test_x)):
            cnn_x.append(test_x[i][0])
            lstm_x.append(test_x[i][1])
        test_x = [np.array(cnn_x), np.array(lstm_x)]

        return test_x, test_y, data_y

    def get_data_features(self, df):   # 这段代码基于pandas方法的优化
        norm_data = df.reset_index()
        feature_data = norm_data[:-self.opt.predict_points]
        label_data = norm_data[self.opt.predict_points:].reset_index(drop=True)
        time_step = self.opt.Model["time_step"]
        time_step_loc = time_step - 1
        train_num = int(len(feature_data))
        time_rp = [feature_data.loc[i:i + time_step_loc, 'C_TIME':'C_WD_INST50'] for i in range(train_num - time_step)]
        nwp = [label_data.loc[i:i + time_step_loc, 'C_T':] for i in range(train_num - time_step)]
        features_x, features_x1, features_y = [], [], []
        for row in zip(time_rp, nwp):
            row0 = row[0]   # row0是时间+rp+环境
            row1 = row[1]  # row1是nwp
            row0.set_index('C_TIME', inplace=True, drop=False)
            row0["C_TIME"] = row0["C_TIME"].apply(datetime_to_timestr)
            row0_ = row0.loc[:, ['C_TIME', 'C_REAL_VALUE']]
            row0_.reset_index(drop=True, inplace=True)
            row1.reset_index(drop=True, inplace=True)
            rowx = pd.concat([row0_, row1], axis=1)  # rowx是时间+rp+nwp
            features_x.append([row0, rowx])
        self.columns = row0.columns.tolist()
        self.columns_cnn = row0.columns.tolist()
        self.columns_lstm = rowx.columns.tolist()

        features_y = [label_data.loc[i:i + time_step_loc, ['C_TIME', 'C_REAL_VALUE']] for i in range(train_num - time_step)]

        return features_x, features_y

    def norm_features(self, data: np.ndarray):
        for i, d in enumerate(data):
            mean = np.array([self.mean[col] for col in self.columns_cnn])
            std = np.array([self.std[col] for col in self.columns_cnn])
            d[0] = (d[0] - mean) / std  # 归一化
            mean = np.array([self.mean[col] for col in self.columns_lstm])
            std = np.array([self.std[col] for col in self.columns_lstm])
            d[1] = (d[1] - mean) / std  # 归一化
            data[i] = d
            self.opt.input_size_lstm = len(self.columns_lstm)
            self.opt.input_size_cnn = len(self.columns_cnn)
        return data

    def norm_label(self, label_data: np.ndarray):
        return (label_data - self.mean['C_REAL_VALUE']) / self.std['C_REAL_VALUE']