{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","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"}],"dockerImageVersionId":30886,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import tensorflow as tf\nfrom tensorflow import keras\nfrom tensorflow.keras.layers import Conv2D, Conv2DTranspose, BatchNormalization, LeakyReLU, ReLU, Input\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport os\nimport PIL\nimport shutil\nfrom kaggle_datasets import KaggleDatasets","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Define the data path\nGCS_PATH = KaggleDatasets().get_gcs_path()\n# Get the list of Monet and Photo image files\nMONET_FILENAMES = tf.io.gfile.glob(str(GCS_PATH + '/monet_jpg/*.jpg'))\nPHOTO_FILENAMES = tf.io.gfile.glob(str(GCS_PATH + '/photo_jpg/*.jpg'))\n\n# Print the number of files found\nprint('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\n# if 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#     print(\"File path:\", example_monet_path)\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#     print(\"File path:\", example_photo_path)\n    \n#     # Display the images side by side\n#     plt.figure(figsize=(10, 5)) # Set the figure size\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#     # plt.axis('off')\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#     # plt.axis('off')\n\n#     plt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# All the images for the competition are already sized to 256x256. As these images are RGB images, \n# set the channel to 3. Additionally, we need to scale the images to a [-1, 1] scale.\nIMAGE_SIZE = [256, 256]\ndef decode_image(image_path):\n    image = tf.io.read_file(image_path)\n    image = tf.image.decode_jpeg(image, channels=3)\n    image = (tf.cast(image, tf.float32) / 127.5) - 1\n    image = tf.reshape(image, [*IMAGE_SIZE, 3])\n    return image\n\n# Convert the list of file paths to a TensorFlow dataset where each element is a file path string\nmonet_ds = tf.data.Dataset.from_tensor_slices(MONET_FILENAMES)\nphoto_ds = tf.data.Dataset.from_tensor_slices(PHOTO_FILENAMES)\n\n# Apply the preprocessing function with parallelism for training or further processing\nmonet_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)\n\n# Optimize dataset performance for training (batching, shuffling, prefetching)\nBATCH_SIZE = 1  # Define an appropriate batch size\n\nmonet_ds = monet_ds.shuffle(buffer_size=len(MONET_FILENAMES)).batch(BATCH_SIZE).prefetch(buffer_size=tf.data.AUTOTUNE)\nphoto_ds = photo_ds.shuffle(buffer_size=len(PHOTO_FILENAMES)).batch(BATCH_SIZE).prefetch(buffer_size=tf.data.AUTOTUNE)\n\n# Print dataset summary\nprint(f\"Monet dataset batches: {len(MONET_FILENAMES) // BATCH_SIZE}\")\nprint(f\"Photo dataset batches: {len(PHOTO_FILENAMES) // BATCH_SIZE}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Define 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    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# Define Discriminator\ndef build_discriminator():\n    inputs = Input(shape=(256, 256, 3))\n    x = Conv2D(32, (4, 4), strides=2, padding='same')(inputs)\n    x = LeakyReLU(negative_slope=0.2)(x)\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":"code","source":"# Define 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# Define 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":"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: Real photo → Monet-style\n            \"generator_F\": build_generator(),  # Generator: 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),\n            \"gen_F\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1),\n            \"disc_X\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1),\n            \"disc_Y\": keras.optimizers.Adam(learning_rate=learning_rate, beta_1=beta_1)\n        }\n    \n    def train_step(self, real_X, real_Y):\n        with tf.GradientTape(persistent=True) as tape:\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            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            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        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        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    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\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":"# _, ax = plt.subplots(3, 2, figsize=(12, 12))\n# for i, img in enumerate(photo_ds.take(3)):\n#     prediction = cycle_gan.models[\"generator_G\"](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[i, 0].imshow(img)\n#     ax[i, 1].imshow(prediction)\n#     ax[i, 0].set_title(\"Input Photo\")\n#     ax[i, 1].set_title(\"Monet-esque\")\n#     ax[i, 0].axis(\"off\")\n#     ax[i, 1].axis(\"off\")\n# plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"! mkdir ../images\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}]}