{"cells":[{"metadata":{},"cell_type":"markdown","source":"## CyclicGAN for Monet Generation\n\nBasically the main reason of using **CyclicGAN** for this task is that the data is **Unpaired** and we have to learn mappings in both directions i.e. *photo to monet* and *monet to photo* in order to generate the best and desired monets. You can learn more about CyclicGANs from the official [Tensorflow tutorials](https://www.tensorflow.org/tutorials/generative/cyclegan). Also, I've created an [article](https://towardsdatascience.com/gans-leveraging-technology-for-a-better-tomorrow-ea192087b4e4) that summarizes most of the information required to know about GANs, its variants and their applications. Please check out that also.\n\nThis notebook is inspired from [Amy Jang's](https://www.kaggle.com/amyjang) [notebook](https://www.kaggle.com/amyjang/monet-cyclegan-tutorial/).\n\n<body>\n    <p style=\"color:red\"><b>Caution: It is highly recommended to have a basic info about GANs and specifically CyclicGANs before moving down. I would also try to explain most things in this notebook.</b></p>\n</body>"},{"metadata":{},"cell_type":"markdown","source":"## Imports and Setting up the environment"},{"metadata":{"trusted":true},"cell_type":"code","source":"!pip install git+https://github.com/tensorflow/examples.git","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"#IMPORTS\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport tensorflow as tf\nimport os, time\nfrom kaggle_datasets import KaggleDatasets\nfrom tensorflow_examples.models.pix2pix import pix2pix\nfrom IPython.display import clear_output","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(tf.__version__)","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"# Configuration\nclass Configuration:\n    \"\"\"Class containing most of the parameters or hyperparameters used\n    throughout the notebook.\"\"\"\n    \n    epochs = 30\n    MONET_TFREC = \"/monet_tfrec/*.tfrec\"\n    MONET_JPG = \"/monet_jpg/*.jpg\"\n    PHOTO_TFREC = \"/photo_tfrec/*.tfrec\"\n    PHOTO_JPG = \"/photo_jpg/*.jpg\"\n    BATCH_SIZE = 8\n    IMAGE_SIZE = [256, 256]\n    BUFFER = 10000\n    steps_per_epoch = 0\n    loss = tf.keras.losses.BinaryCrossentropy(from_logits=True, reduction=tf.keras.losses.Reduction.NONE)\n\ncfg = Configuration()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Setting up the TPU\ntry:\n    tpu = tf.distribute.cluster_resolver.TPUClusterResolver()\n    print('Device:', tpu.master())\n    tf.config.experimental_connect_to_cluster(tpu)\n    tf.tpu.experimental.initialize_tpu_system(tpu)\n    strategy = tf.distribute.experimental.TPUStrategy(tpu)\nexcept:\n    strategy = tf.distribute.get_strategy()\nprint('Number of replicas:', strategy.num_replicas_in_sync)\n\nAUTOTUNE = tf.data.experimental.AUTOTUNE","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Now since we would be going to make the dataset infinitely repeatable, it would be necessary for us to define the number of steps that has to be taken at each epoch. So let's just set up that also."},{"metadata":{"trusted":true},"cell_type":"code","source":"monet_jpg = tf.io.gfile.glob(\"../input/gan-getting-started/monet_jpg/*.jpg\")\ncfg.steps_per_epoch = len(monet_jpg)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Setting up the Dataset\nWe would use the object oriented way to create and manage dataset. This is more efficient and cleaner way to write code.\n\nNote that we've used random-jittering as a preprocessing step for the dataset. **Random Jittering** is basically an augmentation technique which is accomplished by following steps:\n1. Resize the image to higher dimension \n2. Randomly crop the image to the original shape\n3. Occasionally flip the image.    "},{"metadata":{"trusted":true},"cell_type":"code","source":"class MonetDataset:\n        def __init__(self,config):\n            \"\"\"Creates a data of TFRecord files.\"\"\"\n            self.cfg = config\n            gcs_path = KaggleDatasets().get_gcs_path()\n            self.monet_files = tf.io.gfile.glob(gcs_path+self.cfg.MONET_TFREC)\n            self.photo_files = tf.io.gfile.glob(gcs_path + self.cfg.PHOTO_TFREC)\n            \n        def decode_image(self, image):\n            \"\"\"Function to preprocess the image prior to training.\"\"\"\n            img = tf.image.decode_jpeg(image, channels=3)\n            img = tf.cast(img, tf.float32)\n            img = img/127.5 - 1\n            img = tf.reshape(img, [*self.cfg.IMAGE_SIZE, 3])\n            return img\n        \n        def read_tfrecord(self, instance):\n            \"\"\"Function to extract data from TFRecordDataset Instance.\"\"\"\n            tfrecordformat = {\n                    \"image_name\": tf.io.FixedLenFeature([], tf.string),\n                    \"image\": tf.io.FixedLenFeature([], tf.string),\n                    \"target\": tf.io.FixedLenFeature([], tf.string)\n                   }\n            example = tf.io.parse_single_example(instance, tfrecordformat)\n            return self.decode_image(example[\"image\"])\n        \n        def prepare_dataset(self, monet=True):\n            \"\"\"Main function to prepare the input pipeline.\n            Args: \n            monet- bool value\n            Determines if we wanna generate monet dataset or the photo dataset\"\"\"\n            dataset = tf.data.TFRecordDataset(self.monet_files if monet else self.photo_files, num_parallel_reads=AUTOTUNE)\n            dataset = dataset.map(self.read_tfrecord, num_parallel_calls=AUTOTUNE)\n            dataset = dataset.map(self.random_jitter, num_parallel_calls=AUTOTUNE)\n            dataset = dataset.repeat()\n            dataset = dataset.shuffle(self.cfg.BUFFER)\n            dataset = dataset.batch(self.cfg.BATCH_SIZE)\n            dataset = dataset.prefetch(AUTOTUNE)\n            return dataset\n\n        def random_crop(self, image):\n            \"\"\"Function to perform random cropping.\"\"\"\n            image = tf.image.random_crop(image, [*self.cfg.IMAGE_SIZE, 3])\n            return image\n        \n        def random_jitter(self, image):\n            \"\"\"Function to perform random jittering.\"\"\"\n            image = tf.image.resize(image, [286, 286])\n            image = self.random_crop(image)\n            \n            if tf.random.uniform([], 0, 1) > 0.5:\n                image = tf.image.random_flip_left_right(image)\n            return image\n        \n        def visualize_data(self, data):\n            \"\"\"Utility function to visualize the samples in the dataset instance being provided.\"\"\"\n            fig, ax = plt.subplots(2, self.cfg.BATCH_SIZE//2, figsize=(16, 4)) # Figsize->W x H\n            ax = ax.flatten()\n            for i, im in zip(range(self.cfg.BATCH_SIZE), data):\n                im = im*0.5 + 0.5\n                ax[i].imshow(im)\n                ax[i].axis(\"off\")\n            plt.show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Creating instance of dataset\ndataset = MonetDataset(Configuration())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Creating seperate monet and photo dataset.\nmonet_dataset = dataset.prepare_dataset(monet=True)\nphoto_dataset = dataset.prepare_dataset(monet=False)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"raw","source":"Let us visualize the Monets and Real photographs."},{"metadata":{"trusted":true},"cell_type":"code","source":"example = next(iter(monet_dataset))\ndataset.visualize_data(example)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"dataset.visualize_data(next(iter(photo_dataset)))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Prepare Model\n\nThe Generator network follows the **U-Net** architecture. A U-Net architecture is a vanilla encoder-decoder network with an addition of **skip connections** in between. The reason for using skip-connections is that the problem of [Vanishing Gradients](https://en.wikipedia.org/wiki/Vanishing_gradient_problem) would be resolved.\n\nThe Discriminator network uses the **PatchGAN** architecture. PatchGANs are a class of networks which use patches extracted from the input image to classify it. In our case the patch extracted from the input image (real image and the generated image) would be used to classify if the image is real or fake. A *higher intense* patch denotes real image whereas a *lower intense* patch determines a fake or generated image. Here a **70x70** patch extracted from the input image would be used to classify it.\n\nThe reason of using PatchGAN is basically the simplicity of the architecture which allows us to train the network faster and also as the **L1 norm** (used as a loss function) tackles the low level features in the data, we only need to take care of high level features which we accomplish by using PatchGANs.\n\nFor more info on-\n\nU-Net architecture - https://arxiv.org/pdf/1505.04597.pdf<br></br>\nPatchGAN - https://paperswithcode.com/method/patchgan\n\n**Following the architecture of CyclicGANs, we would create two Generator instances and two Discriminator instances.**"},{"metadata":{"trusted":true},"cell_type":"code","source":"class CyclicGAN(tf.keras.Model):\n    \"\"\"Class to build and train custom CyclicGAN architecture.\"\"\"\n    def __init__(self, \n                monet_generator, \n                photo_generator,\n                monet_discriminator,\n                photo_discriminator,\n                lambda_cyclic = 10):\n        super(CyclicGAN, self).__init__()\n        self.m_gen = monet_generator\n        self.p_gen = photo_generator\n        self.m_disc = monet_discriminator\n        self.p_disc = photo_discriminator\n        self.lambda_ = lambda_cyclic\n    \n    def compile(self,\n                m_gen_optimizer,\n                p_gen_optimizer,\n                m_disc_optimizer,\n                p_disc_optimizer,\n                gen_loss,\n                disc_loss,\n                cyclic_loss,\n                identity_loss\n               ):\n        \"\"\"Function to set the optimizers and metrics used for the model training.\"\"\"\n        super(CyclicGAN, self).compile()\n        self.m_gen_optimizer= m_gen_optimizer\n        self.p_gen_optimizer = p_gen_optimizer\n        self.m_disc_optimizer = m_disc_optimizer\n        self.p_disc_optimizer = p_disc_optimizer\n        self.gen_loss = gen_loss\n        self.disc_loss = disc_loss\n        self.cyclic_loss = cyclic_loss\n        self.identity_loss = identity_loss\n    \n    def train_step(self, batch_data):\n        \"\"\"Function to run a single step of training.\"\"\"\n        real_monet, real_photo = batch_data\n        with tf.GradientTape(persistent=True) as tape:\n            \n            # Getting Generator and Discriminator output. \n            fake_monet = self.m_gen(real_photo, training=True)\n            cycled_photo = self.p_gen(fake_monet, training=True)\n            \n            fake_photo = self.p_gen(real_monet, training=True)\n            cycled_monet = self.m_gen(fake_photo, training=True)\n            \n            same_monet = self.m_gen(real_monet, training=True)\n            same_photo = self.p_gen(real_photo, training=True)\n            \n            disc_real_monet = self.m_disc(real_monet, training=True)\n            disc_fake_monet = self.m_disc(fake_monet, training=True)\n            \n            disc_real_photo = self.p_disc(real_photo, training=True)\n            disc_fake_photo = self.p_disc(fake_photo, training=True)\n            \n            # Calculate Losses\n            cycle_loss = self.lambda_*(self.cyclic_loss(real_monet, cycled_monet)+self.cyclic_loss(real_photo, cycled_photo))\n            identity_loss = self.lambda_ * (self.identity_loss(real_monet, same_monet) + self.identity_loss(real_photo, same_photo))\n            \n            monet_gen_loss = self.gen_loss(disc_fake_monet)\n            photo_gen_loss = self.gen_loss(disc_fake_photo)\n            \n            total_monet_gen_loss = monet_gen_loss + cycle_loss + identity_loss\n            total_photo_gen_loss = photo_gen_loss + cycle_loss + identity_loss\n            \n            monet_disc_loss = self.disc_loss(disc_real_monet, disc_fake_monet)\n            photo_disc_loss = self.disc_loss(disc_real_photo, disc_fake_photo)\n            \n            # Calculate Gradients\n            monet_gen_gradient = tape.gradient(total_monet_gen_loss, self.m_gen.trainable_variables)\n            photo_gen_gradient = tape.gradient(total_photo_gen_loss, self.p_gen.trainable_variables)\n            \n            monet_disc_gradient = tape.gradient(monet_disc_loss, self.m_disc.trainable_variables)\n            photo_disc_gradient = tape.gradient(photo_disc_loss, self.p_disc.trainable_variables)\n            \n            self.m_gen_optimizer.apply_gradients(zip(monet_gen_gradient, self.m_gen.trainable_variables))\n            self.p_gen_optimizer.apply_gradients(zip(photo_gen_gradient, self.p_gen.trainable_variables))\n            \n            # Apply Gradients\n            self.m_disc_optimizer.apply_gradients(zip(monet_disc_gradient, self.m_disc.trainable_variables))\n            self.p_disc_optimizer.apply_gradients(zip(photo_disc_gradient, self.p_disc.trainable_variables))\n            \n            return {\n                \"monet_generator_loss\": total_monet_gen_loss,\n                \"monet_discriminator_loss\": monet_disc_loss,\n                \"photo_generator_loss\": total_photo_gen_loss,\n                \"photo_discriminator_loss\": photo_disc_loss\n            }","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"OUTPUT_CHANNELS = 3\nwith strategy.scope():\n    monet_gen = pix2pix.unet_generator(OUTPUT_CHANNELS, norm_type=\"instancenorm\")\n    photo_gen = pix2pix.unet_generator(OUTPUT_CHANNELS, norm_type=\"instancenorm\")\n\n    monet_disc = pix2pix.discriminator(norm_type=\"instancenorm\", target=False)\n    photo_disc = pix2pix.discriminator(norm_type=\"instancenorm\", target=False)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"**To get an intuition about Instance Normalization refer [this](https://www.tensorflow.org/addons/tutorials/layers_normalizations).**"},{"metadata":{"trusted":true},"cell_type":"code","source":"with strategy.scope():\n    model = CyclicGAN(monet_gen, photo_gen, monet_disc, photo_disc)\n    \n    # Prepairing the loss functions.\n    def generator_loss(generated_op):\n        return cfg.loss(tf.ones_like(generated_op), generated_op)\n    \n    def discriminator_loss(disc_real_op, disc_fake_op):\n        real_loss = cfg.loss(tf.ones_like(disc_real_op), disc_real_op)\n        fake_loss = cfg.loss(tf.zeros_like(disc_fake_op), disc_fake_op)\n        total_loss = real_loss + fake_loss\n        return total_loss * 0.5\n    \n    def cyclic_loss(real_image, cycled_image):\n        return tf.reduce_mean(tf.abs(real_image - cycled_image))\n    \n    def identity_loss(real_image, same_image):\n        return tf.reduce_mean(tf.abs(real_image-same_image))\n    \n    m_gen_optim = tf.keras.optimizers.Adam(lr=2e-04, beta_1=0.5)\n    p_gen_optim = tf.keras.optimizers.Adam(lr=2e-04, beta_1=0.5)\n    m_disc_optim = tf.keras.optimizers.Adam(lr=2e-04, beta_1=0.5)\n    p_disc_optim = tf.keras.optimizers.Adam(lr=2e-04, beta_1=0.5)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"with strategy.scope():\n    model.compile(m_gen_optim,\n                 p_gen_optim,\n                 m_disc_optim,\n                 p_disc_optim,\n                 generator_loss,\n                 discriminator_loss,\n                 cyclic_loss,\n                 identity_loss)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Visualize the model\nIn this section we would look up at how the U-Net and PatchGAN architectures look like."},{"metadata":{"trusted":true},"cell_type":"code","source":"# Visualizing Generator model.\ntf.keras.utils.plot_model(monet_gen, show_shapes=True, dpi=64)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"This could look different from what you've seen in a typical U-Net architecture but it is similar to that. Its also doing downsampling and then upsampling with skip connections but in a better way. "},{"metadata":{"trusted":true},"cell_type":"code","source":"# Visualizing Discriminator model\ntf.keras.utils.plot_model(monet_disc, show_shapes=True, dpi=96)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Train the model"},{"metadata":{"trusted":true},"cell_type":"code","source":"with strategy.scope():\n    model.fit(tf.data.Dataset.zip((monet_dataset, photo_dataset)), epochs=cfg.epochs,\n             steps_per_epoch = cfg.steps_per_epoch)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Visualizing Predictions\n\nLet us know visualize what the model has learnt so far."},{"metadata":{"trusted":true},"cell_type":"code","source":"photo_example = next(iter(photo_dataset))\npredict_img = monet_gen.predict(tf.expand_dims(photo_example[0], axis=0))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"fig, ax = plt.subplots(1, 2, figsize=(8,8))\nax = ax.flatten()\nax[0].imshow(photo_example[0]*0.5 + 0.5)\nout  = (predict_img[0]*127.5 + 127.5).astype(np.uint8)\nax[1].imshow(out)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"# Submission"},{"metadata":{"trusted":true},"cell_type":"code","source":"import PIL\n!mkdir ../images","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"photo_jpg = tf.io.gfile.glob(\"../input/gan-getting-started/photo_jpg/*.jpg\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"for i, image in zip(range(1, len(photo_jpg)+1), photo_dataset):\n    prediction = monet_gen(image, training=False)[0].numpy()\n    prediction = (prediction*127.5 + 127.5).astype(np.uint8)\n    im = PIL.Image.fromarray(prediction)\n    im.save(f\"../images/{i}.jpg\")\n    if(i%100==0):\n        print(f\"Processed {i} images\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import shutil\nshutil.make_archive(\"/kaggle/working/images\", \"zip\", \"/kaggle/images\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"<b>--------------------------------------------------------------------------------------------------------------------------------------------------</b>"},{"metadata":{},"cell_type":"markdown","source":"### Areas to modify:\nThese are some areas which could be modified to make the notebook more interactive and efficient.\n1. Visualize the performance of the model while training.\n2. Introduce K-Fold cross validation.\n3. Improve hyperparameter set."},{"metadata":{},"cell_type":"markdown","source":"<body>\n    <p style=\"color:#bf800a;font-size:18px\"><b>I hope you liked and learnt something new from my kernel. If you liked it, please upvote to keep me motivated.</b></p>\n    <p style=\"color:#c650eb;font-size:15px\">Also feel free to write your thoughts or suggestions in the comments below.</p>\n</body>"},{"metadata":{"trusted":true},"cell_type":"code","source":"","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}