TrafficWheel/dataloader/STEPdataloader.py

from lib.normalization import normalize_dataset

import numpy as np
import gc
import os
import torch
import h5py


def get_dataloader(args, normalizer='std', single=True):
    """STEP模型的数据加载器
    
    Args:
        args: 配置参数
        normalizer: 标准化方法
        single: 是否为单步预测
        
    Returns:
        train_dataloader, val_dataloader, test_dataloader, scaler
    """
    data = load_st_dataset(args['type'], args['sample'])  # 加载数据
    L, N, F = data.shape  # 数据形状

    # Step 1: data -> x,y
    x = add_window_x(data, args['lag'], args['horizon'], single)
    y = add_window_y(data, args['lag'], args['horizon'], single)

    del data
    gc.collect()

    # Step 2: time_in_day, day_in_week -> day, week
    time_in_day = [i % args['steps_per_day'] / args['steps_per_day'] for i in range(L)]
    time_in_day = np.tile(np.array(time_in_day), [1, N, 1]).transpose((2, 1, 0))
    day_in_week = [(i // args['steps_per_day']) % args['days_per_week'] for i in range(L)]
    day_in_week = np.tile(np.array(day_in_week), [1, N, 1]).transpose((2, 1, 0))

    x_day = add_window_x(time_in_day, args['lag'], args['horizon'], single)
    x_week = add_window_x(day_in_week, args['lag'], args['horizon'], single)

    # Step 3 day, week, x, y --> x, y
    x = np.concatenate([x, x_day, x_week], axis=-1)

    del x_day, x_week
    gc.collect()

    # Step 4 x,y --> x_train, x_val, x_test, y_train, y_val, y_test
    if args['test_ratio'] > 1:
        x_train, x_val, x_test = split_data_by_days(x, args['val_ratio'], args['test_ratio'])
    else:
        x_train, x_val, x_test = split_data_by_ratio(x, args['val_ratio'], args['test_ratio'])

    del x
    gc.collect()

    # Normalization
    scaler = normalize_dataset(x_train[..., :args['input_dim']], normalizer, args['column_wise'])
    x_train[..., :args['input_dim']] = scaler.transform(x_train[..., :args['input_dim']])
    x_val[..., :args['input_dim']] = scaler.transform(x_val[..., :args['input_dim']])
    x_test[..., :args['input_dim']] = scaler.transform(x_test[..., :args['input_dim']])

    y_day = add_window_y(time_in_day, args['lag'], args['horizon'], single)
    y_week = add_window_y(day_in_week, args['lag'], args['horizon'], single)

    del time_in_day, day_in_week
    gc.collect()

    y = np.concatenate([y, y_day, y_week], axis=-1)

    del y_day, y_week
    gc.collect()

    # Split Y
    if args['test_ratio'] > 1:
        y_train, y_val, y_test = split_data_by_days(y, args['val_ratio'], args['test_ratio'])
    else:
        y_train, y_val, y_test = split_data_by_ratio(y, args['val_ratio'], args['test_ratio'])

    del y
    gc.collect()

    # Step 5: x_train y_train x_val y_val x_test y_test --> train val test
    train_dataloader = data_loader(x_train, y_train, args['batch_size'], shuffle=True, drop_last=True)

    del x_train, y_train
    gc.collect()

    val_dataloader = data_loader(x_val, y_val, args['batch_size'], shuffle=False, drop_last=True)

    del x_val, y_val
    gc.collect()

    test_dataloader = data_loader(x_test, y_test, args['batch_size'], shuffle=False, drop_last=False)

    del x_test, y_test
    gc.collect()

    return train_dataloader, val_dataloader, test_dataloader, scaler


def load_st_dataset(dataset, sample):
    # output L, N, F
    match dataset:
        case 'PEMSD3':
            data_path = os.path.join('./data/PEMS03/PEMS03.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case 'PEMSD4':
            data_path = os.path.join('./data/PEMS04/PEMS04.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case 'PEMSD7':
            data_path = os.path.join('./data/PEMS07/PEMS07.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case 'PEMSD8':
            data_path = os.path.join('./data/PEMS08/PEMS08.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case 'METR-LA':
            data_path = os.path.join('./data/METR-LA/METR-LA.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case 'METR-BAY':
            data_path = os.path.join('./data/METR-BAY/METR-BAY.npz')
            data = np.load(data_path)['data']  # (L, N, F)
        case _:
            raise ValueError(f"Unknown dataset: {dataset}")
    
    if sample:
        data = data[:sample]
    
    return data


def add_window_x(data, lag, horizon, single):
    """
    Add window to data for x
    """
    L, N, F = data.shape
    if single:
        x = np.zeros((L - lag - horizon + 1, lag, N, F))
        for i in range(L - lag - horizon + 1):
            x[i] = data[i:i + lag]
    else:
        x = np.zeros((L - lag - horizon + 1, lag, N, F))
        for i in range(L - lag - horizon + 1):
            x[i] = data[i:i + lag]
    return x


def add_window_y(data, lag, horizon, single):
    """
    Add window to data for y
    """
    L, N, F = data.shape
    if single:
        y = np.zeros((L - lag - horizon + 1, horizon, N, F))
        for i in range(L - lag - horizon + 1):
            y[i] = data[i + lag:i + lag + horizon]
    else:
        y = np.zeros((L - lag - horizon + 1, horizon, N, F))
        for i in range(L - lag - horizon + 1):
            y[i] = data[i + lag:i + lag + horizon]
    return y


def split_data_by_ratio(data, val_ratio, test_ratio):
    """
    Split data by ratio
    """
    L = data.shape[0]
    val_len = int(L * val_ratio)
    test_len = int(L * test_ratio)
    train_len = L - val_len - test_len
    
    train_data = data[:train_len]
    val_data = data[train_len:train_len + val_len]
    test_data = data[train_len + val_len:]
    
    return train_data, val_data, test_data


def split_data_by_days(data, val_days, test_days):
    """
    Split data by days
    """
    L = data.shape[0]
    val_len = val_days * 288  # 288 time steps per day
    test_len = test_days * 288
    train_len = L - val_len - test_len
    
    train_data = data[:train_len]
    val_data = data[train_len:train_len + val_len]
    test_data = data[train_len + val_len:]
    
    return train_data, val_data, test_data


def data_loader(x, y, batch_size, shuffle=True, drop_last=True):
    """
    Create data loader
    """
    dataset = torch.utils.data.TensorDataset(torch.FloatTensor(x), torch.FloatTensor(y))
    dataloader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=shuffle, drop_last=drop_last)
    return dataloader