{"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":"gpu","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"dockerImageVersionId":30762,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import re\nimport os\nimport tensorflow as tf\nimport numpy as np\nimport matplotlib.pyplot as plt\n\nimport time\nfrom tqdm import tqdm\nfrom IPython import display\n\nfrom tensorflow.keras.layers import (\n    Input, Dense, Conv2D, Conv2DTranspose, LeakyReLU,\n    BatchNormalization, Reshape, Flatten, Dropout\n)","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2024-09-12T12:42:24.209186Z","iopub.execute_input":"2024-09-12T12:42:24.210159Z","iopub.status.idle":"2024-09-12T12:42:36.552079Z","shell.execute_reply.started":"2024-09-12T12:42:24.210094Z","shell.execute_reply":"2024-09-12T12:42:36.551286Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"IMAGE_PHOTO_DIR = '/kaggle/input/gan-getting-started/photo_jpg'\nimage_filenames = []\n\nfor dirname, _, filenames in os.walk(IMAGE_PHOTO_DIR):\n    for filename in filenames:\n        image_filenames.append(os.path.join(dirname, filename))\n        \nprint(len(image_filenames), ' files loaded')","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:42:36.554031Z","iopub.execute_input":"2024-09-12T12:42:36.554824Z","iopub.status.idle":"2024-09-12T12:42:40.524876Z","shell.execute_reply.started":"2024-09-12T12:42:36.554774Z","shell.execute_reply":"2024-09-12T12:42:40.523974Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"BUFFER_SIZE = 7038\nBATCH_SIZE = 16\n\ndef preprocess_jpg_image(image):\n    img = tf.io.read_file(image)\n    img = tf.image.decode_jpeg(img, channels=3)\n    img = tf.cast(img, tf.float32)\n    img = (img - 127.5) / 127.5\n    return img\n\n# Preprocess images and create dataset\nprocessed_jpg_photos = list(map(preprocess_jpg_image, image_filenames))\ndataset = tf.data.Dataset.from_tensor_slices(processed_jpg_photos)\ndataset = dataset.shuffle(buffer_size=BUFFER_SIZE).batch(BATCH_SIZE)\n    \nprint(len(dataset), ' image dataset created')","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:42:40.526222Z","iopub.execute_input":"2024-09-12T12:42:40.526619Z","iopub.status.idle":"2024-09-12T12:43:16.428987Z","shell.execute_reply.started":"2024-09-12T12:42:40.526574Z","shell.execute_reply":"2024-09-12T12:43:16.428074Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_latest_checkpoint(directory):\n    # Define the pattern for the filenames\n    pattern = re.compile(r'ckpt-(\\d+)\\.weights\\.h5')\n    \n    latest_file = None\n    latest_index = -1\n\n    try:\n        # List all files in the directory\n        for filename in os.listdir(directory):\n            match = pattern.match(filename)\n            if match:\n                # Extract the number from the filename\n                index = int(match.group(1))\n                if index > latest_index:\n                    latest_index = index\n                    latest_file = filename\n\n    except FileNotFoundError:\n        return latest_file\n    \n    return latest_file","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.431106Z","iopub.execute_input":"2024-09-12T12:43:16.431424Z","iopub.status.idle":"2024-09-12T12:43:16.437283Z","shell.execute_reply.started":"2024-09-12T12:43:16.43139Z","shell.execute_reply":"2024-09-12T12:43:16.436452Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"GEN_CHECKPOINT_DIR = \"generator_checkpoints\"\nDISC_CHECKPOINT_DIR = \"discriminator_checkpoints\"\nCHECKPOINT_PREFIX = 'ckpt'\n\nos.makedirs(GEN_CHECKPOINT_DIR, exist_ok=True)\nos.makedirs(DISC_CHECKPOINT_DIR, exist_ok=True)\n\ndef create_checkpoint(epoch, directory, model, name):\n    # Ensure the directory exists\n    if not os.path.exists(directory):\n        os.makedirs(directory)\n        \n    checkpoint_path = os.path.join(directory, f'{name}-{CHECKPOINT_PREFIX}-{epoch}.weights.h5')\n\n    print(f\"{checkpoint_path} created\")\n    model.save_weights(checkpoint_path)","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.438471Z","iopub.execute_input":"2024-09-12T12:43:16.438749Z","iopub.status.idle":"2024-09-12T12:43:16.457385Z","shell.execute_reply.started":"2024-09-12T12:43:16.438718Z","shell.execute_reply":"2024-09-12T12:43:16.456583Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"NOISE_DIM = 100\n\ndef create_generator():\n    noise_input = Input(shape=(NOISE_DIM,))\n\n    # Dense layer to reshape into a 16x16x512 tensor\n    x = Dense(16 * 16 * 512, use_bias=False)(noise_input)  # Use bias=False for BatchNormalization\n    x = Reshape((16, 16, 512))(x)\n    \n    # Upsampling blocks\n    x = Conv2DTranspose(256, kernel_size=(3, 3), strides=(2, 2), padding='same', use_bias=False)(x)\n    x = BatchNormalization()(x)\n    x = LeakyReLU(negative_slope=0.2)(x)  # 32x32x256\n    \n    x = Conv2DTranspose(128, kernel_size=(3, 3), strides=(2, 2), padding='same', use_bias=False)(x)\n    x = BatchNormalization()(x)\n    x = LeakyReLU(negative_slope=0.2)(x)  # 64x64x128\n    \n    x = Conv2DTranspose(64, kernel_size=(3, 3), strides=(2, 2), padding='same', use_bias=False)(x)\n    x = BatchNormalization()(x)\n    x = LeakyReLU(negative_slope=0.2)(x)  # 128x128x64\n    \n    x = Conv2DTranspose(32, kernel_size=(3, 3), strides=(2, 2), padding='same', use_bias=False)(x)\n    x = BatchNormalization()(x)\n    x = LeakyReLU(negative_slope=0.2)(x)  # 256x256x32\n    \n    output = Conv2DTranspose(3, kernel_size=(3, 3), strides=(1, 1), padding='same', activation='tanh')(x)  # Output layer\n\n    return tf.keras.Model(inputs=noise_input, outputs=output)\n\ngenerator_model = create_generator()\ngenerator_weights = get_latest_checkpoint(GEN_CHECKPOINT_DIR)\n\nif generator_weights:\n    generator_model.load_weights(generator_weights)\n    print(f\"Generator loaded from - {generator_weights}\")","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.458395Z","iopub.execute_input":"2024-09-12T12:43:16.458654Z","iopub.status.idle":"2024-09-12T12:43:16.604364Z","shell.execute_reply.started":"2024-09-12T12:43:16.458624Z","shell.execute_reply":"2024-09-12T12:43:16.60356Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def create_discriminator():\n    image_input = Input(shape=(256, 256, 3))\n\n    # First convolutional layer\n    x = Conv2D(32, kernel_size=(3, 3), strides=(2, 2), padding='same')(image_input)\n    x = LeakyReLU(negative_slope=0.2)(x)\n    x = BatchNormalization()(x)\n\n    # Second convolutional layer\n    x = Conv2D(64, kernel_size=(3, 3), strides=(2, 2), padding='same')(x)\n    x = LeakyReLU(negative_slope=0.2)(x)\n    x = BatchNormalization()(x)\n\n    # Third convolutional layer\n    x = Conv2D(128, kernel_size=(3, 3), strides=(2, 2), padding='same')(x)\n    x = LeakyReLU(negative_slope=0.2)(x)\n    x = BatchNormalization()(x)\n\n    # Fourth convolutional layer\n    x = Conv2D(256, kernel_size=(3, 3), strides=(2, 2), padding='same')(x)\n    x = LeakyReLU(negative_slope=0.2)(x)\n    x = BatchNormalization()(x)\n\n    # Flatten and fully connected layers\n    x = Flatten()(x)\n    x = Dense(512)(x)  # Reduced number of units\n    x = LeakyReLU(negative_slope=0.2)(x)\n    x = Dropout(0.3)(x)  # Adjust dropout rate for regularization\n\n    # Output layer\n    output = Dense(1, activation='sigmoid')(x)  # Probability of being real\n\n    return tf.keras.Model(inputs=image_input, outputs=output)\n\ndiscriminator_model = create_discriminator()\ndiscriminator_weights = get_latest_checkpoint(DISC_CHECKPOINT_DIR)\n\nif discriminator_weights:\n    discriminator_model.load_weights(discriminator_weights)\n    print(f\"Discriminator loaded from - {discriminator_weights}\")","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.605476Z","iopub.execute_input":"2024-09-12T12:43:16.605768Z","iopub.status.idle":"2024-09-12T12:43:16.708942Z","shell.execute_reply.started":"2024-09-12T12:43:16.605737Z","shell.execute_reply":"2024-09-12T12:43:16.708171Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"generator_model.summary()\ndiscriminator_model.summary()","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.710099Z","iopub.execute_input":"2024-09-12T12:43:16.710487Z","iopub.status.idle":"2024-09-12T12:43:16.774426Z","shell.execute_reply.started":"2024-09-12T12:43:16.710444Z","shell.execute_reply":"2024-09-12T12:43:16.773573Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generate_image():\n    # Generate random noise as input\n    noise = np.random.normal(0, 1, (1, NOISE_DIM))\n\n    # Generate an image from the noise\n    return generator_model.predict(noise)\n\ndef show_generated_image(image):\n    image = ((image * 127.5) + 127.5).reshape((256, 256, 3))\n    image = image.astype(np.uint8)\n    \n    plt.axis('off')\n    plt.imshow(image)\n    plt.show()\n    \nshow_generated_image(generate_image())","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:16.775738Z","iopub.execute_input":"2024-09-12T12:43:16.7761Z","iopub.status.idle":"2024-09-12T12:43:18.588632Z","shell.execute_reply.started":"2024-09-12T12:43:16.776057Z","shell.execute_reply":"2024-09-12T12:43:18.587282Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generator_loss(fake_output):\n    return tf.reduce_mean(tf.keras.losses.binary_crossentropy(tf.ones_like(fake_output), fake_output))\n\ndef discriminator_loss(real_output, fake_output):\n    real_loss = tf.reduce_mean(tf.keras.losses.binary_crossentropy(tf.ones_like(real_output), real_output))\n    fake_loss = tf.reduce_mean(tf.keras.losses.binary_crossentropy(tf.zeros_like(fake_output), fake_output))\n    return real_loss + fake_loss\n\ngenerator_optimizer = tf.keras.optimizers.Adam(learning_rate=0.0005, beta_1=0.5, beta_2=0.999)\ndiscriminator_optimizer = tf.keras.optimizers.Adam(learning_rate=0.0005, beta_1=0.5, beta_2=0.999)","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:18.593357Z","iopub.execute_input":"2024-09-12T12:43:18.594402Z","iopub.status.idle":"2024-09-12T12:43:18.619039Z","shell.execute_reply.started":"2024-09-12T12:43:18.594338Z","shell.execute_reply":"2024-09-12T12:43:18.617802Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"@tf.function\ndef train_step(generator, discriminator, images):\n    # Generate random noise\n    noise = tf.random.normal([images.shape[0], NOISE_DIM])\n\n    with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:\n        # Generate images from the noise\n        generated_images = generator(noise, training=True)\n\n        # Discriminate between real and generated images\n        real_output = discriminator(images, training=True)\n        fake_output = discriminator(generated_images, training=True)\n        \n        # Calculate losses\n        gen_loss = generator_loss(fake_output)\n        disc_loss = discriminator_loss(real_output, fake_output)\n    \n    # Calculate gradients\n    gradients_of_generator = gen_tape.gradient(gen_loss, generator.trainable_variables)\n    gradients_of_discriminator = disc_tape.gradient(disc_loss, discriminator.trainable_variables)\n    \n    # Apply gradients to update weights\n    generator_optimizer.apply_gradients(zip(gradients_of_generator, generator.trainable_variables))\n    discriminator_optimizer.apply_gradients(zip(gradients_of_discriminator, discriminator.trainable_variables))\n    \n    return gen_loss, disc_loss","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:18.620903Z","iopub.execute_input":"2024-09-12T12:43:18.621915Z","iopub.status.idle":"2024-09-12T12:43:18.63662Z","shell.execute_reply.started":"2024-09-12T12:43:18.621852Z","shell.execute_reply":"2024-09-12T12:43:18.635267Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def generate_and_save_images(epoch=1, num_samples=5**2, noise_dim=NOISE_DIM, save=True):\n    # Create directory for saving images if it doesn't exist\n    if not os.path.exists('epoch_images'):\n        os.makedirs('epoch_images')\n    \n    # Generate noise\n    noise = tf.random.normal([num_samples, noise_dim])  # Shape: (9, 100)\n    \n    # Generate images using the generator model\n    predictions = generator_model(noise, training=False)\n#     predictions = (predictions * 127.5) + 127.5\n    predictions = (predictions + 1) / 2\n    \n    # Create a figure for saving images\n    num_cols = 5\n    num_rows = 5\n    \n    fig = plt.figure(figsize=(num_cols * 2, num_rows * 2))\n    \n    for i in range(num_samples):\n        plt.subplot(num_rows, num_cols, i + 1)\n        plt.imshow(predictions[i])\n        plt.axis('off')\n    \n    # Save the figure\n    if save:\n        plt.savefig(os.path.join('epoch_images', f\"img_epoch_{epoch:04d}.jpg\"))\n    plt.show()\n\n# Example usage\ngenerate_and_save_images(save=False)","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:18.638438Z","iopub.execute_input":"2024-09-12T12:43:18.638957Z","iopub.status.idle":"2024-09-12T12:43:20.222185Z","shell.execute_reply.started":"2024-09-12T12:43:18.638895Z","shell.execute_reply":"2024-09-12T12:43:20.221197Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"EPOCHS = 1000\n\ndef train(dataset, epochs, start=0):\n    generate_and_save_images(epoch=start, save=True)\n    create_checkpoint(start, GEN_CHECKPOINT_DIR, generator_model, 'gen')\n    create_checkpoint(start, DISC_CHECKPOINT_DIR, discriminator_model, 'disc')\n    \n    for epoch in range(start+1, epochs):\n        start = time.time()\n        \n        for image_batch in dataset:\n            # Train the model with the current batch\n            train_step(generator_model, discriminator_model, image_batch)\n\n        display.clear_output(wait=True)\n        \n        # Clear the output and generate images every epoch\n        generate_and_save_images(epoch=epoch+1, save=True)\n    \n        # Save the model checkpoint every 10 epochs\n        if (epoch + 1) % 10 == 0:\n            create_checkpoint(epoch + 1, GEN_CHECKPOINT_DIR, generator_model, name='gen')\n            create_checkpoint(epoch + 1, DISC_CHECKPOINT_DIR, discriminator_model, name='disc')\n                \n        print(f\"Epoch {epoch + 1}: {time.time() - start} seconds\")\n\n    create_checkpoint(epochs, GEN_CHECKPOINT_DIR, generator_model, 'gen')\n    create_checkpoint(epochs, DISC_CHECKPOINT_DIR, discriminator_model, 'disc')\n        \n    # Generate images at the end of training\n    generate_and_save_images(epoch=epochs, save=True)","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:20.223497Z","iopub.execute_input":"2024-09-12T12:43:20.22381Z","iopub.status.idle":"2024-09-12T12:43:20.232659Z","shell.execute_reply.started":"2024-09-12T12:43:20.223776Z","shell.execute_reply":"2024-09-12T12:43:20.231571Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train(dataset, EPOCHS)","metadata":{"execution":{"iopub.status.busy":"2024-09-12T12:43:20.233775Z","iopub.execute_input":"2024-09-12T12:43:20.234092Z","iopub.status.idle":"2024-09-12T13:32:57.058411Z","shell.execute_reply.started":"2024-09-12T12:43:20.234059Z","shell.execute_reply":"2024-09-12T13:32:57.056855Z"},"trusted":true},"execution_count":null,"outputs":[]}]}