From b3e92b1e7f0172761b68cd84a0ad72501a19c5b4 Mon Sep 17 00:00:00 2001 From: Brian Delhaisse Date: Tue, 9 Apr 2019 04:02:51 +0200 Subject: [PATCH] update NN models (DNN, MLP, CNN, AE, VAE, GAN) --- pyrobolearn/models/basics/linear.py | 17 ++ pyrobolearn/models/model.py | 17 ++ pyrobolearn/models/nn/__init__.py | 10 +- pyrobolearn/models/nn/ae.py | 192 ++++++++++---- pyrobolearn/models/nn/cnn.py | 257 ++++++++++++++++++- pyrobolearn/models/nn/dnn.py | 123 +++++---- pyrobolearn/models/nn/gan.py | 381 +++++++++++++++++++++++++++- pyrobolearn/models/nn/mlp.py | 110 +++----- pyrobolearn/models/nn/rcnn.py | 16 +- pyrobolearn/models/nn/rnn.py | 113 ++++++++- pyrobolearn/models/nn/vae.py | 229 +++++++++++++++-- 11 files changed, 1227 insertions(+), 238 deletions(-) diff --git a/pyrobolearn/models/basics/linear.py b/pyrobolearn/models/basics/linear.py index ab45dfd..a9d7f77 100644 --- a/pyrobolearn/models/basics/linear.py +++ b/pyrobolearn/models/basics/linear.py @@ -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() diff --git a/pyrobolearn/models/model.py b/pyrobolearn/models/model.py index 53fc90a..fe70669 100644 --- a/pyrobolearn/models/model.py +++ b/pyrobolearn/models/model.py @@ -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.""" diff --git a/pyrobolearn/models/nn/__init__.py b/pyrobolearn/models/nn/__init__.py index 7e113f9..1e8f85e 100644 --- a/pyrobolearn/models/nn/__init__.py +++ b/pyrobolearn/models/nn/__init__.py @@ -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 * diff --git a/pyrobolearn/models/nn/ae.py b/pyrobolearn/models/nn/ae.py index 08ddd74..bed9113 100644 --- a/pyrobolearn/models/nn/ae.py +++ b/pyrobolearn/models/nn/ae.py @@ -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)) diff --git a/pyrobolearn/models/nn/cnn.py b/pyrobolearn/models/nn/cnn.py index d12f0f3..714fa81 100644 --- a/pyrobolearn/models/nn/cnn.py +++ b/pyrobolearn/models/nn/cnn.py @@ -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)) diff --git a/pyrobolearn/models/nn/dnn.py b/pyrobolearn/models/nn/dnn.py index bb6b993..1d91461 100644 --- a/pyrobolearn/models/nn/dnn.py +++ b/pyrobolearn/models/nn/dnn.py @@ -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) diff --git a/pyrobolearn/models/nn/gan.py b/pyrobolearn/models/nn/gan.py index 8cec17d..006cde2 100644 --- a/pyrobolearn/models/nn/gan.py +++ b/pyrobolearn/models/nn/gan.py @@ -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) diff --git a/pyrobolearn/models/nn/mlp.py b/pyrobolearn/models/nn/mlp.py index d0b40d8..b40845d 100644 --- a/pyrobolearn/models/nn/mlp.py +++ b/pyrobolearn/models/nn/mlp.py @@ -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)) diff --git a/pyrobolearn/models/nn/rcnn.py b/pyrobolearn/models/nn/rcnn.py index 4f0393b..07fb243 100644 --- a/pyrobolearn/models/nn/rcnn.py +++ b/pyrobolearn/models/nn/rcnn.py @@ -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 diff --git a/pyrobolearn/models/nn/rnn.py b/pyrobolearn/models/nn/rnn.py index 1fa3972..75c29c1 100644 --- a/pyrobolearn/models/nn/rnn.py +++ b/pyrobolearn/models/nn/rnn.py @@ -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)) diff --git a/pyrobolearn/models/nn/vae.py b/pyrobolearn/models/nn/vae.py index da7ede0..8fba9b5 100644 --- a/pyrobolearn/models/nn/vae.py +++ b/pyrobolearn/models/nn/vae.py @@ -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))