mtsscaling/utils.py at main · lpphd/mtsscaling · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from sklearn.preprocessing import MinMaxScaler
from sklearn.preprocessing import MaxAbsScaler
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import RobustScaler
from sklearn.preprocessing import QuantileTransformer
from sklearn.preprocessing import PowerTransformer
from sklearn.preprocessing import Normalizer

import numpy as np


def ScaleData(train_x, test_x, scaling_method, dimension, random_seed):
    if scaling_method == "none":
        return train_x, test_x
    if np.isinf(train_x).any() or np.isinf(test_x).any():
        print("Train or test set contains infinity.")
        exit(-1)
    scaling_dict = {"minmax": MinMaxScaler(), "maxabs": MaxAbsScaler(), "standard": StandardScaler(),
                    "robust": RobustScaler(), "quantile": QuantileTransformer(random_state=random_seed),
                    "powert": PowerTransformer(), 'normalize': Normalizer()}
    if scaling_method not in scaling_dict.keys():
        print(f"Scaling method {scaling_method} not found.")
        exit(-1)
    if dimension not in ['timesteps', 'channels', 'all', 'both']:
        print(f"Dimension {dimension} not found.")
        exit(-1)

    dim1 = -1
    dim2 = 1
    if scaling_method == 'normalize':
        dim1 = 1
        dim2 = -1
    out_train_x = np.zeros_like(train_x, dtype=np.float64)
    out_test_x = np.zeros_like(test_x, dtype=np.float64)
    train_shape = train_x.shape
    test_shape = test_x.shape
    if dimension == 'all':
        out_train_x = scaling_dict[scaling_method].fit_transform(train_x.reshape((dim1, dim2))).reshape(train_shape)
        if scaling_method == 'normalize':
            out_test_x = scaling_dict[scaling_method].fit_transform(test_x.reshape((dim1, dim2))).reshape(test_shape)
        else:
            out_test_x = scaling_dict[scaling_method].transform(test_x.reshape((dim1, dim2))).reshape(test_shape)
    else:
        if dimension == 'channels':
            train_channel_shape = train_x[:, 0, :].shape
            test_channel_shape = test_x[:, 0, :].shape
            for i in range(train_x.shape[1]):
                out_train_x[:, i, :] = scaling_dict[scaling_method].fit_transform(
                    train_x[:, i, :].reshape((dim1, dim2))).reshape(
                    train_channel_shape)
                if scaling_method == 'normalize':
                    out_test_x[:, i, :] = scaling_dict[scaling_method].fit_transform(
                        test_x[:, i, :].reshape((dim1, dim2))).reshape(
                        test_channel_shape)
                else:
                    out_test_x[:, i, :] = scaling_dict[scaling_method].transform(
                        test_x[:, i, :].reshape((dim1, dim2))).reshape(
                        test_channel_shape)
        elif dimension == 'timesteps':
            train_timest_shape = train_x[:, :, 0].shape
            test_timest_shape = test_x[:, :, 0].shape
            for i in range(train_x.shape[2]):
                out_train_x[:, :, i] = scaling_dict[scaling_method].fit_transform(
                    train_x[:, :, i].reshape((dim1, dim2))).reshape(
                    train_timest_shape)
                if scaling_method == 'normalize':
                    out_test_x[:, :, i] = scaling_dict[scaling_method].fit_transform(
                        test_x[:, :, i].reshape((dim1, dim2))).reshape(
                        test_timest_shape)
                else:
                    out_test_x[:, :, i] = scaling_dict[scaling_method].transform(
                        test_x[:, :, i].reshape((dim1, dim2))).reshape(
                        test_timest_shape)
        elif dimension == 'both':
            train_both_shape = train_x[:, 0, 0].shape
            test_both_shape = test_x[:, 0, 0].shape
            for i in range(train_x.shape[1]):
                for j in range(train_x.shape[2]):
                    out_train_x[:, i, j] = scaling_dict[scaling_method].fit_transform(
                        train_x[:, i, j].reshape((dim1, dim2))).reshape(
                        train_both_shape)
                    if scaling_method == 'normalize':
                        out_test_x[:, i, j] = scaling_dict[scaling_method].fit_transform(
                            test_x[:, i, j].reshape((dim1, dim2))).reshape(
                            test_both_shape)
                    else:
                        out_test_x[:, i, j] = scaling_dict[scaling_method].transform(
                            test_x[:, i, j].reshape((dim1, dim2))).reshape(
                            test_both_shape)
        else:
            print(f"Dimension {dimension} not found.")
            exit(-1)
    return out_train_x, out_test_x