{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:08.58292Z","iopub.execute_input":"2025-03-09T19:19:08.583249Z","iopub.status.idle":"2025-03-09T19:19:20.386573Z","shell.execute_reply.started":"2025-03-09T19:19:08.583219Z","shell.execute_reply":"2025-03-09T19:19:20.385638Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Define the data path from Kaggle Datasets\nGCS_PATH = KaggleDatasets().get_gcs_path()","metadata":{"trusted":true,"_kg_hide-input":true,"execution":{"iopub.status.busy":"2025-03-09T19:19:20.387724Z","iopub.execute_input":"2025-03-09T19:19:20.388233Z","iopub.status.idle":"2025-03-09T19:19:20.804472Z","shell.execute_reply.started":"2025-03-09T19:19:20.388181Z","shell.execute_reply":"2025-03-09T19:19:20.803546Z"}},"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:20.806324Z","iopub.execute_input":"2025-03-09T19:19:20.80657Z","iopub.status.idle":"2025-03-09T19:19:21.850936Z","shell.execute_reply.started":"2025-03-09T19:19:20.806551Z","shell.execute_reply":"2025-03-09T19:19:21.850271Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print('Monet jpg Files:', len(MONET_FILENAMES))\nprint('Photo jpg Files:', len(PHOTO_FILENAMES))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-09T19:19:21.851891Z","iopub.execute_input":"2025-03-09T19:19:21.852094Z","iopub.status.idle":"2025-03-09T19:19:21.85716Z","shell.execute_reply.started":"2025-03-09T19:19:21.852077Z","shell.execute_reply":"2025-03-09T19:19:21.856461Z"}},"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:21.85788Z","iopub.execute_input":"2025-03-09T19:19:21.858086Z","iopub.status.idle":"2025-03-09T19:19:23.770243Z","shell.execute_reply.started":"2025-03-09T19:19:21.858067Z","shell.execute_reply":"2025-03-09T19:19:23.769343Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.771127Z","iopub.execute_input":"2025-03-09T19:19:23.77146Z","iopub.status.idle":"2025-03-09T19:19:23.775798Z","shell.execute_reply.started":"2025-03-09T19:19:23.771437Z","shell.execute_reply":"2025-03-09T19:19:23.774963Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.776776Z","iopub.execute_input":"2025-03-09T19:19:23.777051Z","iopub.status.idle":"2025-03-09T19:19:23.811378Z","shell.execute_reply.started":"2025-03-09T19:19:23.777024Z","shell.execute_reply":"2025-03-09T19:19:23.810734Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.81307Z","iopub.execute_input":"2025-03-09T19:19:23.813305Z","iopub.status.idle":"2025-03-09T19:19:23.887711Z","shell.execute_reply.started":"2025-03-09T19:19:23.813286Z","shell.execute_reply":"2025-03-09T19:19:23.88706Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.88863Z","iopub.execute_input":"2025-03-09T19:19:23.88886Z","iopub.status.idle":"2025-03-09T19:19:23.902081Z","shell.execute_reply.started":"2025-03-09T19:19:23.88884Z","shell.execute_reply":"2025-03-09T19:19:23.901471Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.902924Z","iopub.execute_input":"2025-03-09T19:19:23.903222Z","iopub.status.idle":"2025-03-09T19:19:23.91171Z","shell.execute_reply.started":"2025-03-09T19:19:23.903178Z","shell.execute_reply":"2025-03-09T19:19:23.91097Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.912405Z","iopub.execute_input":"2025-03-09T19:19:23.912633Z","iopub.status.idle":"2025-03-09T19:19:23.928141Z","shell.execute_reply.started":"2025-03-09T19:19:23.912615Z","shell.execute_reply":"2025-03-09T19:19:23.927325Z"}},"outputs":[],"execution_count":null},{"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.928955Z","iopub.execute_input":"2025-03-09T19:19:23.929208Z","iopub.status.idle":"2025-03-09T19:19:23.946094Z","shell.execute_reply.started":"2025-03-09T19:19:23.929163Z","shell.execute_reply":"2025-03-09T19:19:23.945153Z"}},"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,"execution":{"iopub.status.busy":"2025-03-09T19:19:23.946919Z","iopub.execute_input":"2025-03-09T19:19:23.94715Z"}},"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}]}