forked from taolei87/rcnn
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathextended_layers.py
165 lines (137 loc) · 5.46 KB
/
extended_layers.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
import numpy as np
import theano
import theano.tensor as T
from theano.sandbox.rng_mrg import MRG_RandomStreams
from nn import create_optimization_updates, get_activation_by_name, sigmoid, linear
from nn import EmbeddingLayer, Layer, RecurrentLayer, LSTM, RCNN, apply_dropout, default_rng
from nn import create_shared, random_init
class ExtRCNN(RCNN):
def forward(self, x_t, mask_t, hc_tm1):
hc_t = super(ExtRCNN, self).forward(x_t, hc_tm1)
hc_t = mask_t * hc_t + (1-mask_t) * hc_tm1
return hc_t
def forward_all(self, x, mask, h0=None, return_c=False):
if h0 is None:
if x.ndim > 1:
h0 = T.zeros((x.shape[1], self.n_out*(self.order+1)), dtype=theano.config.floatX)
else:
h0 = T.zeros((self.n_out*(self.order+1),), dtype=theano.config.floatX)
h, _ = theano.scan(
fn = self.forward,
sequences = [ x, mask ],
outputs_info = [ h0 ]
)
if return_c:
return h
elif x.ndim > 1:
return h[:,:,self.n_out*self.order:]
else:
return h[:,self.n_out*self.order:]
def copy_params(self, from_obj):
self.internal_layers = from_obj.internal_layers
self.bias = from_obj.bias
class ExtLSTM(LSTM):
def forward(self, x_t, mask_t, hc_tm1):
hc_t = super(LSTM, self).forward(x_t, hc_tm1)
hc_t = mask_t * hc_t + (1-mask_t) * hc_tm1
return hc_t
def forward_all(self, x, mask, h0=None, return_c=False):
if h0 is None:
if x.ndim > 1:
h0 = T.zeros((x.shape[1], self.n_out*(self.order+1)), dtype=theano.config.floatX)
else:
h0 = T.zeros((self.n_out*(self.order+1),), dtype=theano.config.floatX)
h, _ = theano.scan(
fn = self.forward,
sequences = [ x, mask ],
outputs_info = [ h0 ]
)
if return_c:
return h
elif x.ndim > 1:
return h[:,:,self.n_out*self.order:]
else:
return h[:,self.n_out*self.order:]
def copy_params(self, from_obj):
self.internal_layers = from_obj.internal_layers
class ZLayer(object):
def __init__(self, n_in, n_hidden, activation):
self.n_in, self.n_hidden, self.activation = \
n_in, n_hidden, activation
self.MRG_rng = MRG_RandomStreams()
self.create_parameters()
def create_parameters(self):
n_in, n_hidden = self.n_in, self.n_hidden
activation = self.activation
self.w1 = create_shared(random_init((n_in,)), name="w1")
self.w2 = create_shared(random_init((n_hidden,)), name="w2")
bias_val = random_init((1,))[0]
self.bias = theano.shared(np.cast[theano.config.floatX](bias_val))
rlayer = RCNN((n_in+1), n_hidden, activation=activation, order=2)
self.rlayer = rlayer
self.layers = [ rlayer ]
def forward(self, x_t, z_t, h_tm1, pz_tm1):
print "z_t", z_t.ndim
pz_t = sigmoid(
T.dot(x_t, self.w1) +
T.dot(h_tm1[:,-self.n_hidden:], self.w2) +
self.bias
)
xz_t = T.concatenate([x_t, z_t.reshape((-1,1))], axis=1)
h_t = self.rlayer.forward(xz_t, h_tm1)
# batch
return h_t, pz_t
def forward_all(self, x, z):
assert x.ndim == 3
assert z.ndim == 2
xz = T.concatenate([x, z.dimshuffle((0,1,"x"))], axis=2)
h0 = T.zeros((1, x.shape[1], self.n_hidden), dtype=theano.config.floatX)
h = self.rlayer.forward_all(xz)
h_prev = T.concatenate([h0, h[:-1]], axis=0)
assert h.ndim == 3
assert h_prev.ndim == 3
pz = sigmoid(
T.dot(x, self.w1) +
T.dot(h_prev, self.w2) +
self.bias
)
assert pz.ndim == 2
return pz
def sample(self, x_t, z_tm1, h_tm1):
print "z_tm1", z_tm1.ndim, type(z_tm1)
pz_t = sigmoid(
T.dot(x_t, self.w1) +
T.dot(h_tm1[:,-self.n_hidden:], self.w2) +
self.bias
)
# batch
pz_t = pz_t.ravel()
z_t = T.cast(self.MRG_rng.binomial(size=pz_t.shape,
p=pz_t), theano.config.floatX)
xz_t = T.concatenate([x_t, z_t.reshape((-1,1))], axis=1)
h_t = self.rlayer.forward(xz_t, h_tm1)
return z_t, h_t
def sample_all(self, x):
h0 = T.zeros((x.shape[1], self.n_hidden*(self.rlayer.order+1)), dtype=theano.config.floatX)
z0 = T.zeros((x.shape[1],), dtype=theano.config.floatX)
([ z, h ], updates) = theano.scan(
fn = self.sample,
sequences = [ x ],
outputs_info = [ z0, h0 ]
)
assert z.ndim == 2
return z, updates
@property
def params(self):
return [ x for layer in self.layers for x in layer.params ] + \
[ self.w1, self.w2, self.bias ]
@params.setter
def params(self, param_list):
start = 0
for layer in self.layers:
end = start + len(layer.params)
layer.params = param_list[start:end]
start = end
self.w1.set_value(param_list[-3].get_value())
self.w2.set_value(param_list[-2].get_value())
self.bias.set_value(param_list[-1].get_value())