{"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"}],"dockerImageVersionId":30823,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import tensorflow as tf\nfrom tensorflow.keras import layers, models\n\nimport os\nimport pathlib\nimport time\nimport datetime\n\nfrom matplotlib import pyplot as plt\nfrom IPython import display\n\nimport PIL\n\nAUTOTUNE = tf.data.experimental.AUTOTUNE\n\ntf.__version__","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#load dataset \nTRAIN_FILES = []\nTEST_FILES = []\nfor dirname, _, filenames in os.walk('../input/gan-getting-started/monet_tfrec'):\n    for filename in filenames:\n         TRAIN_FILES.append(os.path.join(dirname, filename))\n            \nfor dirname, _, filenames in os.walk('../input/gan-getting-started/photo_tfrec'):\n    for filename in filenames:\n         TEST_FILES.append(os.path.join(dirname, filename))\n\nTRAIN_FILES","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"raw_train_ds = tf.data.TFRecordDataset(TRAIN_FILES)\nraw_test_ds = tf.data.TFRecordDataset(TEST_FILES)\n\nraw_train_ds","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import numpy as np\n\nfor raw_record in raw_train_ds.take(1):\n    example = tf.train.Example()\n    example.ParseFromString(raw_record.numpy())\n\n    result = {}\n\n# example.features.feature is the dictionary\nfor key, feature in example.features.feature.items():\n    # The values are the Feature objects which contain a `kind` which contains:\n    # one of three fields: bytes_list, float_list, int64_list\n    kind = feature.WhichOneof('kind')\n    result[key] = np.array(getattr(feature, kind).value).dtype\n\nresult","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Create a description of the features.\nfeature_description = {\n    'target': tf.io.FixedLenFeature([], tf.string, default_value=''),\n    'image_name': tf.io.FixedLenFeature([], tf.string, default_value=''),\n    'image': tf.io.FixedLenFeature([], tf.string, default_value=''),\n}\n\ndef _parse_image(example_proto):\n  # Parse the input tf.train.Example proto using the dictionary above.\n  return tf.io.parse_single_example(example_proto, feature_description)\n\ntrain_ds = raw_train_ds.map(_parse_image)\ntest_ds = raw_test_ds.map(_parse_image)\n\ntrain_ds","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for image_features in train_ds.take(1):\n    image = image_features['image'].numpy()\n    \n    image_name = image_features['image_name'].numpy()\n    filename = image_name.decode('utf-8')\n    \n    target = image_features['target'].numpy()\n    label = target.decode('utf-8')\n    \n    # Display the image\n    display.display(display.Image(data=image))\n\n# Decode an image file to a uint8 tensor\nsample_image = tf.io.decode_jpeg(image)\nsample_image.shape\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Preprocessing\n#Each image is 256x256 in size\nIMAGE_SIZE = 256\nCHANNELS = 3\nEPOCHS = 30\n\ndef random_crop(image):\n    cropped_image = tf.image.random_crop(image, size=[IMAGE_SIZE, IMAGE_SIZE, CHANNELS])\n\n    return cropped_image\n\n# normalizing the images to [-1, 1]\ndef normalize(image):\n    image = tf.cast(image, tf.float32)\n    image = (image / 127.5) - 1\n    return image\n\ndef random_jitter(image):\n    # resizing to 286 x 286 x 3\n    image = tf.image.resize(image, [286, 286], method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)\n\n    # randomly cropping to 256 x 256 x 3\n    image = random_crop(image)\n\n    # random mirroring\n    image = tf.image.random_flip_left_right(image)\n\n    return image","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# The facade training set consist of 300 images\nBUFFER_SIZE = 300\nBATCH_SIZE = 1\n\ndef preprocess_image_train(batch):\n    image = batch['image']\n    image = tf.io.decode_jpeg(image, channels=CHANNELS)\n    image = random_jitter(image)\n    image = normalize(image)\n    return image\n\ndef preprocess_image_test(batch):\n    image = batch['image']\n    image = tf.io.decode_jpeg(image, channels=CHANNELS)\n    image = normalize(image)\n    return image\n\ndef create_dataset_train(dataset):\n    ds = dataset.map(lambda batch: preprocess_image_train(batch), num_parallel_calls=AUTOTUNE)\n    ds = ds.cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE)\n    return ds.prefetch(AUTOTUNE)\n\ndef create_dataset_test(dataset):\n    ds = dataset.map(lambda batch: preprocess_image_test(batch), num_parallel_calls=AUTOTUNE)\n    ds = ds.cache().shuffle(BUFFER_SIZE).batch(BATCH_SIZE)\n    return ds.prefetch(AUTOTUNE)\n\ntrain_ds = create_dataset_train(train_ds)\ntest_ds = create_dataset_test(test_ds)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"plt.subplot(121)\nplt.title('Monet art')\nplt.imshow(sample_image)\n\nplt.subplot(122)\nplt.title('Monet art with random jitter')\nplt.imshow(random_jitter(sample_image))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Build the CycleGAN Model\n#Build the downsampler(encoder)\ndef downsample(filters, size, apply_norm=True):\n    initializer = tf.random_normal_initializer(0., 0.02)\n    result = models.Sequential()\n    result.add(layers.Conv2D(filters, size, strides=2, padding='same',\n                             kernel_initializer=initializer, use_bias=False))\n    if apply_norm:\n        result.add(layers.GroupNormalization(groups=-1))\n    result.add(layers.LeakyReLU())\n    return result","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Build the upsampler(encoder)\ndef upsample(filters, size, apply_dropout=False):\n    initializer = tf.random_normal_initializer(0., 0.02)\n    result = models.Sequential()\n    result.add(layers.Conv2DTranspose(filters, size, strides=2, padding='same',\n                                      kernel_initializer=initializer, use_bias=False))\n    result.add(layers.GroupNormalization(groups=-1))\n    if apply_dropout:\n        result.add(layers.Dropout(0.5))\n    result.add(layers.ReLU())\n    return result","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Generator\ndef make_generator():\n    inputs = layers.Input(shape=[IMAGE_SIZE, IMAGE_SIZE, CHANNELS])\n\n    down_stack = [\n        downsample(64, 4, apply_norm=False),\n        downsample(128, 4),\n        downsample(256, 4),\n        downsample(512, 4),\n        downsample(512, 4),\n        downsample(512, 4),\n        downsample(512, 4),\n        downsample(512, 4),\n    ]\n    \n    up_stack = [\n        upsample(512, 4, apply_dropout=True),\n        upsample(512, 4, apply_dropout=True),\n        upsample(512, 4, apply_dropout=True),\n        upsample(512, 4),\n        upsample(256, 4),\n        upsample(128, 4),\n        upsample(64, 4),\n    ]\n    \n    initializer = tf.random_normal_initializer(0., 0.02)\n    last = layers.Conv2DTranspose(3, 4, strides=2, padding='same',\n                                  kernel_initializer=initializer, activation='tanh')\n    \n    x = inputs\n    skips = []\n    for down in down_stack:\n        x = down(x)\n        skips.append(x)\n    \n    skips = reversed(skips[:-1])\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    return models.Model(inputs=inputs, outputs=x)#","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Discriminator\ndef make_discriminator():\n    initializer = tf.random_normal_initializer(0., 0.02)\n    inputs = layers.Input(shape=[IMAGE_SIZE, IMAGE_SIZE, 3], name='input_image')\n    \n    x = inputs\n    x = downsample(64, 4, False)(x)\n    x = downsample(128, 4)(x)\n    x = downsample(256, 4)(x)\n    x = layers.ZeroPadding2D()(x)\n    x = layers.Conv2D(512, 4, strides=1, kernel_initializer=initializer, use_bias=False)(x)\n    x = layers.GroupNormalization(groups=-1)(x)\n    x = layers.LeakyReLU()(x)\n    x = layers.ZeroPadding2D()(x)\n    x = layers.Conv2D(1, 4, strides=1, kernel_initializer=initializer)(x)\n    return models.Model(inputs=inputs, outputs=x)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Create generators and Discriminator \nmonet_generator = make_generator()\nphoto_generator = make_generator()\nmonet_discriminator = make_discriminator()\nphoto_discriminator = make_discriminator()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Loss function\ncross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Discriminator Loss\ndef discriminator_loss(real_output, fake_output):\n    real_loss = cross_entropy(tf.ones_like(real_output), real_output)\n    fake_loss = cross_entropy(tf.zeros_like(fake_output), fake_output)\n    total_loss = real_loss + fake_loss\n    return total_loss * 0.5\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#generator Loss\ndef generator_loss(fake_output):\n    return cross_entropy(tf.ones_like(fake_output), fake_output)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"LAMBDA = 10\n\ndef compute_cycle_loss(real_img, cycled_img):\n    loss = tf.reduce_mean(tf.abs(real_img - cycled_img))\n    return LAMBDA * loss","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def identity_loss(real_img, same_img):\n    loss = tf.reduce_mean(tf.abs(real_img - same_img))\n    return LAMBDA * 0.5 * loss","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from tensorflow.keras import optimizers\n\n# Optimizers\nmonet_generator_optimizer = optimizers.Adam(2e-4, beta_1=0.5)\nphoto_generator_optimizer = optimizers.Adam(2e-4, beta_1=0.5)\nmonet_discriminator_optimizer = optimizers.Adam(2e-4, beta_1=0.5)\nphoto_discriminator_optimizer = optimizers.Adam(2e-4, beta_1=0.5)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Checkpoints\nCHECKPOINT_DIR = '../working/checkpoints/train'\n\nckpt = tf.train.Checkpoint(monet_generator=monet_generator,\n                           photo_generator=photo_generator,\n                           monet_discriminator=monet_discriminator,\n                           photo_discriminator=photo_discriminator,\n                           monet_generator_optimizer=monet_generator_optimizer,\n                           photo_generator_optimizer=photo_generator_optimizer,\n                           monet_discriminator_optimizer=monet_discriminator_optimizer,\n                           photo_discriminator_optimizer=photo_discriminator_optimizer)\n\nckpt_manager = tf.train.CheckpointManager(ckpt, CHECKPOINT_DIR, max_to_keep=5)\n\n# if a checkpoint exists, restore the latest checkpoint.\nif ckpt_manager.latest_checkpoint:\n  ckpt.restore(ckpt_manager.latest_checkpoint)\n  print ('Latest checkpoint restored!!')","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#training\n@tf.function\ndef train_step(real_monet, real_photo):\n    with tf.GradientTape(persistent=True) as tape:\n        # real monet -> fake photo\n        fake_photo = monet_generator(real_monet, training=True)\n        # fake photo -> monet\n        cycled_monet = photo_generator(fake_photo, training=True)\n        # real photo -> fake monet\n        fake_monet = photo_generator(real_photo, training=True)\n        # fake monet -> photo\n        cycled_photo = monet_generator(fake_monet, training=True)\n        # use for identity loss\n        same_monet = monet_generator(real_monet, training=True)\n        same_photo = photo_generator(real_photo, training=True)\n        \n        # The discriminators evaluate both real and fake images in each domain\n        disc_real_monet = monet_discriminator(real_monet, training=True)\n        disc_real_photo = photo_discriminator(real_photo, training=True)\n        disc_fake_monet = monet_discriminator(fake_monet, training=True)\n        disc_fake_photo = photo_discriminator(fake_photo, training=True)\n        \n        # Generator losses: How well the generators fool the discriminators\n        monet_gen_loss = generator_loss(disc_fake_photo)\n        photo_gen_loss = generator_loss(disc_fake_monet)\n        # Cycle consistency loss: Ensures that cycled images are similar to originals\n        total_cycle_loss = compute_cycle_loss(real_monet, cycled_monet) + compute_cycle_loss(real_photo, cycled_photo)\n        # Identity loss: Encourages generators to preserve input when it's already in the target domain\n        total_monet_gen_loss = monet_gen_loss + total_cycle_loss + identity_loss(real_monet, same_monet)\n        total_photo_gen_loss = photo_gen_loss + total_cycle_loss + identity_loss(real_photo, same_photo)\n        # Discriminator losses: How well the discriminators distinguish real from fake images\n        monet_disc_loss = discriminator_loss(disc_real_monet, disc_fake_monet)\n        photo_disc_loss = discriminator_loss(disc_real_photo, disc_fake_photo)\n    \n    # Computes the gradients for each model with respect to its loss.\n    monet_generator_gradients = tape.gradient(total_monet_gen_loss, monet_generator.trainable_variables)\n    photo_generator_gradients = tape.gradient(total_photo_gen_loss, photo_generator.trainable_variables)\n    monet_discriminator_gradients = tape.gradient(monet_disc_loss, monet_discriminator.trainable_variables)\n    photo_discriminator_gradients = tape.gradient(photo_disc_loss, photo_discriminator.trainable_variables)\n    \n    # Updates the model parameters using the calculated gradients.\n    monet_generator_optimizer.apply_gradients(zip(monet_generator_gradients, monet_generator.trainable_variables))\n    photo_generator_optimizer.apply_gradients(zip(photo_generator_gradients, photo_generator.trainable_variables))\n    monet_discriminator_optimizer.apply_gradients(zip(monet_discriminator_gradients, monet_discriminator.trainable_variables))\n    photo_discriminator_optimizer.apply_gradients(zip(photo_discriminator_gradients, photo_discriminator.trainable_variables))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from IPython.display import clear_output\n\ndef train(monet_dataset, photo_dataset, epochs):\n    n = 0\n    for epoch in range(epochs):\n        start = time.time()\n        for monet_image, photo_image in tf.data.Dataset.zip((monet_dataset, photo_dataset)):\n            train_step(monet_image, photo_image)\n            if n % 10 == 0:\n                print ('.', end='')\n            n += 1\n        clear_output(wait=True)\n        if (epoch + 1) % 5 == 0:\n            ckpt_save_path = ckpt_manager.save()\n            print ('Saving checkpoint for epoch {} at {}'.format(epoch+1,ckpt_save_path))\n        \n        print(f'Time taken for epoch {epoch + 1} is {time.time() - start} sec')","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def generate_images(model, test_input):\n    prediction = model(test_input)\n    plt.figure(figsize=(12, 12))\n    display_list = [test_input[0], prediction[0]]\n    title = ['Input Image', 'Monet-style Image']\n    for i in range(2):\n        plt.subplot(1, 2, i+1)\n        plt.title(title[i])\n        plt.imshow(display_list[i] * 0.5 + 0.5)\n        plt.axis('off')\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Train the model\ntrain(train_ds, test_ds, epochs=EPOCHS)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Generate Monet-style image from 5 test photos\nfor photo in test_ds.take(5):\n    generate_images(monet_generator, photo)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#submission\n! mkdir ../images","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import shutil\n\ni = 1\nfor photo in test_ds:\n    prediction = monet_generator(photo, 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}]}