{"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":"none","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"dockerImageVersionId":30822,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"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 os\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport torch\nimport torch.nn as nn\nimport torchvision\nfrom torch.utils.data import Dataset, DataLoader, random_split\nfrom torchvision.utils import make_grid\nfrom PIL import Image\n\nfrom tqdm import tqdm\nimport shutil\n\nfrom pathlib import Path\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"data_root = '/kaggle/input/gan-getting-started/'\nprint(data_root)\nprint(os.listdir(data_root))\nmonet_path = 'monet_jpg'\nphoto_path = 'photo_jpg'","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class MonetPhoto(Dataset):\n    def __init__(self, data_root, monet_path, photo_path, transform=None):\n        \"\"\"\n        Initialize the MonetPhoto dataset with paths to Monet and photo images and optional transformations.\n        \n        Args:\n            data_root (str): Root directory where the data is stored.\n            monet_path (str): Subdirectory within data_root containing Monet-style images.\n            photo_path (str): Subdirectory within data_root containing photo images.\n            transform (callable, optional): Optional transform to be applied on a sample.\n        \"\"\"\n        self.data_root = data_root  # Set the root directory for data\n        self.monet_path = monet_path  # Set the subdirectory for Monet images\n        self.monet_images = os.listdir(os.path.join(data_root, monet_path))  # List all Monet image filenames\n        self.photo_path = photo_path  # Set the subdirectory for photo images\n        self.photo_images = os.listdir(os.path.join(data_root, photo_path))  # List all photo image filenames\n        self.transform = transform  # Set the transformation to be applied on images\n\n    def __len__(self):\n        \"\"\"\n        Return the total number of samples in the dataset.\n        The length is the maximum of the number of Monet and photo images to ensure pairing.\n        \n        Returns:\n            int: Total number of samples.\n        \"\"\"\n        return max(len(self.monet_images), len(self.photo_images))  # Ensure all images are paired\n\n    def __getitem__(self, idx):\n        \"\"\"\n        Retrieve a sample from the dataset at the given index.\n        It pairs a Monet image with a photo image. If indices exceed the number of available images,\n        it wraps around using modulo operation.\n        \n        Args:\n            idx (int): Index of the sample to retrieve.\n        \n        Returns:\n            tuple: A pair of Monet and photo images after applying transformations.\n        \"\"\"\n        # Open the Monet image at the given index, wrapping around if idx exceeds length\n        monet_image = Image.open(os.path.join(self.data_root, self.monet_path, self.monet_images[idx % len(self.monet_images)]))\n        # Open the photo image at the given index, wrapping around if idx exceeds length\n        photo_image = Image.open(os.path.join(self.data_root, self.photo_path, self.photo_images[idx % len(self.photo_images)]))\n        \n        # Apply transformations if any are provided\n        if self.transform:\n            monet_image = self.transform(monet_image)  # Transform the Monet image\n            photo_image = self.transform(photo_image)  # Transform the photo image\n        \n        return monet_image, photo_image  # Return the transformed image pair\n\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Define a sequence of transformations to apply to the images\ntransforms = torchvision.transforms.Compose([\n    torchvision.transforms.Resize((256, 256)),  # Resize images to 256x256 pixels\n    torchvision.transforms.ToTensor(),  # Convert PIL images to PyTorch tensors\n    torchvision.transforms.Normalize(0.5, 0.5)  # Normalize tensor values to [-1, 1]\n])\n\n# Create an instance of the MonetPhoto dataset with the specified paths and transformations\ndataset = MonetPhoto(data_root, monet_path, photo_path, transform=transforms)\n\n# Get the total number of samples in the dataset\nlen(dataset)\n\n# Calculate the number of samples for training (80%) and testing (20%) splits\ntrain_size, test_size = int(len(dataset) * 0.8), len(dataset) - int(len(dataset) * 0.8)\n# Split the dataset into training and testing subsets\ntraining_data, testing_data = random_split(dataset, [train_size, test_size])\n\n# Get the number of samples in the training subset\nlen(training_data)\n\n# Create a DataLoader for the training data with a batch size of 32, shuffling enabled, and using 2 worker processes\ntrain_dataloader = DataLoader(training_data.dataset, batch_size=32, shuffle=True, num_workers=2)\n# Create a DataLoader for the testing data with a batch size of 32, shuffling enabled, and using 2 worker processes\ntest_dataloader = DataLoader(testing_data.dataset, batch_size=32, shuffle=True, num_workers=2)\n\n# Retrieve the first batch of Monet and photo images from the training DataLoader\nmones, photos = next(iter(train_dataloader))\n\n# Plot the first Monet image in the batch\nplt.imshow(mones[0].permute(1, 2, 0).numpy() * 0.5 + 0.5)  # Convert tensor to numpy array and denormalize for display","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class CNNBlock(nn.Module):\n    def __init__(self, in_channels, out_channels, stride=2, device=\"cuda\"):\n        \"\"\"\n        Initialize the CNNBlock with convolution, batch normalization, and activation layers.\n        \n        Args:\n            in_channels (int): Number of input channels.\n            out_channels (int): Number of output channels.\n            stride (int, optional): Stride for the convolution. Defaults to 2.\n            device (str, optional): Device to run the operations on ('cuda' or 'cpu'). Defaults to \"cuda\".\n        \"\"\"\n        super().__init__()  # Initialize the parent class\n        self.device = device  # Set the device\n        self.conv = nn.Sequential(\n            nn.Conv2d(\n                in_channels, \n                out_channels, \n                kernel_size=4, \n                stride=stride, \n                bias=False, \n                padding_mode='reflect', \n                device=self.device\n            ),  # Convolutional layer with specified parameters\n            nn.BatchNorm2d(out_channels, device=self.device),  # Batch normalization to stabilize and accelerate training\n            nn.LeakyReLU(0.2),  # Leaky ReLU activation with negative slope of 0.2\n        )","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def forward(self, x):\n        \"\"\"\n        Forward pass through the CNNBlock.\n        \n        Args:\n            x (torch.Tensor): Input tensor.\n        \n        Returns:\n            torch.Tensor: Output tensor after convolution, batch normalization, and activation.\n        \"\"\"\n        return self.conv(x)  # Apply the sequential layers to the input\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Discriminator(nn.Module):\n    def __init__(self, in_channels=3, features=[64, 128, 256, 512], device=\"cuda\"):\n        \"\"\"\n        Initialize the Discriminator model using a series of CNNBlocks.\n        \n        Args:\n            in_channels (int, optional): Number of input channels. Defaults to 3 (e.g., RGB images).\n            features (list, optional): List of feature sizes for each CNNBlock. Defaults to [64, 128, 256, 512].\n            device (str, optional): Device to run the operations on ('cuda' or 'cpu'). Defaults to \"cuda\".\n        \"\"\"\n        super().__init__()  # Initialize the parent class\n        self.device = device  # Set the device\n        self.initial = nn.Sequential(\n            nn.Conv2d(\n                in_channels, \n                features[0], \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                padding_mode='reflect', \n                device=self.device\n            ),  # Initial convolutional layer\n            nn.LeakyReLU(0.2)  # Leaky ReLU activation\n        )\n        \n        layers = []  # Initialize a list to hold subsequent CNNBlocks\n        in_channels = features[0]  # Set the initial number of input channels\n        for feature in features[1:]:\n            layers.append(\n                CNNBlock(in_channels, feature, stride=1 if feature == features[-1] else 2, device=self.device),\n            )  # Append CNNBlock with appropriate stride\n            in_channels = feature  # Update the number of input channels for the next block\n        \n        layers.append(\n            nn.Conv2d(\n                in_channels, \n                1, \n                kernel_size=4, \n                stride=1, \n                padding=1, \n                padding_mode='reflect', \n                device=self.device\n            )\n        )  # Final convolutional layer to produce a single output channel (real/fake classification)\n            \n        self.model = nn.Sequential(*layers)  # Combine all layers into a sequential model\n        ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":" def forward(self, x):\n        \"\"\"\n        Forward pass through the Discriminator.\n        \n        Args:\n            x (torch.Tensor): Input tensor (e.g., an image).\n        \n        Returns:\n            torch.Tensor: Output tensor representing the discriminator's prediction.\n        \"\"\"\n        x = self.initial(x)  # Apply the initial convolution and activation\n        return self.model(x)  # Apply the subsequent CNNBlocks and final convolution\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Block(nn.Module):\n    def __init__(self, in_channels, out_channels, down=True, act='relu', use_dropout=False, device=\"cuda\"):\n        \"\"\"\n        Initialize a Block, which can act as either a downsampling or upsampling layer.\n        \n        Args:\n            in_channels (int): Number of input channels.\n            out_channels (int): Number of output channels.\n            down (bool, optional): If True, performs downsampling using Conv2d; else, upsampling using ConvTranspose2d. Defaults to True.\n            act (str, optional): Activation function to use ('relu' or 'leaky'). Defaults to 'relu'.\n            use_dropout (bool, optional): Whether to include a dropout layer. Defaults to False.\n            device (str, optional): Device to run the operations on ('cuda' or 'cpu'). Defaults to \"cuda\".\n        \"\"\"\n        super().__init__()  # Initialize the parent class\n        self.device = device  # Set the device\n        self.conv = nn.Sequential(\n            nn.Conv2d(\n                in_channels, \n                out_channels, \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                bias=False, \n                padding_mode='reflect', \n                device=self.device\n            ) if down else nn.ConvTranspose2d(\n                in_channels, \n                out_channels, \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                bias=False, \n                device=self.device\n            ),  # Choose Conv2d for downsampling or ConvTranspose2d for upsampling\n            nn.BatchNorm2d(out_channels, device=self.device),  # Batch normalization\n            nn.ReLU() if act == 'relu' else nn.LeakyReLU(0.2),  # Activation function\n        )\n        self.use_dropout = use_dropout  # Set whether to use dropout\n        self.dropout = nn.Dropout(0.5)  # Define a dropout layer with 50% dropout rate\n    ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":" def forward(self, x):\n        \"\"\"\n        Forward pass through the Block.\n        \n        Args:\n            x (torch.Tensor): Input tensor.\n        \n        Returns:\n            torch.Tensor: Output tensor after convolution, normalization, activation, and optional dropout.\n        \"\"\"\n        x = self.conv(x)  # Apply convolution, normalization, and activation\n        return self.dropout(x) if self.use_dropout else x  # Apply dropout if enabled, else return the tensor\n\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Generator(nn.Module):\n    def __init__(self, in_channels=3, features=64, device=\"cuda\"):\n        \"\"\"\n        Initialize the Generator model using a series of downsampling and upsampling Blocks.\n        \n        Args:\n            in_channels (int, optional): Number of input channels. Defaults to 3 (e.g., RGB images).\n            features (int, optional): Base number of feature channels. Defaults to 64.\n            device (str, optional): Device to run the operations on ('cuda' or 'cpu'). Defaults to \"cuda\".\n        \"\"\"\n        super().__init__()  # Initialize the parent class\n        self.device = device  # Set the device\n        self.initial_down = nn.Sequential(\n            nn.Conv2d(\n                in_channels, \n                features, \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                padding_mode='reflect', \n                device=self.device\n            ),  # Initial convolutional layer\n            nn.LeakyReLU(0.2),  # Leaky ReLU activation\n        )  # Output size: 128\n        \n        # Define downsampling Blocks\n        self.down1 = Block(features, features*2, down=True, act='leaky', use_dropout=False)   # Output size: 64\n        self.down2 = Block(features*2, features*4, down=True, act='leaky', use_dropout=False) # Output size: 32\n        self.down3 = Block(features*4, features*8, down=True, act='leaky', use_dropout=False) # Output size: 16\n        self.down4 = Block(features*8, features*8, down=True, act='leaky', use_dropout=False) # Output size: 8\n        self.down5 = Block(features*8, features*8, down=True, act='leaky', use_dropout=False) # Output size: 4\n        self.down6 = Block(features*8, features*8, down=True, act='leaky', use_dropout=False) # Output size: 2\n        \n        # Bottleneck layer\n        self.bottleneck = nn.Sequential(\n            nn.Conv2d(\n                features*8, \n                features*8, \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                padding_mode='reflect', \n                device=self.device\n            ),  # Convolutional layer at the bottleneck\n            nn.ReLU(),  # ReLU activation\n        )  # Output size: 1x1\n        \n        # Define upsampling Blocks with skip connections\n        self.up1 = Block(features*8, features*8, down=False, act='relu', use_dropout=True)  # Output size: 2\n        self.up2 = Block(features*8*2, features*8, down=False, act='relu', use_dropout=True)  # Output size: 4\n        self.up3 = Block(features*8*2, features*8, down=False, act='relu', use_dropout=True)  # Output size: 8\n        self.up4 = Block(features*8*2, features*8, down=False, act='relu', use_dropout=False) # Output size: 16\n        self.up5 = Block(features*8*2, features*4, down=False, act='relu', use_dropout=False) # Output size: 32\n        self.up6 = Block(features*4*2, features*2, down=False, act='relu', use_dropout=False) # Output size: 64\n        self.up7 = Block(features*2*2, features, down=False, act='relu', use_dropout=False)   # Output size: 128\n\ndef forward(self, x):\n        \"\"\"\n        Forward pass through the Generator, implementing the U-Net architecture with skip connections.\n        \n        Args:\n            x (torch.Tensor): Input tensor (e.g., an image).\n        \n        Returns:\n            torch.Tensor: Output tensor representing the generated image.\n        \"\"\"\n        d1 = self.initial_down(x)  # Apply initial downsampling\n        d2 = self.down1(d1)  # Apply first downsampling Block\n        d3 = self.down2(d2)  # Apply second downsampling Block\n        d4 = self.down3(d3)  # Apply third downsampling Block\n        d5 = self.down4(d4)  # Apply fourth downsampling Block\n        d6 = self.down5(d5)  # Apply fifth downsampling Block\n        d7 = self.down6(d6)  # Apply sixth downsampling Block\n        \n        up1 = self.up1(bottleneck)  # Apply first upsampling Block\n        up2 = self.up2(torch.cat([up1, d7], 1))  # Concatenate with corresponding downsampled feature map and apply second upsampling Block\n        up3 = self.up3(torch.cat([up2, d6], 1))  # Concatenate with corresponding downsampled feature map and apply third upsampling Block\n        up4 = self.up4(torch.cat([up3, d5], 1))  # Concatenate with corresponding downsampled feature map and apply fourth upsampling Block\n        up5 = self.up5(torch.cat([up4, d4], 1))  # Concatenate with corresponding downsampled feature map and apply fifth upsampling Block\n        up6 = self.up6(torch.cat([up5, d3], 1))  # Concatenate with corresponding downsampled feature map and apply sixth upsampling Block\n        up7 = self.up7(torch.cat([up6, d2], 1))  # Concatenate with corresponding downsampled feature map and apply seventh upsampling Block\n\n        # Concatenate with initial downsampled feature map and apply final upsampling to generate the output image\n        return self.finil_up(torch.cat([up7, d1], 1))\n        ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Trainer:\n    def __init__(\n        self,\n        train_data: DataLoader,\n        val_data: DataLoader,\n        generator: torch.nn.Module,\n        discriminator: torch.nn.Module,\n        nb_epochs: int = 5,\n        device: str = \"cuda\",\n        save_path: str = None,\n    ):\n        \"\"\"\n        Initialize the Trainer with data loaders, models, training parameters, and device configuration.\n        \n        Args:\n            train_data (DataLoader): DataLoader for training data.\n            val_data (DataLoader): DataLoader for validation data.\n            generator (torch.nn.Module): Generator model.\n            discriminator (torch.nn.Module): Discriminator model.\n            nb_epochs (int, optional): Number of training epochs. Defaults to 5.\n            device (str, optional): Device to run the training on ('cuda' or 'cpu'). Defaults to \"cuda\".\n            save_path (str, optional): Path to save the trained models. Defaults to None.\n        \"\"\"\n        self.train_data = train_data        # Assign training data loader\n        self.val_data = val_data            # Assign validation data loader\n        self.generator = generator          # Assign generator model\n        self.discriminator = discriminator  # Assign discriminator model\n        self.nb_epochs = nb_epochs          # Set number of epochs\n        self.device = device                # Set device for computation\n        self.save_path = Path(save_path) if save_path else save_path  # Set save path as Path object if provided\n\n        # Initialize a fixed set of validation samples for monitoring generator progress\n        self.z = next(iter(self.val_data))[0][:32].to(self.device)\n        \n        # Initialize a dictionary to store training logs\n        self.logs = {\n            \"Step\": [],\n            \"Train_g_loss\": [],\n            \"Train_d_loss\": [],\n            \"Val_g_loss\": [],\n            \"Val_d_loss\": [],\n            \"Samples\": [],\n        }\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def init_optimizers(self, lr: float=3e-4, betas: tuple=(0.5, 0.999)):\n        \"\"\"\n        Initialize the optimizers for the generator and discriminator.\n        \n        Args:\n            lr (float, optional): Learning rate for the optimizers. Defaults to 3e-4.\n            betas (tuple, optional): Beta parameters for the Adam optimizer. Defaults to (0.5, 0.999).\n        \"\"\"\n        # Initialize Adam optimizer for the generator\n        self.g_optimizer = torch.optim.Adam(\n            self.generator.parameters(), lr=lr, betas=betas\n        )\n        # Initialize Adam optimizer for the discriminator\n        self.d_optimizer = torch.optim.Adam(\n            self.discriminator.parameters(), lr=lr, betas=betas\n        )\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n    def train(self):\n        \"\"\"\n        Execute the training loop for the GAN, including training and validation phases,\n        logging, and model checkpointing.\n        \n        Returns:\n            dict: Dictionary containing training logs.\n        \"\"\"\n        # Ensure that both optimizers are initialized before training\n        assert (self.g_optimizer is not None) and (\n            self.d_optimizer is not None\n        ), \"Please run Trainer().init_optimizer()\"\n\n        # If a saved model exists at the save path, load the model weights\n        if self.save_path and self.save_path.exists():\n            self.load_model()\n            \n        best_score = torch.inf  # Initialize the best validation score to infinity\n\n        # Iterate over each epoch\n        for i in range(self.nb_epochs):\n            # Initialize cumulative losses for the epoch\n            train_d_loss, train_g_loss, val_d_loss, val_g_loss = 0, 0, 0, 0\n            self.generator.train()      # Set generator to training mode\n            self.discriminator.train()  # Set discriminator to training mode\n            \n            # Training loop with progress bar\n            loop = tqdm(\n                enumerate(self.train_data),\n                desc=f\"Epoch {i + 1}/{self.nb_epochs} train\",\n                leave=False,\n                total=len(self.train_data),\n            )\n            for step, (x, y) in loop:\n                x = x.to(self.device)  # Move input data to the specified device\n                y = y.to(self.device)  # Move target data to the specified device\n\n                # Perform a training step and retrieve generator and discriminator losses\n                g_loss, d_loss = self.train_step(x, y)\n\n                # Accumulate the losses\n                train_g_loss += g_loss\n                train_d_loss += d_loss\n\n                # Update the progress bar with average losses\n                loop.set_postfix_str(\n                    f\"g_loss: {train_g_loss / (step + 1) :.2f}, d_loss: {train_d_loss / (step + 1) :.2f}\"\n                )\n\n            # Validation phase\n            self.generator.eval()  # Set generator to evaluation mode\n            self.discriminator.eval()  # Set discriminator to evaluation mode\n\n            # Validation loop with progress bar\n            loop = tqdm(\n                enumerate(self.val_data),\n                desc=f\"Epoch {i + 1}/{self.nb_epochs} validation\",\n                leave=True,\n                total=len(self.val_data),\n            )\n            for step, (x, y) in loop:\n                x = x.to(self.device)  # Move input data to the specified device\n                y = y.to(self.device)  # Move target data to the specified device\n\n                # Perform a validation step and retrieve generator and discriminator losses\n                g_loss, d_loss = self.val_step(x, y)\n\n                # Accumulate the validation losses\n                val_g_loss += g_loss\n                val_d_loss += d_loss\n\n                # Update the progress bar with average validation losses\n                loop.set_postfix_str(\n                    f\"g_loss: {val_g_loss / (step + 1) :.2f} d_loss: {val_d_loss / (step + 1) :.2f}\"\n                )\n\n            # Check if the current validation loss is the best so far\n            if self.save_path and best_score > val_g_loss:\n                best_score = val_g_loss  # Update the best score\n                self.save_model()  # Save the current model as the best model\n\n            # Log the metrics for the current epoch\n            self.log_metrics(\n                step=i,\n                train_g_loss=train_g_loss,\n                train_d_loss=train_d_loss,\n                val_g_loss=val_g_loss,\n                val_d_loss=val_d_loss,\n            )\n            # Generate and display sample images from the generator\n            fake_img = self.generator((photos[0].unsqueeze(0)).to('cuda'))\n            fake_img = fake_img[0].cpu().detach()\n            \n            # Plot the original photo and the generated fake image\n            fig, ax = plt.subplots(1, 2, figsize=(20, 4))\n            ax[0].imshow(photos[0].permute(1, 2, 0).numpy() * 0.5 + 0.5)\n            ax[0].set_title(\"Photo\")\n            ax[0].set(xticks=[], yticks=[])\n            ax[1].imshow(fake_img.permute(1, 2, 0).numpy() * 0.5 + 0.5)\n            ax[1].set_title(\"FakeMonet\")\n            ax[1].set(xticks=[], yticks=[])\n            plt.show()\n            \n        return self.logs  # Return the training logs\n\n    def train_step(\n        self,\n        x: torch.Tensor,\n        y: torch.Tensor\n    ) -> tuple:\n        \"\"\"\n        Perform a single training step for both generator and discriminator.\n        \n        Args:\n            x (torch.Tensor): Input tensor (e.g., photos).\n            y (torch.Tensor): Target tensor (e.g., Monet paintings).\n            \n        Returns:\n            tuple: Generator loss and discriminator loss.\n        \"\"\"\n        self.g_optimizer.zero_grad(set_to_none=True)  # Reset generator gradients\n        self.d_optimizer.zero_grad(set_to_none=True)  # Reset discriminator gradients\n        \n        # Generate fake images and classify them with the discriminator\n        fake_ = self.discriminator(\n            self.generator(x)\n        )\n        # Classify real images with the discriminator\n        real = self.discriminator(y)\n        # Calculate discriminator loss: real images should be classified as ones and fake as zeros\n        d_loss = (\n            torch.nn.functional.mse_loss(real, torch.ones_like(real, device=self.device)) + \n            torch.nn.functional.mse_loss(fake_, torch.zeros_like(fake_, device=self.device))\n        )\n\n        d_loss.backward()  # Backpropagate discriminator loss\n        self.d_optimizer.step()  # Update discriminator weights\n        \n        # Re-generate fake images for generator training\n        fake = self.discriminator(\n            self.generator(x)\n        )\n        \n        # Calculate generator loss: fake images should be classified as ones\n        g_loss = torch.nn.functional.mse_loss(fake, torch.ones_like(fake, device=self.device))\n        g_loss.backward()  # Backpropagate generator loss\n        self.g_optimizer.step()  # Update generator weights\n\n        return g_loss.item(), d_loss.item()  # Return scalar losses\n@torch.no_grad()\n    def val_step(\n        self,\n        x: torch.Tensor,\n        y: torch.Tensor,\n    ) -> tuple:\n        \"\"\"\n        Perform a single validation step without updating the model.\n        \n        Args:\n            x (torch.Tensor): Input tensor (e.g., photos).\n            y (torch.Tensor): Target tensor (e.g., Monet paintings).\n            \n        Returns:\n            tuple: Generator loss and discriminator loss.\n        \"\"\"\n        # Generate fake images and classify them with the discriminator\n        fake = self.discriminator(\n            self.generator(x)\n        )\n        # Classify real images with the discriminator\n        real = self.discriminator(y)\n        # Calculate generator loss: fake images should be classified as ones\n        g_loss = torch.nn.functional.mse_loss(fake, torch.ones_like(fake, device=self.device))\n        # Calculate discriminator loss: real images as ones and fake as zeros\n        d_loss = (\n            torch.nn.functional.mse_loss(real, torch.ones_like(real, device=self.device)) + \n            torch.nn.functional.mse_loss(fake, torch.zeros_like(fake, device=self.device))\n        )\n        return g_loss.item(), d_loss.item()  # Return scalar losses\n\n    @torch.no_grad()\n    def log_metrics(\n        self,\n        step: int,\n        train_g_loss: torch.Tensor,\n        train_d_loss: torch.Tensor,\n        val_g_loss: torch.Tensor,\n        val_d_loss: torch.Tensor,\n    ):\n        \"\"\"\n        Log the training and validation metrics for the current epoch.\n        \n        Args:\n            step (int): Current epoch number.\n            train_g_loss (torch.Tensor): Cumulative generator loss during training.\n            train_d_loss (torch.Tensor): Cumulative discriminator loss during training.\n            val_g_loss (torch.Tensor): Cumulative generator loss during validation.\n            val_d_loss (torch.Tensor): Cumulative discriminator loss during validation.\n        \"\"\"\n        self.logs[\"Step\"].append(step)  # Log the current epoch\n        self.logs[\"Train_g_loss\"].append(train_g_loss / len(self.train_data))  # Log average training generator loss\n        self.logs[\"Train_d_loss\"].append(train_d_loss / len(self.train_data))  # Log average training discriminator loss\n        self.logs[\"Val_g_loss\"].append(val_g_loss / len(self.val_data))  # Log average validation generator loss\n        self.logs[\"Val_d_loss\"].append(val_d_loss / len(self.val_data))  # Log average validation discriminator loss\n        self.logs[\"Samples\"].append(make_grid(self.generator(self.z).cpu() * 0.5 + 0.5, normalize=True))  # Log generated sample images\n\n    def save_model(self, full: bool = False):\n        \"\"\"\n        Save the generator and discriminator models.\n        \n        Args:\n            full (bool, optional): Whether to save the full model or just the state dictionaries. Defaults to False.\n        \"\"\"\n        if full:\n            # Save the entire generator and discriminator models\n            torch.save(self.generator, Path(self.save_path) / \"generator.pth\")\n            torch.save(self.discriminator, Path(self.save_path) / \"discriminator.pth\")\n        else:\n            # Save only the state dictionaries (weights) of the models\n            torch.save(\n                self.generator.state_dict(),\n                Path(self.save_path) / \"generator_weights.pth\",\n            )\n            torch.save(\n                self.discriminator.state_dict(),\n                Path(self.save_path) / \"discriminator_weights.pth\",\n            )\n\n    def load_model(self, full: bool = False):\n        \"\"\"\n        Load the generator and discriminator models.\n        \n        Args:\n            full (bool, optional): Whether to load the full model or just the state dictionaries. Defaults to False.\n        \"\"\"\n        if (\n            full\n            and (self.save_path / \"generator.pth\").is_file()\n            and (self.save_path / \"discriminator.pth\").is_file()\n        ):\n            # Load the entire generator and discriminator models\n            self.generator = torch.load(self.save_path / \"generator.pth\")\n            self.discriminator = torch.load(self.save_path / \"discriminator.pth\")\n        elif (\n            (self.save_path / \"generator_weights.pth\").is_file() and\n            (self.save_path / \"discriminator_weights.pth\").is_file()\n        ):\n            # Load only the state dictionaries (weights) of the models\n            self.generator.load_state_dict(torch.load(self.save_path / \"generator_weights.pth\"))\n            self.discriminator.load_state_dict(torch.load(self.save_path / \"discriminator_weights.pth\"))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T07:15:31.775324Z","iopub.execute_input":"2024-12-21T07:15:31.775770Z","iopub.status.idle":"2024-12-21T07:15:31.800490Z","shell.execute_reply.started":"2024-12-21T07:15:31.775729Z","shell.execute_reply":"2024-12-21T07:15:31.799106Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!mkdir models","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T07:15:38.865899Z","iopub.execute_input":"2024-12-21T07:15:38.866325Z","iopub.status.idle":"2024-12-21T07:15:39.000527Z","shell.execute_reply.started":"2024-12-21T07:15:38.866290Z","shell.execute_reply":"2024-12-21T07:15:38.998359Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nprint(\"CUDA Available:\", torch.cuda.is_available())\nprint(\"PyTorch Version:\", torch.__version__)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader\n\n# Definisi Generator\nclass Generator(nn.Module):\n    def __init__(self, in_channels, features):\n        super(Generator, self).__init__()\n        self.conv1 = nn.Conv2d(in_channels, features, kernel_size=3, stride=1, padding=1)\n        self.conv2 = nn.Conv2d(features, 3, kernel_size=3, stride=1, padding=1)\n\n    def forward(self, x):\n        x = torch.relu(self.conv1(x))\n        x = torch.tanh(self.conv2(x))\n        return x\n\n# Definisi Discriminator\nclass Discriminator(nn.Module):\n    def __init__(self):\n        super(Discriminator, self).__init__()\n        self.conv1 = nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1)\n        self.fc = nn.Linear(64 * 64 * 64, 1)  # Sesuaikan ukuran input sesuai data Anda\n\n    def forward(self, x):\n        x = torch.relu(self.conv1(x))\n        x = x.view(x.size(0), -1)  # Flattening\n        x = torch.sigmoid(self.fc(x))\n        return x\n\n# Definisi Trainer\nclass Trainer:\n    def __init__(self, train_data, val_data, generator, discriminator, nb_epochs, device, save_path):\n        self.train_data = train_data\n        self.val_data = val_data\n        self.generator = generator.to(device)\n        self.discriminator = discriminator.to(device)\n        self.nb_epochs = nb_epochs\n        self.device = device\n        self.save_path = save_path\n\n    def load_model(self):\n        # Implementasi untuk memuat model yang sudah dilatih sebelumnya (jika ada)\n        pass\n\n    def init_optimizers(self):\n        # Inisialisasi optimizer\n        self.g_optimizer = optim.Adam(self.generator.parameters(), lr=0.0002, betas=(0.5, 0.999))\n        self.d_optimizer = optim.Adam(self.discriminator.parameters(), lr=0.0002, betas=(0.5, 0.999))\n\n    def train(self):\n        logs = []  # Menyimpan log pelatihan\n        for epoch in range(self.nb_epochs):\n            # Implementasi pelatihan untuk setiap epoch\n            print(f\"Epoch {epoch + 1}/{self.nb_epochs}\")\n            # Simulasi log pelatihan\n            logs.append({'g_loss': 0.5, 'd_loss': 0.4})  # Contoh loss\n        return logs\n\n# Tentukan perangkat yang akan digunakan (GPU jika tersedia, jika tidak gunakan CPU)\ndevice = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n\n# Siapkan DataLoader\ntrain_dataloader = DataLoader(...)  # Gantilah dengan DataLoader Anda\ntest_dataloader = DataLoader(...)   # Gantilah dengan DataLoader Anda\n\n# Inisialisasi model dan trainer\ngenerator = Generator(in_channels=3, features=64).to(device)\ndiscriminator = Discriminator().to(device)\n\ntrainer = Trainer(\n    train_data=train_dataloader,\n    val_data=test_dataloader,\n    generator=generator,\n    discriminator=discriminator,\n    nb_epochs=20,\n    device=device,\n    save_path='./models/'\n)\n\n# Memulai pelatihan\nlogs = trainer.train()\nprint(logs)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\nimport matplotlib.pyplot as plt\n\n# Asumsikan logs adalah list of dictionaries\n# Contoh format logs yang benar\nlogs = [\n    {\"Step\": 1, \"Train_g_loss\": 0.4, \"Train_d_loss\": 0.3, \"Val_g_loss\": 0.5, \"Val_d_loss\": 0.6},\n    {\"Step\": 2, \"Train_g_loss\": 0.5, \"Train_d_loss\": 0.4, \"Val_g_loss\": 0.6, \"Val_d_loss\": 0.7},\n    # Tambahkan data log lainnya sesuai epoch\n]\n\n# Mengonversi logs ke DataFrame\nlogs_df = pd.DataFrame(logs)\n\n# Plotting\nfig, ax = plt.subplots(1, 2, figsize=(20, 4))\n\n# Plotkan masing-masing data loss\nax[0].plot(logs_df[\"Step\"], logs_df[\"Train_g_loss\"], label=\"Train_g_loss\")\nax[0].plot(logs_df[\"Step\"], logs_df[\"Val_g_loss\"], label=\"Val_g_loss\")\nax[1].plot(logs_df[\"Step\"], logs_df[\"Train_d_loss\"], label=\"Train_d_loss\")\nax[1].plot(logs_df[\"Step\"], logs_df[\"Val_d_loss\"], label=\"Val_d_loss\")\n\n# Set title dan legend\nax[0].set_title(\"Generator loss\")\nax[1].set_title(\"Discriminator loss\")\nax[0].legend()\nax[1].legend()\n\n# Menampilkan plot\nplt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"plt.imshow(photos[1].permute(1, 2, 0).numpy() * 0.5 + 0.5)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Generate the fake image on the correct device (CPU or GPU)\nfake_img = trainer.generator((photos[1].unsqueeze(0)).to(device))\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!mkdir ../images","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport matplotlib.pyplot as plt\n\n# Memastikan device yang digunakan adalah GPU jika tersedia, atau CPU jika tidak ada GPU\ndevice = \"cuda\" if torch.cuda.is_available() else \"cpu\"\n\n# Menambahkan dimensi batch pada foto dan memindahkannya ke perangkat (GPU/CPU)\nfake_img = trainer.generator(photos[1].unsqueeze(0).to(device))\n\n# Mengambil gambar pertama (karena hasil generator bisa berupa batch, kita ambil gambar pertama)\nfake_img = fake_img[0].cpu().detach()\n\n# Menampilkan gambar hasil generator\nplt.imshow(fake_img.permute(1, 2, 0).numpy() * 0.5 + 0.5)\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"with torch.no_grad():\n    for i, (_, img) in enumerate(DataLoader(dataset, batch_size=1, shuffle=False)):\n        output = trainer.generator(img.to(device))  # Output: (1, 64, H, W)\n        \n        # Ambil channel pertama untuk grayscale\n        prediction = output[0, 0].cpu().numpy() * 0.5 + 0.5  # Normalisasi ke [0, 1]\n        im = Image.fromarray((prediction * 255).astype(np.uint8), mode=\"L\")  # Mode \"L\" untuk grayscale\n        im.save(f\"../images/{i}.jpg\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import shutil\nshutil.make_archive(\"/kaggle/working/images\", 'zip', \"/kaggle/images\")\n","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}