{"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"},{"sourceId":10228317,"sourceType":"datasetVersion","datasetId":6323854}],"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,"execution":{"iopub.status.busy":"2024-12-18T09:40:07.874722Z","iopub.execute_input":"2024-12-18T09:40:07.875107Z","iopub.status.idle":"2024-12-18T09:40:12.221282Z","shell.execute_reply.started":"2024-12-18T09:40:07.875074Z","shell.execute_reply":"2024-12-18T09:40:12.220165Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Setup\n\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nimport itertools\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision.utils import save_image\nfrom torch.autograd.variable import Variable \nfrom torchvision import transforms, datasets\nimport matplotlib.pyplot as plt\nimport os\nimport numpy as np\nfrom PIL import Image\nimport zipfile\n\nprint(\"Success\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-18T09:40:21.641170Z","iopub.execute_input":"2024-12-18T09:40:21.641645Z","iopub.status.idle":"2024-12-18T09:40:24.923225Z","shell.execute_reply.started":"2024-12-18T09:40:21.641611Z","shell.execute_reply":"2024-12-18T09:40:24.921929Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================\n# 2. Helper Functions\n# =============================================\n# Dataset Class\nclass GANDataset(Dataset):\n    def __init__(self, root_dir, transform=None):\n        self.root_dir = root_dir\n        self.transform = transform\n        # Filter only image files (e.g., .jpg, .jpeg, .png)\n        self.image_paths = [\n            os.path.join(root_dir, fname)\n            for fname in os.listdir(root_dir)\n            if fname.lower().endswith(('.jpg', '.jpeg', '.png'))\n        ]\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\n# Common Transformations\ndef get_transform():\n    return transforms.Compose([\n        transforms.Resize(int(256 * 1.12)),  # Resize to 1.12 times 256\n        transforms.RandomCrop(256),          # Random Crop to 256x256\n        transforms.RandomHorizontalFlip(),   # Random Horizontal Flip\n        transforms.ToTensor(),               # Convert to Tensor\n        transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))  # Normalize to [-1, 1]\n    ])\n\n# DataLoader Function\ndef create_dataloader(root_dir, transform, batch_size=4, shuffle=True):\n    dataset = GANDataset(root_dir=root_dir, transform=transform)\n    return DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)\n\n# Paths\nmonet_path = '/kaggle/input/gan-getting-started/monet_jpg/'\nphoto_path = '/kaggle/input/gan-getting-started/photo_jpg/'\n\n# Transform\ntransform = get_transform()\n\n# DataLoaders\nmonet_loader = create_dataloader(root_dir=monet_path, transform=transform)\nphoto_loader = create_dataloader(root_dir=photo_path, transform=transform)\n\n# Example Usage: Iterate over batches\nfor images in monet_loader:\n    print(images.shape)  # Example output: torch.Size([4, 3, 256, 256])\n    break\nprint(\"Success\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-18T09:40:26.779118Z","iopub.execute_input":"2024-12-18T09:40:26.779926Z","iopub.status.idle":"2024-12-18T09:40:26.980050Z","shell.execute_reply.started":"2024-12-18T09:40:26.779851Z","shell.execute_reply":"2024-12-18T09:40:26.978978Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Residual Block\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\n# Generator\nclass Generator(nn.Module):\n    def __init__(self, input_channels, output_channels, num_residual_blocks=9):\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(num_residual_blocks)],\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\n# Discriminator\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\n# Weight Initialization\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\nhyperparams = {\n    'input_channels': 3,\n    'output_channels': 3,\n    'epochs': 100,\n    'batch_size': 1,\n    'lr': 0.0002,\n    'lambda_cycle': 10.0,\n    'lambda_identity': 5.0  # 0.5 * lambda_cycle\n}\n\n# Device Configuration\ndef get_device():\n    return torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# Model Builder\ndef build_models(hyperparams, device):\n    G_Monet = Generator(input_channels=hyperparams['input_channels'], output_channels=hyperparams['output_channels'])\n    G_Photo = Generator(input_channels=hyperparams['input_channels'], output_channels=hyperparams['output_channels'])\n    D_Monet = Discriminator(input_channels=hyperparams['input_channels'])\n    D_Photo = Discriminator(input_channels=hyperparams['input_channels'])\n\n    # Move to device\n    G_Monet.to(device)\n    G_Photo.to(device)\n    D_Monet.to(device)\n    D_Photo.to(device)\n\n    # Apply weight initialization\n    G_Monet.apply(weights_init_normal)\n    G_Photo.apply(weights_init_normal)\n    D_Monet.apply(weights_init_normal)\n    D_Photo.apply(weights_init_normal)\n\n    return G_Monet, G_Photo, D_Monet, D_Photo\n\n# Instantiate Models\ndevice = get_device()\nG_Monet, G_Photo, D_Monet, D_Photo = build_models(hyperparams, device)\n\n# Print device and model summary\nprint(f\"Device: {device}\")\nprint(\"Models built and initialized.\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-18T09:41:00.477777Z","iopub.execute_input":"2024-12-18T09:41:00.478244Z","iopub.status.idle":"2024-12-18T09:41:00.994194Z","shell.execute_reply.started":"2024-12-18T09:41:00.478197Z","shell.execute_reply":"2024-12-18T09:41:00.992760Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"criterion_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))\n\ndef train_generators(real_photos, real_monets):\n    \"\"\"Train the generators and calculate total loss.\"\"\"\n    optimizer_G.zero_grad()\n\n    # Forward Pass: Generate and Reconstruct\n    fake_monets = G_Monet(real_photos)\n    reconstructed_photos = G_Photo(fake_monets)\n    fake_photos = G_Photo(real_monets)\n    reconstructed_monets = G_Monet(fake_photos)\n\n    # 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    # Total Generator Loss\n    loss_G = (\n        loss_GAN_Monet + loss_GAN_Photo \n        + loss_cycle_photo + loss_cycle_monet \n        + loss_identity_photo + loss_identity_monet\n    )\n\n    # Backward and Optimize\n    loss_G.backward()\n    gradient_clipping([G_Monet, G_Photo])\n    optimizer_G.step()\n\n    return loss_G.item()\n\ndef train_discriminator(discriminator, real_images, fake_images, optimizer):\n    \"\"\"Train a single discriminator.\"\"\"\n    optimizer.zero_grad()\n\n    # Real loss\n    pred_real = discriminator(real_images)\n    loss_real = criterion_GAN(pred_real, torch.ones_like(pred_real))\n\n    # Fake loss\n    pred_fake = discriminator(fake_images.detach())\n    loss_fake = criterion_GAN(pred_fake, torch.zeros_like(pred_fake))\n\n    # Total Discriminator Loss\n    loss_D = (loss_real + loss_fake) * 0.5\n\n    # Backward and Optimize\n    loss_D.backward()\n    gradient_clipping([discriminator])\n    optimizer.step()\n\n    return loss_D.item()\n\ndef gradient_clipping(models, max_norm=1.0):\n    \"\"\"Apply gradient clipping to prevent exploding gradients.\"\"\"\n    for model in models:\n        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm)\n\n# Training Loop\nnum_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        loss_G = train_generators(real_photos, real_monets)\n\n        # Train Discriminators\n        loss_D_Monet = train_discriminator(D_Monet, real_monets, G_Monet(real_photos), optimizer_D_Monet)\n        loss_D_Photo = train_discriminator(D_Photo, real_photos, G_Photo(real_monets), optimizer_D_Photo)\n\n    # Logging\n    print(f\"Epoch [{epoch+1}/{num_epochs}], \"\n          f\"Generator Loss: {loss_G:.4f}, \"\n          f\"D_Monet Loss: {loss_D_Monet:.4f}, \"\n          f\"D_Photo Loss: {loss_D_Photo:.4f}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-18T09:42:01.803411Z","iopub.execute_input":"2024-12-18T09:42:01.803850Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport torch\nimport numpy as np\nfrom PIL import Image\nimport zipfile\n\n# Directory to save generated images\nOUTPUT_DIR = 'images'\nos.makedirs(OUTPUT_DIR, exist_ok=True)\n\ndef save_generated_images(generator, data_loader, total_images, device, output_dir):\n\n    generator.eval()\n    images_generated = 0\n    batch_size = data_loader.batch_size\n\n    with torch.no_grad():\n        for batch in data_loader:\n            real_photo = batch.to(device)\n            fake_monet_batch = generator(real_photo)\n\n            for i in range(fake_monet_batch.size(0)):\n                fake_monet = fake_monet_batch[i].permute(1, 2, 0).cpu().numpy()\n                fake_monet = (fake_monet * 0.5 + 0.5) * 255  # Unnormalize\n                fake_monet = fake_monet.astype(np.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                if images_generated >= total_images:\n                    return images_generated\n    return images_generated\n\ndef zip_images(output_dir, zip_filename):\n\n    with zipfile.ZipFile(zip_filename, 'w', zipfile.ZIP_DEFLATED) as img_zip:\n        for filename in os.listdir(output_dir):\n            file_path = os.path.join(output_dir, filename)\n            img_zip.write(file_path, arcname=filename)\n\ndef main():\n    # Configuration\n    total_images = 7000\n    zip_filename = \"images.zip\"\n\n    # Generate and save images\n    images_generated = save_generated_images(G_Monet, photo_loader, total_images, device, OUTPUT_DIR)\n    print(f\"Generated {images_generated}/{total_images} images saved to {OUTPUT_DIR}\")\n\n    # Zip images\n    zip_images(OUTPUT_DIR, zip_filename)\n    print(f\"All images zipped into {zip_filename}\")\n\n# Run the main function\nif __name__ == '__main__':\n    main()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}