{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.14","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"},{"sourceId":1123351,"sourceType":"datasetVersion","datasetId":514871}],"dockerImageVersionId":30786,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport glob, math, os, sys, datetime\n\nimport tensorflow as tf\nfrom tensorflow import keras\nfrom tensorflow.keras import layers, initializers, activations, callbacks\n\ntf.config.run_functions_eagerly(True)","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19"},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"classes = ['monet']\nimg_size = (256, 256, 3)\nlearning_rate = 2e-4\nepochs = 50\nbatch_size = 2\nseed = 1234\ntrain_split = 0.7\nstyles_path = '/kaggle/input/gan-getting-started/monet_jpg/*'\nphotos_path = '/kaggle/input/gan-getting-started/photo_jpg/*'\noutputs_path = datetime.datetime.now(datetime.timezone.utc).isoformat()\nos.makedirs(outputs_path, exist_ok=True)\n\nprint(outputs_path)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"@tf.function\ndef loader(img_path):\n    image = tf.io.read_file(img_path)\n    image = tf.io.decode_jpeg(image, channels=3)\n    image = tf.image.resize(image, img_size[:2])\n    image = tf.cast(image, tf.float32) / 255.0\n\n    return image","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"styles = tf.data.Dataset.list_files(styles_path)\nstyles = styles.shuffle(buffer_size=10, seed=seed)\nstyles = styles.map(loader)\nstyles = styles.batch(batch_size)\n\nphotos = tf.data.Dataset.list_files(photos_path)\nphotos = photos.shuffle(buffer_size=10, seed=seed)\n\ntest_photos = photos.skip(math.ceil(len(photos) * train_split))\ntest_photos = test_photos.map(loader)\ntest_photos = test_photos.batch(batch_size)\n\nphotos = photos.take(math.ceil(len(photos) * train_split))\nphotos = photos.map(loader)\nphotos = photos.batch(batch_size)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"for exemple in styles.take(1):\n    plt.imshow(exemple[0])\n    plt.show()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"for exemple in test_photos.take(1):\n    plt.imshow(exemple[0])\n    plt.show()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Pix2Pix","metadata":{}},{"cell_type":"code","source":"def downsample(filters, size, apply_batchnorm=True):\n    initializer = tf.random_normal_initializer(0., 0.02)\n    result = tf.keras.Sequential()\n    result.add(\n        tf.keras.layers.Conv2D(filters, size, strides=2, padding='same',\n                                 kernel_initializer=initializer, use_bias=False))\n\n    if apply_batchnorm:\n        result.add(tf.keras.layers.BatchNormalization())\n\n    result.add(tf.keras.layers.LeakyReLU())\n    return result\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def upsample(filters, size, apply_dropout=False):\n    initializer = tf.random_normal_initializer(0., 0.02)\n    result = tf.keras.Sequential()\n    result.add(tf.keras.layers.Conv2DTranspose(filters, \n                                               size, \n                                               strides=2,\n                                               padding='same',\n                                               kernel_initializer=initializer,\n                                               use_bias=False))\n\n    result.add(tf.keras.layers.BatchNormalization())\n    if apply_dropout:\n        result.add(tf.keras.layers.Dropout(0.5))\n\n    result.add(tf.keras.layers.ReLU())\n    return result\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generator(img_shape):\n    inputs = tf.keras.layers.Input(shape=img_shape)\n\n    down_stack = [\n        downsample(64, 4, apply_batchnorm=False),  # (batch_size, 128, 128, 64)\n        downsample(128, 4),  # (batch_size, 64, 64, 128)\n        downsample(256, 4),  # (batch_size, 32, 32, 256)\n        downsample(512, 4),  # (batch_size, 16, 16, 512)\n        downsample(512, 4),  # (batch_size, 8, 8, 512)\n        downsample(512, 4),  # (batch_size, 4, 4, 512)\n        downsample(512, 4),  # (batch_size, 2, 2, 512)\n        downsample(512, 4),  # (batch_size, 1, 1, 512)\n    ]\n\n    up_stack = [\n        upsample(512, 4, apply_dropout=True),  # (batch_size, 2, 2, 1024)\n        upsample(512, 4, apply_dropout=True),  # (batch_size, 4, 4, 1024)\n        upsample(512, 4, apply_dropout=True),  # (batch_size, 8, 8, 1024)\n        upsample(512, 4),  # (batch_size, 16, 16, 1024)\n        upsample(256, 4),  # (batch_size, 32, 32, 512)\n        upsample(128, 4),  # (batch_size, 64, 64, 256)\n        upsample(64, 4),  # (batch_size, 128, 128, 128)\n    ]\n\n    initializer = tf.random_normal_initializer(0., 0.02)\n    last = tf.keras.layers.Conv2DTranspose(OUTPUT_CHANNELS, 4,\n                                         strides=2,\n                                         padding='same',\n                                         kernel_initializer=initializer,\n                                         activation='tanh')  # (batch_size, 256, 256, 3)\n\n    x = inputs\n    skips = []\n    for down in down_stack:\n        x = down(x)\n        skips.append(x)\n    skips = reversed(skips[:-1])\n\n    for up, skip in zip(up_stack, skips):\n        x = up(x)\n        x = tf.keras.layers.Concatenate()([x, skip])\n\n    x = last(x)\n    return tf.keras.Model(inputs=inputs, outputs=x)\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generator_loss(disc_generated_output, gen_output, target):\n    gan_loss = loss_object(tf.ones_like(disc_generated_output), disc_generated_output)\n\n    # Mean absolute error\n    l1_loss = tf.reduce_mean(tf.abs(target - gen_output))\n\n    total_gen_loss = gan_loss + (LAMBDA * l1_loss)\n\n    return total_gen_loss, gan_loss, l1_loss\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def Discriminator(img_shape):\n    initializer = tf.random_normal_initializer(0., 0.02)\n\n    inp = tf.keras.layers.Input(shape=img_shape, name='input_image')\n    tar = tf.keras.layers.Input(shape=img_shape, name='target_image')\n\n    x = tf.keras.layers.concatenate([inp, tar])  # (batch_size, 256, 256, channels*2)\n\n    down1 = downsample(64, 4, False)(x)  # (batch_size, 128, 128, 64)\n    down2 = downsample(128, 4)(down1)  # (batch_size, 64, 64, 128)\n    down3 = downsample(256, 4)(down2)  # (batch_size, 32, 32, 256)\n\n    zero_pad1 = tf.keras.layers.ZeroPadding2D()(down3)  # (batch_size, 34, 34, 256)\n    conv = tf.keras.layers.Conv2D(512, 4, strides=1,\n                                kernel_initializer=initializer,\n                                use_bias=False)(zero_pad1)  # (batch_size, 31, 31, 512)\n\n    batchnorm1 = tf.keras.layers.BatchNormalization()(conv)\n\n    leaky_relu = tf.keras.layers.LeakyReLU()(batchnorm1)\n\n    zero_pad2 = tf.keras.layers.ZeroPadding2D()(leaky_relu)  # (batch_size, 33, 33, 512)\n\n    last = tf.keras.layers.Conv2D(1, 4, strides=1,\n                                kernel_initializer=initializer)(zero_pad2)  # (batch_size, 30, 30, 1)\n\n    return tf.keras.Model(inputs=[inp, tar], outputs=last)\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def discriminator_loss(disc_real_output, disc_generated_output):\n    real_loss = loss_object(tf.ones_like(disc_real_output), disc_real_output)\n    generated_loss = loss_object(tf.zeros_like(disc_generated_output), disc_generated_output)\n    total_disc_loss = real_loss + generated_loss\n    return total_disc_loss\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"@tf.function\ndef train_step(input_image, target, step):\n    with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:\n        gen_output = generator(input_image, training=True)\n\n        disc_real_output = discriminator([input_image, target], training=True)\n        disc_generated_output = discriminator([input_image, gen_output], training=True)\n\n        gen_total_loss, gen_gan_loss, gen_l1_loss = generator_loss(disc_generated_output, gen_output, target)\n        disc_loss = discriminator_loss(disc_real_output, disc_generated_output)\n\n    generator_gradients = gen_tape.gradient(gen_total_loss,\n                                          generator.trainable_variables)\n    discriminator_gradients = disc_tape.gradient(disc_loss,\n                                               discriminator.trainable_variables)\n\n    generator_optimizer.apply_gradients(zip(generator_gradients,\n                                          generator.trainable_variables))\n    discriminator_optimizer.apply_gradients(zip(discriminator_gradients,\n                                              discriminator.trainable_variables))\n\n    with summary_writer.as_default():\n        tf.summary.scalar('gen_total_loss', gen_total_loss, step=step//1000)\n        tf.summary.scalar('gen_gan_loss', gen_gan_loss, step=step//1000)\n        tf.summary.scalar('gen_l1_loss', gen_l1_loss, step=step//1000)\n        tf.summary.scalar('disc_loss', disc_loss, step=step//1000)\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def fit(train_ds, test_ds, steps):\n    example_input, example_target = next(iter(test_ds.take(1)))\n    start = time.time()\n\n    for step, (input_image, target) in train_ds.repeat().take(steps).enumerate():\n    if (step) % 1000 == 0:\n        display.clear_output(wait=True)\n\n        if step != 0:\n            print(f'Time taken for 1000 steps: {time.time()-start:.2f} sec\\n')\n\n        start = time.time()\n\n        generate_images(generator, example_input, example_target)\n        print(f\"Step: {step//1000}k\")\n\n    train_step(input_image, target, step)\n\n    # Training step\n    if (step+1) % 10 == 0:\n        print('.', end='', flush=True)\n\n\n    # Save (checkpoint) the model every 5k steps\n    if (step + 1) % 5000 == 0:\n        checkpoint.save(file_prefix=checkpoint_prefix)\n","metadata":{},"execution_count":null,"outputs":[]}]}