venus_volcanoes/data_loader.py at master · marinapts/venus_volcanoes · 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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
from src.pyvov import ChipsIndex
from random import shuffle
import numpy as np
from sklearn.model_selection import train_test_split


class DataLoader:
    def __init__(self, experiment_names=['C1', 'D4'], validation_ratio=0.1, testset_ratio=0.1, seed=8):
        ci = ChipsIndex()
        if experiment_names == ['C1', 'D4']:
            self.full_dataset, self.all_labels = ci.get_all()
            self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(self.full_dataset, self.all_labels,
                                                                                    test_size=testset_ratio,
                                                                                    random_state=seed)
            self.X_train, self.X_val, self.y_train, self.y_val = train_test_split(self.X_train, self.y_train,
                                                                                  test_size=validation_ratio,
                                                                                  random_state=seed)

        else:
            self.X_train, self.X_test, self.y_train, self.y_test = ci.get_specific(experiment_names)
            #self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(self.full_dataset, self.all_labels, test_size=testset_ratio, random_state=seed)
            #self.X_train, self.X_val, self.y_train, self.y_val = train_test_split(self.X_train, self.y_train, test_size=validation_ratio, random_state=seed)

    def normalise_array(self, array):
        normalised_array = array - np.mean(array)
        normalised_array = normalised_array / np.std(normalised_array)
        return normalised_array

    def normalise_split(self, split, split_labels):
        # Indices of constant images
        indices = [i for i in np.arange(len(split)) if np.std(split[i]) == 0]

        # Remove constant images
        split = [split[i] for i in np.arange(len(split)) if i not in indices]
        split_labels = [split_labels[i] for i in np.arange(len(split_labels)) if i not in indices]

        # Normalise the remaining data
        split = [self.normalise_array(image) for image in split]

        return split, split_labels

    def preprocess_data(self):
        self.X_train, self.y_train = self.normalise_split(self.X_train, self.y_train)
        self.X_val, self.y_val = self.normalise_split(self.X_val, self.y_val)
        self.X_test, self.y_test = self.normalise_split(self.X_test, self.y_test)

    def convert_to_numpy_sets(self, binary_class=False):
        """
        Returns the train, validation and test sets and labels as numpy arrays

        :param self: volcano data object that contains all the data as lists containing square images as
        flattened lists
        :param binary_class: boolean that determines whether we want to perform binary classification on
        class 0 versus all of 1, 2, 3, 4
        :return: (X_train, y_train), (X_val, y_val), (X_test, y_test): (input, output) numpy ndarray pairs
        of the format (n_samples, n_channels, n_rows, n_columns)
        """
        N_train = len(self.y_train)
        N_val = len(self.y_val)
        N_test = len(self.y_test)

        img_len = len(self.X_test[0])  # obtain length of flattened image
        num_rows = int(np.sqrt(img_len))
        num_cols = int(np.sqrt(img_len))
        num_channels = 1  # only have pixel intensities, no colour

        X_train = np.asarray(self.X_train).reshape((N_train, num_channels, num_rows, num_cols))
        X_val = np.asarray(self.X_val).reshape((N_val, num_channels, num_rows, num_cols))
        X_test = np.asarray(self.X_test).reshape((N_test, num_channels, num_rows, num_cols))

        y_train = np.asarray(self.y_train)
        y_val = np.asarray(self.y_val)
        y_test = np.asarray(self.y_test)

        if binary_class is True:
            np.place(y_train, mask=y_train > 0, vals=1)
            np.place(y_val, mask=y_val > 0, vals=1)
            np.place(y_test, mask=y_test > 0, vals=1)

        return X_train, y_train, X_val, y_val, X_test, y_test

    def get_training_set(self):
        return self.X_train, self.y_train

    def get_validation_set(self):
        return self.X_val, self.y_val

    def get_testing_set(self):
        return self.X_test, self.y_test

    def get_full_dataset(self):
        return self.full_dataset, self.all_labels

    def get_data_tuple(self):
        """
        Acts like a getter, but returns all sets at once as a tuple
        :return: tuple of all splits and corresponding labels as (set, labels), (set, labels), ...
        """
        return self.X_train, self.y_train, self.X_val, self.y_val, \
               self.X_test, self.y_test

    def get_training_set_positives(self):
        """Returns the training set with only the positive examples (volcanoes with labels 1-4)

        Returns:
            X_train_volcanoes: Training examples of class volcano
            y_train_volcanoes: List of labels ranging from 1 to 4
        """
        X_train_volcanoes = []
        y_train_volcanoes = []

        for training_example, label in zip(self.X_train, self.y_train):
            if label != 0:
                X_train_volcanoes.append(training_example)
                y_train_volcanoes.append(label)

        return X_train_volcanoes, y_train_volcanoes


if __name__ == "__main__":
    data = DataLoader()
    train, labels = data.get_training_set()
    val, labels = data.get_validation_set()
    test, labels = data.get_testing_set()
    full, labels = data.get_full_dataset()
    # print(len(data.get_training_set()), len(data.get_validation_set()), len(data.get_testing_set()))
    print(len(train), len(val), len(test))
    print(len(full))

    data = DataLoader('A1')

    train_a1 = data.get_training_set()
    print(len(train_a1))