import numpy as np
import cudarray as ca
from ..feedforward.layers import Activation, FullyConnected
from ..loss import Loss
from ..base import Model, PickleMixin
from ..input import Input
from ..parameter import Parameter
[docs]class Autoencoder(Model, PickleMixin):
def __init__(self, n_out, weights, bias=0.0, bias_prime=0.0,
activation='sigmoid', loss='bce'):
self.name = 'autoenc'
self.n_out = n_out
self.activation = Activation(activation)
self.activation_decode = Activation(activation)
self.loss = Loss.from_any(loss)
self.weights = Parameter.from_any(weights)
self.bias = Parameter.from_any(bias)
self.bias_prime = Parameter.from_any(bias_prime)
self._initialized = False
self._tmp_x = None
self._tmp_y = None
def _setup(self, x_shape):
if self._initialized:
return
n_in = x_shape[1]
self.weights._setup((n_in, self.n_out))
if not self.weights.name:
self.weights.name = self.name + '_w'
self.bias._setup(self.n_out)
if not self.bias.name:
self.bias.name = self.name + '_b'
self.bias_prime._setup(n_in)
if not self.bias_prime.name:
self.bias_prime.name = self.name + '_b_prime'
self.loss._setup((x_shape[0], self.n_out))
self._initialized = True
@property
def _params(self):
return self.weights, self.bias, self.bias_prime
@_params.setter
def _params(self, params):
self.weights, self.bias, self.bias_prime = params
[docs] def output_shape(self, input_shape):
return (input_shape[0], self.n_out)
[docs] def encode(self, x):
self._tmp_x = x
y = ca.dot(x, self.weights.array) + self.bias.array
return self.activation.fprop(y)
[docs] def decode(self, y):
self._tmp_y = y
x = ca.dot(y, self.weights.array.T) + self.bias_prime.array
return self.activation_decode.fprop(x)
[docs] def decode_bprop(self, x_grad):
x_grad = self.activation_decode.bprop(x_grad)
ca.dot(x_grad.T, self._tmp_y, out=self.weights.grad_array)
ca.sum(x_grad, axis=0, out=self.bias_prime.grad_array)
return ca.dot(x_grad, self.weights.array)
[docs] def encode_bprop(self, y_grad):
y_grad = self.activation.bprop(y_grad)
# Because the weight gradient has already been updated by
# decode_bprop() we must add the contribution.
w_grad = self.weights.grad_array
w_grad += ca.dot(self._tmp_x.T, y_grad)
ca.sum(y_grad, axis=0, out=self.bias.grad_array)
return ca.dot(y_grad, self.weights.array.T)
def _update(self, x):
y_prime = self.encode(x)
x_prime = self.decode(y_prime)
x_prime_grad = self.loss.grad(x_prime, x)
y_grad = self.decode_bprop(x_prime_grad)
self.encode_bprop(y_grad)
return self.loss.loss(x_prime, x)
def _reconstruct_batch(self, x):
y = self.encode(x)
return self.decode(y)
[docs] def reconstruct(self, input):
""" Returns the reconstructed input. """
input = Input.from_any(input)
x_prime = np.empty(input.x.shape)
offset = 0
for x_batch in input.batches():
x_prime_batch = np.array(self._reconstruct_batch(x_batch))
batch_size = x_prime_batch.shape[0]
x_prime[offset:offset+batch_size, ...] = x_prime_batch
offset += batch_size
return x_prime
def _embed_batch(self, x):
return self.encode(x)
[docs] def embed(self, input):
""" Returns the embedding of the input. """
input = Input.from_any(input)
y = np.empty(self.output_shape(input.x.shape))
offset = 0
for x_batch in input.batches():
y_batch = np.array(self._embed_batch(x_batch))
batch_size = y_batch.shape[0]
y[offset:offset+batch_size, ...] = y_batch
offset += batch_size
return y
[docs] def feedforward_layers(self):
return [FullyConnected(self.n_out, self.weights.array,
self.bias.array),
self.activation]
[docs]class DenoisingAutoencoder(Autoencoder):
def __init__(self, n_out, weights, bias=0.0, bias_prime=0.0,
corruption=0.25, activation='sigmoid', loss='bce'):
super(DenoisingAutoencoder, self).__init__(
n_out=n_out, weights=weights, bias=bias, bias_prime=bias_prime,
activation=activation, loss=loss
)
self.corruption = corruption
[docs] def corrupt(self, x):
mask = ca.random.uniform(size=x.shape) < (1-self.corruption)
return x * mask
def _update(self, x):
x_tilde = self.corrupt(x)
y_prime = self.encode(x_tilde)
x_prime = self.decode(y_prime)
x_prime_grad = self.loss.grad(x_prime, x)
y_grad = self.decode_bprop(x_prime_grad)
self.encode_bprop(y_grad)
return self.loss.loss(x_prime, x)