GraphFeatureAutoencoder/eval.py at master · RaminHasibi/GraphFeatureAutoencoder · 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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
import torch
from sklearn.metrics import mean_squared_error as scimse
from torch_geometric.utils import to_undirected
import numpy as np
import networkx as nx
from sklearn.model_selection import KFold
from train_test import train_epoch, test
import copy
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from utils.functions import index_to_mask
from magic import MAGIC

def supervised_prediction_eval(model_class, data, opts):

    loss_train = []
    criterion = torch.nn.MSELoss()
    kf = KFold(n_splits=3, random_state=opts.seed)
    kf_feats = KFold(n_splits=3, random_state=opts.seed)

    mse = []

    for k, train_test_indices in enumerate(kf.split(data.x)):
        print('Fold number: {:d}'.format(k))
        y_pred = []
        train_index, test_index = train_test_indices
        eval_data = copy.deepcopy(data)
        if opts.random_graph:
            print('Random Graph used')
            G_rand = nx.gnp_random_graph(data.x.shape[0],opts.random_graph_alpha)
            eval_data.edge_index = to_undirected(torch.tensor(np.array(G_rand.edges()).T).to(opts.device))
            print(eval_data)
        train_feats_indeces, test_feats_indeces = next(kf_feats.split(np.arange(data.y.size(1))))
        if not opts.no_features:
            eval_data.x = data.x[:, train_feats_indeces]
        eval_data.y = data.y[:, test_feats_indeces]
        eval_data.train_mask = index_to_mask(train_index, eval_data.x.size(0))
        eval_data.test_mask = index_to_mask(test_index, eval_data.x.size(0))
        for exp_num in range(eval_data.y.size(1)):
            if (model_class == LinearRegression) | (model_class == RandomForestRegressor):
                model = model_class()
                model.fit(eval_data.x[eval_data.train_mask], eval_data.y[eval_data.train_mask, exp_num])
                pred = model.predict(eval_data.x[eval_data.test_mask])
                test_loss = scimse(pred,
                       eval_data.y[eval_data.test_mask, exp_num])
                print('Exp: {:03d}, Loss: {:.5f}'
                      .format(exp_num, test_loss))
                y_pred.append(pred)
            else:
                torch.manual_seed(opts.seed)
                if torch.cuda.is_available():
                    torch.cuda.manual_seed_all(opts.seed)

                model = model_class(eval_data.num_features, opts).to(opts.device)
                optimizer = torch.optim.Adam(model.parameters(), lr=opts.learning_rate)
                best_loss = 1e9
                for epoch in range(1, opts.epochs + 1):
                    loss_train = train_epoch(model, eval_data, optimizer, opts, exp_num, criterion)
                    if loss_train < best_loss:
                        best_loss = loss_train
                        best_model = copy.deepcopy(model)
                loss_test = test(best_model, eval_data, exp_num, criterion, opts)
                print('Exp: {:03d}, Loss: {:.5f}, TestLoss: {:.5f}'.
                      format(exp_num, loss_train, loss_test))
                with torch.no_grad():
                    y_pred.append(best_model(eval_data))

        for i in range(eval_data.y.size(1)):
            if (model_class == LinearRegression) | (model_class == RandomForestRegressor):
                mse.append(scimse(y_pred[i],
                                  eval_data.y[eval_data.test_mask, i]))
            else:
                mse.append(scimse(y_pred[i][eval_data.test_mask.cpu().numpy()].cpu().numpy(),
                                  eval_data.y[eval_data.test_mask, i].cpu().numpy().reshape([-1, 1])))
    print('Average+-std Error for test expression values: {:.5f}+-{:.5f}'.format(np.mean(mse), np.std(mse)))
    return mse

