-
Notifications
You must be signed in to change notification settings - Fork 3
Expand file tree
/
Copy pathutils.py
More file actions
158 lines (132 loc) · 5.11 KB
/
utils.py
File metadata and controls
158 lines (132 loc) · 5.11 KB
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
import os
import numpy as np
from math import sqrt
from scipy import stats
from torch_geometric.data import InMemoryDataset
from torch_geometric.loader import DataLoader
from torch_geometric import data as DATA
import torch
class TestbedDataset(InMemoryDataset):
def __init__(self, root='/tmp', dataset='davis',
xd=None, xt=None, y=None, transform=None,
pre_transform=None,smile_graph=None):
#root is required for save preprocessed data, default is '/tmp'
super(TestbedDataset, self).__init__(root, transform, pre_transform)
# benchmark dataset, default = 'davis'
self.dataset = dataset
if os.path.isfile(self.processed_paths[0]):
print('Pre-processed data found: {}, loading ...'.format(self.processed_paths[0]))
self.data, self.slices = torch.load(self.processed_paths[0])
else:
print('Pre-processed data {} not found, doing pre-processing...'.format(self.processed_paths[0]))
self.process(xd, xt, y,smile_graph)
self.data, self.slices = torch.load(self.processed_paths[0])
@property
def raw_file_names(self):
pass
#return ['some_file_1', 'some_file_2', ...]
@property
def processed_file_names(self):
return [self.dataset + '.pt']
def download(self):
# Download to `self.raw_dir`.
pass
def _download(self):
pass
def _process(self):
if not os.path.exists(self.processed_dir):
os.makedirs(self.processed_dir)
# Customize the process method to fit the task of drug-target affinity prediction
# Inputs:
# XD - list of SMILES, XT: list of encoded target (categorical or one-hot),
# Y: list of labels (i.e. affinity)
# Return: PyTorch-Geometric format processed data
def process(self, xd, xt, y,smile_graph):
assert (len(xd) == len(xt) and len(xt) == len(y)), "The three lists must be the same length!"
data_list = []
data_len = len(xd)
for i in range(data_len):
print('Converting SMILES to graph: {}/{}'.format(i+1, data_len))
smiles = xd[i]
target = xt[i]
labels = y[i]
# convert SMILES to molecular representation using rdkit
c_size, features, edge_index = smile_graph[smiles]
# make the graph ready for PyTorch Geometrics GCN algorithms:
GCNData = DATA.Data(x=torch.Tensor(features),
edge_index=torch.LongTensor(edge_index).transpose(1, 0),
y=torch.FloatTensor([labels]))
GCNData.target = torch.LongTensor([target])
GCNData.__setitem__('c_size', torch.LongTensor([c_size]))
# append graph, label and target sequence to data list
data_list.append(GCNData)
if self.pre_filter is not None:
data_list = [data for data in data_list if self.pre_filter(data)]
if self.pre_transform is not None:
data_list = [self.pre_transform(data) for data in data_list]
print('Graph construction done. Saving to file.')
data, slices = self.collate(data_list)
# save preprocessed data:
torch.save((data, slices), self.processed_paths[0])
def rmse(y,f):
rmse = sqrt(((y - f)**2).mean(axis=0))
return rmse
def r_squared_error(y_obs, y_pred):
y_obs = np.array(y_obs)
y_pred = np.array(y_pred)
y_obs_mean = [np.mean(y_obs) for y in y_obs]
y_pred_mean = [np.mean(y_pred) for y in y_pred]
mult = sum((y_pred - y_pred_mean) * (y_obs - y_obs_mean))
mult = mult * mult
y_obs_sq = sum((y_obs - y_obs_mean) * (y_obs - y_obs_mean))
y_pred_sq = sum((y_pred - y_pred_mean) * (y_pred - y_pred_mean))
return mult / float(y_obs_sq * y_pred_sq)
def get_k(y_obs, y_pred):
y_obs = np.array(y_obs)
y_pred = np.array(y_pred)
return sum(y_obs * y_pred) / float(sum(y_pred * y_pred))
def squared_error_zero(y_obs, y_pred):
k = get_k(y_obs, y_pred)
y_obs = np.array(y_obs)
y_pred = np.array(y_pred)
y_obs_mean = [np.mean(y_obs) for y in y_obs]
upp = sum((y_obs - (k * y_pred)) * (y_obs - (k * y_pred)))
down = sum((y_obs - y_obs_mean) * (y_obs - y_obs_mean))
return 1 - (upp / float(down))
def get_rm2(ys_orig, ys_line):
ys_orig = np.concatenate(ys_orig)
ys_line = np.concatenate(ys_line)
r2 = r_squared_error(ys_orig, ys_line)
r02 = squared_error_zero(ys_orig, ys_line)
return r2 * (1 - np.sqrt(np.absolute((r2 * r2) - (r02 * r02))))
def mse(y,f):
mse = ((y - f)**2).mean(axis=0)
return mse
def pearson(y,f):
rp = np.corrcoef(y, f)[0,1]
return rp
def spearman(y,f):
rs = stats.spearmanr(y, f)[0]
return rs
def ci(y,f):
ind = np.argsort(y)
y = y[ind]
f = f[ind]
i = len(y)-1
j = i-1
z = 0.0
S = 0.0
while i > 0:
while j >= 0:
if y[i] > y[j]:
z = z+1
u = f[i] - f[j]
if u > 0:
S = S + 1
elif u == 0:
S = S + 0.5
j = j - 1
i = i - 1
j = i-1
ci = S/z
return ci