{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","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":30664,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Part 1: Description of the problem and data\n\nThe primary challenge is to develop a Generative Adversarial Network capable of producing images that emulate the distinctive style of Claude Monet. This task involves effectively training a GAN comprising a generator and a discriminator to generate high quality, diverse images without simply mimicking existing works. Success is measured using the Memorization informed Fréchet Inception Distance (MiFID), a metric assessing the similarity to Monet's style while ensuring the images are not just copies.\n\n**Data Overview**\nThe dataset is split into four directories, each serving a distinct purpose for the GAN project:\n* monet_jpg: Contains 300 JPEG images of Claude Monet's paintings.\n* monet_tfrec: Parallel to monet_jpg, includes the same 300 Monet paintings but in TFRecord format.\n* photo_jpg: Comprises 7,028 JPEG images of regular photos.\n* photo_tfrec: Mirrors photo_jpg with the same 7,028 photos, but in TFRecord format.\n\n**Purpose of Monet Images:** Both monet_jpg and monet_tfrec are used for training the GAN to learn and replicate Monet's artistic style.\n\n**Purpose of Photo Images:** The photo_jpg and photo_tfrec images are intended to be transformed by the GAN into Monet-style artwork, forming the basis of the project's output.","metadata":{}},{"cell_type":"markdown","source":"# Some Libraries","metadata":{}},{"cell_type":"code","source":"import numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\nimport os\n\nimport torch \nimport torch.nn as nn\nfrom torch.utils.data import DataLoader, Dataset\nimport torch.optim as optim\nimport torchvision\nfrom torchvision import transforms\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\n\nfrom tqdm import tqdm\n\nfrom PIL import Image\nimport matplotlib.pyplot as plt","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Part 2: EDA and data preparation\n\nFirst, we will use the following function to dipsplay a batch of images using PyTorch and Matplotlib. It first detaches the images from the PyTorch computational graph and moves them to the CPU. It then normalizes these images from a range of [-1, 1] to [0, 1], suitable for display. The function arranges the images in a grid with a specified number of rows (defaulting to 5) and then uses Matplotlib to plot this grid. It sets the figure size, removes axis labels for a cleaner look, adds an optional title, and displays the image grid.","metadata":{}},{"cell_type":"code","source":"def show_images(img, nrow=5, title=\"\", normalization_range=(-1, 1)):\n    if not isinstance(img, torch.Tensor):\n        raise ValueError(\"Input img should be a torch.Tensor\")\n\n    # Normalization adjustment based on specified range\n    scale = normalization_range[1] - normalization_range[0]\n    img = img.detach().cpu() * 0.5 + 0.5\n\n    img_grid = torchvision.utils.make_grid(img, nrow=nrow).permute(1, 2, 0)\n    plt.figure(figsize=(8, 5))\n    plt.imshow(img_grid)\n    plt.axis(\"off\")\n    plt.title(title)\n    plt.show()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Next, we are defining a custom dataset class named MonetPhotoDataset for your GAN project, which inherits from PyTorch's Dataset class. This class is designed to handle loading and transforming two types of image datasets: Monet paintings and regular photos. In the constructor (__init__), the class initializes paths to the directories containing Monet and photo images, along with an optional transformation. It then populates lists of file paths for both Monet paintings and regular photos, filtering for specific image file extensions. The __getitem__ method is responsible for loading and optionally transforming a pair of images (one Monet painting and one photo) based on the provided index. It ensures that each pair is loaded and transformed correctly, which is crucial for training the GAN. The __len__ method returns the maximum length of the two datasets, ensuring that the DataLoader can iterate over the dataset correctly.","metadata":{}},{"cell_type":"code","source":"import os\nfrom PIL import Image\nimport numpy as np\nimport torch\nfrom torch.utils.data import Dataset\n\nclass MonetPhotoDataset(Dataset):\n    def __init__(self, monet_dir, photo_dir, apply_transform=None):\n        if not os.path.isdir(monet_dir) or not os.path.isdir(photo_dir):\n            raise ValueError(\"Provided folder paths are invalid or do not exist.\")\n        \n        self.monet_dir = monet_dir\n        self.photo_dir = photo_dir\n        self.apply_transform = apply_transform\n        self.monet_images = [os.path.join(monet_dir, file) for file in os.listdir(monet_dir) if file.endswith(('.jpg', '.jpeg', '.png'))]\n        self.photo_images = [os.path.join(photo_dir, file) for file in os.listdir(photo_dir) if file.endswith(('.jpg', '.jpeg', '.png'))]\n\n        if not self.monet_images or not self.photo_images:\n            raise ValueError(\"No images found in the provided directory paths.\")\n        \n    def __getitem__(self, index):\n        monet_img_path = self.monet_images[index % len(self.monet_images)]\n        photo_img_path = self.photo_images[index % len(self.photo_images)]\n\n        try:\n            monet_img = np.array(Image.open(monet_img_path).convert('RGB')).astype('float32')\n            photo_img = np.array(Image.open(photo_img_path).convert('RGB')).astype('float32')\n        except IOError:\n            raise IOError(\"Error occurred when reading images.\")\n\n        if self.apply_transform:\n            try:\n                transformed_images = self.apply_transform(image=monet_img, image0=photo_img)\n                monet_img = transformed_images[\"image\"]\n                photo_img = transformed_images[\"image0\"]\n            except Exception as e:\n                raise RuntimeError(\"An error occurred during transformation: \" + str(e))\n\n        return monet_img, photo_img\n\n    def __len__(self):\n        return max(len(self.monet_images), len(self.photo_images))\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Finally, we are setting up the configuration for the deep learning model and preparing the data for training. Initially, we determine the device for training ('cuda' for GPU or 'cpu'), which is crucial for efficient computation. Then, we define several hyperparameters, such as the number of training epochs, learning rates for the discriminator and generator, the number of workers for data loading, batch size, image size, and the number of channels in the images. Additional parameters like lambda values for cycle consistency and identity losses, and a flag for saving the model, are also set, indicating their role in the training process. We then specify file names for saving the checkpoints of different models (discriminator for Monet, discriminator for photos, generator for Monet, and generator for photos).\n\n","metadata":{}},{"cell_type":"code","source":"# Device Configuration\ndevice = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n\n# Hyperparameters Configuration\nhyperparams = {\n    \"num_epochs\": 35,\n    \"learning_rate_discriminator\": 2e-4,\n    \"learning_rate_generator\": 2e-4,\n    \"num_workers\": 2,\n    \"batch_size\": 8,\n    \"image_dimensions\": {\"size\": 128, \"channels\": 3},\n    \"lambda_cycle_consistency\": 10.0,\n    \"lambda_identity\": 5.0,\n    \"save_model\": True\n}\n\n# Model Checkpoint Paths\ncheckpoint_paths = {\n    \"discriminator_monet\": \"discM.pth.tar\",\n    \"discriminator_photo\": \"discP.pth.tar\",\n    \"generator_monet\": \"genM.pth.tar\",\n    \"generator_photo\": \"genP.pth.tar\"\n}\n\n# Image Transformations\nimage_transforms = A.Compose(\n    [\n        A.Resize(hyperparams[\"image_dimensions\"][\"size\"], hyperparams[\"image_dimensions\"][\"size\"]),\n        A.HorizontalFlip(),\n        A.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5], max_pixel_value=255.0),\n        ToTensorV2(),\n    ], additional_targets={\"image0\": \"image\"}\n)\n\n# Dataset and DataLoader Setup\ntrain_dataset = MonetPhotoDataset(\"/kaggle/input/gan-getting-started/monet_jpg\", \"/kaggle/input/gan-getting-started/photo_jpg\", apply_transform=image_transforms)\ntrain_loader = DataLoader(train_dataset, batch_size=hyperparams[\"batch_size\"], shuffle=True, num_workers=hyperparams[\"num_workers\"])\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(f\"Total monet images: {len(train_dataset)}\")\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"num_images_to_show = 5 \n\nmonet_img, photo_img = next(iter(train_loader))\nshow_images(monet_img[:num_images_to_show], nrow=5, title=\"Monet Images\")\nshow_images(photo_img[:num_images_to_show], nrow=5, title=\"Photo Images\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Computing Basic Statistics","metadata":{}},{"cell_type":"code","source":"def compute_basic_image_statistics(image_paths):\n    stats = {'min': [], 'max': [], 'mean': [], 'median': []}\n    for path in image_paths[:5]:  # Sample a few images\n        image = np.array(Image.open(path))\n        stats['min'].append(image.min())\n        stats['max'].append(image.max())\n        stats['mean'].append(image.mean())\n        stats['median'].append(np.median(image))\n\n    for key in stats:\n        print(f\"{key.capitalize()} across sampled images: {stats[key]}\")\n\n# Usage\ncompute_basic_image_statistics(train_dataset.monet_images)\ncompute_basic_image_statistics(train_dataset.photo_images)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"From these basic statistics we can draw a few important conclusions:\n\n- Both Monet images and regular photos have pixel values ranging from 0 to 255.\n\n- The mean values indicate the average brightness of the images. For Monet images, the means seem generally higher, suggesting that these images might be brighter on average than the regular photos. This could be due to Monet's preference for brighter colors or outdoor scenes. In contrast, the regular photos show a lower average brightness, indicating these images might be overall darker or have more variation in lighting conditions.\n\n- The median values provide a robust measure of central tendency. Similar to the mean, the higher median values in Monet images suggest a generally brighter central value. The variation in the median values for both sets also suggests diversity in image content and brightness.\n\n- The differences in brightness between the two sets of images can be a critical factor in how the GAN learns to transform regular photos into Monet-style images.","metadata":{}},{"cell_type":"markdown","source":"# Part 3: Model Building and training ","metadata":{}},{"cell_type":"markdown","source":"The architecture of the model is designed to progressively break down and analyze the input image, extracting increasingly abstract and relevant features, and finally making a binary decision. This process is integral to the GAN's ability to learn to generate images that are indistinguishable from real ones, as it provides the necessary feedback loop for the generator to improve its outputs.\n\nFirst, we are defining a Discriminator model for a GAN using PyTorch. The model is structured with two main components: BlockDisc and Discriminator.\n\n- BlockDisc Class: This is a custom block used in building the Discriminator. It consists of a convolutional layer followed by instance normalization and a LeakyReLU activation. The forward method defines how the data flows through this block.\n\n- Discriminator Class: This is the main discriminator model. It starts with an initial convolutional layer followed by a LeakyReLU activation. Then, it dynamically builds more layers based on the features list. Each subsequent layer is an instance of BlockDisc. The final layer is a convolution that maps to a single output channel, suitable for binary classification (real vs. fake). The output is passed through a sigmoid activation function to obtain a probability score.","metadata":{}},{"cell_type":"code","source":"class BlockDisc(nn.Module):\n    def __init__(self, in_channels, out_channels, stride=2):\n        super().__init__()\n        # A convolutional block with normalization and LeakyReLU activation\n        self.conv = nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, 4, stride, 1, bias=True, padding_mode=\"reflect\"),\n            nn.InstanceNorm2d(out_channels),\n            nn.LeakyReLU(0.2),\n        )\n\n    def forward(self, x):\n        return self.conv(x)\n\nclass Discriminator(nn.Module):\n    def __init__(self, in_channels, features=[64, 128, 256, 512]):\n        super().__init__()\n\n        # Initial layer without normalization\n        self.initial = nn.Sequential(\n            nn.Conv2d(in_channels, features[0], kernel_size=4, stride=2, padding=1, padding_mode=\"reflect\"),\n            nn.LeakyReLU(0.2),\n        )\n\n        layers = [self.initial]\n        in_channel = features[0]\n\n        # Building intermediate layers dynamically\n        for feature in features[1:]:\n            layers.append(\n                BlockDisc(in_channel, feature, stride=1 if feature == features[-1] else 2),\n            )\n            in_channel = feature\n\n        # Final convolution layer to produce a single-channel output\n        layers.append(\n            nn.Conv2d(in_channel, 1, kernel_size=4, stride=1, padding=1, padding_mode=\"reflect\")\n        )\n\n        self.model = nn.Sequential(*layers)\n\n    def forward(self, x):\n        return torch.sigmoid(self.model(x))\n\n    # Optional: Advanced weight initialization\n    def _initialize_weights(self):\n        for m in self.modules():\n            if isinstance(m, nn.Conv2d):\n                nn.init.xavier_uniform_(m.weight)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## The generator\nThis part of the code is structured to generate images from input data through a series of convolutional transformations. The Generator is composed of several types of blocks:\n\n- ConvBlock: This serves as the core building block, capable of performing either standard or transposed convolutions, controlled by the is_encoder flag. It also includes instance normalization and ReLU activation, this is great for stabilizing training and introducing non-linearity.\n\n- ResidualBlock: These blocks, each consisting of two convolutional layers and a skip connection, are used for feature transformation. The skip connection helps mitigate the vanishing gradient problem in deep networks by allowing an alternate pathway for the gradient.\n\n- Generator Architecture: The Generator starts with an initial convolutional layer (initial_layer), followed by a series of encoding ConvBlocks that downsample the input. It then passes the data through multiple ResidualBlocks for deeper feature extraction. Finally, a series of decoding ConvBlocks upsample the feature map back to the original image size. The final_layer uses a Tanh activation function to output the generated image in the expected range.\n\n- Forward Pass: The forward method defines the data flow through these layers, ensuring the transformation of the input to the output image.\n\n- Output Shape Validation: An assertion checks that the output image has the correct dimensions (256x256 in this case).\n\n- Generator Test: The generator_test function is a simple test to ensure the Generator can process an input tensor and produce an output of the expected shape.","metadata":{}},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\n\nclass ConvBlock(nn.Module):\n    \"\"\"\n    A convolutional block used in the Generator. \n    It can perform either standard convolution or transposed convolution \n    based on the 'is_encoder' flag.\n    Includes instance normalization and ReLU activation.\n    \"\"\"\n    def __init__(self, in_channels, out_channels, is_encoder=True, **kwargs):\n        super().__init__()\n        if is_encoder:\n            self.conv = nn.Conv2d(in_channels, out_channels, padding_mode=\"reflect\", **kwargs)\n        else:\n            self.conv = nn.ConvTranspose2d(in_channels, out_channels, **kwargs)\n        self.normalization = nn.InstanceNorm2d(out_channels)\n        self.activation = nn.ReLU(inplace=True)\n\n    def forward(self, x):\n        return self.activation(self.normalization(self.conv(x)))\n\n\nclass ResidualBlock(nn.Module):\n    \"\"\"\n    Implements a residual block with two convolutional layers and a skip connection.\n    Helps in preserving the network's ability to effectively train deeper architectures.\n    \"\"\"\n    def __init__(self, channels):\n        super().__init__()\n        self.block = nn.Sequential(\n            nn.Conv2d(channels, channels, kernel_size=3, padding=1, stride=1, padding_mode=\"reflect\"),\n            nn.InstanceNorm2d(channels),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(channels, channels, kernel_size=3, padding=1, stride=1, padding_mode=\"reflect\"),\n            nn.InstanceNorm2d(channels),\n        )\n\n    def forward(self, x):\n        return x + self.block(x)\n\n\nclass Generator(nn.Module):\n    \"\"\"\n    Generator model in GAN. Composed of an initial convolution, \n    a series of ConvBlocks for encoding and decoding, \n    several ResidualBlocks for transformation, and a final convolution.\n    The model outputs generated images with characteristics learned from the input data.\n    \"\"\"\n    def __init__(self, in_channels, num_features=64, num_residuals=6):\n        super().__init__()\n        self.initial_layer = nn.Sequential(\n            nn.Conv2d(in_channels, num_features, kernel_size=7, stride=1, padding=3, padding_mode=\"reflect\"),\n            nn.InstanceNorm2d(num_features),\n            nn.ReLU(inplace=True),\n        )\n\n        self.encoder_blocks = nn.ModuleList([\n            ConvBlock(num_features, num_features*2, is_encoder=True, kernel_size=3, stride=2, padding=1),\n            ConvBlock(num_features*2, num_features*4, is_encoder=True, kernel_size=3, stride=2, padding=1),\n        ])\n\n        self.residual_blocks = nn.Sequential(\n            *[ResidualBlock(num_features*4) for _ in range(num_residuals)]\n        )\n\n        self.decoder_blocks = nn.ModuleList([\n            ConvBlock(num_features*4, num_features*2, is_encoder=False, kernel_size=3, stride=2, padding=1, output_padding=1),\n            ConvBlock(num_features*2, num_features, is_encoder=False, kernel_size=3, stride=2, padding=1, output_padding=1),\n        ])\n\n        self.final_layer = nn.Sequential(\n            nn.Conv2d(num_features, in_channels, kernel_size=7, stride=1, padding=3, padding_mode=\"reflect\"),\n            nn.Tanh(),\n        )\n\n    def forward(self, x):\n        x = self.initial_layer(x)\n        for layer in self.encoder_blocks:\n            x = layer(x)\n        x = self.residual_blocks(x)\n        for layer in self.decoder_blocks:\n            x = layer(x)\n        x = self.final_layer(x)\n#         assert x.shape[2:4] == (256, 256), f\"The output shape is not 256x256, instead we got {x.shape[2:4]}\"\n        return x\n\n# Test function to validate the Generator's output\ndef generator_test():\n    x = torch.randn((10, 3, 128, 128))\n    generator = Generator(in_channels=3, num_residuals=6)\n    generated_img = generator(x)\n    print(f\"Generated Image Shape: {generated_img.shape}\")\n    \ngenerator_test()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Supporting functions\n- Plotting: We want to create a way to visualize the training losses of both the Generator and Discriminator. Here we creates a line plot where the x-axis represents the number of iterations, and the y-axis shows the loss. This visualization is crucial for monitoring the training process, ensuring both networks are learning effectively and identifying potential issues like model convergence or oscillation in loss.\n\n- Save Checkpoint: With these models it is crucial that we can save the state of a model and its optimizer at a particular point during training. This is important for preserving the model's learned parameters, allowing for training to be paused and resumed without loss of progress. It's also useful for later evaluation or fine tuning of the model.","metadata":{}},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport torch\n\ndef plot(g_loss, d_loss, smoothing=False):\n    plt.figure(figsize=(10, 5))\n    plt.title(\"Generator and Discriminator Loss During Training\")\n\n    if smoothing:\n        g_loss = np.convolve(g_loss, np.ones((10,))/10, mode='valid')\n        d_loss = np.convolve(d_loss, np.ones((10,))/10, mode='valid')\n\n    plt.plot(g_loss, label=\"Generator\")\n    plt.plot(d_loss, label=\"Discriminator\")\n    plt.xlabel(\"Iterations\")\n    plt.ylabel(\"Loss\")\n    plt.legend()\n    plt.savefig('training_loss.png')  # Save the plot as a file\n    plt.show()\n\n# Enhanced save_checkpoint function\ndef save_checkpoint(model, optimizer, file_name=\"my_checkpoint.pth.tar\"):\n    print(\"Saving checkpoint...\")\n    checkpoint = {\n        \"state_dict\": model.state_dict(),\n        \"optimizer\": optimizer.state_dict(),\n        # Add more elements as needed\n    }\n    try:\n        torch.save(checkpoint, file_name)\n        print(\"Checkpoint saved successfully to\", file_name)\n    except Exception as e:\n        print(\"Error saving checkpoint:\", e)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Now we are preparing and configuring key components for the training of the model. This setup involves initializing two discriminator models (disc_P and disc_M) and two generator models (gen_P and gen_M), each designed to work on images with three channels. The discriminators aim to distinguish between real and generated images, while the generators are tasked with creating believable fake images. Both the discriminators and generators are transferred to the designated compute device. For training these models, we set up two Adam optimizers one for the discriminators (opt_disc) and one for the generators (opt_gen) with specific learning rates and momentum terms. Additionally, we define L1 and mean squared error (MSE) loss functions. We also introduce learning rate schedulers (lr_scheduler_D and lr_scheduler_G) with a lambda function determining the learning rate decay, to adjust the learning rate as training progresses. This setup is essential for effectively training the GAN, as it controls how the models learn and adapt over time, aiming to produce high quality generated images.","metadata":{}},{"cell_type":"code","source":"import itertools\n\n# Configuration dictionary\nconfig = {\n    \"device\": \"cuda\" if torch.cuda.is_available() else \"cpu\",\n    \"lr_disc\": 2e-4,\n    \"lr_gen\": 2e-4,\n    \"betas\": (0.5, 0.999),\n    \"num_residuals\": 9,\n    \"decay_epoch\": 20,\n    \"num_epochs\": 35,\n}\n\n# Function to create models\ndef create_model(model_class, in_channels):\n    if model_class == Generator:\n        return model_class(in_channels, num_residuals=config[\"num_residuals\"]).to(config[\"device\"])\n    elif model_class == Discriminator:\n        return model_class(in_channels).to(config[\"device\"])\n    \n# Function to create optimizers\ndef create_optimizer(model_parameters, lr):\n    return optim.Adam(model_parameters, lr=lr, betas=config[\"betas\"])\n\n# Initializing models\ndisc_P = create_model(Discriminator, in_channels=3)\ndisc_M = create_model(Discriminator, in_channels=3)\ngen_P = create_model(Generator, in_channels=3)\ngen_M = create_model(Generator, in_channels=3)\n\n# Initializing optimizers\nopt_disc = create_optimizer(itertools.chain(disc_P.parameters(), disc_M.parameters()), config[\"lr_disc\"])\nopt_gen = create_optimizer(itertools.chain(gen_P.parameters(), gen_M.parameters()), config[\"lr_gen\"])\n\n# Loss functions\nL1_loss = nn.L1Loss()\nmse_loss = nn.MSELoss()\n\n# Lambda function for learning rate decay\nlambda_lr_decay = lambda epoch: 1 - max(0, epoch - config[\"decay_epoch\"]) / (config[\"num_epochs\"] - config[\"decay_epoch\"])\n\n# Learning rate schedulers\nlr_scheduler_D = torch.optim.lr_scheduler.LambdaLR(opt_disc, lr_lambda=lambda_lr_decay)\nlr_scheduler_G = torch.optim.lr_scheduler.LambdaLR(opt_gen, lr_lambda=lambda_lr_decay)\n\n# Test the device setup\nprint(f\"Using device: {config['device']}\")\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Executing the model\n\nNow, we executethe model. Here is a breakdown of our approach and functions:\n\n- Discriminator and Generator Loss Functions: The calculate_discriminator_loss function computes the loss for the discriminator by assessing how well it differentiates between real images and fake images produced by the generator.The calculate_generator_loss function calculates the generator's loss. This loss is multifaceted: it includes not only the generator's ability to fool the discriminator but also ensures the generated image retains certain characteristics of the original image and maintains its core identity.\n\n- Training Procedure: During training, the model iterates over batches of images. For each batch, the generator creates 'fake' images from the input images.The discriminator is then updated to get better at distinguishing real images from these fakes. This update involves calculating the discriminator loss and adjusting the model parameters accordingly. Similarly, the generator is updated to improve its ability to create convincing fake images that the discriminator will classify as real. This update uses the generator loss, incorporating aspects like image fidelity and content preservation.\n\n- Efficiency and Monitoring: Mixed precision training, via torch.cuda.amp.GradScaler, is used here to make the training process faster and more memory efficient.The training loop includes logging of losses and periodic image display, which allows for monitoring the model's performance and progression visually.\n\n- Learning Rate Schedulers and Model Checkpoints: Learning rate schedulers adjust the learning rate during training, which can help in improving the model's training efficiency and effectiveness.The model's state is saved at regular intervals. This practice, known as checkpointing, is crucial for long training processes as it allows for recovery in case of interruptions and provides snapshots of the model at different stages of training.","metadata":{}},{"cell_type":"code","source":"def calculate_discriminator_loss(disc, real, fake, loss_function):\n    real_output = disc(real)\n    fake_output = disc(fake.detach())\n    \n    # Creating target tensors that match the output size of the discriminator\n    real_target = torch.ones_like(real_output)\n    fake_target = torch.zeros_like(fake_output)\n    \n    real_loss = loss_function(real_output, real_target)\n    fake_loss = loss_function(fake_output, fake_target)\n    \n    return real_loss + fake_loss\n\ndef calculate_generator_loss(disc, fake, loss_function, cycle_loss_function, identity_loss_function, real, cycle, identity, lambda_cycle, lambda_identity):\n    disc_output = disc(fake)\n    target = torch.ones_like(disc_output)\n\n    adversarial_loss = loss_function(disc_output, target)\n    cycle_loss = cycle_loss_function(real, cycle)\n    identity_loss = identity_loss_function(real, identity)\n\n    return adversarial_loss + lambda_cycle * cycle_loss + lambda_identity * identity_loss\n\n\ndef display_images(img, title, nrow=5):\n    with torch.no_grad():\n        show_images(img=img[:5], nrow=nrow, title=title)\n\ndef save_model(model, optimizer, file_name):\n    if hyperparams[\"save_model\"]:\n        save_checkpoint(model, optimizer, file_name=file_name)\n        \nd_scaler = torch.cuda.amp.GradScaler()\ng_scaler = torch.cuda.amp.GradScaler()\n\n\nD_LOSS = []\nG_LOSS = []\n\n# Training Loop\nfor epoch in range(config['num_epochs']):\n    loop = tqdm(train_loader, leave=True)\n    print(f\"EPOCH {epoch+1}\")\n\n    for idx, (monet, photo) in enumerate(loop):\n        monet, photo = monet.to(device), photo.to(device)\n\n        # DISCRIMINATOR UPDATE\n        with torch.cuda.amp.autocast():\n            fake_monet = gen_M(photo)\n            fake_photo = gen_P(monet)\n            d_m_loss = calculate_discriminator_loss(disc_M, monet, fake_monet, mse_loss)\n            d_p_loss = calculate_discriminator_loss(disc_P, photo, fake_photo, mse_loss)\n            d_loss = (d_m_loss + d_p_loss) / 2\n\n        opt_disc.zero_grad()\n        d_scaler.scale(d_loss).backward()\n        d_scaler.step(opt_disc)\n        d_scaler.update()\n\n        # GENERATOR UPDATE\n        with torch.cuda.amp.autocast():\n            g_m_loss = calculate_generator_loss(disc_M, fake_monet, mse_loss, L1_loss, L1_loss, monet, gen_P(fake_monet), gen_M(monet), hyperparams[\"lambda_cycle_consistency\"], hyperparams[\"lambda_identity\"])\n            g_p_loss = calculate_generator_loss(disc_P, fake_photo, mse_loss, L1_loss, L1_loss, photo, gen_M(fake_photo), gen_P(photo), hyperparams[\"lambda_cycle_consistency\"], hyperparams[\"lambda_identity\"])\n            g_loss = (g_m_loss + g_p_loss) / 2\n\n        opt_gen.zero_grad()\n        g_scaler.scale(g_loss).backward()\n        g_scaler.step(opt_gen)\n        g_scaler.update()\n\n        # LOGGING\n        D_LOSS.append(d_loss.item())\n        G_LOSS.append(g_loss.item())\n\n        if idx % 200 == 0:\n            loop.set_postfix(d_loss=d_loss.item(), g_loss=g_loss.item())\n\n    display_images(photo, \"Real Photos\")\n    display_images(fake_monet.type(torch.float64), \"Monets\")\n\n    lr_scheduler_D.step()\n    lr_scheduler_G.step()\n\n    # Save Checkpoints\n    save_model(disc_M, opt_disc, checkpoint_paths[\"discriminator_monet\"])\n    save_model(disc_P, opt_disc, checkpoint_paths[\"discriminator_photo\"])\n    save_model(gen_M, opt_gen, checkpoint_paths[\"generator_monet\"])\n    save_model(gen_P, opt_gen, checkpoint_paths[\"generator_photo\"])\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"## Loss plot\n\nplot(G_LOSS, D_LOSS)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Part 4: Results and conclusions\n","metadata":{}},{"cell_type":"markdown","source":"### Submission\n\nFor our submission we first standardize and transform the images into tensors, then we feed them into the generator model. The model generates new images based on its training (like style transfer). These generated images are then converted back to regular image format and saved in a the images directory. The process is done in batches to efficiently handle large sets of images, particularly useful for high volume image processing tasks while managing system memory usage effectively.","metadata":{}},{"cell_type":"code","source":"## Creating save_dir if it doesnt exsist\nsave_dir = '../images'\nif not os.path.exists(save_dir):\n    os.makedirs(save_dir)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import torch\nimport os\nfrom PIL import Image\nfrom torchvision import transforms\n\ngenerate_transforms = transforms.Compose([\n    transforms.ToTensor(),\n    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))\n])\n\nto_image = transforms.ToPILImage()\n\ngen_M.eval()\n\ndata_dir = \"/kaggle/input/gan-getting-started/photo_jpg\"\nfiles = [os.path.join(data_dir, name) for name in os.listdir(data_dir)]\n\nTensor = torch.cuda.FloatTensor if torch.cuda.is_available() else torch.Tensor\n\nfor i in range(0, len(files), batch_size):\n    imgs = torch.zeros((min(batch_size, len(files) - i), 3, IMAGE_HEIGHT, IMAGE_WIDTH)).type(Tensor)\n    \n    for j in range(batch_size):\n        idx = i + j\n        if idx >= len(files):\n            break\n        with Image.open(files[idx]) as img:\n            imgs[j] = generate_transforms(img)\n\n    fake_imgs = gen_M(imgs).detach().cpu()\n\n    for j, img in enumerate(fake_imgs):\n        to_image(img.squeeze()).save(os.path.join(save_dir, os.path.basename(files[i + j])))\n","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"## taking the files from images directory and converting them to a zip file\nimport shutil\nshutil.make_archive(\"/kaggle/working/images\", 'zip', \"/kaggle/images\")","metadata":{},"execution_count":null,"outputs":[]}]}