{"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\n\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,"execution":{"iopub.status.busy":"2024-12-21T05:45:24.719319Z","iopub.execute_input":"2024-12-21T05:45:24.719713Z","iopub.status.idle":"2024-12-21T05:45:30.745527Z","shell.execute_reply.started":"2024-12-21T05:45:24.719685Z","shell.execute_reply":"2024-12-21T05:45:30.744298Z"}},"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","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T05:45:30.746812Z","iopub.execute_input":"2024-12-21T05:45:30.747287Z","iopub.status.idle":"2024-12-21T05:45:32.843060Z","shell.execute_reply.started":"2024-12-21T05:45:30.747220Z","shell.execute_reply":"2024-12-21T05:45:32.841966Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"dataset_path = '/kaggle/input/gan-getting-started/'\nprint(dataset_path)\nprint(os.listdir(dataset_path))\nmonet_directory = 'monet_jpg'\nphoto_directory = 'photo_jpg'\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class MonetPhotoDataset(Dataset):\n    def __init__(self, root_dir, monet_dir, photo_dir, transform=None):\n        \"\"\"\n        Initialize the dataset with paths to Monet and photo images and optional transformations.\n\n        Args:\n            root_dir (str): Root directory containing the dataset.\n            monet_dir (str): Subdirectory with Monet images.\n            photo_dir (str): Subdirectory with photo images.\n            transform (callable, optional): Transformations to be applied on the images.\n        \"\"\"\n        self.root_dir = root_dir\n        self.monet_dir = monet_dir\n        self.photo_dir = photo_dir\n        self.monet_images = os.listdir(os.path.join(root_dir, monet_dir))\n        self.photo_images = os.listdir(os.path.join(root_dir, photo_dir))\n        self.transform = transform\n\n    def __len__(self):\n        return max(len(self.monet_images), len(self.photo_images))\n\n    def __getitem__(self, idx):\n        monet_image_path = os.path.join(\n            self.root_dir, self.monet_dir, self.monet_images[idx % len(self.monet_images)]\n        )\n        photo_image_path = os.path.join(\n            self.root_dir, self.photo_dir, self.photo_images[idx % len(self.photo_images)]\n        )\n\n        monet_image = Image.open(monet_image_path)\n        photo_image = Image.open(photo_image_path)\n\n        if self.transform:\n            monet_image = self.transform(monet_image)\n            photo_image = self.transform(photo_image)\n\n        return monet_image, photo_image\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"image_transforms = torchvision.transforms.Compose([\n    torchvision.transforms.Resize((256, 256)),\n    torchvision.transforms.ToTensor(),\n    torchvision.transforms.Normalize((0.5,), (0.5,)),\n])\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"dataset_path = '/kaggle/input/gan-getting-started/'\nmonet_subdir = 'monet_jpg'\nphoto_subdir = 'photo_jpg'\n\nmonet_photo_dataset = MonetPhotoDataset(\n    root_dir=dataset_path,\n    monet_dir=monet_subdir,\n    photo_dir=photo_subdir,\n    transform=image_transforms\n)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_len = int(len(monet_photo_dataset) * 0.8)\ntest_len = len(monet_photo_dataset) - train_len\n\ntrain_data, test_data = random_split(\n    monet_photo_dataset, [train_len, test_len]\n)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_loader = DataLoader(\n    train_data, batch_size=32, shuffle=True, num_workers=2\n)\n\ntest_loader = DataLoader(\n    test_data, batch_size=32, shuffle=True, num_workers=2\n)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"monet_batch, _ = next(iter(train_loader))\nmonet_image = monet_batch[0].permute(1, 2, 0).numpy() * 0.5 + 0.5\n\nplt.imshow(monet_image)\nplt.show()\n","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, kernel_size=4, padding_mode='reflect', 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            kernel_size (int, optional): Size of the convolutional kernel. Defaults to 4.\n            padding_mode (str, optional): Padding mode for the convolution. Defaults to 'reflect'.\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=kernel_size, \n                stride=stride, \n                bias=False, \n                padding=kernel_size // 2,  # Ensure same padding\n                padding_mode=padding_mode, \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        )\n\n    def forward(self, x):\n        \"\"\"\n        Forward pass for the CNNBlock.\n\n        Args:\n            x (torch.Tensor): Input tensor.\n\n        Returns:\n            torch.Tensor: Output tensor after applying the block.\n        \"\"\"\n        return self.conv(x)\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(Discriminator, self).__init__()  # Initialize the parent class\n\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            ),  # Initial convolutional layer\n            nn.LeakyReLU(0.2)  # Leaky ReLU activation\n        )\n\n        # Construct the intermediate layers\n        self.blocks = self._build_blocks(features)\n\n        # Final output layer to produce a single output (real/fake classification)\n        self.final = nn.Conv2d(\n            features[-1],  # Last feature size\n            1,  # Output one channel (real/fake classification)\n            kernel_size=4, \n            stride=1, \n            padding=1, \n            padding_mode='reflect'\n        )\n        \n    def _build_blocks(self, features):\n        \"\"\"\n        Builds the intermediate convolutional blocks with proper strides and padding.\n        \n        Args:\n            features (list): List of feature sizes for each layer.\n        \n        Returns:\n            nn.Sequential: A sequential model of CNN blocks.\n        \"\"\"\n        layers = []\n        in_channels = features[0]  # Set the initial number of input channels\n        for feature in features[1:]:\n            stride = 1 if feature == features[-1] else 2  # Use stride 1 for the last layer\n            layers.append(\n                CNNBlock(in_channels, feature, stride=stride, device=self.device)\n            )  # Add CNNBlock with the specified stride\n            in_channels = feature  # Update the number of input channels for the next block\n        return nn.Sequential(*layers)  # Return the combined sequential model\n\n    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        x = self.blocks(x)  # Apply the intermediate CNNBlocks\n        return self.final(x)  # Apply the final convolution for output\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(Block, self).__init__()\n\n        self.device = device  # Set the device\n        self.down = down  # Determine if this block is for downsampling or upsampling\n        self.use_dropout = use_dropout  # Whether to use dropout\n        self.act = act  # Activation function type\n        \n        # Conv layer (downsampling or upsampling)\n        self.conv_layer = self._get_conv_layer(in_channels, out_channels)\n        \n        # BatchNorm layer\n        self.batch_norm = nn.BatchNorm2d(out_channels, device=self.device)\n        \n        # Activation function\n        self.activation = self._get_activation()\n        \n        # Dropout layer\n        self.dropout = nn.Dropout(0.5) if self.use_dropout else None\n\n    def _get_conv_layer(self, in_channels, out_channels):\n        \"\"\"\n        Returns a Conv2d or ConvTranspose2d layer based on the downsampling flag.\n        \n        Args:\n            in_channels (int): Number of input channels.\n            out_channels (int): Number of output channels.\n        \n        Returns:\n            nn.Module: A convolutional layer (either Conv2d or ConvTranspose2d).\n        \"\"\"\n        if self.down:\n            return 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            )\n        else:\n            return 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            )\n\n    def _get_activation(self):\n        \"\"\"\n        Returns the activation function (ReLU or LeakyReLU) based on the specified activation type.\n        \n        Returns:\n            nn.Module: The activation function.\n        \"\"\"\n        if self.act == 'relu':\n            return nn.ReLU()\n        return nn.LeakyReLU(0.2)\n\n    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_layer(x)  # Apply the convolution (downsampling or upsampling)\n        x = self.batch_norm(x)  # Apply batch normalization\n        x = self.activation(x)  # Apply activation function\n        \n        if self.use_dropout:\n            x = self.dropout(x)  # Apply dropout if enabled\n        \n        return x\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\n        # Initial downsampling layer\n        self.initial_down = self._initial_downsampling(in_channels, features)\n\n        # Downsampling blocks\n        self.down_blocks = self._create_downsampling_blocks(features)\n\n        # Bottleneck layer\n        self.bottleneck = self._create_bottleneck(features)\n\n        # Upsampling blocks with skip connections\n        self.up_blocks = self._create_upsampling_blocks(features)\n\n    def _initial_downsampling(self, in_channels, features):\n        \"\"\"\n        Create the initial downsampling layer.\n        \n        Args:\n            in_channels (int): Number of input channels.\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.Sequential: The initial downsampling layer.\n        \"\"\"\n        return 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            ),\n            nn.LeakyReLU(0.2)\n        )\n\n    def _create_downsampling_blocks(self, features):\n        \"\"\"\n        Create a series of downsampling blocks.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.ModuleList: A list of downsampling blocks.\n        \"\"\"\n        down_blocks = []\n        for i in range(5):\n            in_channels = features * (2 ** i)\n            out_channels = features * (2 ** (i + 1))\n            down_blocks.append(Block(in_channels, out_channels, down=True, act='leaky', use_dropout=False, device=self.device))\n        return nn.ModuleList(down_blocks)\n\n    def _create_bottleneck(self, features):\n        \"\"\"\n        Create the bottleneck layer.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.Sequential: The bottleneck layer.\n        \"\"\"\n        return 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            ),\n            nn.ReLU()\n        )\n\n    def _create_upsampling_blocks(self, features):\n        \"\"\"\n        Create a series of upsampling blocks with skip connections.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.ModuleList: A list of upsampling blocks.\n        \"\"\"\n        up_blocks = []\n        in_channels = features * 8\n        for i in range(7):\n            out_channels = features * (8 // (2 ** i)) if i < 4 else features * (4 // (2 ** (i - 4)))\n            use_dropout = True if i < 3 else False\n            up_blocks.append(Block(in_channels, out_channels, down=False, act='relu', use_dropout=use_dropout, device=self.device))\n            in_channels = out_channels * 2  # Skip connections for concatenation\n        return nn.ModuleList(up_blocks)\n\n    def forward(self, x):\n        \"\"\"\n        Forward pass through the Generator.\n        \n        Args:\n            x (torch.Tensor): Input tensor (e.g., a latent vector or noise).\n        \n        Returns:\n            torch.Tensor: The generated image.\n        \"\"\"\n        # Initial downsampling\n        x = self.initial_down(x)\n\n        # Downsampling through blocks\n        for down_block in self.down_blocks:\n            x = down_block(x)\n\n        # Bottleneck layer\n        x = self.bottleneck(x)\n\n        # Upsampling through blocks\n        for up_block in self.up_blocks:\n            x = up_block(x)\n\n        return x\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__()\n        self.device = device\n\n        # Initial downsampling layer\n        self.initial_down = self._initial_downsampling(in_channels, features)\n\n        # Downsampling blocks\n        self.down_blocks = self._create_downsampling_blocks(features)\n\n        # Bottleneck layer\n        self.bottleneck = self._create_bottleneck(features)\n\n        # Upsampling blocks with skip connections\n        self.up_blocks = self._create_upsampling_blocks(features)\n\n        # Final upsampling layer\n        self.final_up = self._create_final_upsampling(features, in_channels)\n\n    def _initial_downsampling(self, in_channels, features):\n        \"\"\"\n        Create the initial downsampling layer.\n        \n        Args:\n            in_channels (int): Number of input channels.\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.Sequential: The initial downsampling layer.\n        \"\"\"\n        return 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            ),\n            nn.LeakyReLU(0.2)\n        )\n\n    def _create_downsampling_blocks(self, features):\n        \"\"\"\n        Create a series of downsampling blocks.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.ModuleList: A list of downsampling blocks.\n        \"\"\"\n        down_blocks = []\n        for i in range(5):\n            in_channels = features * (2 ** i)\n            out_channels = features * (2 ** (i + 1))\n            down_blocks.append(Block(in_channels, out_channels, down=True, act='leaky', use_dropout=False, device=self.device))\n        return nn.ModuleList(down_blocks)\n\n    def _create_bottleneck(self, features):\n        \"\"\"\n        Create the bottleneck layer.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.Sequential: The bottleneck layer.\n        \"\"\"\n        return 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            ),\n            nn.ReLU()\n        )\n\n    def _create_upsampling_blocks(self, features):\n        \"\"\"\n        Create a series of upsampling blocks with skip connections.\n        \n        Args:\n            features (int): Base number of feature channels.\n        \n        Returns:\n            nn.ModuleList: A list of upsampling blocks.\n        \"\"\"\n        up_blocks = []\n        in_channels = features * 8\n        for i in range(7):\n            out_channels = features * (8 // (2 ** i)) if i < 4 else features * (4 // (2 ** (i - 4)))\n            use_dropout = True if i < 3 else False\n            up_blocks.append(Block(in_channels, out_channels, down=False, act='relu', use_dropout=use_dropout, device=self.device))\n            in_channels = out_channels * 2  # Skip connections for concatenation\n        return nn.ModuleList(up_blocks)\n\n    def _create_final_upsampling(self, features, in_channels):\n        \"\"\"\n        Create the final upsampling layer.\n        \n        Args:\n            features (int): Base number of feature channels.\n            in_channels (int): Number of input channels.\n        \n        Returns:\n            nn.Sequential: The final upsampling layer.\n        \"\"\"\n        return nn.Sequential(\n            nn.ConvTranspose2d(\n                features * 2, \n                in_channels, \n                kernel_size=4, \n                stride=2, \n                padding=1, \n                device=self.device\n            ),\n            nn.Tanh()  # Use Tanh activation to scale the output between -1 and 1\n        )\n\n    def 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        # Downsampling through initial and downsampling blocks\n        d1 = self.initial_down(x)  # Apply initial downsampling\n        d2 = self.down_blocks[0](d1)  # Apply first downsampling Block\n        d3 = self.down_blocks[1](d2)  # Apply second downsampling Block\n        d4 = self.down_blocks[2](d3)  # Apply third downsampling Block\n        d5 = self.down_blocks[3](d4)  # Apply fourth downsampling Block\n        d6 = self.down_blocks[4](d5)  # Apply fifth downsampling Block\n        d7 = self.down_blocks[5](d6)  # Apply sixth downsampling Block\n\n        # Bottleneck layer\n        bottleneck = self.bottleneck(d7)\n\n        # Upsampling through blocks with skip connections\n        up1 = self.up_blocks[0](bottleneck)  # First upsampling Block\n        up2 = self.up_blocks[1](torch.cat([up1, d7], 1))  # Skip connection and second upsampling Block\n        up3 = self.up_blocks[2](torch.cat([up2, d6], 1))  # Skip connection and third upsampling Block\n        up4 = self.up_blocks[3](torch.cat([up3, d5], 1))  # Skip connection and fourth upsampling Block\n        up5 = self.up_blocks[4](torch.cat([up4, d4], 1))  # Skip connection and fifth upsampling Block\n        up6 = self.up_blocks[5](torch.cat([up5, d3], 1))  # Skip connection and sixth upsampling Block\n        up7 = self.up_blocks[6](torch.cat([up6, d2], 1))  # Skip connection and seventh upsampling Block\n\n        # Final upsampling with skip connection to generate the output image\n        return self.final_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        self.train_data = train_data\n        self.val_data = val_data\n        self.generator = generator\n        self.discriminator = discriminator\n        self.nb_epochs = nb_epochs\n        self.device = device\n        self.save_path = Path(save_path) if save_path else save_path\n        self.z = next(iter(self.val_data))[0][:32].to(self.device)  # Fixed validation samples\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        self.g_optimizer = torch.optim.Adam(\n            self.generator.parameters(), lr=lr, betas=betas\n        )\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":"    def train(self):\n        \"\"\"\n        Execute the training loop for the GAN, including training and validation phases.\n        \"\"\"\n        assert (self.g_optimizer is not None) and (self.d_optimizer is not None), \"Optimizers not initialized\"\n\n        if self.save_path and self.save_path.exists():\n            self.load_model()  # Load saved model weights if available\n\n        best_score = torch.inf  # Initialize the best validation score to infinity\n\n        for epoch in range(self.nb_epochs):\n            train_d_loss, train_g_loss, val_d_loss, val_g_loss = 0, 0, 0, 0\n            self.generator.train()\n            self.discriminator.train()\n\n            # Training phase\n            train_loop = tqdm(enumerate(self.train_data), total=len(self.train_data), desc=f\"Epoch {epoch+1}/{self.nb_epochs}\")\n            for step, (x, y) in train_loop:\n                x, y = x.to(self.device), y.to(self.device)\n                g_loss, d_loss = self.train_step(x, y)\n                train_g_loss += g_loss\n                train_d_loss += d_loss\n                train_loop.set_postfix_str(f\"g_loss: {train_g_loss/(step+1):.2f}, d_loss: {train_d_loss/(step+1):.2f}\")\n\n            # Validation phase\n            self.generator.eval()\n            self.discriminator.eval()\n            val_loop = tqdm(enumerate(self.val_data), total=len(self.val_data), desc=f\"Epoch {epoch+1}/{self.nb_epochs} validation\")\n            for step, (x, y) in val_loop:\n                x, y = x.to(self.device), y.to(self.device)\n                g_loss, d_loss = self.val_step(x, y)\n                val_g_loss += g_loss\n                val_d_loss += d_loss\n                val_loop.set_postfix_str(f\"g_loss: {val_g_loss/(step+1):.2f}, d_loss: {val_d_loss/(step+1):.2f}\")\n\n            # Save model if validation loss improves\n            if self.save_path and val_g_loss < best_score:\n                best_score = val_g_loss\n                self.save_model()\n\n            # Log metrics for the current epoch\n            self.log_metrics(epoch, train_g_loss, train_d_loss, val_g_loss, val_d_loss)\n            self.visualize_samples()  # Display sample images\n\n        return self.logs  # Return the training logs\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"    def train_step(self, x: torch.Tensor, y: torch.Tensor) -> tuple:\n        \"\"\"\n        Perform a single training step for both generator and discriminator.\n        \"\"\"\n        self.g_optimizer.zero_grad()\n        self.d_optimizer.zero_grad()\n\n        fake = self.generator(x)\n        fake_ = self.discriminator(fake)\n        real = self.discriminator(y)\n\n        # Discriminator loss (real vs fake)\n        d_loss = torch.nn.functional.mse_loss(real, torch.ones_like(real)) + torch.nn.functional.mse_loss(fake_, torch.zeros_like(fake_))\n        d_loss.backward()\n        self.d_optimizer.step()\n\n        # Generator loss (fake should be classified as real)\n        fake_ = self.discriminator(self.generator(x))\n        g_loss = torch.nn.functional.mse_loss(fake_, torch.ones_like(fake_))\n        g_loss.backward()\n        self.g_optimizer.step()\n\n        return g_loss.item(), d_loss.item()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"@torch.no_grad()\ndef val_step(self, x: torch.Tensor, y: torch.Tensor) -> 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(self.generator(x))\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","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!mkdir models\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nprint(\"CUDA Available:\", torch.cuda.is_available())\nprint(\"PyTorch Version:\", torch.__version__)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T07:13:03.319937Z","iopub.execute_input":"2024-12-21T07:13:03.320278Z","iopub.status.idle":"2024-12-21T07:13:09.399288Z","shell.execute_reply.started":"2024-12-21T07:13:03.320256Z","shell.execute_reply":"2024-12-21T07:13:09.397811Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","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)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\n\n# Misalnya logs berupa list of dictionaries\nlogs = [{\"Step\": 1, \"Train_g_loss\": 0.4, \"Val_g_loss\": 0.5, \"Train_d_loss\": 0.3, \"Val_d_loss\": 0.4},\n        {\"Step\": 2, \"Train_g_loss\": 0.6, \"Val_g_loss\": 0.7, \"Train_d_loss\": 0.5, \"Val_d_loss\": 0.6},\n        # Tambahkan lebih banyak data di sini\n       ]\n\n# Ubah menjadi DataFrame\nlogs_df = pd.DataFrame(logs)\n\n# Sekarang Anda bisa mengakses kolom-kolom dengan nama string\nprint(logs_df.head())\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T05:49:56.202239Z","iopub.execute_input":"2024-12-21T05:49:56.202524Z","iopub.status.idle":"2024-12-21T05:49:56.226206Z","shell.execute_reply.started":"2024-12-21T05:49:56.202505Z","shell.execute_reply":"2024-12-21T05:49:56.224955Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"fig, ax = plt.subplots(1, 2, figsize=(20, 4))\n\n# Plotting both the generator and discriminator loss for each subplot\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\")\n\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 titles\nax[0].set_title(\"Generator loss\")\nax[1].set_title(\"Discriminator loss\")\n\n# Adding legends to each subplot\nax[0].legend()\nax[1].legend()\n\n# Display the plot\nplt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T05:50:06.062309Z","iopub.execute_input":"2024-12-21T05:50:06.062587Z","iopub.status.idle":"2024-12-21T05:50:06.363149Z","shell.execute_reply.started":"2024-12-21T05:50:06.062568Z","shell.execute_reply":"2024-12-21T05:50:06.362371Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nfrom torch.utils.data import DataLoader\nimport torchvision\nimport torchvision.transforms as transforms\n\n# Transformasi untuk normalisasi gambar\ntransform = transforms.Compose([\n    transforms.ToTensor(),   # Mengubah gambar menjadi tensor\n    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))  # Normalisasi dengan mean dan std\n])\n\n# Menggunakan CIFAR-10 sebagai contoh dataset\ntrain_dataset = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)\n\n# Membuat DataLoader\ntrain_loader = DataLoader(train_dataset, batch_size=4, shuffle=True)\n\n# Memastikan DataLoader bekerja dengan baik\nphotos, labels = next(iter(train_loader))  # Mengambil batch pertama\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T05:55:42.399504Z","iopub.execute_input":"2024-12-21T05:55:42.399824Z","iopub.status.idle":"2024-12-21T05:55:57.679544Z","shell.execute_reply.started":"2024-12-21T05:55:42.399804Z","shell.execute_reply":"2024-12-21T05:55:57.678283Z"}},"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()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T05:57:31.765013Z","iopub.execute_input":"2024-12-21T05:57:31.765432Z","iopub.status.idle":"2024-12-21T05:57:32.092666Z","shell.execute_reply.started":"2024-12-21T05:57:31.765404Z","shell.execute_reply":"2024-12-21T05:57:32.091372Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport zipfile\nfrom PIL import Image\nimport numpy as np\n\n# Tentukan direktori untuk gambar yang akan disimpan\noutput_dir = '/path/to/output/images'\nzip_file_path = '/kaggle/working/images.zip'\n\n# Membuat direktori output jika belum ada\nos.makedirs(output_dir, exist_ok=True)\n\n# Fungsi untuk mengubah ukuran dan menyimpan gambar sebagai JPEG\ndef save_resized_images(num_images):\n    for i in range(num_images):\n        # Buat gambar acak sebagai contoh (anda bisa mengganti ini dengan dataset asli)\n        img_array = np.random.rand(256, 256, 3) * 255  # Gambar acak berukuran 256x256\n        img = Image.fromarray(img_array.astype(np.uint8))\n\n        # Menyimpan gambar dalam format JPEG\n        img.save(os.path.join(output_dir, f'{i}.jpg'), 'JPEG')\n\n# Tentukan jumlah gambar (antara 7000-10000 gambar)\nnum_images = 7000  # Ubah sesuai kebutuhan\n\n# Simpan gambar dengan ukuran 256x256\nsave_resized_images(num_images)\n\n# Membuat file ZIP untuk mengarsipkan gambar\nwith zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf:\n    # Menambahkan semua gambar ke dalam file ZIP\n    for i in range(num_images):\n        img_path = os.path.join(output_dir, f'{i}.jpg')\n        zipf.write(img_path, os.path.basename(img_path))\n\nprint(f\"File ZIP dengan {num_images} gambar telah dibuat di {zip_file_path}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T06:26:29.072036Z","iopub.execute_input":"2024-12-21T06:26:29.072412Z","iopub.status.idle":"2024-12-21T06:26:49.564010Z","shell.execute_reply.started":"2024-12-21T06:26:29.072385Z","shell.execute_reply":"2024-12-21T06:26:49.563135Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!mkdir ../images\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T07:14:05.298784Z","iopub.execute_input":"2024-12-21T07:14:05.299074Z","iopub.status.idle":"2024-12-21T07:14:05.434452Z","shell.execute_reply.started":"2024-12-21T07:14:05.299054Z","shell.execute_reply":"2024-12-21T07:14:05.433136Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nfrom torch.utils.data import DataLoader, Dataset\nfrom PIL import Image\nimport numpy as np\nimport os\n\n# Dummy Dataset Example (replace this with your actual dataset)\nclass DummyDataset(Dataset):\n    def __init__(self, transform=None):\n        self.transform = transform\n        # Example dummy data: 10 images of random noise\n        self.data = torch.randn(10, 3, 64, 64)  # 10 images of size (3, 64, 64)\n\n    def __len__(self):\n        return len(self.data)\n\n    def __getitem__(self, idx):\n        image = self.data[idx]\n        if self.transform:\n            image = self.transform(image)\n        return image, 0  # Return image and a dummy label\n\n# Create the output directory if it doesn't exist\nos.makedirs('../images', exist_ok=True)\n\n# Assuming 'trainer.generator' is the generator model\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# Create a dataset and DataLoader\ndataset = DummyDataset()  # Replace with your actual dataset\ndata_loader = DataLoader(dataset, batch_size=1, shuffle=False)\n\n# Make sure the model is in evaluation mode\ntrainer.generator.eval()\n\n# Disable gradient calculations\nwith torch.no_grad():\n    # Iterate over the dataset (batch_size=1 here)\n    for i, (img, _) in enumerate(data_loader):\n        # Transfer image to the device (GPU or CPU)\n        img = img.to(device)\n\n        # Generate the fake image using the generator\n        prediction = trainer.generator(img)[0].cpu().permute(1, 2, 0).numpy() * 0.5 + 0.5\n\n        # Convert the generated image to uint8 format\n        im = Image.fromarray((prediction * 255).astype(np.uint8))\n\n        # Save the image to the '../images/' directory with the current index as filename\n        im.save(f\"../images/{i}.jpg\")\n\nprint(\"Image generation and saving complete!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T07:14:23.170121Z","iopub.execute_input":"2024-12-21T07:14:23.170440Z","iopub.status.idle":"2024-12-21T07:14:23.204099Z","shell.execute_reply.started":"2024-12-21T07:14:23.170420Z","shell.execute_reply":"2024-12-21T07:14:23.202979Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import shutil\n\n# Create a zip archive of the '/kaggle/images' directory and save it as '/kaggle/working/images.zip'\nshutil.make_archive('/kaggle/working/images', 'zip', '/kaggle/images')\n\nprint(\"Archive created at: '/kaggle/working/images.zip'\")\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-21T06:00:28.642146Z","iopub.execute_input":"2024-12-21T06:00:28.642490Z","iopub.status.idle":"2024-12-21T06:00:28.652597Z","shell.execute_reply.started":"2024-12-21T06:00:28.642465Z","shell.execute_reply":"2024-12-21T06:00:28.651318Z"}},"outputs":[],"execution_count":null}]}