{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport PIL\nimport shutil\nimport numpy as np\nimport tensorflow as tf\nfrom tensorflow import keras\nfrom tensorflow.keras.layers import Conv2D, Conv2DTranspose, BatchNormalization, LeakyReLU, ReLU, Input\nimport matplotlib.pyplot as plt\nfrom kaggle_datasets import KaggleDatasets","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"The following code block is used to retrieve image file paths from a dataset stored in Google Cloud Storage (GCS), specifically from Kaggle Datasets, and count the number of images in two categories: **Monet paintings** and **Photos**.","metadata":{}},{"cell_type":"code","source":"# Define the data path from Kaggle Datasets\nGCS_PATH = KaggleDatasets().get_gcs_path()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Retrieve image file paths\nMONET_FILENAMES = tf.io.gfile.glob(str(GCS_PATH + '/monet_jpg/*.jpg'))\nPHOTO_FILENAMES = tf.io.gfile.glob(str(GCS_PATH + '/photo_jpg/*.jpg'))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"- `tf.io.gfile.glob(pattern)` is a TensorFlow function that finds all files matching a given pattern.\n\n- `GCS_PATH + '/monet_jpg/*.jpg` specifies that it should find all `.jpg` files inside the `monet_jpg` folder.\n\n- `GCS_PATH + '/photo_jpg/*.jpg` does the same for the `photo_jpg` folder.","metadata":{}},{"cell_type":"code","source":"print('Monet jpg Files:', len(MONET_FILENAMES))\nprint('Photo jpg Files:', len(PHOTO_FILENAMES))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Load and display an example Monet image & Photo image\nif MONET_FILENAMES and PHOTO_FILENAMES:\n    example_monet_path = MONET_FILENAMES[0]\n    monet_image = tf.io.read_file(example_monet_path)\n    monet_image = tf.image.decode_jpeg(monet_image, channels=3)\n    print(\"Monet image shape:\", monet_image.shape)\n\n    example_photo_path = PHOTO_FILENAMES[0]\n    photo_image = tf.io.read_file(example_photo_path)\n    photo_image = tf.image.decode_jpeg(photo_image, channels=3)\n    print(\"Photo image shape:\", photo_image.shape)\n    \n    # Display the images side by side\n    plt.figure(figsize=(10, 5)) \n\n    # Show Monet image in the first subplot\n    plt.subplot(1, 2, 2)\n    plt.imshow(photo_image.numpy())\n    plt.title(\"Example Photo Image\")\n\n    # Show Photo image in the second subplot\n    plt.subplot(1, 2, 1)\n    plt.imshow(monet_image.numpy())\n    plt.title(\"Example Monet Image\")\n\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"According to the description given in the data section, all images are of size 256x256 in `JPEG` format. So there is no need to resize images. \nSince these images are in RGB format, we set the number of channels to 3. Additionally, we scale the pixel values to the range [-1, 1] for proper normalization.","metadata":{}},{"cell_type":"code","source":"IMAGE_SIZE = [256, 256]\n# This function is responsible for loading, decoding\n# normalizing, and reshaping an image from a given file path.\ndef decode_image(image_path):\n    image = tf.io.read_file(image_path) # loads image as binary string from image path\n    image = tf.image.decode_jpeg(image, channels=3) # decodes jpeg image into a TF tensor with 3 color channels\n    image = (tf.cast(image, tf.float32)/127.5) - 1 # image tensor to tf.float32; Scaling pixel values\n    image = tf.reshape(image, [*IMAGE_SIZE, 3]) # Ensures shape of (256, 256, 3)\n    return image","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"The following code converts lists of image file paths (stored in MONET_FILENAMES and PHOTO_FILENAMES) into TensorFlow datasets which can be efficiently processed for ML tasks.\n\n`tf.data.Dataset.from_tensor_slices()` creates a TensorFlow dataset from a list (or tensor).","metadata":{}},{"cell_type":"code","source":"monet_ds = tf.data.Dataset.from_tensor_slices(MONET_FILENAMES)\nphoto_ds = tf.data.Dataset.from_tensor_slices(PHOTO_FILENAMES)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"We will now apply `decode_image` preprocessing function to each file path in the dataset using TF's `map()` function. We will also apply parallel processing to speed up data loading.","metadata":{}},{"cell_type":"code","source":"monet_ds = monet_ds.map(decode_image, num_parallel_calls=tf.data.AUTOTUNE)\nphoto_ds = photo_ds.map(decode_image, num_parallel_calls=tf.data.AUTOTUNE)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Following block optimizes the dataset pipeline for training by applying shuffling, batching, and prefetching to improve performance.","metadata":{}},{"cell_type":"code","source":"BATCH_SIZE = 1  # Each batch contains one image\n# monet_ds = monet_ds.shuffle(buffer_size=len(MONET_FILENAMES)) # setting buffer size to length of filenames for perfect shuffling\n# photo_ds = photo_ds.shuffle(buffer_size=len(PHOTO_FILENAMES)) # same as above\n\n# monet_ds = monet_ds.prefetch(buffer_size=tf.data.AUTOTUNE) # Preloads batches of data into memory before they are needed during training\n# photo_ds = photo_ds.prefetch(buffer_size=tf.data.AUTOTUNE) # same as above\n\nmonet_ds = monet_ds.shuffle(len(MONET_FILENAMES)).batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)\nphoto_ds = photo_ds.shuffle(len(PHOTO_FILENAMES)).batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"In the following steps we are going to define Generator and Discriminatior models used in GANs specifically for image-to-image translation tasks like CycleGAN.","metadata":{}},{"cell_type":"code","source":"# Generator\ndef build_generator():\n    inputs = Input(shape=(256, 256, 3))\n    \n    # Downsampling layers\n    x = Conv2D(32, (7, 7), padding='same')(inputs)\n    x = BatchNormalization()(x)\n    x = ReLU()(x)\n    \n    for _ in range(1):\n        x = Conv2D(64, (3, 3), strides=2, padding='same')(x)\n        x = BatchNormalization()(x)\n        x = ReLU()(x)\n\n    # Residual blocks\n    # Adds residual connections to prevent information loss during deep learning.\n    for _ in range(2):\n        res = Conv2D(64, (3, 3), padding='same')(x)\n        res = BatchNormalization()(res)\n        res = ReLU()(res)\n        res = Conv2D(64, (3, 3), padding='same')(res)\n        res = BatchNormalization()(res)\n        x = tf.keras.layers.Add()([x, res])\n    \n    # Upsampling layers\n    for _ in range(1):\n        x = Conv2DTranspose(64, (3, 3), strides=2, padding='same')(x)\n        x = BatchNormalization()(x)\n        x = ReLU()(x)\n    \n    x = Conv2D(3, (7, 7), padding='same', activation='tanh')(x)\n    return keras.Model(inputs, x, name=\"Generator\")\n\n# Discriminator\ndef build_discriminator():\n    inputs = Input(shape=(256, 256, 3))\n\n    # Downsampling\n    x = Conv2D(32, (4, 4), strides=2, padding='same')(inputs)\n    x = LeakyReLU(negative_slope=0.2)(x)\n\n    \n    for filters in [64, 128]:\n        x = Conv2D(filters, (4, 4), strides=2, padding='same')(x)\n        x = BatchNormalization()(x)\n        x = LeakyReLU(negative_slope=0.2)(x)\n    \n    x = Conv2D(1, (4, 4), padding='same')(x)\n    return keras.Model(inputs, x, name=\"Discriminator\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### **Summary of Above Code**\n**Generator (build_generator())**\n- Transforms an image (e.g., a real photo) into a stylized image (e.g., a Monet painting).\n- Uses downsampling → residual blocks → upsampling.\n- Final output activation is tanh to match normalized image pixel values.\n\n\n**Discriminator (build_discriminator())**\n- Determines whether an image is real or fake.\n- Uses multiple convolutional layers to extract features.\n- Ends with a single-channel feature map for patch-based classification.","metadata":{}},{"cell_type":"markdown","source":"Now we will define loss functions for training a GAN which is used for image-to-image translation. In this case for converting a photo into a Monet Painting and vice versa.","metadata":{}},{"cell_type":"code","source":"# Loss functions\nloss_object = keras.losses.BinaryCrossentropy(from_logits=True)\n\ndef generator_loss(fake_output):\n    return loss_object(tf.ones_like(fake_output), fake_output)\n\ndef discriminator_loss(real_output, fake_output):\n    real_loss = loss_object(tf.ones_like(real_output), real_output)\n    fake_loss = loss_object(tf.zeros_like(fake_output), fake_output)\n    return real_loss + fake_loss\n\n# Cycle consistency loss\ndef cycle_loss(real_image, cycled_image):\n    return tf.reduce_mean(tf.abs(real_image - cycled_image))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"- Generator Loss: Ensures the generator fools the discriminator by making fake images realistic.\n- Discriminator Loss: Ensures the discriminator correctly classifies real vs. fake images.\n- Cycle Loss (for CycleGAN): Ensures the translated image can be reconstructed to match the original.","metadata":{}},{"cell_type":"code","source":"class CycleGAN:\n    def __init__(self, learning_rate=0.0002, beta_1=0.5):\n        # Initialize models\n        self.models = {\n            \"generator_G\": build_generator(),  # Generator: Converts Real photo → Monet-style\n            \"generator_F\": build_generator(),  # Generator: Converts Monet-style → Real photo\n            \"discriminator_X\": build_discriminator(),  # Discriminator: Checks if photo is real\n            \"discriminator_Y\": build_discriminator()  # Discriminator: Checks if Monet painting is real\n        }\n        \n        # Initialize optimizers\n        self.optimizers = {\n            \"gen_G\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1), # Optimizer for Generator_G\n            \"gen_F\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1), # Optimizer for Generator_F\n            \"disc_X\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1), # Optimizer for Discriminator_X\n            \"disc_Y\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1) # Optimizer for Discriminator_Y\n        }\n\n    # Training one batch\n    def train_step(self, real_X, real_Y):\n        with tf.GradientTape(persistent=True) as tape:\n            #Generate Fake and Cycled Images\n            fake_Y = self.models[\"generator_G\"](real_X, training=True)\n            cycled_X = self.models[\"generator_F\"](fake_Y, training=True)\n            \n            fake_X = self.models[\"generator_F\"](real_Y, training=True)\n            cycled_Y = self.models[\"generator_G\"](fake_X, training=True)\n\n            # Compute Discriminator Outputs\n            real_output_X = self.models[\"discriminator_X\"](real_X, training=True)\n            fake_output_X = self.models[\"discriminator_X\"](fake_X, training=True)\n            \n            real_output_Y = self.models[\"discriminator_Y\"](real_Y, training=True)\n            fake_output_Y = self.models[\"discriminator_Y\"](fake_Y, training=True)\n\n            # Compute Losses\n            gen_G_loss = generator_loss(fake_output_Y) + cycle_loss(real_X, cycled_X)\n            gen_F_loss = generator_loss(fake_output_X) + cycle_loss(real_Y, cycled_Y)\n            \n            disc_X_loss = discriminator_loss(real_output_X, fake_output_X)\n            disc_Y_loss = discriminator_loss(real_output_Y, fake_output_Y)\n\n        # Computed gradients of losses w.r.t. model parameters\n        gradients = {\n            \"gen_G\": tape.gradient(gen_G_loss, self.models[\"generator_G\"].trainable_variables),\n            \"gen_F\": tape.gradient(gen_F_loss, self.models[\"generator_F\"].trainable_variables),\n            \"disc_X\": tape.gradient(disc_X_loss, self.models[\"discriminator_X\"].trainable_variables),\n            \"disc_Y\": tape.gradient(disc_Y_loss, self.models[\"discriminator_Y\"].trainable_variables)\n        }\n\n        # Applies gradient updates using the respective Adam optimizer        \n        for key, grads in gradients.items():\n            valid_grads = [(grad, var) for grad, var in zip(grads, self.models[key.replace(\"gen_\", \"generator_\").replace(\"disc_\", \"discriminator_\")].trainable_variables) if grad is not None]\n            self.optimizers[key].apply_gradients(valid_grads)\n        \n        return gen_G_loss, gen_F_loss, disc_X_loss, disc_Y_loss\n\n    # Trains CycleGAN model over multiple epochs\n    def train(self, dataset_X, dataset_Y, epochs):\n        for epoch in range(epochs):\n            for real_X, real_Y in zip(dataset_X, dataset_Y):\n                gen_G_loss, gen_F_loss, disc_X_loss, disc_Y_loss = self.train_step(real_X, real_Y)\n            print(f\"Epoch {epoch+1}, Gen G Loss: {gen_G_loss.numpy()}, Gen F Loss: {gen_F_loss.numpy()}, Disc X Loss: {disc_X_loss.numpy()}, Disc Y Loss: {disc_Y_loss.numpy()}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Start Training with appropriate parameters\ncycle_gan = CycleGAN(learning_rate=0.0002, beta_1=0.5)\ncycle_gan.train(photo_ds, monet_ds, epochs=20)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"! mkdir ../images\n\n# Saving for submission\n\ni = 1\nfor img in photo_ds:\n    prediction = cycle_gan.models[\"generator_G\"](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\")\n    i += 1\n\nshutil.make_archive(\"/kaggle/working/images\", 'zip', \"/kaggle/images\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}