3_convolutional.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convolutional Neural Networks\n",
    "\n",
    "Convolutional Neural Networks (CNN) are a class of networks, that are typically employed in the field of neural image processing.\n",
    "Since fully connected layers for a large image would be prohibitively expensive, convolutions are used to vastly reduce the amount of parameters needed to process an image.\n",
    "\n",
    "Instead of connecting every output to every input with individual weights, we reduce the number of parameters by sharing the weights and applying the same ones to different parts of the input image. This reduces the number of weights to be learned in a single layer and the number of layers by taking into account the nature of the input data (2D pixel map).\n",
    "\n",
    "We use the Sequential model with Dense layers from before and add $\\texttt{Conv2D}$, $\\texttt{MaxPooling2D}$, $\\texttt{Dropout}$ and $\\texttt{Flatten}$ layers.\n",
    "\n",
    "\n",
    "As an example task we classify the images in the MNIST dataset, a set of images of handwritten digits, with respect to which digit an image supposedly shows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import h5py\n",
    "import numpy as np\n",
    "import scipy.signal as scis\n",
    "import tensorflow as tf\n",
    "import matplotlib.pyplot as plt\n",
    "plt.style.use('seaborn')\n",
    "\n",
    "SEED = 42\n",
    "np.random.seed(SEED)\n",
    "tf.set_random_seed(SEED)\n",
    "from keras.callbacks import Callback\n",
    "from keras.models import Sequential\n",
    "from keras.layers import Conv2D, Dense, Dropout, Flatten, MaxPooling2D\n",
    "from keras.utils import to_categorical"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_data(path='mnist.h5'):\n",
    "    \"\"\"\n",
    "    Loads a dataset and its supervisor labels from the HDF5 file specified by the path.\n",
    "    It is assumed that the HDF5 dataset containing the data is called 'data' and the labels are called 'labels'.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    path : str, optional\n",
    "        The absolute or relative path to the HDF5 file, defaults to mnist.h5.\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    data_and_labels : tuple(np.array[samples, width, height], np.array[samples])\n",
    "        a tuple with two numpy array containing the data and labels\n",
    "    \"\"\"\n",
    "    with h5py.File(path, 'r') as handle:\n",
    "        return np.array(handle['data']), np.array(handle['labels'])\n",
    "    \n",
    "data, labels = load_data()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A single discrete convolution is not different from a filter in traditional image processing. They allow you for example to blur, sharpen or detect edges within an image. \n",
    "\n",
    "A filter is computed by extracting every possible sub-image, e.g $3\\times 3$ pixels, of the entire image and computing the weighted sum of the sub-image and filter. This will produce a single corresponding output pixel. Look at the following example filters and try to understand what they do."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "gauss_filter = scis.convolve2d(data[0], np.array([\n",
    "    [1, 2, 1],\n",
    "    [2, 4, 2],\n",
    "    [1, 2, 1]\n",
    "]), mode='same')\n",
    "\n",
    "laplacian_filter = scis.convolve2d(data[0], np.array([\n",
    "    [0,  1, 0],\n",
    "    [1, -4, 1],\n",
    "    [0,  1, 0]\n",
    "]), mode='same')\n",
    "\n",
    "high_pass_filter = scis.convolve2d(data[0], np.array([\n",
    "    [-1, -1, -1],\n",
    "    [-1,  9, -1],\n",
    "    [-1, -1, -1]\n",
    "]), mode='same')\n",
    "\n",
    "sharpen = scis.convolve2d(data[0], np.array([\n",
    "    [-0.5, -0.5, -0.5],\n",
    "    [-0.5,  8.5, -0.5],\n",
    "    [-0.5, -0.5, -0.5]\n",
    "]), mode='same')\n",
    "\n",
    "sobel_x = scis.convolve2d(data[0], np.array([\n",
    "    [1, 0, -1],\n",
    "    [2, 0, -2],\n",
    "    [1, 0, -1]\n",
    "]), mode='same')\n",
    "\n",
    "sobel_y = scis.convolve2d(data[0], np.array([\n",
    "    [ 1,  2,  1],\n",
    "    [ 0,  0,  0],\n",
    "    [-1, -2, -1,]\n",
    "]), mode='same')\n",
    "\n",
    "sobel = np.sqrt(sobel_x ** 2 + sobel_y ** 2)\n",
    "\n",
    "emboss_filter = scis.convolve2d(data[0], np.array([\n",
    "    [-1, -1,  0],\n",
    "    [-1,  0,  1],\n",
    "    [ 0,  1,  1,]\n",
    "]), mode='same')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filters also exist in larger version, with increased kernel-size, usually enhancing the 'intensity' of the particular effect."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "gauss_filter_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [1,  4,  7,  2,  1],\n",
    "    [4, 16, 26, 16,  4],\n",
    "    [7, 26, 41, 26,  7],\n",
    "    [4, 16, 26, 16,  4],\n",
    "    [1,  4,  7,  4,  1],\n",
    "]), mode='same')\n",
    "\n",
    "laplacian_filter_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [0, 0,   1, 0, 0],\n",
    "    [0, 1,   2, 1, 0],\n",
    "    [1, 2, -16, 2, 1],\n",
    "    [0, 1,   2, 1, 0],\n",
    "    [0, 0,   1, 0, 0]\n",
    "]), mode='same')\n",
    "\n",
    "high_pass_filter_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [ 0, -1, -1, -1,  0],\n",
    "    [-1,  2, -4,  2, -1],\n",
    "    [-1, -4, 13, -4, -1],\n",
    "    [-1,  2, -4,  2, -1],\n",
    "    [ 0, -1, -1, -1,  0]\n",
    "]), mode='same')\n",
    "\n",
    "sharpen_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [-0.5, -0.5,  -0.5, -0.5, -0.5],\n",
    "    [-0.5, -0.5,  -0.5, -0.5, -0.5],\n",
    "    [-0.5, -0.5,  24.5, -0.5, -0.5],\n",
    "    [-0.5, -0.5,  -0.5, -0.5, -0.5],\n",
    "    [-0.5, -0.5,  -0.5, -0.5, -0.5],\n",
    "]), mode='same')\n",
    "\n",
    "sobel_x_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [2, 1, 0, -1, -2],\n",
    "    [2, 1, 0, -1, -2],\n",
    "    [4, 2, 0, -2, -4],\n",
    "    [2, 1, 0, -1, -2],\n",
    "    [2, 1, 0, -1, -2]\n",
    "]), mode='same')\n",
    "\n",
    "sobel_y_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [ 1,  1,  4,  1,  1],\n",
    "    [ 1,  1,  2,  1,  1],\n",
    "    [ 0,  0,  0,  0,  0],\n",
    "    [-1, -1, -2, -1, -1],\n",
    "    [-1, -1, -4, -1, -1],\n",
    "]), mode='same')\n",
    "\n",
    "sobel_5x5 = np.sqrt(sobel_x_5x5 ** 2 + sobel_y_5x5 ** 2)\n",
    "\n",
    "emboss_filter_5x5 = scis.convolve2d(data[0], np.array([\n",
    "    [-1, -1, -1, -1, 0],\n",
    "    [-1, -1, -1,  0, 1],\n",
    "    [-1, -1,  0,  1, 1],\n",
    "    [-1,  0,  1,  1, 1],\n",
    "    [ 0,  1,  1,  1, 1],\n",
    "]), mode='same')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def show_examples(data):\n",
    "    \"\"\"\n",
    "    Plots example images.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    data : np.array[grid_width, grid_height, image_width, image_height]\n",
    "        the image dataset\n",
    "    \"\"\"\n",
    "    height = data.shape[0]\n",
    "    width = data.shape[1]\n",
    "    figure, axes = plt.subplots(height, width, figsize=(16, 4), sharex=True, sharey=True)\n",
    "    \n",
    "    for h in range(height):\n",
    "        for w in range(width):\n",
    "            axis = axes[h][w]\n",
    "            axis.grid(False)\n",
    "            axis.imshow(data[h, w, :, :], cmap='gist_gray')\n",
    "    plt.show()\n",
    "\n",
    "    \n",
    "filtered_images = np.array([\n",
    "    [data[0], gauss_filter,     laplacian_filter,     high_pass_filter,     sharpen,     sobel,     emboss_filter],\n",
    "    [data[0], gauss_filter_5x5, laplacian_filter_5x5, high_pass_filter_5x5, sharpen_5x5, sobel_5x5, emboss_filter_5x5]\n",
    "])\n",
    "show_examples(filtered_images)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Digit Recognition\n",
    "\n",
    "To determine which of these filters is beneficial for a classification task and which particular weights to use is a manual and very labor intensive task. Therefore, the idea of a CNN is to determines these weights automatically. Analogous to the fully-connected neural network, this is done through optimization of a loss function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess_data(data, labels):\n",
    "    \"\"\"\n",
    "    Flattens the each image in the data to be a one-dimensional feature vector and encodes the labels in one-hot encoding.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    data : np.array[samples, width, height]\n",
    "        the image dataset\n",
    "    labels : np.array[samples]\n",
    "        the corresponding labels\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    data_and_labels : tuple(np.array[samples, width * height], np.array[samples, classes])\n",
    "        a tuple with two numpy array containing the flattened data and one-hot encoded labels\n",
    "    \"\"\"\n",
    "    ##############\n",
    "    return the flattened images and labels\n",
    "##############\n",
    "\n",
    "preprocessed_data, preprocessed_labels = preprocess_data(data, labels)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In practice, several convolutional layers with multiple filters each, are applied directly to the image. Hence, a convolutional layers produces multiple image as output, where the number of filters in the layers corresponds to the number of produced images, also often refered to as \"color\" channels. Convolutions by themselves will slightly reduce the size of the image, as you require a number of \"frame\" pixels around the convolution (half the convolution filter size, floored). To counteract this effect, an image may be padded with additional pixels, e.g. a constant value like 0 or 1, or by repeating the border pixels (padding='same').\n",
    "\n",
    "Another way is to use pooling layers (e.g. $\\texttt{MaxPooling2D}$) to deliberately combine several outputs of a previous layer to a single input for the next one. In case of a MaxPooling layer with a kernel size of $2\\times 2$ pixel and a stride of $2\\times 2$ for example, all non-overlapping $2\\times 2$ pixel sub-images are reduced to their contained maximum value. The amount of pixels in the output is reduced to a quarter.\n",
    "\n",
    "Any activation function that is aplicable for a dense layer works for also for a convolutional one. In recent year, however, the community has introduced a number of new activation functions that are a) faster to compute and b) do not suffer as heavily from the vanishing gradient problem, like the sigmoid function does. One of the is the *Recitified Linear Unit (ReLU)*, defined as: $$f(x) = max(x, 0)$$\n",
    "\n",
    "For classification tasks on images, like digit recognition, a final dense layer is still needed to do the actual classification of the previously filtered image, similar to fully-connected neural network.\n",
    "\n",
    "Build an image classification model, that takes the correct input shape from the data and passes it through some convolutional, pooling and finally a dense layer and outputs the probability of the image belonging to each class.\n",
    "\n",
    "Train and test the model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def build_model(data, classes):\n",
    "    \"\"\"\n",
    "    Constructs a convolutional neural network model for the given data and number of classes\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    data : np.array[samples, width * height]\n",
    "        the image dataset\n",
    "    classes : int\n",
    "        the number of unique classes in the dataset\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    model : keras.Model\n",
    "        the fully-connected neural network\n",
    "    \"\"\"\n",
    "    model = Sequential()\n",
    "    \n",
    "    ##############\n",
    "    add all the layers\n",
    "    compile the model with sgd optimizer, categorical crossentropy loss and accuracy printing\n",
    "    ##############\n",
    "    \n",
    "    return model\n",
    "\n",
    "model = build_model(preprocessed_data, classes=preprocessed_labels.shape[1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TrainingHistory(Callback):\n",
    "    \"\"\"\n",
    "    Class for tracking the training progress/history of the neural network. Implements the keras.Callback interface.\n",
    "    \"\"\"\n",
    "    def on_train_begin(self, logs):\n",
    "        self.loss = []\n",
    "        self.acc = []\n",
    "        self.validation_loss = []\n",
    "        self.validation_acc = []\n",
    "        \n",
    "    def on_batch_end(self, _, logs):\n",
    "        \"\"\"\n",
    "        Callback invoked after each training batch.\n",
    "        Should track the training loss and accuracy in the respective members.\n",
    "\n",
    "        Parameters\n",
    "        ----------\n",
    "        _ : int\n",
    "            unused, int corresponding to the batch number\n",
    "        logs : dict{str -> float}\n",
    "            a dictionary mapping from the observed quantity to the actual valu\n",
    "        \"\"\"\n",
    "        if 'loss' in logs:\n",
    "            self.loss.append(logs['loss'])\n",
    "        if 'acc' in logs:\n",
    "            self.acc.append(logs['acc'])\n",
    "            \n",
    "    def on_epoch_end(self, _, logs):\n",
    "        if 'val_loss' in logs:\n",
    "            self.validation_loss.append(logs['val_loss'])\n",
    "        if 'val_acc' in logs:\n",
    "            self.validation_acc.append(logs['val_acc'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def train_model(model, data, labels, epochs=25, batch_size=64, train_fraction=0.8):\n",
    "    \"\"\"\n",
    "    Trains a convolutional neural network given the data and labels.\n",
    "    This time we employ the automatic train and test set functionality of Keras.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    model : keras.Model\n",
    "        the fully-connected neural network\n",
    "    data : np.array[samples, width * height]\n",
    "        the entire data\n",
    "    labels : np.array[samples, classes]\n",
    "        the one-hot encoded training labels\n",
    "    epoch: positive int, optional\n",
    "        the number of epochs for which the neural network is trained, defaults to 50\n",
    "    batch_size: positive int, optional\n",
    "        the size of the training batches, defaults to 64\n",
    "    train_fraction: positive float, optional\n",
    "        the fraction of data to be used as training data, defaults to 0.8\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    history : TrainingHistory\n",
    "        the tracked training and test history\n",
    "    \"\"\"\n",
    "    history = TrainingHistory()\n",
    "    model.fit(\n",
    "        data, \n",
    "        labels, \n",
    "        epochs=epochs, \n",
    "        batch_size=batch_size, \n",
    "        validation_split=1.0 - train_fraction, \n",
    "        shuffle=True,\n",
    "        callbacks=[history]\n",
    "    )\n",
    "    return history\n",
    "    \n",
    "history = train_model(model, preprocessed_data, preprocessed_labels)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Visualization of the Training Progress\n",
    "\n",
    "Using the previously created and filled TrainingHistory instance, we are able to plot the loss and accuracy of the training batches and test epochs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_history(history):\n",
    "    \"\"\"\n",
    "    Plots the training (batch-wise) and test (epoch-wise) loss and accuracy.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    history : TrainingHistory\n",
    "        an instance of TrainingHistory monitoring callback\n",
    "    \"\"\"\n",
    "    figure, (batch_axis, epoch_axis) = plt.subplots(1, 2, figsize=(16, 5), sharey=True)\n",
    "    \n",
    "    # plot the training loss and accuracy\n",
    "    batch_axis.set_xlabel('batch number')\n",
    "    training_batches = np.arange(len(history.loss))\n",
    "    \n",
    "    batch_axis.plot(training_batches, history.loss, color='C0', label='loss')\n",
    "    batch_axis.set_ylabel('loss')\n",
    "    \n",
    "    batch_acc_axis = batch_axis.twinx()\n",
    "    batch_acc_axis.grid(False)\n",
    "    batch_acc_axis.set_ylabel('accuracy')\n",
    "    batch_acc_axis.set_ylim(bottom=0.0)\n",
    "    batch_acc_axis.plot(training_batches, history.acc, color='C4', label='accuracy')\n",
    "    \n",
    "    # plot the training loss and accuracy\n",
    "    epoch_axis.set_xlabel('epoch number')\n",
    "    validation_epochs = np.arange(len(history.validation_loss))\n",
    "    \n",
    "    epoch_axis.plot(validation_epochs, history.validation_loss, color='C0')\n",
    "    epoch_axis.set_ylabel('validation loss')\n",
    "    \n",
    "    epoch_acc_axis = epoch_axis.twinx()\n",
    "    epoch_acc_axis.grid(False)\n",
    "    epoch_acc_axis.set_ylabel('validation accuracy')\n",
    "    epoch_acc_axis.set_ylim(bottom=0.0)\n",
    "    epoch_acc_axis.plot(validation_epochs, history.validation_acc, color='C4')\n",
    "    \n",
    "    # display a legend\n",
    "    figure.legend(loc=8)\n",
    "    plt.show()\n",
    "\n",
    "plot_history(history)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next: add $\\texttt{Dropout}$ layers and observe their effect on the history.\n",
    "\\emph{Dropout} is a technique to reduce overfitting to the training data by randomly selecting neurons and setting their activation to $0$.\n",
    "The $\\texttt{Dropout}$ layer in Keras just passes through its input, except in randomly selected positions, where it passes through $0$."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Your Convolutions\n",
    "\n",
    "Extract the learned filters from CNN and apply them to the image. Try to compare them with the previously introduced example filters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def apply_network_convolutions(image, weights):\n",
    "    \"\"\"\n",
    "    Applies the passed convolutional weights to the image.\n",
    "\n",
    "    Parameters\n",
    "    ----------\n",
    "    image : np.array[width, height]\n",
    "        the image to be filtered\n",
    "    weights : np.array[filters, filter_width, filter_height]\n",
    "        the convolutional filter weights\n",
    "\n",
    "    Returns\n",
    "    -------\n",
    "    filtered images : np.array[filters, width, height]\n",
    "        the filtered images\n",
    "    \"\"\"\n",
    "    height, width = image.shape\n",
    "    kernels_y, kernels_x = weights.shape[:2]\n",
    "    \n",
    "    output = np.zeros((kernels_y, kernels_x + 1, height, width), dtype=image.dtype)\n",
    "    ##############\n",
    "    apply the convolutions to the image and store the output in output\n",
    "    ##############\n",
    "    return output\n",
    "\n",
    "weights = np.moveaxis(model.get_weights()[0], 0, -1).reshape(4, 8, 5, 5)\n",
    "model_convolutions = apply_network_convolutions(data[0], weights)\n",
    "show_examples(model_convolutions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Activate the filtered images additionally with ReLU function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "activated = model_convolutions.copy()\n",
    "activated[activated < 0] = 0\n",
    "show_examples(activated)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bonus:\n",
    "Add a [naive inception layer](https://www.cs.unc.edu/~wliu/papers/GoogLeNet.pdf) to your CNN and investigate its effect.\n",
    "Simplified, an inception layer are multiple convolutional layers with different filter sizes in parallel that are supposed to 'look' at the image on different resolution levels.\n",
    "\n",
    "In order to be able to implement this network, you will need advanced features of the Keras library, like the [functional API](https://keras.io/getting-started/functional-api-guide/), which allows you to specify layer connections, and the $\\texttt{Input}$ and $\\texttt{Model}$ classes."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}