{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.16","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"tpu1vmV38","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"dockerImageVersionId":31012,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# CycleGAN | Photo to Monet\nThis notebook uses the CycleGAN generative adversarial network to stylize photos in the style of Monet as described in paper [Unpaired Image-to-Image Translation using Cycle-Consistent Adversarial Networks](https://arxiv.org/pdf/1703.10593).\n\nImage-to-image translation tasks are easy for humans. Humans  have no trouble imagining the scene that Monet painted as a true landscape, or vice versa. However, for machines this task is more challenging. Often there are no labeled examples for which to train a model (for example, a photograph and the resulting landscape painting). CycleGAN proposes an adversarial network that trains a generator to fool a discriminator that is trying to detect generated images. Over time, the generator learns the appropriate style to achieve the image-to-image translation task.","metadata":{}},{"cell_type":"code","source":"import tensorflow as tf\nfrom tensorflow import keras\nfrom tensorflow.keras import layers\nimport matplotlib.pyplot as plt\nimport PIL\nimport shutil\nfrom tqdm import tqdm\nimport numpy as np\n\nAUTOTUNE = tf.data.AUTOTUNE\n\nprint(tf.__version__)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:01.686006Z","iopub.execute_input":"2025-04-17T22:26:01.686334Z","iopub.status.idle":"2025-04-17T22:26:24.777396Z","shell.execute_reply.started":"2025-04-17T22:26:01.686308Z","shell.execute_reply":"2025-04-17T22:26:24.772923Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"try:\n    tpu = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='local')\n    tf.tpu.experimental.initialize_tpu_system(tpu)\n    strategy = tf.distribute.TPUStrategy(tpu)\n    BATCH_SIZE = 16 * strategy.num_replicas_in_sync\n    print(\"✅ TPU initialized\")\nexcept Exception as e:\n    strategy = tf.distribute.get_strategy()  # fallback to CPU/GPU\n    BATCH_SIZE = 1\n    print(\"❌ TPU not found, using default strategy\")\n    print(f\"Error: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:28.090525Z","iopub.execute_input":"2025-04-17T22:26:28.091186Z","iopub.status.idle":"2025-04-17T22:26:37.129499Z","shell.execute_reply.started":"2025-04-17T22:26:28.091131Z","shell.execute_reply":"2025-04-17T22:26:37.123670Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Exploratory Data Analysis\nThe dataset contains four directories: `monet_tfrec`, `photo_tfrec`, `monet_jpg`, and `photo_jpg`, offering the same data in two formats: TFRecords and JPEG. For processing efficiency, I'll use the TFRecord files. The directories include\n- monet_tfrec - 300 Monet paintings sized 256x256 in TFRecord format\n- photo_tfrec - 7028 photos sized 256x256 in TFRecord format","metadata":{}},{"cell_type":"code","source":"monet_tf_path = '/kaggle/input/gan-getting-started/monet_tfrec/*.tfrec'\nphoto_tf_path = '/kaggle/input/gan-getting-started/photo_tfrec/*.tfrec'\n\nmonet_tf = tf.data.Dataset.list_files(monet_tf_path)\nphoto_tf = tf.data.Dataset.list_files(photo_tf_path)\n\nprint('Monet TFRecord Files: ', len(monet_tf))\nprint('Photo TFRecord Files: ', len(photo_tf))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:39.822637Z","iopub.execute_input":"2025-04-17T22:26:39.822963Z","iopub.status.idle":"2025-04-17T22:26:39.880314Z","shell.execute_reply.started":"2025-04-17T22:26:39.822934Z","shell.execute_reply":"2025-04-17T22:26:39.876505Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Data Preparation\nFollowing the [TensorFlow CycleGAN tutorial](https://www.tensorflow.org/tutorials/generative/cyclegan) and [CycleGAN paper](https://arxiv.org/abs/1703.10593) referenced therein, I'll load the datasets from TFRecords. For training data, I'll apply the tranformations to avoid overfitting including\n- random jittering - image is resized to 286 x 286 and then randomly cropped to 256 x 256.\n- random mirroring - image is randomly flipped horizontally i.e., left to right.\n\nThe data preparation code will be organized in my `ImagePreprocessor` class.","metadata":{}},{"cell_type":"code","source":"IMAGE_SIZE = [256, 256]\n\nclass ImagePreprocessor:\n    def __init__(self, training=True):\n        self.training = training\n\n    def decode(self, image):\n        img = tf.image.decode_jpeg(image, channels=3)\n        return img\n\n    def read_tfrecord(self, tfrecord):\n        format_tf = {\n            'image_name': tf.io.FixedLenFeature([], tf.string),\n            'image': tf.io.FixedLenFeature([], tf.string),\n            'target': tf.io.FixedLenFeature([], tf.string),\n        }\n        ex = tf.io.parse_single_example(tfrecord, format_tf)\n        image = self.decode(ex['image'])\n        return image\n\n    def random_crop(self, image):\n        return tf.image.random_crop(image, size=[IMAGE_SIZE[0], IMAGE_SIZE[1], 3])\n\n    def normalize(self, image):\n        image = tf.cast(image, tf.float32)\n        return (image / 127.5) - 1\n\n    def random_jitter(self, image):\n        image = tf.image.resize(image, [286, 286], method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)\n        image = self.random_crop(image)\n        image = tf.image.random_flip_left_right(image)\n        return image\n\n    def preprocess_image_train(self, image):\n        image = self.random_jitter(image)\n        return self.normalize(image)\n\n    def preprocess_image_test(self, image):\n        return self.normalize(image)\n\n    def preprocess_image(self, tfrecord):\n        image = self.read_tfrecord(tfrecord)\n        if self.training:\n            return self.preprocess_image_train(image)\n        else:\n            return self.preprocess_image_test(image)\n\n    def load_dataset(self, tf_files, batch_size=BATCH_SIZE, repeat=True):\n        dataset = tf.data.TFRecordDataset(tf_files)\n        dataset = dataset.map(lambda x: self.preprocess_image(x), num_parallel_calls=AUTOTUNE)\n        if repeat:\n            dataset = dataset.repeat()\n        dataset = dataset.shuffle(1024)\n        dataset = dataset.batch(batch_size, drop_remainder=True)\n        dataset = dataset.prefetch(AUTOTUNE)\n        return dataset\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:42.325208Z","iopub.execute_input":"2025-04-17T22:26:42.325489Z","iopub.status.idle":"2025-04-17T22:26:42.344110Z","shell.execute_reply.started":"2025-04-17T22:26:42.325465Z","shell.execute_reply":"2025-04-17T22:26:42.338676Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_preprocessor = ImagePreprocessor(training=True)\ntest_preprocessor = ImagePreprocessor(training=False)\n\nmonet_ds = train_preprocessor.load_dataset(monet_tf)\nphoto_ds = train_preprocessor.load_dataset(photo_tf)\ntest_monet_ds = test_preprocessor.load_dataset(monet_tf, batch_size=1, repeat=False)\ntest_photo_ds = test_preprocessor.load_dataset(photo_tf, batch_size=1, repeat=False)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:44.318202Z","iopub.execute_input":"2025-04-17T22:26:44.318475Z","iopub.status.idle":"2025-04-17T22:26:44.694723Z","shell.execute_reply.started":"2025-04-17T22:26:44.318452Z","shell.execute_reply":"2025-04-17T22:26:44.690431Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(\"Monet Images: \", sum(1 for _ in tf.data.TFRecordDataset(monet_tf)))\nprint('Photo Images: ', sum(1 for _ in tf.data.TFRecordDataset(photo_tf)))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:26:44.696526Z","iopub.execute_input":"2025-04-17T22:26:44.696735Z","iopub.status.idle":"2025-04-17T22:26:48.123660Z","shell.execute_reply.started":"2025-04-17T22:26:44.696714Z","shell.execute_reply":"2025-04-17T22:26:48.117224Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def show_img(ds, title):\n    plt.figure(figsize=(12,6))\n    for i, img in enumerate(ds.take(4)):\n        img = (img[0] * 127.5 + 127.5).numpy().astype(np.uint8)\n        plt.subplot(1, 4, i+1)\n        plt.imshow(img)\n        plt.axis('off')\n        plt.title(title) if i == 0 else ''\n    plt.tight_layout()\n    plt.show()\n\nshow_img(test_monet_ds, 'Monet')\nshow_img(test_photo_ds, 'Photo')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:28:07.039576Z","iopub.execute_input":"2025-04-17T22:28:07.040263Z","iopub.status.idle":"2025-04-17T22:28:07.931428Z","shell.execute_reply.started":"2025-04-17T22:28:07.040200Z","shell.execute_reply":"2025-04-17T22:28:07.925115Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Model Architecture (Analysis Plan)\nCycleGAN is an adversarial approach to translating an image from one domain (e.g., a photo) to another domain (e.g., the painting style of Monet). As described in paper [Unpaired Image-to-Image Translation using Cycle-Consistent Adversarial Networks](https://arxiv.org/pdf/1703.10593) the algorithm learns a mapping $G: X \\to Y$ (i.e., from photos to Monet) with an inverse mapping $F: Y \\to X$ (i.e., from Monet to photos) under the constraint $F(G(X)) \\approx X$ such that cycling an image through will translate from one domain back to its own domain. This is akin to what humans do when imagining the actual landscape that Monet painted, or the way Monet might paint a landscape.\n\nThe algorithm is trained by training both the mapping G and F simultaneously, and adding a cycle consistency loss, where a descriminator learns to distinguish real from generated examples in each domain. The process is shown below\n\n![img](https://www.tensorflow.org/static/tutorials/generative/images/cycle_loss.png)\n\nThe CycleGAN paper uses a modified resnet based generator. I instead implement a modified unet generator for simplicity. Inspiration for implementation provided by Amy Jang's implementation in the [CycleGAN Tutorial](https://www.kaggle.com/code/amyjang/monet-cyclegan-tutorial/notebook) notebook.","metadata":{}},{"cell_type":"markdown","source":"### Generator\nThe generator first downsamples the image by striding across the image with stride=2 such that the width and height are reduced by half with each pass. Upsampling does the opposite. The idea is to create the stepped architecture such that the image is encoded to the latent space coding and then decoded in the new domain (learned through interaction with the discriminator). Instance normalization is applied for model stability and consistent style across imagesl dropout is applied for improved model generalization.  Skip connections bypass the vanishing gradient problem by concatenating the output of a layer to multiple layers instead of only one.","metadata":{}},{"cell_type":"code","source":"OUTPUT_CHANNELS = 3\n\nclass Generator:\n        \n    def downsample(self, filters, size, apply_instancenorm=True):\n        initializer = tf.random_normal_initializer(0., 0.02)\n        result = keras.Sequential()\n        result.add(layers.Conv2D(filters, size, strides=2, padding='same',\n                                 kernel_initializer=initializer, use_bias=False))\n    \n        if apply_instancenorm:\n            result.add(layers.GroupNormalization(1)) # instance normalization \n    \n        result.add(layers.LeakyReLU())\n    \n        return result\n\n    def upsample(self, filters, size, apply_dropout=False):\n        initializer = tf.random_normal_initializer(0., 0.02)\n        result = keras.Sequential()\n        result.add(layers.Conv2DTranspose(filters, size, strides=2,\n                                          padding='same',\n                                          kernel_initializer=initializer,\n                                          use_bias=False))\n        result.add(layers.GroupNormalization(1)) # instance normalization \n    \n        if apply_dropout:\n            result.add(layers.Dropout(0.5))\n    \n        result.add(layers.ReLU())\n    \n        return result\n\n    def build(self):\n        inputs = layers.Input(shape=[IMAGE_SIZE[0], IMAGE_SIZE[1], 3])\n\n        # bs = batch size\n        down_stack = [\n            self.downsample(64, 4, apply_instancenorm=False), # (bs, 128, 128, 64)\n            self.downsample(128, 4), # (bs, 64, 64, 128)\n            self.downsample(256, 4), # (bs, 32, 32, 256)\n            self.downsample(512, 4), # (bs, 16, 16, 512)\n            self.downsample(512, 4), # (bs, 8, 8, 512)\n            self.downsample(512, 4), # (bs, 4, 4, 512)\n            self.downsample(512, 4), # (bs, 2, 2, 512)\n            self.downsample(512, 4), # (bs, 1, 1, 512)\n        ]\n    \n        up_stack = [\n            self.upsample(512, 4, apply_dropout=True), # (bs, 2, 2, 1024)\n            self.upsample(512, 4, apply_dropout=True), # (bs, 4, 4, 1024)\n            self.upsample(512, 4, apply_dropout=True), # (bs, 8, 8, 1024)\n            self.upsample(512, 4), # (bs, 16, 16, 1024)\n            self.upsample(256, 4), # (bs, 32, 32, 512)\n            self.upsample(128, 4), # (bs, 64, 64, 256)\n            self.upsample(64, 4), # (bs, 128, 128, 128)\n        ]\n    \n        initializer = tf.random_normal_initializer(0., 0.02)\n        last = layers.Conv2DTranspose(OUTPUT_CHANNELS, 4,\n                                      strides=2,\n                                      padding='same',\n                                      kernel_initializer=initializer,\n                                      activation='tanh') # (bs, 256, 256, 3)\n    \n        x = inputs\n    \n        # Downsampling through the model\n        skips = []\n        for down in down_stack:\n            x = down(x)\n            skips.append(x)\n    \n        skips = reversed(skips[:-1])\n    \n        # Upsampling and establishing the skip connections\n        for up, skip in zip(up_stack, skips):\n            x = up(x)\n            x = layers.Concatenate()([x, skip])\n    \n        x = last(x)\n    \n        return keras.Model(inputs=inputs, outputs=x, name='generator')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.579409Z","iopub.execute_input":"2025-04-17T22:02:25.579705Z","iopub.status.idle":"2025-04-17T22:02:25.599272Z","shell.execute_reply.started":"2025-04-17T22:02:25.579681Z","shell.execute_reply":"2025-04-17T22:02:25.595637Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### Discriminator\nThe discriminator learns to distinguish the difference between real and generated images. In this implementation, instead of a binary prediction of \"Real\" or \"Fake\", the discriminator outputs a smaller 2D image with higher pixel values indicating a real classification and lower values indicating a fake classification.","metadata":{}},{"cell_type":"code","source":"class Discriminator:\n\n    def downsample(self, filters, size, apply_instancenorm=True):\n        initializer = tf.random_normal_initializer(0., 0.02)\n        result = keras.Sequential()\n        result.add(layers.Conv2D(filters, size, strides=2, padding='same',\n                                 kernel_initializer=initializer, use_bias=False))\n    \n        if apply_instancenorm:\n            result.add(layers.GroupNormalization(1)) # instance normalization \n    \n        result.add(layers.LeakyReLU())\n    \n        return result\n\n    def build(self):    \n        inp = layers.Input(shape=[IMAGE_SIZE[0], IMAGE_SIZE[1], 3], name='input_image')\n        x = inp\n    \n        ds_stack = [\n            self.downsample(64, 4, apply_instancenorm=False), # batch, 128, 128, 64\n            self.downsample(128, 4), # batch, 64, 64, 128\n            self.downsample(256, 4), # batch, 32, 32, 256\n        ]\n    \n        for ds in ds_stack:\n            x = ds(x)\n        \n        result = keras.Sequential([\n            layers.Conv2D(512, 4, strides=1, padding='same'), # batch, 32, 32, 512\n            layers.GroupNormalization(1), # instance normalization\n            layers.LeakyReLU(),\n            layers.Conv2D(1, 4, strides=1, padding='same'), # batch, 32, 32, 1 <-- our patch map (Real/Fake)\n        ])\n    \n        x = result(x)\n    \n        return tf.keras.Model(inputs=inp, outputs=x, name='discriminator')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.602525Z","iopub.execute_input":"2025-04-17T22:02:25.602744Z","iopub.status.idle":"2025-04-17T22:02:25.622465Z","shell.execute_reply.started":"2025-04-17T22:02:25.602723Z","shell.execute_reply":"2025-04-17T22:02:25.618036Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Now I'll build the Generator and Discriminator. `strategy.scope()` ensures TPU can be used if available.","metadata":{}},{"cell_type":"code","source":"with strategy.scope():\n    monet_generator = Generator().build()\n    photo_generator = Generator().build()\n\n    monet_discriminator = Discriminator().build()\n    photo_discriminator = Discriminator().build()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.624469Z","iopub.execute_input":"2025-04-17T22:02:25.624717Z","iopub.status.idle":"2025-04-17T22:02:25.654229Z","shell.execute_reply.started":"2025-04-17T22:02:25.624695Z","shell.execute_reply":"2025-04-17T22:02:25.649921Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"monet_generator.summary()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.656003Z","iopub.status.idle":"2025-04-17T22:02:25.656786Z","shell.execute_reply.started":"2025-04-17T22:02:25.656155Z","shell.execute_reply":"2025-04-17T22:02:25.656168Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"monet_discriminator.summary()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.657679Z","iopub.status.idle":"2025-04-17T22:02:25.659360Z","shell.execute_reply.started":"2025-04-17T22:02:25.657854Z","shell.execute_reply":"2025-04-17T22:02:25.657868Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### CycleGAN Model\nSubclassing a `keras` model, I'll build the custom CycleGAN model.","metadata":{}},{"cell_type":"code","source":"class CycleGan(keras.Model):\n    def __init__(\n        self,\n        monet_generator,\n        photo_generator,\n        monet_discriminator,\n        photo_discriminator,\n        lambda_cycle=10,\n    ):\n        super(CycleGan, 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_cycle = lambda_cycle\n        \n    def compile(\n        self,\n        m_gen_optimizer,\n        p_gen_optimizer,\n        m_disc_optimizer,\n        p_disc_optimizer,\n        gen_loss_fn,\n        disc_loss_fn,\n        cycle_loss_fn,\n        identity_loss_fn\n    ):\n        super(CycleGan, 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_fn = gen_loss_fn\n        self.disc_loss_fn = disc_loss_fn\n        self.cycle_loss_fn = cycle_loss_fn\n        self.identity_loss_fn = identity_loss_fn\n        \n    def train_step(self, batch_data):\n        real_monet, real_photo = batch_data\n        \n        with tf.GradientTape(persistent=True) as tape:\n            # photo to monet back to photo\n            fake_monet = self.m_gen(real_photo, training=True)\n            cycled_photo = self.p_gen(fake_monet, training=True)\n\n            # monet to photo back to monet\n            fake_photo = self.p_gen(real_monet, training=True)\n            cycled_monet = self.m_gen(fake_photo, training=True)\n\n            # generating itself\n            same_monet = self.m_gen(real_monet, training=True)\n            same_photo = self.p_gen(real_photo, training=True)\n\n            # discriminator used to check, inputing real images\n            disc_real_monet = self.m_disc(real_monet, training=True)\n            disc_real_photo = self.p_disc(real_photo, training=True)\n\n            # discriminator used to check, inputing fake images\n            disc_fake_monet = self.m_disc(fake_monet, training=True)\n            disc_fake_photo = self.p_disc(fake_photo, training=True)\n\n            # evaluates generator loss\n            monet_gen_loss = self.gen_loss_fn(disc_fake_monet)\n            photo_gen_loss = self.gen_loss_fn(disc_fake_photo)\n\n            # evaluates total cycle consistency loss\n            total_cycle_loss = (\n                self.cycle_loss_fn(real_monet, cycled_monet, self.lambda_cycle) + \n                self.cycle_loss_fn(real_photo, cycled_photo, self.lambda_cycle)\n            )\n\n            # evaluates total generator loss\n            total_monet_gen_loss = (\n                monet_gen_loss + \n                total_cycle_loss + \n                self.identity_loss_fn(real_monet, same_monet, self.lambda_cycle)\n            )\n            total_photo_gen_loss = (\n                photo_gen_loss + \n                total_cycle_loss + \n                self.identity_loss_fn(real_photo, same_photo, self.lambda_cycle)\n            )\n\n            # evaluates discriminator loss\n            monet_disc_loss = self.disc_loss_fn(disc_real_monet, disc_fake_monet)\n            photo_disc_loss = self.disc_loss_fn(disc_real_photo, disc_fake_photo)\n\n        # Calculate the gradients for generator and discriminator\n        monet_generator_gradients = tape.gradient(total_monet_gen_loss,\n                                                  self.m_gen.trainable_variables)\n        photo_generator_gradients = tape.gradient(total_photo_gen_loss,\n                                                  self.p_gen.trainable_variables)\n\n        monet_discriminator_gradients = tape.gradient(monet_disc_loss,\n                                                      self.m_disc.trainable_variables)\n        photo_discriminator_gradients = tape.gradient(photo_disc_loss,\n                                                      self.p_disc.trainable_variables)\n\n        # Apply the gradients to the optimizer\n        self.m_gen_optimizer.apply_gradients(zip(monet_generator_gradients,\n                                                 self.m_gen.trainable_variables))\n\n        self.p_gen_optimizer.apply_gradients(zip(photo_generator_gradients,\n                                                 self.p_gen.trainable_variables))\n\n        self.m_disc_optimizer.apply_gradients(zip(monet_discriminator_gradients,\n                                                  self.m_disc.trainable_variables))\n\n        self.p_disc_optimizer.apply_gradients(zip(photo_discriminator_gradients,\n                                                  self.p_disc.trainable_variables))\n        \n        return {\n            \"monet_gen_loss\": total_monet_gen_loss,\n            \"photo_gen_loss\": total_photo_gen_loss,\n            \"monet_disc_loss\": monet_disc_loss,\n            \"photo_disc_loss\": photo_disc_loss\n        }\n\n    def call(self, inputs, training=False):\n        # Provide a simple forward pass — used for model.build()\n        return self.m_gen(inputs, training=training)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.661731Z","iopub.status.idle":"2025-04-17T22:02:25.662378Z","shell.execute_reply.started":"2025-04-17T22:02:25.661888Z","shell.execute_reply":"2025-04-17T22:02:25.661903Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#### Loss Function\nThe authors propose cycle consistency loss as the loss function for updating weights in the model. This simply means that when translating from one domain to another and back again, the output should be similar to the input $F(G(X) \\approx X$ and $G(F(Y)) \\approx Y$.\n\nRecall the loss will be pixel-by-pixel rather than binary. The perfect discriminator will output all 1s for real images and all 0s for fake images. The discriminator loss outputs the average of the real and generated loss. Meanwhile, the generator is trying to fool the discriminator into returning all 1s for a generated image. Cycle consistency loss averages the difference between the real and cycled image (e.g., $X \\to Y \\to X$). Identity loss compares the photo created by its own generator with itself, ideally these should be equivalent (all 1s).","metadata":{}},{"cell_type":"code","source":"with strategy.scope():\n    def discriminator_loss(real, generated):\n        real_loss = tf.keras.losses.BinaryCrossentropy(\n            from_logits=True, \n            reduction=tf.keras.losses.Reduction.NONE\n        )(tf.ones_like(real), real)\n\n        generated_loss = tf.keras.losses.BinaryCrossentropy(\n            from_logits=True, \n            reduction=tf.keras.losses.Reduction.NONE\n        )(tf.zeros_like(generated), generated)\n\n        total_disc_loss = real_loss + generated_loss\n\n        return total_disc_loss * 0.5\n\n    def generator_loss(generated):\n        return tf.keras.losses.BinaryCrossentropy(\n            from_logits=True, \n            reduction=tf.keras.losses.Reduction.NONE\n        )(tf.ones_like(generated), generated)\n\n    def calc_cycle_loss(real_image, cycled_image, LAMBDA):\n        loss1 = tf.reduce_mean(tf.abs(real_image - cycled_image))\n\n        return LAMBDA * loss1\n\n    def identity_loss(real_image, same_image, LAMBDA):\n        loss = tf.reduce_mean(tf.abs(real_image - same_image))\n        return LAMBDA * 0.5 * loss","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.663519Z","iopub.status.idle":"2025-04-17T22:02:25.665793Z","shell.execute_reply.started":"2025-04-17T22:02:25.664434Z","shell.execute_reply":"2025-04-17T22:02:25.664448Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#### Model training\nThe authors of the CycleGAN paper use the Adam optimizer with learning rate of `2e-4`, so I will follow suit to fit the model.","metadata":{}},{"cell_type":"code","source":"with strategy.scope():\n    monet_generator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)\n    photo_generator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)\n\n    monet_discriminator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)\n    photo_discriminator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)\n\n    cycle_gan_model = CycleGan(\n        monet_generator, \n        photo_generator, \n        monet_discriminator, \n        photo_discriminator\n    )\n    \n    cycle_gan_model.compile(\n        m_gen_optimizer = monet_generator_optimizer,\n        p_gen_optimizer = photo_generator_optimizer,\n        m_disc_optimizer = monet_discriminator_optimizer,\n        p_disc_optimizer = photo_discriminator_optimizer,\n        gen_loss_fn = generator_loss,\n        disc_loss_fn = discriminator_loss,\n        cycle_loss_fn = calc_cycle_loss,\n        identity_loss_fn = identity_loss\n    )","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.668625Z","iopub.status.idle":"2025-04-17T22:02:25.670487Z","shell.execute_reply.started":"2025-04-17T22:02:25.669787Z","shell.execute_reply":"2025-04-17T22:02:25.669802Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"history = cycle_gan_model.fit(\n    tf.data.Dataset.zip((monet_ds, photo_ds)),\n    epochs=200,\n    steps_per_epoch=300 // BATCH_SIZE\n)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.671824Z","iopub.status.idle":"2025-04-17T22:02:25.672829Z","shell.execute_reply.started":"2025-04-17T22:02:25.671991Z","shell.execute_reply":"2025-04-17T22:02:25.672004Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Results & Analysis","metadata":{}},{"cell_type":"code","source":"fig, ax = plt.subplots(2, 4, figsize=(12, 6))\n\nfor i, img in enumerate(photo_ds.take(4)):\n    prediction = monet_generator(img, training=False)[0].numpy()\n    prediction = (prediction * 127.5 + 127.5).astype(np.uint8)\n    img = (img[0] * 127.5 + 127.5).numpy().astype(np.uint8)\n\n    ax[0, i].imshow(img)\n    ax[1, i].imshow(prediction)\n    ax[0, i].set_title(\"Input Photo\")\n    ax[1, i].set_title(\"Monet-esque\")\n    ax[0, i].axis(\"off\")\n    ax[1, i].axis(\"off\")\n\nplt.tight_layout()\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.675793Z","iopub.status.idle":"2025-04-17T22:02:25.677350Z","shell.execute_reply.started":"2025-04-17T22:02:25.676047Z","shell.execute_reply":"2025-04-17T22:02:25.676061Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for name, losses in history.history.items():\n    scalar_losses = [tf.reduce_mean(l).numpy() for l in losses]\n    plt.plot(range(1, len(scalar_losses)+1), scalar_losses, label=name)\n\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.title('CycleGAN Loss per Epoch')\nplt.legend()\nplt.grid(True)\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.678483Z","iopub.status.idle":"2025-04-17T22:02:25.678937Z","shell.execute_reply.started":"2025-04-17T22:02:25.678637Z","shell.execute_reply":"2025-04-17T22:02:25.678650Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Submission\nAs recommended, I'll create an images directory and zip all generated images from the photo directory for submission. ","metadata":{}},{"cell_type":"code","source":"! mkdir ../images","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.679957Z","iopub.status.idle":"2025-04-17T22:02:25.681859Z","shell.execute_reply.started":"2025-04-17T22:02:25.680110Z","shell.execute_reply":"2025-04-17T22:02:25.680125Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for i, img in enumerate(tqdm(test_photo_ds, total=7028), 1):\n    prediction = monet_generator(img, training=False)[0].numpy()\n    prediction = (prediction * 127.5 + 127.5).astype(np.uint8)\n    im = PIL.Image.fromarray(prediction)\n    im.save(\"../images/\" + str(i) + \".jpg\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.683721Z","iopub.status.idle":"2025-04-17T22:02:25.684467Z","shell.execute_reply.started":"2025-04-17T22:02:25.683866Z","shell.execute_reply":"2025-04-17T22:02:25.683879Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"shutil.make_archive(\"/kaggle/working/images\", 'zip', \"/kaggle/images\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-17T22:02:25.685197Z","iopub.status.idle":"2025-04-17T22:02:25.685996Z","shell.execute_reply.started":"2025-04-17T22:02:25.685352Z","shell.execute_reply":"2025-04-17T22:02:25.685366Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## References\n- https://www.tensorflow.org/tutorials/generative/cyclegan\n- https://colab.research.google.com/github/tensorflow/datasets/blob/master/docs/overview.ipynb#scrollTo=o-cuwvVbeb43\n- https://www.kaggle.com/code/amyjang/monet-cyclegan-tutorial/notebook","metadata":{}},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}