capsule_re_implemented_pawan_notcomplete.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Code for capsule_layers.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/pawan/anaconda2/lib/python2.7/site-packages/h5py/__init__.py:36: FutureWarning: Conversion of the second argument of issubdtype from `float` to `np.floating` is deprecated. In future, it will be treated as `np.float64 == np.dtype(float).type`.\n",
      "  from ._conv import register_converters as _register_converters\n",
      "Using Theano backend.\n",
      "WARNING (theano.tensor.blas): Using NumPy C-API based implementation for BLAS functions.\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "Some key layers used for constructing a Capsule Network. These layers can used to construct CapsNet on other dataset,\n",
    "not just MNIST.\n",
    "*NOTE*: Some functions may be implemented in multiple ways, I keep all of them. You can try them for youself just by\n",
    "uncommenting them and commenting their counterparts.\n",
    "\"\"\"\n",
    "\n",
    "import keras.backend as K\n",
    "import tensorflow as tf\n",
    "from keras import initializers, layers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def squash(vectors, axis=-1):\n",
    "    \"\"\"\n",
    "    The non-linear activation used in Capsule. It drives the length of a large vector to near 1 and small vector to 0\n",
    "    :param vectors: some vectors to be squashed, N-dim tensor\n",
    "    :param axis: the axis to squash\n",
    "    :return: a Tensor with same shape as input vectors\n",
    "    \"\"\"\n",
    "    \n",
    "    s_squared_norm = K.sum(k.square(vectors), axis=axis, keepdims=True)\n",
    "    scale = s_squared_norm / (1+s_squared_norm) / K.sqrt(s_squared_norm+K.epsilon())\n",
    "    return scale*vectors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class CapsuleLayer(layers.Layer):\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def primaryCap(inputs, dim_capsule, n_channels, kernel_size, strides, padding):\n",
    "    \"\"\"\n",
    "    Apply Conv2D `n_channels` times and concatenate all capsules\n",
    "    :param inputs: 4D tensor, shape=[None, width, height, channels]\n",
    "    :param dim_capsule: the dim of the output vector of capsule\n",
    "    :param n_channels: the number of types of capsules\n",
    "    :return: output tensor, shape = [None, num_capsule, dim_capsule]\n",
    "    \"\"\"\n",
    "    \n",
    "    output = layers.Conv2D(filters=dim_capsule*n_channels, kernel_size = kernel_size, strides=strides, padding=padding,\n",
    "                          name='primarycap_conv2d')(inputs)\n",
    "    outputs = layer.Reshape(target_shape=[-1, dim_capsule], name='primarycap_reshape')(output)\n",
    "    return layers.Lambda(squash, name='primarycap_squash')(outputs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Code for capsule_net.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from keras import backend as K\n",
    "from keras import layers, models, optimizers\n",
    "from keras.utils import to_categorical\n",
    "import matplotlib.pyplot as plt\n",
    "from PIL import Image\n",
    "\n",
    "K.set_image_data_format(\"channels_last\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def CapsNet(input_shape, n_class, routings):\n",
    "    \"\"\"\n",
    "    A capsule network on fashion MNIST\n",
    "    :param input_shape: data shape, 3d, [width, height, channels]\n",
    "    :param n_class: number of classes\n",
    "    :routings: number of routing iterations\n",
    "    :return: Two Keras Models, the first one used for training, and the second one for evaluation.\n",
    "            `eval_model` can also be used for training\n",
    "    \"\"\"\n",
    "    x = layers.Input(shape=input_shape)\n",
    "    \n",
    "    # Layer 1: just a convolutional Conv2D layer\n",
    "    conv1 = layers.Conv2D(filters=256, kernel_size=9, strides=1, padding='valid', activation='relu', name='conv1')(x)\n",
    "    \n",
    "    # Layer 2: Conv2D layer with `squash` activation, then reshape to [None, num_capsule, dim_capsule]\n",
    "    primarycaps = PrimaryCap(conv1, dim_capsule=8, n_channels=32, kernel_size = 9, strides=2, padding='valid')\n",
    "    \n",
    "    # Layer 3: Capsule layer. Routing algorithm works here\n",
    "    digitcaps = CapsuleLayer(num_capsule=n_class, dim_capsule=16, routings=routings, name='digitcaps')(primarycaps)\n",
    "    \n",
    "    # Layer 4: This is auxilary layer to replace each capsule with its length. Just to match the true label's shape.\n",
    "    # If using TensorFlow, this will not be necessary. :)\n",
    "    out_caps = Length(name='capsnet')(digitcaps)\n",
    "    \n",
    "    # Decoder network.\n",
    "    y = layers.Input(shape=(n_class,))\n",
    "    masked_by_y = Mask()([digitcaps, y])  # The true label is used to mask the output of capsule layer. (for training)\n",
    "    masked = Mask()(digitcaps)  # Mask using the capsule with maximum length. (for prediction)\n",
    "    \n",
    "    # Shared Decoder Model in training and prediction\n",
    "    decoder = models.Sequential(name='decoder')\n",
    "    decoder.add(layers.Dense(512, activation='relu', input_dim=16*n_class))\n",
    "    decoder.add(layers.Dense(1024, activation='relu'))\n",
    "    decoder.add(layers.Dense(np.prod(input_shape), activation='sigmoid'))\n",
    "    decoder.add(layers.Reshape(target_shape=input_shape, name='out_recon'))\n",
    "    \n",
    "    # Models for training and evaluation (prediction)\n",
    "    train_model = models.Model([x,y], [out_caps, decoder(masked_by_y)])\n",
    "    eval_model = models.Model(x, [out_caps, decoder(masked)])\n",
    "    \n",
    "    # manipulate model\n",
    "    noise = layer.Input(shape=(nclass, 16))\n",
    "    noised_digitcaps = layers.Add()([digitcaps, noise])\n",
    "    masked_noised_y = Mask()([noised_digitcaps, noise])\n",
    "    manipulate_model = models.Model([x, y, noise], decoder(masked_noised_y))\n",
    "    return train_model, eval_model, manipulate_model    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_fashion_mnist():\n",
    "    # the data, shuffled and split between train and test sets\n",
    "    from keras.datasets import fashion_mnist\n",
    "    (x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()\n",
    "    \n",
    "    x_train = x_train.reshape(-1, 28, 28, 1).astype('float32') / 255.0\n",
    "    x_test = x_test.reshape(-1, 28, 28, 1).astype('float32') / 255.0\n",
    "    y_train = to_categorical(y_train.astype('float32'))\n",
    "    y_test = to_categorical(y_test.astype('float32'))\n",
    "    return (x_train, y_train), (x_test, y_test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import argparse\n",
    "from keras.preprocessing.image import ImageDataGenerator\n",
    "from keras import callbacks\n",
    "\n",
    "# setting the hyper parameters\n",
    "parser = argparse.ArgumentParser(description=\"Capsule network on Fashion MNIST\")\n",
    "parser.add_argument('--epochs', default=50, type=int)\n",
    "parser.add_argument('--batch_size', default=100, type=int)\n",
    "parser.add_argument('--lr', default=0.001, type=float, help=\"Initial learning rate\")\n",
    "parser.add_argument('--lr_decay', default=0.9, type=float, help=\"The value multiplied by lr at each epoch. Set a larger value for larger epochs\")\n",
    "parser.add_argument('--lam_recon', default=0.392, type=float, help=\"The cofficient for the loss of decoder\")\n",
    "parser.add_argument('-r', '--routings', default=3, type=int, help=\"Number of iterations used in routing algorithm. Should > 0\")\n",
    "parser.add_argument('--shift_fraction', default=0.1, type=float, help=\"Faction of pixels to shift at most in each direction.\")\n",
    "parser.add_argument('--debug', action='store_true', help=\"Save weights by TensorBoard\")\n",
    "parser.add_argument('--save_dir', default='./result')\n",
    "parser.add_argument('-t', '--testing', action='store_true', help=\"Test the trained model on testing dataset\")\n",
    "parser.add_argument('--digit', default=5, type=int, help=\"Digit to manipulate\")\n",
    "parser.add_argument('-w', '--weights', default=None, help=\"The path of the saved weights. Should be specified when testing.\")\n",
    "args = parser.parse_args([\"--epochs\", \"2\"])\n",
    "print(args)\n",
    "\n",
    "if not os.path.exists(args.save_dir):\n",
    "    os.makedirs(args.save_dir)\n",
    "\n",
    "# load the data\n",
    "(x_train, y_train), (x_test, y_test) = load_fashion_mnist()\n",
    "\n",
    "# define the model\n",
    "model, eval_model, manipulate_model = CapsNet(input_shape=x_train.shape[1:],\n",
    "                                              n_class=len(np.unique(np.argmax(y_train, 1))),\n",
    "                                             routings=args.routings)\n",
    "model.summary()\n",
    "\n",
    "if args.weights is not None:   # init the model weights with provided one\n",
    "    model.load_weights(args.weights)\n",
    "if not args.testing:\n",
    "    train(model=model, data=((x_train, y_train), (x_test, y_test)), args=args)\n",
    "else:\n",
    "    if args.weights is None:\n",
    "        print(\"No weights provided. Will test using random initialized weights.\")\n",
    "    manipulate_latent(manipulate_model, (x_test, y_test), args)\n",
    "    test(model=eval_model, data=(x_test, y_test), args=args)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}