{"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":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 torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torchvision import transforms\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision.utils import save_image\nfrom PIL import Image\nfrom zipfile import ZipFile\nimport matplotlib.pyplot as plt","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Hyperparameters\nIMG_SIZE = 256\nBATCH_SIZE = 16\nLATENT_DIM = 100\nLEARNING_RATE = 0.0002\nEPOCHS = 30\nOUTPUT_DIR = \"../tmp/generated_images\"\nZIP_FILE = \"/kaggle/working/images.zip\"\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Dataset paths\nMONET_DIR = '/kaggle/input/gan-getting-started/monet_jpg/'\nPHOTO_DIR = '/kaggle/input/gan-getting-started/photo_jpg/'","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Transformations\ntransform = transforms.Compose([\n    transforms.Resize((IMG_SIZE, IMG_SIZE)),\n    transforms.ToTensor(),\n    transforms.Normalize([0.5], [0.5])  # Normalize to [-1, 1]\n])","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Custom Dataset\nclass ImageDataset(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) if fname.endswith('.jpg')]\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","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Load datasets\nmonet_dataset = ImageDataset(MONET_DIR, transform=transform)\nphoto_dataset = ImageDataset(PHOTO_DIR, transform=transform)\nmonet_loader = DataLoader(monet_dataset, batch_size=BATCH_SIZE, shuffle=True)\nphoto_loader = DataLoader(photo_dataset, batch_size=BATCH_SIZE, shuffle=True)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Generator\nclass Generator(nn.Module):\n    def __init__(self):\n        super(Generator, self).__init__()\n        self.encoder = nn.Sequential(\n            nn.Conv2d(3, 64, kernel_size=4, stride=2, padding=1),\n            nn.LeakyReLU(0.2, inplace=True),\n            nn.Conv2d(64, 128, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(128),\n            nn.LeakyReLU(0.2, inplace=True),\n            nn.Conv2d(128, 256, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(256),\n            nn.LeakyReLU(0.2, inplace=True)\n        )\n        self.decoder = nn.Sequential(\n            nn.ConvTranspose2d(256, 128, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(128),\n            nn.ReLU(inplace=True),\n            nn.ConvTranspose2d(128, 64, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(64),\n            nn.ReLU(inplace=True),\n            nn.ConvTranspose2d(64, 3, kernel_size=4, stride=2, padding=1),\n            nn.Tanh()\n        )\n\n    def forward(self, x):\n        x = self.encoder(x)\n        x = self.decoder(x)\n        return x","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Discriminator\nclass Discriminator(nn.Module):\n    def __init__(self):\n        super(Discriminator, self).__init__()\n        self.model = nn.Sequential(\n            nn.Conv2d(3, 64, kernel_size=4, stride=2, padding=1),\n            nn.LeakyReLU(0.2, inplace=True),\n            nn.Conv2d(64, 128, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(128),\n            nn.LeakyReLU(0.2, inplace=True),\n            nn.Conv2d(128, 256, kernel_size=4, stride=2, padding=1),\n            nn.BatchNorm2d(256),\n            nn.LeakyReLU(0.2, inplace=True),\n            nn.Conv2d(256, 1, kernel_size=4, stride=1, padding=1)\n        )\n\n    def forward(self, x):\n        return self.model(x)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Initialize models\ngenerator = Generator().to(DEVICE)\ndiscriminator = Discriminator().to(DEVICE)\n\n# Loss and Optimizers\nadversarial_loss = nn.MSELoss()\ncycle_loss = nn.L1Loss()\nidentity_loss = nn.L1Loss()\n\noptimizer_G = optim.Adam(generator.parameters(), lr=LEARNING_RATE, betas=(0.5, 0.999))\noptimizer_D = optim.Adam(discriminator.parameters(), lr=LEARNING_RATE, betas=(0.5, 0.999))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Training Loop\nos.makedirs(OUTPUT_DIR, exist_ok=True)\nimage_counter = 0\nwith ZipFile(ZIP_FILE, 'w') as zipf:\n    for epoch in range(EPOCHS):\n        for i, (photo_batch, monet_batch) in enumerate(zip(photo_loader, monet_loader)):\n            photo_batch, monet_batch = photo_batch.to(DEVICE), monet_batch.to(DEVICE)\n\n            # Train Generator\n            optimizer_G.zero_grad()\n            fake_monets = generator(photo_batch)\n            g_loss = (adversarial_loss(discriminator(fake_monets), torch.ones_like(discriminator(fake_monets))) +\n                     cycle_loss(generator(fake_monets), photo_batch) +\n                     identity_loss(generator(photo_batch), photo_batch))\n            g_loss.backward()\n            optimizer_G.step()\n\n            # Train Discriminator\n            optimizer_D.zero_grad()\n            real_validity = discriminator(monet_batch)\n            fake_validity = discriminator(fake_monets.detach())\n            d_loss = (adversarial_loss(real_validity, torch.ones_like(real_validity)) +\n                     adversarial_loss(fake_validity, torch.zeros_like(fake_validity))) / 2\n            d_loss.backward()\n            optimizer_D.step()\n\n            # Save images and add to ZIP\n            for j, img in enumerate(fake_monets):\n                if image_counter >= 7000:  # Stop once 7,000 images are generated\n                    break\n                img_path = os.path.join(OUTPUT_DIR, f\"img_{image_counter}.jpg\")\n                save_image(img, img_path)\n                zipf.write(img_path, arcname=f\"img_{image_counter}.jpg\")\n                os.remove(img_path)\n                image_counter += 1\n\n            if image_counter >= 7000:  # Stop the training loop once enough images are generated\n                break\n\n        if image_counter >= 7000:\n            break\n\n            # Print progress\n            print(f\"Epoch [{epoch+1}/{EPOCHS}], Batch [{i}], G_Loss: {g_loss.item()}, D_Loss: {d_loss.item()}\")\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Visualize Generated Images\ndef visualize_predictions(loader, generator, num_images=5):\n    generator.eval()\n    with torch.no_grad():\n        for i, photo_batch in enumerate(loader):\n            real_photos = photo_batch.to(DEVICE)\n            fake_monets = generator(real_photos)\n            fake_monets = (fake_monets * 0.5 + 0.5).clamp(0, 1)  # Denormalize to [0, 1]\n\n            for j in range(min(num_images, len(fake_monets))):\n                real_img = transforms.ToPILImage()(real_photos[j].cpu())\n                fake_img = transforms.ToPILImage()(fake_monets[j].cpu())\n\n                plt.figure(figsize=(8, 4))\n                plt.subplot(1, 2, 1)\n                plt.title(\"Real Photo\")\n                plt.imshow(real_img)\n                plt.axis(\"off\")\n\n                plt.subplot(1, 2, 2)\n                plt.title(\"Generated Monet\")\n                plt.imshow(fake_img)\n                plt.axis(\"off\")\n                plt.show()\n\n                num_images -= 1\n                if num_images <= 0:\n                    return\n\nvisualize_predictions(photo_loader, generator)","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}