wassname/pyrobolearn
1
0
Fork 0
mirror of https://github.com/wassname/pyrobolearn.git synced 2026-09-09 11:31:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

update NN models (DNN, MLP, CNN, AE, VAE, GAN)

Browse Source
This commit is contained in:
Brian Delhaisse
2019-04-09 04:02:51 +02:00
parent 3cb65995fb
commit b3e92b1e7f
11 changed files with 1227 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/models/basics/linear.py
+17
View File
@@ -150,6 +150,23 @@ class Linear(object):
# Methods #
###########
def copy_parameters(self, parameters):
"""Copy the given parameters.
Args:
parameters (NN, torch.nn.Module, generator, iterable): the other model's parameters to copy.
"""
if isinstance(parameters, self.__class__):
self.model.load_state_dict(parameters.model.state_dict())
elif isinstance(parameters, torch.nn.Module):
self.model.load_state_dict(parameters.state_dict())
elif isinstance(parameters, (types.GeneratorType, collections.Iterable)):
for model_params, other_params in zip(self.parameters(), parameters):
model_params.data.copy_(other_params.data)
else:
raise TypeError("Expecting the given parameters to be an instance of `NN`, `torch.nn.Module`, `generator`"
", or an iterable object, instead got: {}".format(type(parameters)))
def parameters(self):
"""Returns an iterator over the model parameters."""
return self.model.parameters()
pyrobolearn/models/model.py
+17
View File
@@ -219,6 +219,23 @@ class Model(object):
pass
self._models.append(model)
# def copy_parameters(self, parameters):
# """Copy the given parameters.
#
# Args:
# parameters (NN, torch.nn.Module, generator, iterable): the other model's parameters to copy.
# """
# if isinstance(parameters, self.__class__):
# self.model.load_state_dict(parameters.model.state_dict())
# elif isinstance(parameters, torch.nn.Module):
# self.model.load_state_dict(parameters.state_dict())
# elif isinstance(parameters, (types.GeneratorType, collections.Iterable)):
# for model_params, other_params in zip(self.parameters(), parameters):
# model_params.data.copy_(other_params.data)
# else:
# raise TypeError("Expecting the given parameters to be an instance of `NN`, `torch.nn.Module`, `generator`"
# ", or an iterable object, instead got: {}".format(type(parameters)))
@abstractmethod
def parameters(self):
"""Return an iterator over the parameters of the model."""
pyrobolearn/models/nn/__init__.py
+5 -5
View File
@@ -9,16 +9,16 @@ from .mlp import *
from .neat_model import NEATModel
# import convolutional neural network
# from cnn import *
from .cnn import *
# import recurrent neural network
# from rnn import *
from .rnn import *
# import auto-encoder
# from ae import *
from .ae import *
# import variational auto-encoder
# from vae import *
from .vae import *
# import generative adversarial networks
# from gan import *
from .gan import *
pyrobolearn/models/nn/ae.py
+137 -55
View File
@@ -18,9 +18,6 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN
@@ -38,74 +35,159 @@ __status__ = "Development"
class AE(NN):
r"""Auto-Encoder
Auto-encoders are composed of an encoder and decoder parts. The encoder encodes/projects the input data into a
lower dimensional space, while the decoder projects the latent data back to the output space. The output space is
often the same as the input space.
References:
[1] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
"""
def __init__(self):
pass
class _AETorch(torch.nn.Module):
r"""Auto-Encoder written in Pytorch (that inherits from `torch.nn.Module`)
"""
def __init__(self, layer_sizes=[], activation_fct=None, dropout=None, encoder=None, decoder=None):
super(_AETorch, self).__init__()
if encoder is None and decoder is None:
# nb of layers (the input layer doesn't count)
self.num_layers = len(layer_sizes) - 1
layers = [torch.nn.Linear(layer_sizes[i], layer_sizes[i+1]) for i in range(self.num_layers)]
# check activation function and insert it after each linear layer
if activation_fct is not None:
if isinstance(activation_fct, str):
activation_fct = getattr(torch.nn, activation_fct)()
elif activation_fct.__module__ == 'torch.nn.modules.activation':
if inspect.isclass(activation_fct):
activation_fct = activation_fct()
else:
raise ValueError("activation_fct should be a string or belong to torch.nn.modules.activation")
# add activation layer
for i in range(self.num_layers-1, 0, -1):
layers.insert(activation_fct)
# check dropout
if dropout is not None:
if isinstance(dropout, float):
dropout = torch.nn.Dropout(dropout)
elif dropout.__module__ == 'torch.nn.modules.dropout':
raise ValueError("Dropout should be a float or belong to torch.nn.modules.dropout")
# add dropout layer
for i in range(self.num_layers-1, 0, -2):
layers.insert(dropout)
# Encoder
encoder = torch.nn.Sequential(*layers[:len(layers)//2])
# Decoder
decoder = torch.nn.Sequential(*layers[len(layers)//2:])
def __init__(self, encoder, decoder, input_shape, output_shape):
"""
Initialize the auto-encoder.
Args:
encoder (torch.nn.Module): encoder module.
decoder (torch.nn.Module): decoder module.
input_shape (tuple of int): input shape.
output_shape (tuple of int): output shape.
"""
self.encoder = encoder
self.decoder = decoder
# model = torch.nn.Sequential(*(list(encoder.modules())[1:] + list(decoder.modules())[1:]))
model = torch.nn.Sequential(encoder, decoder)
super(AE, self).__init__(model, input_shape, output_shape)
##################
# Static Methods #
##################
@staticmethod
def is_latent():
"""AEs are latent models."""
return True
###########
# Methods #
###########
def forward(self, x):
"""Forward the input to the encoder and decoder."""
x = self.encoder(x)
x = self.decoder(x)
return x
def encode(self, x):
x = self.encoder(x)
return x
"""Run the encoder."""
return self.encoder(x)
def decode(self, x):
x = self.decoder(x)
return x
"""Run the decoder."""
return self.decoder(x)
class AETorch(NNTorch):
r"""Auto-Encoder in Pytorch
class MLP_AE(AE):
r"""Multi-Layer Peceptron Auto-Encoder
"""
pass
def __init__(self, encoder_units=[], decoder_units=None, activation=None, last_activation=None, dropout=None):
"""
Initialize the MLP AE.
Args:
encoder_units (list/tuple of int): number of units for each layer in the encoder (this includes the input
layer)
decoder_units (list/tuple of int, None): number of units for each layer in the decoder (this includes the
output layer). If None, it will take the encoder units but in the reverse order.
activation (None, str, or list/tuple of str/None): activation function to be applied after each layer.
If list/tuple, then it has to match the number of
hidden layers.
last_activation (None or str): last activation function to be applied. If not specified, it will check
if it is in the list/tuple of activation functions provided for the
previous argument.
dropout (None, float, or list/tuple of float/None): dropout probability.
"""
# check the encoder length
if len(encoder_units) < 2:
raise ValueError("Expecting more than the input layer for the encoder.")
# check decoder units.
if decoder_units is None:
decoder_units = encoder_units[::-1] # reverse
else:
decoder_units = encoder_units[-1:] + decoder_units
if len(decoder_units) == 1:
raise ValueError("Expecting at least the output layer for the decoder.")
num_units = encoder_units + decoder_units[1:]
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
activations.update({act.lower(): act for act in activations})
def check_activation(activation):
if activation is None or activation.lower() == 'linear':
activation = None
else:
if activation not in activations:
raise ValueError("The given activation function is not available")
activation = getattr(torch.nn, activations[activation])
return activation
activation = check_activation(activation)
last_activation = check_activation(last_activation)
# check dropout
dropout_layer = None
if dropout is not None:
dropout_layer = torch.nn.Dropout(dropout)
# build pytorch network
def build_layers(units, encoder=True):
layers = []
size = len(units[:-1]) if encoder else len(units[:-2])
for i in range(size):
# add linear layer
layer = torch.nn.Linear(units[i], units[i + 1])
layers.append(layer)
# add activation layer
if activation is not None:
layers.append(activation())
# add dropout layer
if dropout_layer is not None:
layers.append(dropout_layer)
# last output layer if decoder
if not encoder:
layers.append(torch.nn.Linear(units[-2], units[-1]))
if last_activation is not None:
layers.append(last_activation)
return layers
encoding_layers = build_layers(encoder_units, encoder=True)
decoding_layers = build_layers(decoder_units, encoder=False)
# Encoder
encoder = torch.nn.Sequential(*encoding_layers)
# Decoder
decoder = torch.nn.Sequential(*decoding_layers)
super(MLP_AE, self).__init__(encoder, decoder, input_shape=tuple([num_units[0]]),
output_shape=tuple([num_units[-1]]))
# Tests
if __name__ == '__main__':
# create MLP network
autoencoder = MLP_AE(encoder_units=(4, 3, 2), activation='relu')
print(autoencoder)
x = torch.rand(4)
y = autoencoder(x)
print("Input: {} - Output: {}".format(x, y))
pyrobolearn/models/nn/cnn.py
+248 -9
View File
@@ -18,13 +18,12 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
import math
from pyrobolearn.models.nn.dnn import NN
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
@@ -35,15 +34,255 @@ __email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Flatten(torch.nn.Module):
r"""Flatten layer."""
def forward(self, x):
return x.view(x.size(0), -1)
class CNN(NN):
r"""Convolutional Neural Network
r"""(Feed-forward) Convolutional Neural Networks
Feed-forward CNN.
Convolutional neural networks are neural networks that captures the spatial relationship between data features.
For instance, they are used for pictures where neighboring pixels are related to each other.
References:
[1] "CS231n: Convolutional Neural Networks for Visual Recognition", Li et al., 2019
"""
pass
def __init__(self, model, input_shape, output_shape):
super(CNN, self).__init__(model, input_shape, output_shape)
def forward(self, x):
"""Forward the inputs in the network."""
# reshape the inputs if necessary
unsqueezed = True if len(x.shape) == 3 else False
if unsqueezed:
x = x.unsqueeze(0)
# feed the inputs to the model
x = self.model(x)
# reshape the outputs
if unsqueezed:
x = x.squeeze(0)
# return the output
return x
class CNNTorch(NNTorch):
r"""Convolutional Neural Network in PyTorch
class CNN2D(CNN):
r"""(Feed-forward) Convolutional Neural Networks
Convolutional neural networks are neural networks that captures the spatial relationship between data features.
For instance, they are used for pictures where neighboring pixels are related to each other.
References:
[1] "CS231n: Convolutional Neural Networks for Visual Recognition", Li et al., 2019
"""
pass
def __init__(self, units=[], activation='ReLU', last_activation=None, pool=None, dropout=None, dropout2d=None,
use_batch_norm=False, model=None):
"""
Initialize the convolutional neural network.
Args:
units (list/tuple of int): number of units per layer. For instance,
`units=[(3,32,32), (3,6,5), [2,2], 10, 2]` means the input shape is (3,32,32), the next layer
is a convolutional layer with `in_channels=3`, `out_channels=6`, `kernel_size=5`, the next layer is
a pooling layer with a kernel_size of 2 and a stride of 2, the next layer is a flatten layer which
feeds the output to a linear layer of 10 units to finally finish with 2 units in the output layer.
Use tuples to specify convolutional layers and lists to specify pooling layers.
activation (str, torch.nn.Module, None): activation function to be used after convolution layers and linear
layers.
last_activation (str, torch.nn.Module, None): last activation function to be used at the output layer.
pool (torch.nn.Module, None): pooling layer.
dropout2d (float, None): dropout probability for 2d layers. If None, the probability is 0. This value
should be smaller than the dropout probability for 1d layer (i.e. below at least 0.2).
dropout (float, None): dropout probability for 1d layers. If None, the probability is 0.
use_batch_norm (bool): If True, it will use batch norm.
model (torch.nn.Module, None): If the model is given, it will not use the previous defined
"""
# create convolutional neural network if necessary
if model is None:
# check number of units
if len(units) < 2:
raise ValueError("The num_units list/tuple needs to have at least the input and output layers")
# check that the input shape is 2d
if len(units[0]) != 2 and len(units[0]) != 3:
raise ValueError("Expecting the input shape to be (Height, Width) or (Channel, Height, Width), "
"instead got a length of: {}".format(len(units[0])))
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
activations.update({act.lower(): act for act in activations})
def check_activation(activation):
if activation is None or activation.lower() == 'linear':
activation = None
else:
if activation not in activations:
raise ValueError("The given activation function {} is not available".format(activation))
activation = getattr(torch.nn, activations[activation])
return activation
activation = check_activation(activation)
last_activation = check_activation(last_activation)
# check dropout
dropout_layer, dropout2d_layer = None, None
if dropout is not None:
dropout_layer = torch.nn.Dropout(dropout)
if dropout2d is not None:
dropout2d_layer = torch.nn.Dropout2d(dropout2d)
# check pooling layer
pooling_layer = None
if pool is not None:
if isinstance(pool, torch.nn.Module):
pooling_layer = pool
elif isinstance(pool, str):
pools = {item: item for item in dir(torch.nn) if item[-6:] == 'Pool2d'}
pools.update({item.lower(): item for item in pools})
if pool not in pools:
raise ValueError("The given pooling layer {} has not been implemented".format(pool))
pooling_layer = getattr(torch.nn, pools[pool])
else:
pass
# keep track of (C, H, W) dimensions
if len(units[0]) == 2:
channel = 1
height, width = units[0]
else: # elif len(units[0]):
channel, height, width = units[0]
# build network
layers = []
linear_layer = False
for i in range(1, len(units)):
# check if last layer
if i == len(units) - 1:
if isinstance(units[i], int):
# if first linear layer, add flatten layer
in_features = units[i-1]
if i - 1 > 0 and not isinstance(units[i-1], int):
layers.append(Flatten())
in_features = channel * height * width
# add linear layer
layers.append(torch.nn.Linear(in_features, units[i]))
# get out of the loop
break
# convolution layer
if isinstance(units[i], tuple):
if linear_layer:
raise ValueError("Got a convolution layer after a linear layer... This is not supported.")
# add convolution layer
unit = units[i]
if len(units[i]) == 2:
unit = (channel,) + units[i]
layer = torch.nn.Conv2d(*unit)
layers.append(layer)
# compute new dimensions
channel = layer.out_channels
p, d, k, s = layer.padding, layer.dilation, layer.kernel_size, layer.stride
height = int(math.floor((height + 2. * p[0] - d[0] * (k[0] - 1) - 1.) / s[0] + 1))
width = int(math.floor((width + 2. * p[1] - d[1] * (k[1] - 1) - 1.) / s[1] + 1))
# if use batch normalization
if use_batch_norm:
layers.append(torch.nn.BatchNorm2d(num_features=unit[1]))
# add activation layer
if activation is not None:
layers.append(activation())
# add dropout layer if specified
if dropout2d_layer is not None:
layers.append(dropout2d_layer)
# pooling layer
elif isinstance(units[i], list):
if linear_layer:
raise ValueError("Got a pooling layer after a linear layer... This is not supported.")
if pooling_layer is not None:
layer = pooling_layer(*units[i])
layers.append(layer)
# compute new dimensions
p, d, k, s = layer.padding, layer.dilation, layer.kernel_size, layer.stride
if isinstance(p, int):
p = (p, p)
if isinstance(k, int):
k = (k, k)
if isinstance(s, int):
s = (s, s)
if isinstance(d, int):
d = (d, d)
height = int(math.floor((height + 2. * p[0] - d[0] * (k[0] - 1) - 1.) / s[0] + 1))
width = int(math.floor((width + 2. * p[1] - d[1] * (k[1] - 1) - 1.) / s[1] + 1))
# linear layer
elif isinstance(units[i], int):
linear_layer = True
# if first linear layer, add flatten layer
in_features = units[i-1]
if not isinstance(units[i-1], int): # before last layer was convolutional layer
layers.append(Flatten())
in_features = channel * height * width
# add linear layer
layers.append(torch.nn.Linear(in_features, units[i]))
# add batch normalization layer
if use_batch_norm:
layers.append(torch.nn.BatchNorm1d(num_features=units[i + 1]))
# add activation layer
if activation is not None:
layers.append(activation())
# add dropout layer if specified
if dropout_layer is not None:
layers.append(dropout_layer)
else:
raise TypeError("One of the units is not an int, tuple, or list, instead got: "
"{}".format(type(units[i])))
# add last activation function
if last_activation is not None:
layers.append(last_activation)
# create nn model
model = torch.nn.Sequential(*layers)
# define input and output shapes
input_shape = tuple([units[0]]) if isinstance(units[0], int) else units[0]
output_shape = tuple([units[-1]]) if isinstance(units[-1], int) else units[-1]
super(CNN2D, self).__init__(model, input_shape=input_shape, output_shape=output_shape)
# Tests
if __name__ == '__main__':
input_shape = (3, 32, 32)
# create convolutional neural network
cnn = CNN2D(units=[input_shape, (3, 6, 5), [2, 2], (6, 16, 5), [2, 2], 120, 84, 10], activation='relu',
pool='MaxPool2d')
print(cnn)
x = torch.rand(*input_shape)
y = cnn.forward(x)
print("Input: {} - Output: {}".format(x, y))
pyrobolearn/models/nn/dnn.py
+78 -45
View File
@@ -18,7 +18,8 @@ References:
"""
import copy
import inspect
import types
import collections
import numpy as np
import torch
@@ -63,7 +64,7 @@ class NN(object): # Model
# - nn.Sequential: https://pytorch.org/docs/master/_modules/torch/nn/modules/container.html#Sequential
"""
def __init__(self, model, input_shape, output_shape, framework=None):
def __init__(self, model, input_shape, output_shape): # framework=None):
r"""Initialize the NN model.
Args:
@@ -90,7 +91,7 @@ class NN(object): # Model
self.base_output = None
# TODO: infer the framework based on the model
self.framework = framework
self.framework = 'pytorch' # framework
##############
# Properties #
@@ -161,6 +162,15 @@ class NN(object): # Model
# Static Methods #
##################
@staticmethod
def copy(other, deep=True):
"""Return another copy of the learning model"""
if not isinstance(other, Model):
raise TypeError("Trying to copy an object which is not a Linear model")
if deep:
return copy.deepcopy(other)
return copy.copy(other)
@staticmethod
def is_parametric():
"""A neural network is a parametric model"""
@@ -195,6 +205,11 @@ class NN(object): # Model
for instance, for generative adversarial networks (GANs) and variational auto-encoders (VAEs)."""
return False
@staticmethod
def is_latent(): # unless AE, VAE,...
"""Standard neural networks are not latent models but they can be like (variational) auto-encoders."""
return False
###########
# Methods #
###########
@@ -211,6 +226,23 @@ class NN(object): # Model
# for param in self.model.parameters():
# param.requires_grad = False
def copy_parameters(self, parameters):
"""Copy the given parameters.
Args:
parameters (NN, torch.nn.Module, generator, iterable): the other model's parameters to copy.
"""
if isinstance(parameters, self.__class__):
self.model.load_state_dict(parameters.model.state_dict())
elif isinstance(parameters, torch.nn.Module):
self.model.load_state_dict(parameters.state_dict())
elif isinstance(parameters, (types.GeneratorType, collections.Iterable)):
for model_params, other_params in zip(self.parameters(), parameters):
model_params.data.copy_(other_params.data)
else:
raise TypeError("Expecting the given parameters to be an instance of `NN`, `torch.nn.Module`, `generator`"
", or an iterable object, instead got: {}".format(type(parameters)))
def parameters(self):
"""Return an iterator over the model parameters."""
return self.model.parameters()
@@ -237,21 +269,51 @@ class NN(object): # Model
the activation functions, etc."""
raise NotImplementedError
def predict(self, x=None, to_numpy=False):
"""Predict the output given the input."""
def predict(self, inputs, to_numpy=False):
"""Predict the output given the input :attr:`inputs`.
Args:
inputs (np.ndarray, torch.Tensor): input vector/matrix
to_numpy (bool): if True, return a np.array
Returns:
np.ndarray, torch.Tensor: output vector/matrix
"""
# convert to torch tensor if necessary
if isinstance(x, np.ndarray):
x = torch.from_numpy(x).float()
if isinstance(inputs, np.ndarray):
inputs = torch.from_numpy(inputs).float()
# predict output given input
x = self.model(x)
inputs = self.model(inputs)
# return the output (and convert it to numpy if specified)
if to_numpy:
if x.requires_grad:
return x.detach().numpy()
return x.numpy()
return x
if inputs.requires_grad:
return inputs.detach().numpy()
return inputs.numpy()
return inputs
def forward(self, inputs):
return self.predict(inputs, to_numpy=False)
def save(self, filename):
"""
Save the neural network to the specified file.
Args:
filename (str): filename to save the neural network
"""
torch.save(self.model, filename)
def load(self, filename):
"""
Load the neural network from the specified file.
Args:
filename (str): filename from which to load the neural network
"""
self.model = torch.load(filename)
# check input and output dimensions
#############
# Operators #
@@ -276,6 +338,10 @@ class NN(object): # Model
"""
return self.model[key]
def __call__(self, inputs, to_numpy=True):
"""Predict the output given the input :attr:`inputs`."""
return self.predict(inputs, to_numpy=to_numpy)
def __rshift__(self, other):
"""
Concatenate two NN models in sequence, and return the sequenced model.
@@ -296,36 +362,3 @@ class NN(object): # Model
# return the concatenation
return NN(model)
class NNTorch(NN):
r"""Neural Network written in PyTorch
"""
def __init__(self, model, input_shape, output_shape):
super(NNTorch, self).__init__(model, input_shape, output_shape, framework='pytorch')
def save(self, filename):
"""
Save the neural network to the specified file.
Args:
filename (str): filename to save the neural network
"""
torch.save(self.model, filename)
def load(self, filename):
"""
Load the neural network from the specified file.
Args:
filename (str): filename from which to load the neural network
"""
self.model = torch.load(filename)
# check input and output dimensions
def __str__(self):
"""
Return string describing the NN model.
"""
return str(self.model)
pyrobolearn/models/nn/gan.py
+369 -12
View File
@@ -20,13 +20,11 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
@@ -42,29 +40,388 @@ class GAN(NN):
Type: generative model
Generative adversarial networks are two networks, a generator and a discriminator competing against each other.
The generator tries to produce fake data that are similar to the real data such that the discriminator can not
distinguished the fake data from the real one [1,2]. Once trained the generator is trained to generate samples
that are similar to the original data distribution.
Several GAN models implemented in PyTorch are provided in [3].
Note that with multiple layers, the networks become sensitive to the initial values of the weights and will fail
to train. This problem can be solved by using batch normalization [2] for the layers (except the output layer for
the generator and the input layer of the discriminator). The use of Leaky ReLU for the layers in the discriminator
is also advised in order to propagate the gradients for negative values. The use of ReLU or Leaky ReLU is advised
for the generator except the last layer which has been shown to perform the best with a tanh layer. The last layer
for the discriminator should be a sigmoid function with 0 indicating that the data is fake and 1 the data comes
from the real data distribution.
The loss being minimized for the discriminator is the sum of the losses for real and fake images using the sigmoid
cross-entropy loss for each one of them. That is, the predicted logit outputs of the discriminator on the real
data should be as close as possible to the label 1, while on the fake data the labels should be 0.
As for the generator, the loss being optimized is the sigmoid cross entropy loss taken on the discriminator output
logits but such that the labels are 1 for the fake data.
With these losses, it can be seen that the generator is trying to fool the discriminator while the discriminator
is trying to distinguish between the real and fake data.
.. seealso:: Variational Auto-Encoders
References:
[1] "NIPS:
[2]
[1] "Generative Adversarial Networks", Goodfellow et al., 2014
[2] "Improved Techniques for Training GANs", Salimans et al., 2016
[3] "PyTorch Implementations of GANs", Linder-Noren, 2018
"""
def __init__(self):
pass
def __init__(self, generator, discriminator, input_shape, output_shape):
self.generator = generator
self.discriminator = discriminator
model = torch.nn.Sequential(generator, discriminator)
super(GAN, self).__init__(model, input_shape, output_shape)
##################
# Static methods #
##################
@staticmethod
def isDiscriminative():
def is_discriminative():
"""A neural network is a discriminative model which given inputs predicts some outputs"""
return True
@staticmethod
def isGenerative(): # unless VAE, GAN,...
def is_generative(): # unless VAE, GAN,...
"""Standard neural networks are not generative, and thus we can not sample from it. This is different,
for instance, for generative adversarial networks (GANs) and variational auto-encoders (VAEs)."""
return True
###########
# Methods #
###########
class GANTorch(NNTorch):
r"""Generative Adversarial Network in PyTorch
def forward(self, *inputs):
"""The inputs is discarded."""
x = self.generate_latent(*inputs)
x = self.generate(x)
x = self.discriminator(x)
return x
def generate_latent(self, sample_shape=None):
"""
Generate latent vector / matrix.
Args:
sample_shape (tuple of int, None): shape of the latent vectors to generate.
Returns:
torch.Tensor: latent vector / matrix.
"""
if sample_shape is None:
sample_shape = self.input_shape
return torch.randn(*sample_shape)
def generate(self, latent):
"""Generate data.
Args:
latent (torch.Tensor): latent vector / matrix.
Returns:
torch.Tensor: fake data generated by the generator
"""
# generate the fake data
x = self.generator(latent)
return x
def discriminate(self, data):
"""Discriminate the given data.
Args:
data (torch.Tensor): fake or real data to distinguish.
Returns:
torch.Tensor: value between 0 and 1.
"""
return self.discriminator(data)
class MLP_GAN(GAN):
r"""Generative Multi-Layer Perceptron.
This implements Generative Adversarial Networks (GANs) using Multi-Layer Perceptrons (MLPs).
Generative adversarial networks are two networks, a generator and a discriminator competing against each other.
The generator tries to produce fake data that are similar to the real data such that the discriminator can not
distinguished the fake data from the real one [1,2]. Once trained the generator is trained to generate samples
that are similar to the original data distribution.
Several GAN models implemented in PyTorch are provided in [3].
Note that with multiple layers, the networks become sensitive to the initial values of the weights and will fail
to train. This problem can be solved by using batch normalization [2] for the layers (except the output layer for
the generator and the input layer of the discriminator). The use of Leaky ReLU for the layers in the discriminator
is also advised in order to propagate the gradients for negative values. The use of ReLU or Leaky ReLU is advised
for the generator except the last layer which has been shown to perform the best with a tanh layer. The last layer
for the discriminator should be a sigmoid function with 0 indicating that the data is fake and 1 the data comes
from the real data distribution.
References:
[1] "Generative Adversarial Networks", Goodfellow et al., 2014
[2] "Improved Techniques for Training GANs", Salimans et al., 2016
[3] "PyTorch Implementations of GANs", Linder-Noren, 2018
"""
pass
def __init__(self, generator_units=[], discriminator_units=[],
generator_activation='LeakyReLU', generator_last_activation='tanh',
discriminator_activation='LeakyReLU', discriminator_last_activation='sigmoid',
use_batch_norm=False):
"""
Initialize the MLP-GAN.
Args:
generator_units (list/tuple of int): number of units per layer in the generator, including the data input
dimension.
discriminator_units (list/tuple of int): number of units per layer in the discriminator. The last unit
must be one. If it is not the case, the one will automatically be added.
generator_activation (str, torch.nn.Module, None): activation function to be used at each layer in the
generator. They can be found in `torch.nn.modules.activation.*`. If None, it will use 'LeakyReLU'.
generator_last_activation (str, torch.nn.Module, None): last activation function to be used on the output
of the generator. By default, it will use 'tanh', so the output is between -1 and 1.
discriminator_activation (str, torch.nn.Module): activation function to be used at each layer in the
discriminator. They can be found in `torch.nn.modules.activation.*`. by default, it will use
'LeakyReLU'.
discriminator_last_activation (str, torch.nn.Module, None): last activation function to be used on the
output of the discriminator. By default, it is the sigmoid where 1 means that the discriminator thinks
that the data comes from the real data distribution while 0 means that it comes from the
use_batch_norm (bool): If batch normalization should be used.
"""
# check the generator length
if len(generator_units) < 2:
raise ValueError("Expecting more than the input layer for the generator.")
# check discriminator units.
if discriminator_units is None:
discriminator_units = generator_units[::-1] # reverse
# the last output unit should be 1 for the sigmoid
if discriminator_units[-1] != 1:
discriminator_units = discriminator_units + [1]
# if the number of units in the first layer of the discriminator is not the same as the number of units
# in the last layer of the generator, just append it
if discriminator_units[0] != generator_units[-1]:
discriminator_units = generator_units[-1:] + discriminator_units
if len(discriminator_units) == 0:
raise ValueError("Expecting at least the output layer for the discriminator.")
units = generator_units + discriminator_units[1:]
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
activations.update({act.lower(): act for act in activations})
def check_activation(activation):
if activation is None or activation.lower() == 'linear':
activation = None
elif activation in activations:
activation = getattr(torch.nn, activations[activation])
elif callable(activation):
pass
else:
raise ValueError("The given activation function is not available")
return activation
generator_activation = check_activation(generator_activation)
discriminator_activation = check_activation(discriminator_activation)
generator_last_activation = check_activation(generator_last_activation)
discriminator_last_activation = check_activation(discriminator_last_activation)
# build network
def build_layers(units, activation, last_activation):
layers = []
for i in range(len(units[:-2])):
# add linear layer
layer = torch.nn.Linear(units[i], units[i + 1])
layers.append(layer)
# if use batch normalization
if use_batch_norm:
layers.append(torch.nn.BatchNorm1d(num_features=units[i + 1]))
# add activation layer
if activation is not None:
layers.append(activation())
# last output layer
layers.append(torch.nn.Linear(units[-2], units[-1]))
if last_activation is not None:
layers.append(last_activation())
return layers
# create generator and discriminator layers
generator_layers = build_layers(generator_units, generator_activation, generator_last_activation)
discriminator_layers = build_layers(discriminator_units, discriminator_activation,
discriminator_last_activation)
# create generator and discriminator networks
generator = torch.nn.Sequential(*generator_layers)
discriminator = torch.nn.Sequential(*discriminator_layers)
super(MLP_GAN, self).__init__(generator, discriminator, input_shape=tuple([units[0]]),
output_shape=tuple([units[-1]]))
class DCGAN(GAN):
r"""Deep Convolutional Generative Adversarial Network.
References:
[1] "Unsupervised Representation Learning with Deep Convolutional Generative Adversarial Networks", Radford et
al., 2015
[2] "DCGAN Tutorial": https://pytorch.org/tutorials/beginner/dcgan_faces_tutorial.html
[3] "PyTorch Implementations of GANs", Linder-Noren, 2018
"""
def __init__(self, generator_units=[], discriminator_units=[],
generator_activation='LeakyReLU', generator_last_activation='tanh',
discriminator_activation='LeakyReLU', discriminator_last_activation='sigmoid',
use_batch_norm=True):
"""
Initialize the DCGAN.
Args:
generator_units (list/tuple of int): number of units per layer in the generator, including the data input
dimension.
discriminator_units (list/tuple of int): number of units per layer in the discriminator.
generator_activation (str, torch.nn.Module, None): activation function to be used at each layer in the
generator. They can be found in `torch.nn.modules.activation.*`. If None, it will use 'LeakyReLU'.
generator_last_activation (str, torch.nn.Module, None): last activation function to be used on the output
of the generator. If None, it will use 'tanh', so the output is between -1 and 1.
discriminator_activation (str, torch.nn.Module): activation function to be used at each layer in the
discriminator. They can be found in `torch.nn.modules.activation.*`. If None, it will use 'LeakyReLU'.
discriminator_last_activation (str, torch.nn.Module, None): last activation function to be used on the
output of the discriminator. By default, it is the sigmoid where 1 means that the discriminator thinks
that the data comes from the real data distribution while 0 means that it comes from the
use_batch_norm (bool): If batch normalization should be used.
"""
# check the generator length
if len(generator_units) < 2:
raise ValueError("Expecting more than the input layer for the generator.")
# check discriminator units.
if discriminator_units is None:
discriminator_units = generator_units[::-1] # reverse
# the last output unit should be 1 for the sigmoid
if discriminator_units[-1] != 1:
discriminator_units = discriminator_units + [1]
# if the number of units in the first layer of the discriminator is not the same as the number of units
# in the last layer of the generator, just append it
if discriminator_units[0] != generator_units[-1]:
discriminator_units = generator_units[-1:] + discriminator_units
if len(discriminator_units) == 0:
raise ValueError("Expecting at least the output layer for the discriminator.")
units = generator_units + discriminator_units[1:]
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
activations.update({act.lower(): act for act in activations})
def check_activation(activation):
if activation is None or activation.lower() == 'linear':
activation = None
elif activation in activations:
activation = getattr(torch.nn, activations[activation])
elif callable(activation):
pass
else:
raise ValueError("The given activation function is not available")
return activation
generator_activation = check_activation(generator_activation)
discriminator_activation = check_activation(discriminator_activation)
generator_last_activation = check_activation(generator_last_activation)
discriminator_last_activation = check_activation(discriminator_last_activation)
# keep track of (C, H, W) dimensions
if len(units[0]) == 2:
channel = 1
else:
channel = units[0][0]
# build network
def build_layers(units, activation, last_activation, generator=True):
layers = []
for i in range(1, len(units)):
# currently we only support convolution layers
if not isinstance(units[i], (tuple, list)):
raise ValueError("Expecting each unit to be a list or tuple of 2/3 ints for the `torch.nn.Conv*` "
"and `torch.nn.ConvTranspose*`, instead got: {}".format(units[i]))
# add (transposed) convolution layer based on if we have a generator or discriminator
unit = units[i]
if len(units[i]) == 2:
unit = (channel,) + units[i]
if generator: # generator
layer = torch.nn.Conv2d(*unit)
else: # discriminator
layer = torch.nn.ConvTranspose2d(*unit)
layers.append(layer)
# compute new dimensions
channel = layer.out_channels
# p, d, k, s = layer.padding, layer.dilation, layer.kernel_size, layer.stride
# height = int(math.floor((height + 2. * p[0] - d[0] * (k[0] - 1) - 1.) / s[0] + 1))
# width = int(math.floor((width + 2. * p[1] - d[1] * (k[1] - 1) - 1.) / s[1] + 1))
# if use batch normalization
if use_batch_norm:
layers.append(torch.nn.BatchNorm2d(num_features=unit[1]))
# add activation layer
if activation is not None:
layers.append(activation())
# last output layer
if last_activation is not None:
layers.append(last_activation())
return layers
# create generator and discriminator layers
generator_layers = build_layers(generator_units, generator_activation, generator_last_activation)
discriminator_layers = build_layers(discriminator_units[:-1], discriminator_activation,
discriminator_last_activation, generator=False)
# create generator and discriminator networks
generator = torch.nn.Sequential(*generator_layers)
discriminator = torch.nn.Sequential(*discriminator_layers)
super(DCGAN, self).__init__(generator, discriminator, input_shape=units[0][:3], output_shape=(1,))
# Tests
if __name__ == '__main__':
# create GAN
gan = MLP_GAN(generator_units=[100, 128, 28*28], discriminator_units=[128, 1])
print(gan)
# generate data
z = gan.generate_latent()
x = gan.generate(z)
y = gan.discriminate(x)
print("Latent vector shape: {}".format(z.shape))
print("Generator output shape: {}".format(x.shape))
print("Discriminator output: {}".format(y))
dcgan = DCGAN(generator_units=[(100, 8*64, 4, 1, 0), (8*64, 8*4, 4, 2, 1), (4*64, 2*64, 4, 2, 1),
(2*64, 64, 4, 2, 1), (64, 3, 4, 2, 1)],
discriminator_units=[(3, 64, 4, 2, 1), (64, 2*64, 4, 2, 1), (2*64, 4*64, 4, 2, 1),
(4*64, 8*64, 4, 2, 1), (8*64, 1, 4, 1, 0)])
print(dcgan)
pyrobolearn/models/nn/mlp.py
+30 -80
View File
@@ -17,12 +17,9 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN, NNTorch
from pyrobolearn.models.nn.dnn import NN
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
@@ -51,80 +48,24 @@ class MLP(NN):
The parameters of the neural networks are all the weight matrices and bias vectors.
"""
def __init__(self, num_units=(), activation_fct='Linear', last_activation_fct=None, dropout_prob=None,
framework='pytorch'):
def __init__(self, units=(), activation=None, last_activation=None, dropout=None):
"""
Initialize a MLP network.
Args:
num_units (list/tuple of int): number of units in each layer (this includes the input and output layer)
activation_fct (None, str, or list/tuple of str/None): activation function to be applied after each layer.
units (list/tuple of int): number of units in each layer (this includes the input and output layer)
activation (None, str, or list/tuple of str/None): activation function to be applied after each layer.
If list/tuple, then it has to match the number of
hidden layers.
last_activation_fct (None or str): last activation function to be applied. If not specified, it will check
hidden layers. If None, it is a linear layer.
last_activation (None or str): last activation function to be applied. If not specified, it will check
if it is in the list/tuple of activation functions provided for the
previous argument.
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
framework (str): specifies which framework we want to use between 'pytorch' and 'keras' (default: 'pytorch')
"""
# check framework
framework = framework.lower()
if framework == 'pytorch':
model = MLPTorch(num_units, activation_fct, last_activation_fct, dropout_prob)
elif framework == 'keras':
model = MLPKeras(num_units, activation_fct, last_activation_fct, dropout_prob)
else:
raise ValueError("The current frameworks allowed are pytorch and keras")
# instantiate the super class
super(MLP, self).__init__(model.model, input_shape=tuple([num_units[0]]), output_shape=tuple([num_units[-1]]),
framework=framework)
# rewrite methods
self.save = model.save
self.load = model.load
self.__str__ = model.__str__
class MLPTorch(NNTorch):
r"""Multi-Layer Perceptron in PyTorch
Feed-forward and fully-connected neural network, where linear layers are followed by non-linear activation
functions.
.. math::
h_{l} = f_{l}(W_{l} h_{l-1} + b_{l})
where :math:`l \in [1,...,L]` with :math:`L` is the total number of layers,
:math:`W_{l}` and :math:`b_{l}` are the weight matrix and bias vector at layer :math:`l`, :math:`f_{l}` is
the nonlinear activation function, and :math:`h_{0} = x` and :math:`y = h_{L}` are the input and output vectors.
The parameters of the neural networks are all the weight matrices and bias vectors.
"""
def __init__(self, num_units=(), activation_fct='Linear', last_activation_fct=None, dropout_prob=None):
"""
Initialize a MLP network.
Args:
num_units (list/tuple of int): number of units in each layer (this includes the input and output layer)
activation_fct (None, str, or list/tuple of str/None): activation function to be applied after each layer.
If list/tuple, then it has to match the
last_activation_fct (None or str): last activation function to be applied. If not specified, it will check
if it is in the list/tuple of activation functions provided for the
previous argument.
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
dropout (None, float, or list/tuple of float/None): dropout probability.
"""
# check number of units
if len(num_units) < 2:
if len(units) < 2:
raise ValueError("The num_units list/tuple needs to have at least the input and output layers")
# set the dimensions of the input and output
self.input_dims = num_units[0]
self.output_dims = num_units[-1]
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
@@ -139,43 +80,52 @@ class MLPTorch(NNTorch):
activation = getattr(torch.nn, activations[activation])
return activation
activation_fct = check_activation(activation_fct)
last_activation_fct = check_activation(last_activation_fct)
activation = check_activation(activation)
last_activation = check_activation(last_activation)
# check dropout
dropout_layer = None
if dropout_prob is not None:
dropout_layer = torch.nn.Dropout(dropout_prob)
if dropout is not None:
dropout_layer = torch.nn.Dropout(dropout)
# build pytorch network
layers = []
for i in range(len(num_units[:-2])):
for i in range(len(units[:-2])):
# add linear layer
layer = torch.nn.Linear(num_units[i], num_units[i + 1])
layer = torch.nn.Linear(units[i], units[i + 1])
layers.append(layer)
# add activation layer
if activation_fct is not None:
layers.append(activation_fct())
if activation is not None:
layers.append(activation())
# add dropout layer
if dropout_layer is not None:
layers.append(dropout_layer)
# last output layer
layers.append(torch.nn.Linear(num_units[-2], num_units[-1]))
if last_activation_fct is not None:
layers.append(last_activation_fct)
layers.append(torch.nn.Linear(units[-2], units[-1]))
if last_activation is not None:
layers.append(last_activation)
# create nn model
model = torch.nn.Sequential(*layers)
super(MLPTorch, self).__init__(model, input_shape=num_units[0], output_shape=num_units[-1])
super(MLP, self).__init__(model, input_shape=tuple([units[0]]), output_shape=tuple([units[-1]]))
# rewrite methods
# self.save = model.save
# self.load = model.load
# self.__str__ = model.__str__
# Tests
if __name__ == '__main__':
# create MLP network
mlp = MLPTorch(num_units=(2, 10, 3), activation_fct='relu')
mlp = MLP(units=(2, 10, 3), activation='relu')
print(mlp)
x = torch.rand(2)
y = mlp(x)
print("Input: {} - Output: {}".format(x, y))
pyrobolearn/models/nn/rcnn.py
+8 -8
View File
@@ -19,9 +19,6 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN
@@ -37,11 +34,14 @@ __status__ = "Development"
class RCNN(NN):
r"""Recurrent CNN
"""
pass
r"""Recurrent Convolutional Neural Network
class RCNN(NNTorch):
r"""Recurrent CNN in PyTorch
The Recurrent Convolutional Neural Network (RCNN) is a neural network model used for data that have features that
have a temporal and spatial relationship between them. This includes for instance the processing of videos.
References:
[1] "Recurrent Convolutional Neural Networks for Scene Labeling", Pinheiro et al., 2014
[2] "Recurrent Convolutional Neural Networks for Text Classification", Lai et al., 2015
[3] "Recurrent Convolutional Neural Networks for Object Recognition", Liang et al., 2015
"""
pass
pyrobolearn/models/nn/rnn.py
+106 -7
View File
@@ -18,9 +18,6 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN
@@ -37,10 +34,112 @@ __status__ = "Development"
class RNN(NN):
r"""Recurrent Neural Network
"""
pass
class RNNTorch(NNTorch):
r"""Recurrent Neural Network in PyTorch
A recurrent neural network is a network that captures the sequential nature / aspect of the data. It possesses
an internal memory (i.e. internal state) which is returned at the input.
References:
[1] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
"""
def __init__(self, model, input_shape, output_shape):
super(RNN, self).__init__(model, input_shape, output_shape)
##################
# Static methods #
##################
@staticmethod
def is_recurrent(): # unless RNN
"""RNNs are recurrent models."""
return True
###########
# Methods #
###########
def forward(self, inputs):
pass
class MLP_RNN(RNN):
r"""Recurrent Multi-layer Perceptron using Elman RNNs.
This uses `torch.nn.RNN` and `torch.nn.RNNCell`.
References:
[1] "Finding structure in time.", Elman, 1990
[2] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
"""
def __init__(self, units=(), activation=None, last_activation=None, dropout=None):
"""
Initialize a recurrent MLP network.
Args:
units (list/tuple of int): number of units in each layer (this includes the input and output layer)
activation (None, str, or list/tuple of str/None): activation function to be applied after each layer.
If list/tuple, then it has to match the number of
hidden layers. If None, it is a linear layer.
last_activation (None or str): last activation function to be applied. If not specified, it will check
if it is in the list/tuple of activation functions provided for the
previous argument.
dropout (None, float, or list/tuple of float/None): dropout probability.
"""
pass
# super(MLP_RNN, self).__init__(model, input_shape=tuple([units[0]]), output_shape=tuple([units[-1]]))
class MLP_LSTM(RNN):
r"""Recurrent Multi-Layer Perceptron using Long-Short Term Memories (LSTMs).
This uses `torch.nn.LSTM` and `torch.nn.LSTMCell`.
References:
[1] "Long Short Term Memory", Hochreiter et al., 1997
[2] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
"""
def __init__(self, units=(), activation=None, last_activation=None, dropout=None):
pass
# super(MLP_LSTM, self).__init__(model, input_shape, output_shape)
class MLP_GRU(RNN):
r"""Recurrent Multi-Layer Perceptron using Gated Recurrent Units (GRUs).
This uses `torch.nn.GRU` and `torch.nn.GRUCell`.
References:
[1] "Learning Phrase Representations using RNN Encoder-Decoder for Statistical Machine Translation", Cho et al.,
2014
[2] "Empirical Evaluation of Gated Recurrent Neural Networks on Sequence Modeling", Chung et al., 2014
[3] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
"""
def __init__(self, model, input_shape, output_shape):
pass
# super(MLP_GRU, self).__init__(model, input_shape, output_shape)
# Tests
if __name__ == '__main__':
pass
# # create Recurrent MLP network
# mlp_rnn = MLP_RNN(units=(2, 10, 3), activation='relu')
# mlp_lstm = MLP_LSTM(num_units=(2, 10, 3), activation_fct='relu')
# mlp_gru = MLP_GRU(num_units=(2,10,3), activation_fct='relu')
#
# print("RNN: {}".format(mlp_rnn))
# print("LSTM: {}".format(mlp_lstm))
# print("GRU: {}".format(mlp_gru))
#
# x = torch.rand(2)
#
# for t in range(3):
# y = mlp_rnn.forward(x)
# print("RNN: Input at t{}: {} - Output: {}".format(t, x, y))
# y = mlp_lstm.forward(x)
# print("LSTM: Input at t{}: {} - Output: {}".format(t, x, y))
# y = mlp_gru.forward(x)
# print("GRU: Input at t{}: {} - Output: {}".format(t, x, y))
pyrobolearn/models/nn/vae.py
+212 -17
View File
@@ -18,12 +18,11 @@ References:
[2] PyTorch: https://pytorch.org/
"""
import copy
import inspect
import numpy as np
import torch
from pyrobolearn.models.nn.dnn import NN
from pyrobolearn.distributions.modules import MeanModule, DiagonalCovarianceModule, GaussianModule
from pyrobolearn.models.nn.ae import AE
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
@@ -35,20 +34,68 @@ __email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class VAE(NN):
class VAE(AE):
r"""Variational AutoEncoder
Type: generative model
Variational Auto-encoders (VAEs) are generative latent models that projects the input data to a latent space,
where a probability distribution is defined over it, and latent vector which are sampled are projected back to
the input space. This allows later to generate
The loss being minimized with VAEs is the reconstruction/generation loss and the latent loss which measures how
far the latent distribution produced by the encoder is from a predefined distribution. This predefined latent
distribution is often selected to be a unit Gaussian.
.. seealso:: Generative Adversarial Networks
References:
[1] "Deep Learning" (http://www.deeplearningbook.org/), Goodfellow et al., 2016
[2] "Tutorial on Variational Autoencoder"
[2] "Tutorial on Variational Autoencoders", Doersch, 2016
[3] "Variational Autoencoders Explained" (http://kvfrans.com/variational-autoencoders-explained/), Frans, 2016
"""
def __init__(self, layer_sizes, activation_fct=None, dropout=None):
self.encoder = None
self.decoder = None
def __init__(self, encoder, decoder, latent_distribution, predefined_latent_distribution, input_shape,
output_shape):
"""
Initialize the Variational Auto-Encoder.
Args:
encoder (torch.nn.Module): encoder module.
decoder (torch.nn.Module): decoder module.
latent_distribution (torch.nn.Module): latent distribution module. Given the encoder's output,
the :attr:`latent_distribution` module outputs a `torch.distribution.Distribution`.
See `pyrobolearn/distributions/modules.py` for more choices.
predefined_latent_distribution (torch.distributions.Distribution): Predefined latent distribution to which
the predicted latent distribution should be close to.
input_shape (tuple of int): input shape.
output_shape (tuple of int): output shape.
"""
super(VAE, self).__init__(encoder, decoder, input_shape, output_shape)
# check latent distribution module
if not isinstance(latent_distribution, torch.nn.Module):
raise TypeError("Expecting the given latent distribution to be an instance of `torch.nn.Module`, instead "
"got: {}".format(latent_distribution))
# check predefined latent distribution
if not isinstance(predefined_latent_distribution, torch.distributions.Distribution):
raise TypeError("Expecting the predefined latent distribution to be an instance of "
"`torch.distributions.Distribution`, instead got: "
"{}".format(predefined_latent_distribution))
# check that the distribution returned by the latent distribution module is the same as the predefined latent
# distribution
x = torch.rand(input_shape).unsqueeze(0)
x = self.encode(x)
distribution = latent_distribution(x)
if not isinstance(distribution, predefined_latent_distribution.__class__):
raise ValueError("The distribution returned by the latent distribution module (i.e. {}) is not an instance "
"of the predefined latent distribution "
"(i.e. {})".format(type(distribution), type(predefined_latent_distribution)))
self.latent_distribution = latent_distribution
self.predefined_distribution = predefined_latent_distribution
##################
# Static methods #
@@ -61,20 +108,168 @@ class VAE(NN):
@staticmethod
def is_generative(): # unless VAE, GAN,...
"""Standard neural networks are not generative, and thus we can not sample from it. This is different,
for instance, for generative adversarial networks (GANs) and variational auto-encoders (VAEs)."""
"""VAEs are generative probabilistic models that learn a latent space."""
return True
###########
# Methods #
###########
def sample(self, size=None, seed=None):
"""Sample from the VAE"""
pass
def forward(self, x):
"""Forward the input to the encoder and decoder."""
# go through the encoder, sample from the latent distribution, and decode the samples
shape = (len(x),) if len(x.shape) > 1 else (1,)
x = self.encoder(x)
x = self.latent_distribution(x).rsample(sample_shape=shape).squeeze(0)
x = self.decoder(x)
return x
def encode(self, x):
"""Run the encoder."""
return self.encoder(x)
def decode(self, x):
"""Run the decoder."""
return self.decoder(x)
def sample_latent(self, shape=(), seed=None):
"""Sample latent vectors."""
if seed is not None:
torch.manual_seed(seed)
return self.predefined_distribution.rsample(sample_shape=shape)
def sample(self, shape=(), seed=None):
"""Sample from the VAE."""
if seed is not None:
torch.manual_seed(seed)
x = self.predefined_distribution.rsample(sample_shape=shape)
return self.decoder(x)
class VAETorch(NNTorch):
r"""Variational AutoEncoder in PyTorch
class MLP_VAE(VAE):
r"""Multi-Layer Perceptron Variational Auto-Encoder
"""
pass
def __init__(self, encoder_units=[], decoder_units=None, activation=None, last_activation=None,
dropout=None, latent_distribution=None, predefined_latent_distribution=None):
"""
Initialize the MLP VAE.
Args:
encoder_units (list/tuple of int): number of units for each layer in the encoder (this includes the input
layer)
decoder_units (list/tuple of int, None): number of units for each layer in the decoder (this includes the
output layer). If None, it will take the encoder units but in the reverse order.
activation (None, str, or list/tuple of str/None): activation function to be applied after each layer.
If list/tuple, then it has to match the number of
hidden layers.
last_activation (None or str): last activation function to be applied. If not specified, it will check
if it is in the list/tuple of activation functions provided for the
previous argument.
latent_distribution (torch.nn.Module, None): latent distribution module. Given the encoder's output,
the :attr:`latent_distribution` module outputs a `torch.distribution.Distribution`. If None, it will
create a Gaussian module which outputs a Gaussian distribution based on a learned mean and diagonal
covariance. See `pyrobolearn/distributions/modules.py` for more choices.
predefined_latent_distribution (torch.distributions.Distribution, None): Predefined latent distribution to
which the predicted latent distribution should be close to.
dropout (None, float, or list/tuple of float/None): dropout probability.
"""
# check the encoder length
if len(encoder_units) < 2:
raise ValueError("Expecting more than the input layer for the encoder.")
# check decoder units.
if decoder_units is None:
decoder_units = encoder_units[::-1] # reverse
else:
decoder_units = encoder_units[-1:] + decoder_units
if len(decoder_units) == 1:
raise ValueError("Expecting at least the output layer for the decoder.")
num_units = encoder_units + decoder_units[1:]
# check for activation fcts
activations = dir(torch.nn.modules.activation)
activations = {act: act for act in activations}
activations.update({act.lower(): act for act in activations})
def check_activation(activation):
if activation is None or activation.lower() == 'linear':
activation = None
else:
if activation not in activations:
raise ValueError("The given activation function is not available")
activation = getattr(torch.nn, activations[activation])
return activation
activation = check_activation(activation)
last_activation = check_activation(last_activation)
# check dropout
dropout_layer = None
if dropout is not None:
dropout_layer = torch.nn.Dropout(dropout)
# build pytorch network
def build_layers(units, encoder=True):
layers = []
size = len(units[:-1]) if encoder else len(units[:-2])
for i in range(size):
# add linear layer
layer = torch.nn.Linear(units[i], units[i + 1])
layers.append(layer)
# add activation layer
if activation is not None:
layers.append(activation())
# add dropout layer
if dropout_layer is not None:
layers.append(dropout_layer)
# last output layer if decoder
if not encoder:
layers.append(torch.nn.Linear(units[-2], units[-1]))
if last_activation is not None:
layers.append(last_activation)
return layers
encoding_layers = build_layers(encoder_units, encoder=True)
decoding_layers = build_layers(decoder_units, encoder=False)
# Encoder
encoder = torch.nn.Sequential(*encoding_layers)
# Decoder
decoder = torch.nn.Sequential(*decoding_layers)
# latent distribution module
if latent_distribution is None:
size = encoder_units[-1]
mean = MeanModule(num_inputs=size, num_outputs=size)
covariance = DiagonalCovarianceModule(num_inputs=size, num_outputs=size)
latent_distribution = GaussianModule(mean=mean, covariance=covariance)
# predefined latent distribution
if predefined_latent_distribution is None:
size = encoder_units[-1]
mean = torch.zeros(size)
covariance = torch.diag(torch.ones(size))
predefined_latent_distribution = torch.distributions.MultivariateNormal(loc=mean,
covariance_matrix=covariance)
super(MLP_VAE, self).__init__(encoder, decoder, latent_distribution=latent_distribution,
predefined_latent_distribution=predefined_latent_distribution,
input_shape=tuple([num_units[0]]),
output_shape=tuple([num_units[-1]]))
# Tests
if __name__ == '__main__':
# create MLP network
vae = MLP_VAE(encoder_units=(4, 3, 2), activation='relu')
print(vae)
x = torch.rand(4).unsqueeze(0)
y = vae.forward(x)
print("Input: {} - Output: {}".format(x, y))