09. Learning more about GANs/9.07 Converting photos to paintings using CycleGAN.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Converting photos to paintings using CycleGAN"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We just learned how CycleGAN works and how it converts images from one domain to another without any paired datasets. Now, we will learn how to implement CycleGAN in TensorFlow. We will see how to convert the pictures to paintings using CycleGAN as shown:\n",
    "\n",
    "![image](images/1.png)\n",
    "\n",
    "The dataset used in this section can be downloaded from here https://people.eecs.berkeley.edu/~taesung_park/CycleGAN/datasets/monet2photo.zip. Once you downloaded the dataset, unzip the archive and place it in data folder. The unziped archive will consist of four folders trainA, trainB, testA, and testB with training and testing images. \n",
    "\n",
    "\n",
    "The folder trainA consists of paintings (Monet) and the folder trainB consists of photos. Since we are mapping photos (x) to paintings (y), folder trainB which consists of photos will be our source image and the folder trainA which consists of paintings will be our target. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import the required libraries\n",
    "\n",
    "import the required libraries:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "import numpy as np\n",
    "import tensorflow as tf\n",
    "tf.logging.set_verbosity(tf.logging.ERROR)\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "%matplotlib inline\n",
    "\n",
    "import scipy.misc as misc\n",
    "from PIL import Image\n",
    "import os"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Convolution\n",
    "\n",
    "\n",
    "Define the convolution operation for the discriminator:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def conv(inputs, nums_out, kernel_size, stride, padding, is_dis=False):\n",
    "    c = int(inputs.shape[-1])\n",
    "\n",
    "    weight = tf.get_variable(\"weight\", shape=[kernel_size, kernel_size, c, nums_out], initializer=tf.random_normal_initializer(stddev=0.02))\n",
    "    bias = tf.get_variable(\"bias\", shape=[nums_out], initializer=tf.constant_initializer([0]))\n",
    "    \n",
    "    if is_dis:\n",
    "        return tf.nn.conv2d(inputs, spectral_norm(\"SN\",weight), [1, stride, stride, 1], padding) + bias\n",
    "    else:\n",
    "        return tf.nn.conv2d(inputs, weight, [1, stride, stride, 1], padding) + bias    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Deconvolution\n",
    "\n",
    "\n",
    "Define the deconvolutional operation for the generator:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def deconv(inputs, nums_out, kernel_size, stride):\n",
    "    c = int(inputs.shape[-1])\n",
    "    batch = int(inputs.shape[0])\n",
    "    height = int(inputs.shape[1])\n",
    "    width = int(inputs.shape[2])\n",
    "    \n",
    "    weight = tf.get_variable(\"weight\", shape=[kernel_size, kernel_size, nums_out, c], initializer=tf.random_normal_initializer(stddev=0.02))\n",
    "    bias = tf.get_variable(\"bias\", shape=[nums_out], initializer=tf.constant_initializer([0.]))\n",
    "    \n",
    "    return tf.nn.conv2d_transpose(inputs, weight, output_shape=[batch, height*stride, width*stride, nums_out], strides=[1, stride, stride, 1]) + bias"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Instance Normalization\n",
    "\n",
    "\n",
    "We know that in batch normalization we normalize images across the batches whereas in instance normalization we normalize each batch independently:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "epsilon = 1e-8\n",
    "def InstanceNorm(inputs):\n",
    "    mean, var = tf.nn.moments(inputs, axes=[1, 2], keep_dims=True)\n",
    "    scale = tf.get_variable(\"scale\", shape=mean.shape[-1], initializer=tf.constant_initializer([1.]))\n",
    "    shift = tf.get_variable(\"shift\", shape=mean.shape[-1], initializer=tf.constant_initializer([0.]))\n",
    "    return (inputs - mean) * scale / tf.sqrt(var + epsilon) + shift"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sepctral Normalization\n",
    "\n",
    "\n",
    "\n",
    "We apply spectral normalization on the convolutional layers. The spectral norm is the maximum singular value of a matrix. So, we regularize the weights of the convolutional layer with the largest singular value of weights in that layer.\n",
    "\n",
    "\n",
    "How do we do that?  Miyato et al introduce a method called the power iteration to estimate the spectral norm of each layer. In the power iteration method, we compute the L2 distance between the linear combination of the vector u and the convolutional weights. \n",
    "\n",
    "\n",
    "Let us say we have a random vector $v$ in the domain of our matrix and another vector $u$ in the codomain. Now, we apply the power iteration method:\n",
    "\n",
    "\n",
    "$\\hat{u} = \\frac{W^Tu}{|| W^T u||}$\n",
    "\n",
    "\n",
    "$\\hat{v} = \\frac{W^Tv}{|| W^T v||}$\n",
    "\n",
    "\n",
    "\n",
    "The final weights can be calcuated as:\n",
    "\n",
    "\n",
    "$\\sigma = (\\hat{v} W) \\hat{u}^T$\n",
    "\n",
    "\n",
    "$W = \\frac{W}{\\sigma(W)}$"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def spectral_norm(name, w, iteration=1):\n",
    "  \n",
    "    w_shape = w.shape.as_list()\n",
    "    w = tf.reshape(w, [-1, w_shape[-1]])\n",
    "  \n",
    "    with tf.variable_scope(name, reuse=False):\n",
    "        u = tf.get_variable(\"u\", [1, w_shape[-1]], initializer=tf.truncated_normal_initializer(), trainable=False)\n",
    "    u_hat = u\n",
    "    v_hat = None\n",
    "\n",
    "    \n",
    "    def l2_norm(v, eps=1e-12):\n",
    "        return v / (tf.reduce_sum(v ** 2) ** 0.5 + eps)\n",
    "\n",
    "    for i in range(iteration):\n",
    "        v_ = tf.matmul(u_hat, tf.transpose(w))\n",
    "        v_hat = l2_norm(v_)\n",
    "        u_ = tf.matmul(v_hat, w)\n",
    "        u_hat = l2_norm(u_)\n",
    "    sigma = tf.matmul(tf.matmul(v_hat, w), tf.transpose(u_hat))\n",
    "    w_norm = w / sigma\n",
    "    \n",
    "    with tf.control_dependencies([u.assign(u_hat)]):\n",
    "        w_norm = tf.reshape(w_norm, w_shape)\n",
    "    \n",
    "    return w_norm"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Activation function\n",
    "\n",
    "\n",
    "Define the Leaky ReLU activation function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def leaky_relu(inputs, slope=0.2):\n",
    "    return tf.maximum(slope*inputs, inputs)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Discriminator\n",
    "\n",
    "Discriminator consists of four convolutional layer followed by a fully connected layer:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class discriminator:\n",
    "    def __init__(self, name):\n",
    "        self.name = name\n",
    "\n",
    "    def __call__(self, inputs, reuse = False):\n",
    "        #discriminator\n",
    "        inputs = tf.random_crop(inputs, [batchsize, 70, 70, 3])\n",
    "        with tf.variable_scope(self.name, reuse=reuse):\n",
    "            with tf.variable_scope(\"c64\"):\n",
    "                inputs = leaky_relu(conv(inputs, 64, 5, 2, \"SAME\", True))\n",
    "            with tf.variable_scope(\"c128\"):\n",
    "                inputs = leaky_relu(InstanceNorm(conv(inputs, 128, 5, 2, \"SAME\", True)))\n",
    "            with tf.variable_scope(\"c256\"):\n",
    "                inputs = leaky_relu(InstanceNorm(conv(inputs, 256, 5, 2, \"SAME\", True)))\n",
    "            with tf.variable_scope(\"c512\"):\n",
    "                inputs = leaky_relu(InstanceNorm(conv(inputs, 512, 5, 2, \"SAME\", True)))\n",
    "            with tf.variable_scope(\"fully_conv\"):\n",
    "                kernel_size = np.size(inputs, 1)\n",
    "                inputs = tf.squeeze(conv(inputs, 1, kernel_size, 1, \"VALID\", True), axis=[1, 2, 3])\n",
    "        return inputs\n",
    "\n",
    "    @property\n",
    "    def var(self):\n",
    "        return tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope=self.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Defining Generator\n",
    "\n",
    "Generator consists of deconvolutional layers:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class generator:\n",
    "    def __init__(self, name):\n",
    "        self.name = name\n",
    "\n",
    "    def __call__(self, inputs, reuse=False):\n",
    "        with tf.variable_scope(name_or_scope=self.name, reuse=reuse):\n",
    "            inputs = tf.pad(inputs, tf.constant([[0, 0], [3, 3], [3, 3], [0, 0]]))\n",
    "            with tf.variable_scope(\"c7s1-32\"):\n",
    "                inputs = tf.nn.relu(InstanceNorm(conv(inputs, 32, 7, 1, \"VALID\")))\n",
    "            with tf.variable_scope(\"d64\"):\n",
    "                inputs = tf.nn.relu(InstanceNorm(conv(inputs, 64, 3, 2, \"SAME\")))\n",
    "            with tf.variable_scope(\"d128\"):\n",
    "                inputs = tf.nn.relu(InstanceNorm(conv(inputs, 128, 3, 2, \"SAME\")))\n",
    "            for i in range(6):\n",
    "                with tf.variable_scope(\"R\"+str(i)):\n",
    "                    temp = inputs\n",
    "                    with tf.variable_scope(\"R_conv1\"):\n",
    "                        inputs = tf.pad(inputs, tf.constant([[0, 0], [1, 1], [1, 1], [0, 0]]), \"REFLECT\")\n",
    "                        inputs = tf.nn.relu(InstanceNorm(conv(inputs, 128, 3, 1, \"VALID\")))\n",
    "                    with tf.variable_scope(\"R_conv2\"):\n",
    "                        inputs = tf.pad(inputs, tf.constant([[0, 0], [1, 1], [1, 1], [0, 0]]), \"REFLECT\")\n",
    "                        inputs = InstanceNorm(conv(inputs, 128, 3, 1, \"VALID\"))\n",
    "                    inputs = temp + inputs\n",
    "            with tf.variable_scope(\"u64\"):\n",
    "                inputs = tf.nn.relu(InstanceNorm(deconv(inputs, 64, 3, 2)))\n",
    "            with tf.variable_scope(\"u32\"):\n",
    "                inputs = tf.nn.relu(InstanceNorm(deconv(inputs, 32, 3, 2)))\n",
    "            with tf.variable_scope(\"c7s1-3\"):\n",
    "                inputs = tf.nn.tanh((deconv(inputs, 3, 7, 1)))\n",
    "            return (inputs + 1.) * 127.5\n",
    "\n",
    "    @property\n",
    "    def var(self):\n",
    "        return tf.get_collection(tf.GraphKeys.GLOBAL_VARIABLES, scope=self.name)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define image width, height, number of images and batch size"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "image_width = 128\n",
    "image_height = 128\n",
    "\n",
    "\n",
    "num_images = 8000\n",
    "batchsize = 1"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cycle GAN \n",
    "\n",
    "We will see how to implement cycle gan. Check the comment on each line of code for better understanding:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "class CycleGAN:\n",
    "    \n",
    "    def __init__(self):\n",
    "        self.X = tf.placeholder(\"float\", shape=[batchsize, image_height, image_width, 3])\n",
    "        self.Y = tf.placeholder(\"float\", shape=[batchsize, image_height, image_width, 3])\n",
    "        \n",
    "        #maps x to y \n",
    "        G = generator(\"G\")\n",
    "        \n",
    "        #maps y to x\n",
    "        F = generator(\"F\")\n",
    "        \n",
    "        #discriminate between real source image and fake source image\n",
    "        self.Dx = discriminator(\"Dx\")\n",
    "        \n",
    "        #discriminate between real target image and fake target image\n",
    "        self.Dy = discriminator(\"Dy\")\n",
    "        \n",
    "        #fake source image\n",
    "        self.fake_X = F(self.Y)\n",
    "        \n",
    "        #fake target image\n",
    "        self.fake_Y = G(self.X)\n",
    "        \n",
    "        #real source image logits\n",
    "        self.Dx_logits_real = self.Dx(self.X)\n",
    "        \n",
    "        #fake source image logits\n",
    "        self.Dy_logits_real = self.Dy(self.Y)\n",
    "        \n",
    "        #real target image logits\n",
    "        self.Dx_logits_fake = self.Dx(self.fake_X, True)\n",
    "        \n",
    "        #fake target image logits\n",
    "        self.Dy_logits_fake = self.Dy(self.fake_Y, True)\n",
    "        \n",
    "        #cycle consistency loss\n",
    "        self.cycle_loss = tf.reduce_mean(tf.abs(F(self.fake_Y, True) - self.X)) + \\\n",
    "                        tf.reduce_mean(tf.abs(G(self.fake_X, True) - self.Y))\n",
    "      \n",
    "        \n",
    "        #discriminator loss (refer equations in the chapter)\n",
    "        self.Dy_loss = -tf.reduce_mean(self.Dy_logits_real) + tf.reduce_mean(self.Dy_logits_fake)\n",
    "        self.Dx_loss = -tf.reduce_mean(self.Dx_logits_real) + tf.reduce_mean(self.Dx_logits_fake)\n",
    "        \n",
    "        \n",
    "        #generator loss\n",
    "        self.G_loss = -tf.reduce_mean(self.Dy_logits_fake) + 10. * self.cycle_loss\n",
    "        \n",
    "        #discriminator loss\n",
    "        self.F_loss = -tf.reduce_mean(self.Dx_logits_fake) + 10. * self.cycle_loss\n",
    "        \n",
    "        #optimize discriminator\n",
    "        self.Dx_optimizer = tf.train.AdamOptimizer(2e-4, beta1=0., beta2=0.9).minimize(self.Dx_loss, var_list=[self.Dx.var])\n",
    "        self.Dy_optimizer = tf.train.AdamOptimizer(2e-4, beta1=0., beta2=0.9).minimize(self.Dy_loss, var_list=[self.Dy.var])\n",
    "        \n",
    "        #optimize generator\n",
    "        self.G_optimizer = tf.train.AdamOptimizer(2e-4, beta1=0., beta2=0.9).minimize(self.G_loss, var_list=[G.var])\n",
    "        self.F_optimizer = tf.train.AdamOptimizer(2e-4, beta1=0., beta2=0.9).minimize(self.F_loss, var_list=[F.var])\n",
    "\n",
    "        self.sess = tf.Session()\n",
    "        self.sess.run(tf.global_variables_initializer())\n",
    "        \n",
    "        self.train()\n",
    "\n",
    "\n",
    "    def train(self):\n",
    "        \n",
    "        \n",
    "        #real Images (source images)\n",
    "        X_path = 'data/monet2photo/trainB//'\n",
    "        \n",
    "        #paintings (target images)\n",
    "        Y_path = 'data/monet2photo/trainA//'\n",
    "               \n",
    "        Y = os.listdir(Y_path)[:num_images]\n",
    "        X = os.listdir(X_path)[:num_images]\n",
    "        \n",
    "        #we use only 100 images\n",
    "        X = X[1:100]\n",
    "        Y = Y[1:100]\n",
    "        \n",
    "        nums = Y.__len__()\n",
    "        \n",
    "        saver = tf.train.Saver()\n",
    "        \n",
    "        #create a folder for saving the fake paintings and images generated by the generator\n",
    "        if not os.path.exists('fake_painiting'):\n",
    "            os.makedirs('fake_painiting')\n",
    "\n",
    "        if not os.path.exists('fake_real_image'):\n",
    "            os.makedirs('fake_real_image') \n",
    "            \n",
    "        print 'started training...'\n",
    "        \n",
    "        \n",
    "        #for each epoch\n",
    "        for epoch in range(100000):\n",
    "            \n",
    "            #for every batch\n",
    "            for i in range(int(nums / batchsize) - 1):\n",
    "                \n",
    "                #select batch of images\n",
    "                X_img = np.zeros([batchsize, image_height, image_width, 3])\n",
    "                Y_img = np.zeros([batchsize, image_height, image_width, 3])\n",
    "                \n",
    "                for j in np.arange(i * batchsize, i * batchsize + batchsize, 1):\n",
    "                    \n",
    "                    #resize the source image\n",
    "                    img = misc.imresize(np.array(Image.open(X_path + X[j])), [image_height, image_width])\n",
    "                    X_img[j - i * batchsize, :, :, :] = img\n",
    "                    \n",
    "                    #resize the target image\n",
    "                    img = misc.imresize(np.array(Image.open(Y_path + Y[j])), [image_height, image_width])\n",
    "                    Y_img[j - i * batchsize, :, :, :] = img\n",
    "                \n",
    "                #train the discriminator\n",
    "                self.sess.run(self.Dy_optimizer, feed_dict={self.X: X_img, self.Y: Y_img})\n",
    "                self.sess.run(self.Dx_optimizer, feed_dict={self.X: X_img, self.Y: Y_img})\n",
    "                \n",
    "                #train the generator\n",
    "                self.sess.run(self.G_optimizer, feed_dict={self.X: X_img, self.Y: Y_img})\n",
    "                self.sess.run(self.F_optimizer, feed_dict={self.X: X_img, self.Y: Y_img})\n",
    "                \n",
    "                #compute loss and generate new image on every 50th epochs\n",
    "                if i % 50 == 0:\n",
    "                    \n",
    "                    #compute loss\n",
    "                    [Dx_loss, Dy_loss, G_loss, F_loss, fake_X, fake_Y, cyc_loss] = \\\n",
    "                    self.sess.run([self.Dx_loss, self.Dy_loss, self.G_loss, self.F_loss, \\\n",
    "                       self.fake_X, self.fake_Y, self.cycle_loss], feed_dict={self.X: X_img, self.Y: Y_img})\n",
    "                    \n",
    "                    print(\"Epoch: {}, iteration: {}, Dx Loss: {}, Dy Loss: {}, G Loss: {}, F Loss: {}, Cycle Loss: {}\". \\\n",
    "                          format(epoch, i, Dx_loss, Dy_loss, G_loss, F_loss, cyc_loss))\n",
    "\n",
    "                    print('\\n')\n",
    "                    \n",
    "                    #store the generated images\n",
    "                    Image.fromarray(np.uint8(fake_Y)[0, :, :, :]).save(\".//fake_painiting//\"+str(epoch)+\"_\"+str(i)+\".jpg\")\n",
    "                    Image.fromarray(np.uint8(fake_X)[0, :, :, :]).save(\".//fake_real_image//\" + str(epoch) + \"_\" + str(i) + \".jpg\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Start Training "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "started training...\n",
      "Epoch: 0, iteration: 0, Dx Loss: -4.33659744263, Dy Loss: -2.71015405655, G Loss: 1475.7109375, F Loss: 1474.48071289, Cycle Loss: 147.429473877\n",
      "\n",
      "\n",
      "Epoch: 0, iteration: 50, Dx Loss: 8.18372154236, Dy Loss: -11.3952150345, G Loss: 669.335083008, F Loss: 627.104492188, Cycle Loss: 64.9336242676\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cyc = CycleGAN()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Credits for the code used in this section goes to [MingtaoGuo](https://github.com/MingtaoGuo/CycleGAN).\n",
    "\n",
    "\n",
    "Applications of CycleGAN are endless since it can be use to translate images from one domain to another without the paired dataset. In the next section, we will learn how StackGAN is used for converting the text descriptions to photo realistic images."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python [default]",
   "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.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}