{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Imports","metadata":{}},{"cell_type":"code","source":"# Imports\nimport tensorflow as tf\nimport matplotlib.pyplot as plt\nimport numpy as np\nimport os\nimport time\nfrom IPython.display import clear_output\nAUTOTUNE = tf.data.AUTOTUNE\n\n\n# Function from stack overflow which helped with warnings while running on a GPU\ngpus = tf.config.experimental.list_physical_devices('GPU')\nif gpus:\n    try:\n        # Currently, memory growth needs to be the same across GPUs\n        for gpu in gpus:\n            tf.config.experimental.set_memory_growth(gpu, True)\n        logical_gpus = tf.config.experimental.list_logical_devices('GPU')\n        print(len(gpus), \"Physical GPUs,\", len(logical_gpus), \"Logical GPUs\")\n    except RuntimeError as e:\n        # Memory growth must be set before GPUs have been initialized\n        print(e)\n\n\"\"\"For the generator and discriminator, I chose to use the pix2pix generator and discriminator developed by \nIsola, Phillip, et al. \n(Isola, Phillip, et al. \"Image-to-image translation with conditional adversarial networks.\" \nProceedings of the IEEE conference on computer vision and pattern recognition. 2017.)\nand converted to TensorFlow by TensorFlow (Tensorflow, https://www.tensorflow.org/tutorials/generative/pix2pix)\"\"\"\nfrom shutil import copyfile\ncopyfile(src = \"../input/tensorflow-examples/tensorflow_examples/models/pix2pix/pix2pix.py\", dst = \"../working/pix2pix.py\")\nimport pix2pix","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Data Loading","metadata":{}},{"cell_type":"code","source":"# Data loading setting the batch size to 1 and the image size to 256x256\nbatch_size = 1\nimg_height = 256\nimg_width = 256","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Load in the datasets locally using tf.keras.preprocessing.image_dataset_from_directory() splitting the datasets into\n# training(80%) and validation (20%) sets.\n\n# Photo Datasets\nphotos_dir = '../input/monet-dataset/photo'\ntrain_photo = tf.keras.preprocessing.image_dataset_from_directory(\n    photos_dir,\n    validation_split=0.2,\n    subset='training',\n    seed=123,\n    image_size=(img_height, img_width),\n    batch_size=batch_size\n)\ntest_photo = tf.keras.preprocessing.image_dataset_from_directory(\n    photos_dir,\n    validation_split=0.2,\n    subset='validation',\n    seed=123,\n    image_size=(img_height, img_width),\n    batch_size=batch_size\n)\n\n# Monet Datasets\nmonet_dir = '../input/monet-dataset/monet'\ntrain_monet = tf.keras.preprocessing.image_dataset_from_directory(\n    monet_dir,\n    validation_split=0.2,\n    subset='training',\n    seed=123,\n    image_size=(img_height, img_width),\n    batch_size=batch_size\n)\ntest_monet = tf.keras.preprocessing.image_dataset_from_directory(\n    monet_dir,\n    validation_split=0.2,\n    subset='validation',\n    seed=123,\n    image_size=(img_height, img_width),\n    batch_size=batch_size\n)\n\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Test if photos have been loaded in correctly and display them.\n\nsample_photo_train = next(iter(train_photo))\nsample_monet_train = next(iter(train_monet))\nsample_photo_test = next(iter(test_photo))\nsample_monet_test = next(iter(test_monet))\n\nplt.figure(1)\nplt.title('Photo _Train')\nplt.imshow(sample_photo_train[0][0].numpy().astype(\"uint8\"))\nplt.axis('off')\n\nplt.figure(2)\nplt.title('Monet Train')\nplt.imshow(sample_monet_train[0][0].numpy().astype(\"uint8\"))\nplt.axis('off')\n\nplt.figure(3)\nplt.title('Photo Test')\nplt.imshow(sample_photo_test[0][0].numpy().astype(\"uint8\"))\nplt.axis('off')\n\nplt.figure(4)\nplt.title('Monet Test')\nplt.imshow(sample_monet_test[0][0].numpy().astype(\"uint8\"))\nplt.axis('off')\n\n\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Data Preprocessing\nTo avoid overfitting a random jitter is used on the training set. The random jitter resizes the images to 286x286, randomly crops the image, then performs a random mirror on the image. After the random jitter the images are normalized to -1 to 1 for input into the model. ","metadata":{}},{"cell_type":"code","source":"def random_crop(image):\n    cropped_image = tf.image.random_crop(tf.squeeze(image[0]), size=[img_height, img_width, 3])\n    return cropped_image","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def normalize(image):\n    image = tf.cast(image, tf.float32)\n    image = (image / 127.5) - 1\n    return image","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def random_jitter(image):\n    # resize to 286x286\n    image = tf.image.resize(image, [286, 286], method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)\n\n    # random crop to 256x256\n    image = random_crop(image)\n\n    # random mirror\n    image = tf.image.random_flip_left_right(image)\n\n    return image","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def preprocess_image_train(image, label):\n    image = random_jitter(image)\n    image = normalize(image)\n    return image","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def preprocess_image_test(image, label):\n    image = normalize(image)\n    return image","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Code for testing the random jitter.\n\nplt.figure(1)\nplt.title('Monet painting')\nplt.imshow(sample_monet_train[0][0].numpy().astype(\"uint8\"))\nplt.axis('off')\n\nresized_image = tf.image.resize(sample_monet_train[0].numpy().astype(\"uint8\"), [286, 286], method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)\nplt.figure(2)\nplt.title('Monet painting with resizing')\nplt.imshow(resized_image[0])\nplt.axis('off')\n\nresized_crop = random_crop(resized_image)\nplt.figure(3)\nplt.title('Monet painting with resizing and random crop')\nplt.imshow(resized_crop)\nplt.axis('off')\n\nresized_crop_flip = tf.image.random_flip_left_right(resized_crop)\nplt.figure(4)\nplt.title('Monet painting with resizing, random crop and random flip')\nplt.imshow(resized_crop_flip.numpy().astype(\"uint8\"))\nplt.axis('off')\n\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Apply the preprocessing, shuffling, and caching to the datasets.\n\nbuffer_size = 1000\n\ntrain_photo = train_photo.map(preprocess_image_train, num_parallel_calls=AUTOTUNE).batch(1).cache().shuffle(buffer_size)\n\ntrain_monet = train_monet.map(preprocess_image_train, num_parallel_calls=AUTOTUNE).batch(1).cache().shuffle(buffer_size)\n\ntest_photo = test_photo.map(preprocess_image_test, num_parallel_calls=AUTOTUNE).batch(1).cache().shuffle(buffer_size)\n\ntest_monet = test_monet.map(preprocess_image_test, num_parallel_calls=AUTOTUNE).batch(1).cache().shuffle(buffer_size)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Test if photos have been preprocessed correctly and display them.\n\nsample_photo_train = next(iter(train_photo))\nsample_monet_train = next(iter(train_monet))\nsample_photo_test = next(iter(test_photo))\nsample_monet_test = next(iter(test_monet))\n\n\nplt.figure(1)\nplt.title('Photo _Train')\nplt.imshow(sample_photo_train[0] * 0.5 + 0.5)\nplt.axis('off')\n\nplt.figure(2)\nplt.title('Monet Train')\nplt.imshow(sample_monet_train[0] * 0.5 + 0.5)\nplt.axis('off')\n\nplt.figure(3)\nplt.title('Photo Test')\nplt.imshow(sample_photo_test[0][0] * 0.5 + 0.5)\nplt.axis('off')\n\nplt.figure(4)\nplt.title('Monet Test')\nplt.imshow(sample_monet_test[0][0] * 0.5 + 0.5)\nplt.axis('off')","metadata":{"scrolled":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Import the Tensorflow Example Pix2Pix CycleGAN Models","metadata":{}},{"cell_type":"code","source":"# Load in the pix2pix generator and discriminators. We need two of each for cycle gan. \n\noutput_channels = 3\n\ngenerator_g = pix2pix.unet_generator(output_channels, norm_type='instancenorm')\ngenerator_f = pix2pix.unet_generator(output_channels, norm_type='instancenorm')\n\ndiscriminator_x = pix2pix.discriminator(norm_type='instancenorm', target=False)\ndiscriminator_y = pix2pix.discriminator(norm_type='instancenorm', target=False)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Test the untrained generators on a sample of the training set and display the images. \n\nto_monet = generator_g(sample_photo_train)\nto_photo = generator_f(sample_monet_train)\nplt.figure(figsize=(8, 8))\ncontrast = 8\n\nimgs = [sample_photo_train, to_monet, sample_monet_train, to_photo]\ntitle = ['sample_photo', 'to_monet', 'sample_monet', 'to_photo']\n\nfor i in range(len(imgs)):\n    plt.subplot(2, 2, i+1)\n    plt.axis('off')\n    plt.title(title[i])\n    if i % 2 == 0:\n        plt.imshow(imgs[i][0] * 0.5 + 0.5)\n    else:\n      plt.imshow(imgs[i][0] * 0.5 * contrast + 0.5)\n\nplt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Test the untrained discriminators on a sample of the training set and display the images.\n\nplt.figure(figsize=(8, 8))\n\nplt.subplot(121)\nplt.title('Monet')\nplt.imshow(discriminator_y(sample_photo_train)[0])\n\nplt.subplot(122)\nplt.title('Photo')\nplt.imshow(discriminator_x(sample_monet_train)[0])\n\nplt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Generator, Discriminator, and Cycle Consistency Loss","metadata":{}},{"cell_type":"code","source":"# Using keras BinaryCrossentropy loss (Sigmoid cross entropy loss)\nLAMBDA = 10\nloss_obj = tf.keras.losses.BinaryCrossentropy(from_logits=True)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"The discriminator loss is the calculated on the discriminator of real image and the \ndiscriminator of a generated image. First, the real loss is calculated using the sigmoid \ncross entropy between a tensor of all ones of the same shape as the real image and a \ntensor of the real image. Then the generated loss is calculated by the sigmoid cross \nentropy between a tensor of all ones of the same shape as a generated image and a tensor \nof a generated image. Lastly, the total discriminator loss is calculated by adding the \nreal loss and the generated loss together. The total discriminator is scaled by a factor \nof 0.5 and returned. \n\"\"\"\ndef discriminator_loss(discriminator_real, discriminator_generated):\n    real_loss = loss_obj(tf.ones_like(discriminator_real), discriminator_real)\n    generated_loss = loss_obj(tf.zeros_like(discriminator_generated), discriminator_generated)\n    total_disc_loss = real_loss + generated_loss\n    return total_disc_loss * 0.5","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"The generator loss is calculated on a generated image by creating a tensor of all 1’s \nof the same shape as the tensor of the translated image created by the generator, then \ncalculating the sigmoid cross entropy between the tensor of all ones and the tensor of \nthe translated image.\"\"\"\ndef generator_loss(generated):\n    return loss_obj(tf.ones_like(generated), generated)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"The cycle consistency loss is the mean absolute error calculated between a real image \nand the real image translated through both domains otherwise known as the cycled image. \nFirst, the absolute value tensor of the subtraction of the tensor of the real image minus \nthe tensor of the cycled image is calculated. The absolute value tensor is then reduced \nto its mean. The mean is then multiplied by LAMDA (10) and returned.\"\"\"\ndef calc_cycle_loss(real_image, cycled_image):\n    loss = tf.reduce_mean(tf.abs(real_image - cycled_image)) \n    return LAMBDA * loss","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\"\"\"The identity loss is the mean absolute error calculated between a real image and the \nreal image translated to its original domain through a generator. For example, generator \nF translates Monet paintings to photos, if a photo was inputted into generator F then \ngenerator F should output something similar to the photo image. The mean absolute error \nis the identity loss  between the real image and its translation to the same domain. \nThe absolute value of the tensor of the real image minus the tensor of the same image \ntranslated to the same domain, then the reduced mean is calculated of the absolute value \nformulating the identity loss. In my project the identity loss is multiplied by LAMBDA \nand 0.5 before returning.\"\"\"\ndef identity_loss(real_image, same_image):\n    loss = tf.reduce_mean(tf.abs(real_image - same_image))\n    return LAMBDA * 0.5 * loss","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Initialize the optimizers with a learning rate of 2e-4 and the exponential decay (beta_1) to 0.5\ngenerator_g_optimizer = tf.keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5)\ngenerator_f_optimizer = tf.keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5)\n\ndiscriminator_x_optimizer = tf.keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5)\ndiscriminator_y_optimizer = tf.keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Training the CycleGAN","metadata":{}},{"cell_type":"code","source":"\"\"\"Running this locally on my gpu I could finsh an \nepoch about every 200 seconds I could not get the \nGPU option running on Kaggle. I have switched to \nonly 1 epoch for the Kaggle submission to complete\"\"\"\nEPOCHS = 1","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Generate images using whichever model and input is chosen. \ni = 0\ndef generate_images(model, test_input, epoch):\n    if epoch is None:\n        prediction = model(test_input)\n        plt.figure(figsize=(12, 12))\n        display_list = [test_input[0], prediction[0]]\n        title = ['Input Image', 'Predicted 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()\n    else:\n        prediction = model(test_input)\n        plt.figure(figsize=(12, 12))\n        display_list = [test_input[0], prediction[0]]\n        title = ['Input Image', 'Predicted 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.savefig('output/training/epoch' + str(epoch) + '.jpeg')  # Images were locally saved\n        plt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"@tf.function\ndef train_step(real_x, real_y):\n    with tf.GradientTape(persistent=True) as tape:\n        # Generator G translates photo -> monet\n        # Generator F translates monet -> photo.\n    \n        # Generate real_x -> fake_y -> cycled_x (CycleGAN)\n        fake_y = generator_g(real_x, training=True)\n        cycled_x = generator_f(fake_y, training=True)\n    \n        # Generate real_y -> fake_x -> cycled_y (CycleGAN)\n        fake_x = generator_f(real_y, training=True)\n        cycled_y = generator_g(fake_x, training=True)\n\n        # Generate the same image with its respective generator for identitiy loss\n        same_x = generator_f(real_x, training=True)\n        same_y = generator_g(real_y, training=True)\n        \n        # Generate discriminator images based off real images\n        disc_real_x = discriminator_x(real_x, training=True)\n        disc_real_y = discriminator_y(real_y, training=True)\n        \n        # Generate discriminator images based off fake images\n        disc_fake_x = discriminator_x(fake_x, training=True)\n        disc_fake_y = discriminator_y(fake_y, training=True)\n        \n        # Calculate the loss of each generated based off the discriminator images\n        gen_g_loss = generator_loss(disc_fake_y)\n        gen_f_loss = generator_loss(disc_fake_x)\n        \n        # Calculate the total cycle loss\n        total_cycle_loss = calc_cycle_loss(real_x, cycled_x) + calc_cycle_loss(real_y, cycled_y)\n    \n        # Calculate the total generator losses. Same_x and Same_y are the real image passed\n        # through its respective generator to its current domain.\n        total_gen_g_loss = gen_g_loss + total_cycle_loss + identity_loss(real_y, same_y)\n        total_gen_f_loss = gen_f_loss + total_cycle_loss + identity_loss(real_x, same_x)\n        \n        # Calculate the loss of each discriminator based off the real discriminator images nad the \n        # fake discriminator images\n        disc_x_loss = discriminator_loss(disc_real_x, disc_fake_x)\n        disc_y_loss = discriminator_loss(disc_real_y, disc_fake_y)\n  \n    # Calculate the gradients for generator and discriminator\n    generator_g_gradients = tape.gradient(total_gen_g_loss, generator_g.trainable_variables)\n    generator_f_gradients = tape.gradient(total_gen_f_loss, generator_f.trainable_variables)\n  \n    discriminator_x_gradients = tape.gradient(disc_x_loss, discriminator_x.trainable_variables)\n    discriminator_y_gradients = tape.gradient(disc_y_loss, discriminator_y.trainable_variables)\n  \n    # Apply the gradients to the Adam optimizers\n    generator_g_optimizer.apply_gradients(zip(generator_g_gradients, generator_g.trainable_variables))\n\n    generator_f_optimizer.apply_gradients(zip(generator_f_gradients, generator_f.trainable_variables))\n  \n    discriminator_x_optimizer.apply_gradients(zip(discriminator_x_gradients, discriminator_x.trainable_variables))\n  \n    discriminator_y_optimizer.apply_gradients(zip(discriminator_y_gradients, discriminator_y.trainable_variables))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Run the train_step on the pairs of images for specified EPOCHS \n# Skip training as the notebook was run locally\n\"\"\"for epoch in range(EPOCHS):\n    start_epoch = time.time()\n\n    for image_x, image_y in tf.data.Dataset.zip((train_photo, train_monet)):\n        train_step(image_x, image_y)\n\n    # Generate images showing the models improvement every epoch\n    generate_images(generator_g, sample_photo_train, epoch)\n\n    print ('Time taken for epoch {} is {} sec\\n'.format(epoch + 1, time.time()-start_epoch))\"\"\"","metadata":{"scrolled":true,"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Generate using test dataset","metadata":{}},{"cell_type":"code","source":"# Call the generate_images function on the test photos set to compare data the model has not seen\n# Photo -> Monet\n# Skip output as the notebook was run locally and the output was saved locally.\n\"\"\"for inp in test_photo.take(5):\n    generate_images(generator_g, inp[0], None)\"\"\"","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Call the generate_images function on the test monet set to compare data the model has not seen\n# Monet -> Photo\n# Skip output as the notebook was run locally and the output was saved locally.\n\"\"\"for inp in test_monet.take(5):\n    generate_images(generator_f, inp[0], None)\"\"\"","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Generate 7000 Monet Paintings From Images","metadata":{}},{"cell_type":"code","source":"plt.ioff()\ndef generate_and_save(model, input_train, input_test):\n    i = 0\n    for photo in input_train:\n        plt.imshow(model(photo)[0] * 0.5 + 0.5)\n        plt.axis('off')\n        plt.savefig('output/final_output/' + str(i) + '.jpeg', pad_inches=0, bbox_inches='tight')\n        plt.close()\n        i += 1\n    for photo in input_test:\n        plt.imshow(model(photo[0])[0] * 0.5 + 0.5)\n        plt.axis('off')\n        plt.savefig('output/final_output/' + str(i) + '.jpeg', pad_inches=0, bbox_inches='tight')\n        plt.close()\n        i += 1\n        ","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Skip output as the notebook was run locally and the output was saved locally.\n# generate_and_save(generator_g, train_photo, test_photo)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import shutil\nshutil.make_archive('images', 'zip', '../input/imsomethingofapaintermyselfoutput/output')","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}