def embedding_prediction_eval(model_class, data, opts):
    loss_train = []

    kf = KFold(n_splits=3, random_state=opts.seed, shuffle=True)
    kf_feats = KFold(n_splits=3, random_state=opts.seed, shuffle=True)

    mse_lr = []
    mse_rf = []

    for k, train_test_indices in enumerate(kf.split(data.x)):
        print('Fold number: {:d}'.format(k))
        y_pred = []
        train_index, test_index = train_test_indices
        eval_data = copy.deepcopy(data)
        train_feats_indeces, test_feats_indeces = next(kf_feats.split(np.arange(data.y.size(1))))
        if not opts.no_features:
            eval_data.x = data.x[:, train_feats_indeces]
        eval_data.y = data.y[:, test_feats_indeces]
        eval_data.train_mask = index_to_mask(train_index, eval_data.x.size(0))
        eval_data.test_mask = index_to_mask(test_index, eval_data.x.size(0))
        model = model_class(eval_data.num_features, 32).to(opts.device)
        if torch.cuda.is_available():
            torch.cuda.manual_seed_all(opts.seed)
        optimizer = torch.optim.Adam(model.parameters(), lr=opts.learning_rate)
        print('Training the auto encoder!')
        for epoch in range(1, opts.epochs + 1):
            if epoch % 10 == 0:
                print('Epoch number: {:03d}'.format(epoch))
            train_epoch(model, eval_data, optimizer, opts)
        for exp_num in range(eval_data.y.size(1)):
            torch.manual_seed(opts.seed)
            z = model.encode(eval_data.x, eval_data.edge_index)
            model.fit_predictor(z[eval_data.train_mask].cpu().data.numpy(),
                                eval_data.y[eval_data.train_mask, exp_num].cpu().data.numpy())

            loss_test_lr, loss_test_rf = test(model, eval_data, exp_num, scimse, opts)
            model.eval()
            print('Exp: {:03d}, TestLoss_lr: {:.5f}, TestLoss_rf: {:.5f}'.
                  format(exp_num, loss_test_lr, loss_test_rf))
            with torch.no_grad():
                y_pred.append(model.predict(eval_data.x, eval_data.edge_index))
        for i in range(eval_data.y.size(1)):
            mse_lr.append(scimse(y_pred[i][0][eval_data.test_mask.cpu().numpy()],
                              eval_data.y[eval_data.test_mask, i].cpu().numpy().reshape([-1, 1])))
            mse_rf.append(scimse(y_pred[i][1][eval_data.test_mask.cpu().numpy()],
                                 eval_data.y[eval_data.test_mask, i].cpu().numpy().reshape([-1, 1])))

    print('Average+-std Error for test expression values LR: {:.5f}+-{:.5f}'.format(np.mean(mse_lr), np.std(mse_lr)))
    print('Average+-std Error for test expression values RF: {:.5f}+-{:.5f}'.format(np.mean(mse_rf), np.std(mse_rf)))
    return mse_lr, mse_rf


def imputation_eval(model_class, data, opts):
    if model_class == MAGIC:
        data.x = data.y = data.x.t()
        data.nonzeromask = data.nonzeromask.t()
    criterion = torch.nn.MSELoss()
    kf = KFold(n_splits=3, random_state=opts.seed, shuffle=True)
    loss_test = []
    if opts.dataset == 'Ecoli':
        indices = np.indices([data.x.size(0), data.x.size(1)]).reshape(2, -1)
    else:
        indices = np.array(data.x.cpu().data.numpy().nonzero())
    for k, train_test_indices in enumerate(kf.split(np.arange(len(indices[0])))):
        print('Fold number: {:d}'.format(k))
        train_index, test_index = train_test_indices
        eval_data = copy.deepcopy(data)
        eval_data.train_mask = index_to_mask([indices[0, train_index], indices[1, train_index]],
                                             eval_data.x.size()).to(opts.device)
        eval_data.test_mask = index_to_mask([indices[0, test_index], indices[1, test_index]],
                                            eval_data.x.size()).to(opts.device)
        eval_data.x = eval_data.x * eval_data.train_mask
        if model_class == MAGIC:
            pred = model_class().fit_transform((eval_data.x*eval_data.train_mask).cpu().data.numpy())
            loss_test.append(scimse(pred*eval_data.test_mask.cpu().data.numpy(),
                                    (eval_data.y*eval_data.test_mask).cpu().data.numpy()))
        else:
            model = model_class(eval_data.num_features, opts).to(opts.device)
            optimizer = torch.optim.Adam(model.parameters(), lr=opts.learning_rate)
            best_loss = 1e9
            for epoch in range(1, opts.epochs + 1):
                loss_train = train_epoch(model, eval_data, optimizer, opts, criterion=criterion)
                if loss_train < best_loss:
                    best_loss = loss_train
                    best_model = copy.deepcopy(model)
                if epoch % 10 == 0:
                    print('Epoch number: {:03d}, Train_loss: {:.5f}'.format(epoch, loss_train))
            loss_test.append(test(best_model, eval_data, None, criterion, opts))
            print('Loss: {:.5f}, TestLoss: {:.5f}'.format(loss_train, loss_test[k]))
    print('Average+-std Error for test RNA values: {:.5f}+-{:.5f}'.format(np.mean(loss_test), np.std(loss_test)))
    return np.mean(loss_test)