{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.14","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"dockerImageVersionId":30804,"isInternetEnabled":false,"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 torch\nimport torchvision","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nfrom torchvision import datasets, transforms\nfrom torch.utils.data import DataLoader, Dataset\nfrom PIL import Image\n\nclass GANDataset(Dataset):\n    def __init__(self, root_dir, transform=None):\n        self.root_dir = root_dir\n        self.transform = transform\n        self.image_paths = [os.path.join(root_dir, fname) for fname in os.listdir(root_dir)]\n    \n    def __len__(self):\n        return len(self.image_paths)\n    \n    def __getitem__(self, idx):\n        img_path = self.image_paths[idx]\n        image = Image.open(img_path).convert(\"RGB\")\n        if self.transform:\n            image = self.transform(image)\n        return image\n\ntransform = transforms.Compose([\n    transforms.Resize(int(256 * 1.12)),\n    transforms.RandomCrop(256),\n    transforms.RandomHorizontalFlip(),\n    transforms.ToTensor(),\n    transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))\n])\n\nmonet_dataset = GANDataset(root_dir='/kaggle/input/gan-getting-started/monet_jpg/', transform=transform)\nmonet_loader = DataLoader(monet_dataset, batch_size=4, shuffle=True)\n\nphoto_dataset = GANDataset(root_dir='/kaggle/input/gan-getting-started/photo_jpg/', transform=transform)\nphoto_loader = DataLoader(photo_dataset, batch_size=4, shuffle=True)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# import matplotlib.pyplot as plt\n\n# # Function to show an image\n# def imshow(img):\n#     img = img / 2 + 0.5  # unnormalize\n#     npimg = img.numpy()\n#     plt.imshow(np.transpose(npimg, (1, 2, 0)))\n#     plt.show()\n\n# # Get a batch of training data\n# dataiter = iter(photo_loader)\n# images = next(dataiter)\n\n# # Show images\n# imshow(torchvision.utils.make_grid(images))","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\nfrom torchvision import datasets, transforms\nfrom torchvision.utils import save_image\n\nclass ResidualBlock(nn.Module):\n    def __init__(self, in_channels):\n        super(ResidualBlock, self).__init__()\n        self.conv_block = nn.Sequential(\n            nn.Conv2d(in_channels, in_channels, kernel_size=3, stride=1, padding=1),\n            nn.InstanceNorm2d(in_channels),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(in_channels, in_channels, kernel_size=3, stride=1, padding=1),\n            nn.InstanceNorm2d(in_channels)\n        )\n\n    def forward(self, x):\n        return x + self.conv_block(x)\n\nclass Generator(nn.Module):\n    def __init__(self, input_channels, output_channels):\n        super(Generator, self).__init__()\n        self.model = nn.Sequential(\n            nn.Conv2d(input_channels, 64, kernel_size=7, stride=1, padding=3),\n            nn.InstanceNorm2d(64),\n            nn.ReLU(inplace=True),\n            self.downsample(64, 128),\n            self.downsample(128, 256),\n            *[ResidualBlock(256) for _ in range(9)],\n            self.upsample(256, 128),\n            self.upsample(128, 64),\n            nn.Conv2d(64, output_channels, kernel_size=7, stride=1, padding=3),\n            nn.Tanh()\n        )\n\n    def downsample(self, in_channels, out_channels):\n        return nn.Sequential(\n            nn.Conv2d(in_channels, out_channels, kernel_size=3, stride=2, padding=1),\n            nn.InstanceNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        )\n\n    def upsample(self, in_channels, out_channels):\n        return nn.Sequential(\n            nn.ConvTranspose2d(in_channels, out_channels, kernel_size=3, stride=2, padding=1, output_padding=1),\n            nn.InstanceNorm2d(out_channels),\n            nn.ReLU(inplace=True)\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\nclass Discriminator(nn.Module):\n    def __init__(self, input_channels):\n        super(Discriminator, self).__init__()\n        self.model = nn.Sequential(\n            self.discriminator_block(input_channels, 64, normalization=False),\n            self.discriminator_block(64, 128),\n            self.discriminator_block(128, 256),\n            self.discriminator_block(256, 512),\n            nn.Conv2d(512, 1, kernel_size=3, stride=1, padding=1)\n        )\n\n    def discriminator_block(self, in_channels, out_channels, normalization=True):\n        layers = [nn.Conv2d(in_channels, out_channels, kernel_size=3, stride=2, padding=1)]\n        if normalization:\n            layers.append(nn.InstanceNorm2d(out_channels))\n        layers.append(nn.LeakyReLU(0.2, inplace=True))\n        return nn.Sequential(*layers)\n\n    def forward(self, x):\n        return self.model(x)\n\ndef weights_init_normal(m):\n    classname = m.__class__.__name__\n    if classname.find('Conv') != -1:\n        nn.init.normal_(m.weight.data, 0.0, 0.02)\n    elif classname.find('BatchNorm2d') != -1:\n        nn.init.normal_(m.weight.data, 1.0, 0.02)\n        nn.init.constant_(m.bias.data, 0.0)\n\n# Hyperparameters\ninput_channels = 3\noutput_channels = 3\nepochs = 100\nbatch_size = 1\nlr = 0.0002\nlambda_cycle = 10.0\nlambda_identity = 0.5 * lambda_cycle","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"G_Monet = Generator(input_channels=3, output_channels=3)\nG_Photo = Generator(input_channels=3, output_channels=3)\nD_Monet = Discriminator(input_channels=3)\nD_Photo = Discriminator(input_channels=3)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nG_Monet.to(device)\nG_Photo.to(device)\nD_Monet.to(device)\nD_Photo.to(device)\n\nG_Monet.apply(weights_init_normal)\nG_Photo.apply(weights_init_normal)\nD_Monet.apply(weights_init_normal)\nD_Photo.apply(weights_init_normal)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch.optim as optim\nimport itertools\n# Losses\ncriterion_GAN = nn.MSELoss()\ncriterion_cycle = nn.L1Loss()\ncriterion_identity = nn.L1Loss()\n\n# Optimizers\noptimizer_G = optim.Adam(itertools.chain(G_Monet.parameters(), G_Photo.parameters()), lr=0.0002, betas=(0.5, 0.999))\noptimizer_D_Monet = optim.Adam(D_Monet.parameters(), lr=0.0002, betas=(0.5, 0.999))\noptimizer_D_Photo = optim.Adam(D_Photo.parameters(), lr=0.0002, betas=(0.5, 0.999))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"num_epochs = 5  # Define the number of epochs\nfor epoch in range(num_epochs):\n    for i, (real_photos, real_monets) in enumerate(zip(photo_loader, monet_loader)):\n        # Move data to GPU if available\n        real_photos, real_monets = real_photos.to(device), real_monets.to(device)\n\n        # Train Generators\n        optimizer_G.zero_grad()\n\n        # Generate Monet-style photos and reconstructed photos\n        fake_monets = G_Monet(real_photos)\n        reconstructed_photos = G_Photo(fake_monets)\n\n        # Generate Photo-style Monets and reconstructed Monets\n        fake_photos = G_Photo(real_monets)\n        reconstructed_monets = G_Monet(fake_photos)\n\n        # Calculate Generator losses\n        loss_GAN_Monet = criterion_GAN(D_Monet(fake_monets), torch.ones_like(D_Monet(fake_monets)))\n        loss_GAN_Photo = criterion_GAN(D_Photo(fake_photos), torch.ones_like(D_Photo(fake_photos)))\n        loss_cycle_photo = criterion_cycle(reconstructed_photos, real_photos)\n        loss_cycle_monet = criterion_cycle(reconstructed_monets, real_monets)\n        loss_identity_photo = criterion_identity(G_Photo(real_photos), real_photos)\n        loss_identity_monet = criterion_identity(G_Monet(real_monets), real_monets)\n\n        loss_G = loss_GAN_Monet + loss_GAN_Photo + loss_cycle_photo + loss_cycle_monet + loss_identity_photo + loss_identity_monet\n\n        # Backward pass for Generators\n        loss_G.backward()\n\n        # Apply gradient clipping for Generators\n        torch.nn.utils.clip_grad_norm_(G_Monet.parameters(), 1.0)\n        torch.nn.utils.clip_grad_norm_(G_Photo.parameters(), 1.0)\n\n        optimizer_G.step()\n\n        # Train Discriminators\n        optimizer_D_Monet.zero_grad()\n        loss_D_Monet_real = criterion_GAN(D_Monet(real_monets), torch.ones_like(D_Monet(real_monets)))\n        loss_D_Monet_fake = criterion_GAN(D_Monet(fake_monets.detach()), torch.zeros_like(D_Monet(fake_monets)))\n        loss_D_Monet = (loss_D_Monet_real + loss_D_Monet_fake) * 0.5\n\n        # Backward pass for Discriminator Monet\n        loss_D_Monet.backward()\n\n        # Apply gradient clipping for Discriminator Monet\n        torch.nn.utils.clip_grad_norm_(D_Monet.parameters(), 1.0)\n\n        optimizer_D_Monet.step()\n\n        optimizer_D_Photo.zero_grad()\n        loss_D_Photo_real = criterion_GAN(D_Photo(real_photos), torch.ones_like(D_Photo(real_photos)))\n        loss_D_Photo_fake = criterion_GAN(D_Photo(fake_photos.detach()), torch.zeros_like(D_Photo(fake_photos)))\n        loss_D_Photo = (loss_D_Photo_real + loss_D_Photo_fake) * 0.5\n\n        # Backward pass for Discriminator Photo\n        loss_D_Photo.backward()\n\n        # Apply gradient clipping for Discriminator Photo\n        torch.nn.utils.clip_grad_norm_(D_Photo.parameters(), 1.0)\n\n        optimizer_D_Photo.step()\n\n    print(f\"Epoch [{epoch+1}/{num_epochs}], Generator Loss: {loss_G.item()}, D_Monet Loss: {loss_D_Monet.item()}, D_Photo Loss: {loss_D_Photo.item()}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from PIL import Image\nimport zipfile\n\n# Directory to save generated images\noutput_dir = 'images'\nos.makedirs(output_dir, exist_ok=True)\n\n# Set the model to evaluation mode\nG_Monet.eval()\n\n# Set the number of images to generate (between 7,000-10,000)\ntotal_images = 7000\nbatch_size = 4  # Your batch size\nimages_generated = 0\n\n# Iterate through the DataLoader\nwhile images_generated < total_images:\n    for i, data in enumerate(photo_loader):\n        real_photo = data.to(device)\n        with torch.no_grad():\n            fake_monet_batch = G_Monet(real_photo)\n        \n        # Process each image in the batch\n        for j, fake_monet in enumerate(fake_monet_batch):\n            fake_monet = fake_monet.permute(1, 2, 0)  # Rearrange from CxHxW to HxWxC\n            fake_monet = fake_monet.cpu().numpy()\n            fake_monet = (fake_monet * 0.5 + 0.5) * 255  # Unnormalize\n            fake_monet = fake_monet.astype(np.uint8)  # Convert to uint8\n\n            img = Image.fromarray(fake_monet)\n            img_path = os.path.join(output_dir, f'image_{images_generated + 1}.png')\n            img.save(img_path)\n            images_generated += 1\n\n            # Break if we reach the total_images\n            if images_generated >= total_images:\n                break\n        if images_generated >= total_images:\n            break\n\nprint(f\"Generated {images_generated}/{total_images} images saved to {output_dir}\")\n\n# Zip all the images into a zip file\nzip_filename = \"images.zip\"\nwith zipfile.ZipFile(zip_filename, 'w', zipfile.ZIP_DEFLATED) as img_zip:\n    for filename in os.listdir(output_dir):\n        img_zip.write(os.path.join(output_dir, filename), arcname=filename)\n\nprint(f\"All images zipped into {zip_filename}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}