-
Notifications
You must be signed in to change notification settings - Fork 4
/
encoders.py
230 lines (177 loc) · 6.63 KB
/
encoders.py
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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
#############################################
## Artemis ##
## Copyright (c) 2022-present NAVER Corp. ##
## CC BY-NC-SA 4.0 ##
#############################################
import os
import torch
import torch.nn as nn
import torchtext
import torchvision
from torch.nn.utils.rnn import pack_padded_sequence
from utils import l2norm
from config import TORCH_HOME, GLOVE_DIR
os.environ['TORCH_HOME'] = TORCH_HOME
def get_cnn(arch):
return torchvision.models.__dict__[arch](pretrained=True)
class GeneralizedMeanPooling(nn.Module):
"""
Applies a 2D power-average adaptive pooling over an input signal composed of several input planes.
The function computed is: :math:`f(X) = pow(sum(pow(X, p)), 1/p)`
- At p = infinity, one gets Max Pooling
- At p = 1, one gets Average Pooling
The output is of size H x W, for any input size.
The number of output features is equal to the number of input planes.
Args:
output_size: the target output size of the image of the form H x W.
Can be a tuple (H, W) or a single H for a square image H x H
H and W can be either a ``int``, or ``None`` which means the size will
be the same as that of the input.
"""
def __init__(self, norm, output_size=1, eps=1e-6):
super(GeneralizedMeanPooling, self).__init__()
assert norm > 0
self.p = float(norm)
self.output_size = output_size
self.eps = eps
def forward(self, x):
x = x.clamp(min=self.eps).pow(self.p)
return torch.nn.functional.adaptive_avg_pool2d(x, self.output_size).pow(1. / self.p)
def __repr__(self):
return self.__class__.__name__ + '(' \
+ str(self.p) + ', ' \
+ 'output_size=' + str(self.output_size) + ')'
class EncoderImage(nn.Module):
def __init__(self, opt):
super(EncoderImage, self).__init__()
# general parameters
embed_dim = opt.embed_dim
self.gradcam = opt.gradcam
# backbone CNN
self.cnn = get_cnn(opt.cnn_type)
self.cnn_dim = self.cnn.fc.in_features
self.pool_dim = self.cnn_dim # for avgpool resnet, cnn output and pooling output have the same dim
# replace the avgpool and the last fc layer of the CNN with identity
# then stack new pooling and fc layers at the end
self.cnn.avgpool = nn.Sequential()
self.cnn.fc = nn.Sequential()
self.gemp = GeneralizedMeanPooling(norm=3)
self.fc = nn.Linear(self.pool_dim, embed_dim)
# initialize placeholders for heatmap processing
if opt.gradcam :
# placeholder for activation maps
self.activations = None
# placeholder for the gradients
self.gradients = None
@property
def dtype(self):
return self.cnn.conv1.weight.dtype
def forward(self, images):
# backbone forward
out_7x7 = self.cnn( images ).view(-1, self.cnn_dim, 7, 7).type(self.dtype)
# save intermediary results for further studies (gradcam)
if self.gradcam:
out_7x7.requires_grad_(True)
# register activations
self.register_activations(out_7x7)
# register gradients
h = out_7x7.register_hook(self.activations_hook)
# encoder ending's forward
out = self.gemp(out_7x7).view(-1, self.pool_dim) # pooling
out = self.fc(out)
out = l2norm(out)
return out
# --- gradcam utilitary methods for heatmap processing ---
def activations_hook(self, grad):
""" hook for the gradients of the activations """
self.gradients = grad
def register_activations(self, activations):
self.activations = activations
def get_gradient(self):
""" gradient extraction """
return self.gradients
def get_activation(self):
""" activation extraction """
return self.activations
class EncoderText(nn.Module):
def __init__(self, word2idx, opt):
super(EncoderText, self).__init__()
wemb_type, word_dim, embed_dim = \
opt.wemb_type, opt.word_dim, opt.embed_dim
self.txt_enc_type = opt.txt_enc_type
self.embed_dim = embed_dim
# Word embedding
self.embed = nn.Embedding(len(word2idx), word_dim)
# Sentence embedding
if self.txt_enc_type == "bigru":
self.sent_enc = nn.GRU(word_dim, embed_dim//2, bidirectional=True, batch_first=True)
self.forward = self.forward_bigru
elif self.txt_enc_type == "lstm":
self.lstm_hidden_dim = opt.lstm_hidden_dim
self.sent_enc = nn.Sequential(
nn.LSTM(word_dim, self.lstm_hidden_dim),
nn.Dropout(p=0.1),
nn.Linear(self.lstm_hidden_dim, embed_dim),
)
self.forward = self.forward_lstm
self.init_weights(wemb_type, word2idx, word_dim)
def init_weights(self, wemb_type, word2idx, word_dim):
if wemb_type is None:
print("Word embeddings randomly initialized with xavier")
nn.init.xavier_uniform_(self.embed.weight)
else:
# Load pretrained word embedding
if 'glove' == wemb_type.lower():
wemb = torchtext.vocab.GloVe(cache=GLOVE_DIR)
else:
raise Exception('Unknown word embedding type: {}'.format(wemb_type))
assert wemb.vectors.shape[1] == word_dim
# Get word embeddings + keep track of missing words
missing_words = []
for word, idx in word2idx.items():
if word in wemb.stoi:
self.embed.weight.data[idx] = wemb.vectors[wemb.stoi[word]]
else:
missing_words.append(word)
print('Words: {}/{} found in vocabulary; {} words missing'.format(
len(word2idx)-len(missing_words), len(word2idx), len(missing_words)))
@property
def dtype(self):
return self.embed.weight.data.dtype
@property
def device(self):
return self.embed.weight.data.device
def forward_bigru(self, x, lengths):
# embed word ids to vectors
wemb_out = self.embed(x)
# for pytorch >= 1.7, length.device == 'cpu' (but it worked as a gpu variable in 1.2)
lengths = lengths.cpu()
# forward propagate RNNs
packed = pack_padded_sequence(wemb_out, lengths, batch_first=True)
if torch.cuda.device_count() > 1:
self.sent_enc.flatten_parameters()
_, rnn_out = self.sent_enc(packed)
# reshape output to (batch_size, hidden_size)
rnn_out = rnn_out.permute(1, 0, 2).contiguous().view(-1, self.embed_dim)
out = l2norm(rnn_out)
return out
def forward_lstm(self, x, lengths):
# embed word ids to vectors
wemb_out = self.embed(x) # size (batch, max_length, word_dim)
wemb_out = wemb_out.permute(1, 0, 2) # size (max_length, batch, word_dim)
# lstm
batch_size = wemb_out.size(1)
first_hidden = (torch.zeros(1, batch_size, self.lstm_hidden_dim),
torch.zeros(1, batch_size, self.lstm_hidden_dim))
if torch.cuda.is_available():
first_hidden = (first_hidden[0].cuda(), first_hidden[1].cuda())
lstm_output, last_hidden = self.sent_enc[0](wemb_out, first_hidden)
# extract features
text_features = []
for i in range(batch_size):
text_features.append(lstm_output[:, i, :].max(0)[0])
text_features = torch.stack(text_features)
# output
out = self.sent_enc[1:](text_features)
out = l2norm(out)
return out