dalian-university/ipfcst-forecast-LSTM-v1.0/dataset.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# time: 2023/3/17 10:10
# file: main.py
# author: David
# company: shenyang JY
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
import yaml


class DataSet(object):
    def __init__(self, opt):
        self.std = None
        self.mean = None
        self.opt = opt
        self.time_step = self.opt.Model["time_step"]
        excel_data_path = opt.excel_data_path
        data_format = opt.data_format
        dq_path = excel_data_path + data_format["dq"]
        rp_path = excel_data_path + data_format["rp"]
        envir_path = excel_data_path + data_format["envir"]
        nwp_path = excel_data_path + data_format["nwp"]

        dq_columns = [1, 2]
        rp_columns = [0, 2]
        envir_columns = [0, *[x for x in range(3, 16)]]
        nwp_columns = [x for x in range(1, 27)]

        dq = self.read_data(dq_path, dq_columns)
        rp = self.read_data(rp_path, rp_columns)
        # nwp = self.read_data(nwp_path, nwp_columns)
        # rp_average(rp)    # 计算平均功率
        envir = self.read_data(envir_path, envir_columns)

        self.tables, self.tables_column_name = self.tables_integra(dq, rp, envir)
        # 如果是光
        if opt.is_photovoltaic:
            # self.tables = self.filter_data()
            pass
        self.data_num = self.tables.shape[0]
        self.train_num = int(self.data_num * opt.train_data_rate)

        # 都是在ndarray量纲下进行计算
        self.norm_data = (self.tables[:, 1:] - self.mean) / self.std  # 归一化，去量纲
        # self.norm_data.insert(0, 'C_TIME', self.tables['C_TIME'])
        # self.set_yml({'mean': self.mean.to_dict(), 'std': self.std.to_dict()})
        self.start_num_in_test = 0

    def set_yml(self, yml_dict):
        with open(self.opt.config_yaml, 'r', encoding='utf-8') as f:
            cfg = yaml.safe_load(f)
        for k, v in yml_dict.items():
            cfg[k] = v
        with open(self.opt.config_yaml, 'w') as f:
            yaml.safe_dump(cfg, f, default_flow_style=False)

    def read_data(self, path, cols):
        init_data = pd.read_excel(path, usecols=cols)
        return init_data

    def filter_data(self):
        check_table = self.tables[:, 2]  # 实际功率不能为0，为0代表没发电
        preserve_index = list(np.nonzero(check_table)[0])
        indexs = list(range(len(self.tables)))
        del_index = list(set(indexs) - set(preserve_index))
        self.tables = np.delete(self.tables, del_index, axis=0)
        return self.tables

    def norm(self, tables):
        """
        归一化操作，获取后存储于config.yml
        :param tables:
        :return:
        """
        mean = np.mean(tables.iloc[:, 1:], axis=0)  # 数据的均值
        std = np.std(tables.iloc[:, 1:], axis=0)  # 标准差
        if hasattr(self.opt, 'mean') is False or hasattr(self.opt, 'std') is False:
            self.set_yml({'mean': mean.to_dict(), 'std': std.to_dict()})
        self.mean, self.std = mean.values, std.values

    def tables_integra(self, dq, rp, envir):
        """
        联合表
        :param dq: 短期预测功率
        :param rp: 实际功率
        :param envir: 环境
        :return: 联合表， 列集（不包含第一列时间）
        """
        # 1. 先将 dq rp envir 根据时间联立
        union_tables = pd.merge(dq, rp, on='C_TIME')
        union_tables = union_tables.merge(envir, on='C_TIME')
        self.norm(union_tables)
        return union_tables.values, union_tables.columns.tolist()[1:]

    

    def get_train_and_valid_data(self, case):
        feature_data = self.norm_data[:self.train_num]
        # label_data = self.norm_data[: self.train_num,
        #                             self.opt.label_in_feature_index]    # 将延后几天的数据作为label
        label_data = self.norm_data[self.opt.predict_points: self.opt.predict_points + self.train_num, self.opt.label_in_feature_index]
        time_step = self.opt.Model["time_step"]
        train_x, train_y = [], []
        if not self.opt.do_continue_train:
            # 在非连续训练模式下，每time_step行数据会作为一个样本，两个样本错开一行，比如：1-20行，2-21行。。。。
            if case == 1: # 相当于实际功率+气象
                train_x = [feature_data[i:i + time_step] for i in range(self.train_num - time_step)]
                train_y = [label_data[i:i + time_step] for i in range(self.train_num - time_step)]
            elif case == 2: # 相当于短期+实际功率+气象
                train_rp = [feature_data[i:i + time_step, 1:]for i in range(self.train_num - time_step*2)]
                train_qd = [feature_data[i + time_step: i + 2*time_step, 0][:, np.newaxis] for i in range(self.train_num - time_step*2)]
                train_x = [list(np.append(t[0], t[1], axis=1)) for t in zip(train_rp, train_qd)]
                train_y = [label_data[i:i + time_step] for i in range(self.train_num - time_step*2)]
        else:
            # 在连续训练模式下
            pass
        train_x, train_y = np.array(train_x), np.array(train_y)

        train_x, valid_x, train_y, valid_y = train_test_split(train_x, train_y, test_size=self.opt.valid_data_rate,
                                                              random_state=self.opt.Model["random_seed"],
                                                              shuffle=self.opt.shuffle_train_data)   # 划分训练和验证集，并打乱
        return train_x, valid_x, train_y, valid_y

    def get_test_data(self, return_label_data=False):
        feature_data = self.norm_data[self.train_num:]

        sample_interval = min(feature_data.shape[0], self.time_step*2)     # 防止time_step大于测试集数量
        assert sample_interval == self.time_step*2


        test_x, test_y, dq_y = [], [], []
        if self.opt.is_continuous_predict:
            test_num = len(feature_data)
            test_x = [feature_data[
                       i : i + self.time_step]
                      for i in range(test_num - sample_interval)]
            test_y = [feature_data[
                       i + self.time_step: i + sample_interval, self.opt.label_in_feature_index]
                      for i in range(test_num - sample_interval)]
        else:
            # 在测试数据中，每time_step行数据会作为一个样本，两个样本错开time_step行
            # 比如：1-20行，21-40行。。。到数据末尾。
            # 这个地方要重新获取测试集 刘大为
            self.start_num_in_test = feature_data.shape[0] % sample_interval  # 这些天的数据不够一个sample_interval

            time_step_size = feature_data.shape[0] // sample_interval
            test_x = [feature_data[
                      self.start_num_in_test + i * sample_interval: self.start_num_in_test + i * sample_interval + self.time_step]
                      for i in range(time_step_size)]
            test_y = [feature_data[
                      self.start_num_in_test + i * sample_interval + self.time_step: self.start_num_in_test + (
                                  i + 1) * sample_interval, self.opt.label_in_feature_index]
                      for i in range(time_step_size)]
            dq_y = [feature_data[
                      self.start_num_in_test + i * sample_interval + self.time_step: self.start_num_in_test + (
                                  i + 1) * sample_interval, 0][:, np.newaxis]
                      for i in range(time_step_size)]
            # test_x = [list(np.append(t[0], t[1], axis=1)) for t in zip(test_x, dq_y)]

        print("test_x的长度为：", len(test_x))
        pass
        # 把test_x重新转换成timestamp时间步长
        # for i, x in enumerate(test_x):
        #     p1 = x[0:16, 0]
        #     p2 = x[16:32, 1]
        #     p = [list(t) for t in zip(p1, p2)]
        #     test_x[i] = np.array(p)
        if return_label_data:       # 实际应用中的测试集是没有label数据的
            return np.array(test_x), np.array(test_y), np.array(dq_y)
        return np.array(test_x)


if __name__ == "__main__":
    ds = DataSet()
    # dq = ds.read_data(dq_path, dq_columns)[0]
    # rp = ds.read_data(rp_path, rp_columns)[0]
    # # rp_average(rp)    # 计算平均功率
    # envir = ds.read_data(envir_path, envir_columns)[0]
    # tables = ds.tables_integra(dq, rp, envir)
    # ds.tables_norm_result(tables)