{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[],"dockerImageVersionId":28755,"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\n\n# Use the kagglehub client library to attach Kaggle resources like competitions, datasets, and models to your session\n# Learn more about kagglehub: https://github.com/Kaggle/kagglehub/blob/main/README.md\n\nimport kagglehub\n# kagglehub.dataset_download('/')","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport glob\nimport itertools\nimport torch\nimport torch.nn as nn\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as transforms\nfrom PIL import Image\nfrom tqdm.auto import tqdm\n\n# ==========================================\n# 1. 超参数配置 & Kaggle 路径设定\n# ==========================================\nclass CFG:\n device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n epochs = 30 # Kaggle 建议 30-50 个 Epoch，注意比赛有 9 小时运行限制\n batch_size = 1 # CycleGAN 强烈建议 Batch Size 设为 1，并配合 InstanceNorm\n lr = 0.0002\n img_size = 256\n \n # Kaggle 默认数据集路径\n monet_dir = '../input/competitions/gan-getting-started/monet_jpg'\n photo_dir = '../input/competitions/gan-getting-started/photo_jpg'\n\n# ==========================================\n# 2. 构建非成对数据集 (Unpaired Dataset)\n# ==========================================\nclass UnpairedDataset(Dataset):\n def __init__(self, monet_dir, photo_dir, transform=None):\n self.monet_paths = glob.glob(os.path.join(monet_dir, '*.jpg'))\n self.photo_paths = glob.glob(os.path.join(photo_dir, '*.jpg'))\n self.transform = transform\n # 因为风景照有 7000 张，莫奈画只有 300 张，我们需要让它们的长度对齐\n self.length = max(len(self.monet_paths), len(self.photo_paths))\n\n def __len__(self):\n return self.length\n\n def __getitem__(self, idx):\n # 莫奈画作循环读取\n monet_path = self.monet_paths[idx % len(self.monet_paths)]\n # 风景照顺序读取\n photo_path = self.photo_paths[idx % len(self.photo_paths)]\n\n monet_img = Image.open(monet_path).convert('RGB')\n photo_img = Image.open(photo_path).convert('RGB')\n\n if self.transform:\n monet_img = self.transform(monet_img)\n photo_img = self.transform(photo_img)\n\n return {'monet': monet_img, 'photo': photo_img}\n\n# 数据增强：CycleGAN 经典预处理\ntransform = transforms.Compose([\n transforms.Resize(int(CFG.img_size * 1.12), Image.BICUBIC), # 放大一点\n transforms.RandomCrop(CFG.img_size), # 随机裁剪回 256\n transforms.RandomHorizontalFlip(),\n transforms.ToTensor(),\n transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)) # 归一化到 [-1, 1]\n])\n\n# ==========================================\n# 3. 定义模型架构：残差生成器 (ResNet Generator)\n# ==========================================\nclass ResidualBlock(nn.Module):\n def __init__(self, in_features):\n super().__init__()\n self.block = nn.Sequential(\n nn.ReflectionPad2d(1),\n nn.Conv2d(in_features, in_features, 3),\n nn.InstanceNorm2d(in_features),\n nn.ReLU(inplace=True),\n nn.ReflectionPad2d(1),\n nn.Conv2d(in_features, in_features, 3),\n nn.InstanceNorm2d(in_features)\n )\n def forward(self, x):\n return x + self.block(x)\n\nclass Generator(nn.Module):\n \"\"\"标准的 9-block ResNet 生成器\"\"\"\n def __init__(self, num_res_blocks=9):\n super().__init__()\n # 下采样层 (Downsampling)\n model = [\n nn.ReflectionPad2d(3),\n nn.Conv2d(3, 64, 7),\n nn.InstanceNorm2d(64),\n nn.ReLU(inplace=True),\n nn.Conv2d(64, 128, 3, stride=2, padding=1),\n nn.InstanceNorm2d(128),\n nn.ReLU(inplace=True),\n nn.Conv2d(128, 256, 3, stride=2, padding=1),\n nn.InstanceNorm2d(256),\n nn.ReLU(inplace=True)\n ]\n # 残差块 (Bottleneck)\n for _ in range(num_res_blocks):\n model += [ResidualBlock(256)]\n # 上采样层 (Upsampling)\n model += [\n nn.ConvTranspose2d(256, 128, 3, stride=2, padding=1, output_padding=1),\n nn.InstanceNorm2d(128),\n nn.ReLU(inplace=True),\n nn.ConvTranspose2d(128, 64, 3, stride=2, padding=1, output_padding=1),\n nn.InstanceNorm2d(64),\n nn.ReLU(inplace=True),\n nn.ReflectionPad2d(3),\n nn.Conv2d(64, 3, 7),\n nn.Tanh() # 输出还原到 [-1, 1]\n ]\n self.model = nn.Sequential(*model)\n\n def forward(self, x): return self.model(x)\n\n# ==========================================\n# 4. 定义模型架构：判别器 (PatchGAN Discriminator)\n# ==========================================\nclass Discriminator(nn.Module):\n \"\"\"PatchGAN 判别器，输出一个 30x30 的矩阵，判断局部特征真伪\"\"\"\n def __init__(self):\n super().__init__()\n self.model = nn.Sequential(\n nn.Conv2d(3, 64, 4, stride=2, padding=1),\n nn.LeakyReLU(0.2, inplace=True),\n nn.Conv2d(64, 128, 4, stride=2, padding=1),\n nn.InstanceNorm2d(128),\n nn.LeakyReLU(0.2, inplace=True),\n nn.Conv2d(128, 256, 4, stride=2, padding=1),\n nn.InstanceNorm2d(256),\n nn.LeakyReLU(0.2, inplace=True),\n nn.Conv2d(256, 512, 4, padding=1),\n nn.InstanceNorm2d(512),\n nn.LeakyReLU(0.2, inplace=True),\n nn.Conv2d(512, 1, 4, padding=1)\n )\n def forward(self, x): return self.model(x)\n\n# ==========================================\n# 5. 权重初始化 & 实例化网络\n# ==========================================\ndef weights_init_normal(m):\n classname = m.__class__.__name__\n if classname.find('Conv') != -1:\n torch.nn.init.normal_(m.weight.data, 0.0, 0.02)\n\n# G_m: 照片 -> 莫奈 ; G_p: 莫奈 -> 照片\nG_m = Generator().to(CFG.device)\nG_p = Generator().to(CFG.device)\n# D_m: 鉴别莫奈 ; D_p: 鉴别照片\nD_m = Discriminator().to(CFG.device)\nD_p = Discriminator().to(CFG.device)\n\nG_m.apply(weights_init_normal)\nG_p.apply(weights_init_normal)\nD_m.apply(weights_init_normal)\nD_p.apply(weights_init_normal)\n\n# ==========================================\n# 6. 损失函数与优化器\n# ==========================================\ncriterion_GAN = nn.MSELoss() # LSGAN 损失，比标准 BCE 更稳定\ncriterion_cycle = nn.L1Loss()\ncriterion_identity = nn.L1Loss()\n\n# 生成器优化器绑在一起，判别器独立\noptimizer_G = torch.optim.Adam(itertools.chain(G_m.parameters(), G_p.parameters()), lr=CFG.lr, betas=(0.5, 0.999))\noptimizer_D_m = torch.optim.Adam(D_m.parameters(), lr=CFG.lr, betas=(0.5, 0.999))\noptimizer_D_p = torch.optim.Adam(D_p.parameters(), lr=CFG.lr, betas=(0.5, 0.999))\n\n# ==========================================\n# 7. 核心训练循环 (Training Loop)\n# ==========================================\nif __name__ == '__main__':\n # 比赛为了防止运行超时，可以切片只用部分数据测试跑通\n dataset = UnpairedDataset(CFG.monet_dir, CFG.photo_dir, transform=transform)\n dataloader = DataLoader(dataset, batch_size=CFG.batch_size, shuffle=True, num_workers=2)\n\n print(\"开始训练 CycleGAN...\")\n for epoch in range(CFG.epochs):\n pbar = tqdm(dataloader, desc=f\"Epoch {epoch+1}/{CFG.epochs}\")\n for batch in pbar:\n real_monet = batch['monet'].to(CFG.device)\n real_photo = batch['photo'].to(CFG.device)\n\n # 用于计算 GAN 损失的真假标签矩阵\n valid = torch.ones((real_monet.size(0), 1, 30, 30), device=CFG.device, requires_grad=False)\n fake = torch.zeros((real_monet.size(0), 1, 30, 30), device=CFG.device, requires_grad=False)\n\n # -------------------------\n # 训练生成器 G_m & G_p\n # -------------------------\n G_m.train()\n G_p.train()\n optimizer_G.zero_grad()\n\n # 1. 恒等损失 (Identity Loss) - 防止色调暴走\n id_monet = G_m(real_monet)\n loss_id_m = criterion_identity(id_monet, real_monet) * 5.0\n id_photo = G_p(real_photo)\n loss_id_p = criterion_identity(id_photo, real_photo) * 5.0\n\n # 2. 对抗损失 (Adversarial Loss) - 画得像不像\n fake_monet = G_m(real_photo)\n loss_GAN_m = criterion_GAN(D_m(fake_monet), valid)\n fake_photo = G_p(real_monet)\n loss_GAN_p = criterion_GAN(D_p(fake_photo), valid)\n\n # 3. 循环一致性损失 (Cycle Loss) - 轮廓/内容不丢失\n recov_photo = G_p(fake_monet)\n loss_cycle_photo = criterion_cycle(recov_photo, real_photo) * 10.0\n recov_monet = G_m(fake_photo)\n loss_cycle_monet = criterion_cycle(recov_monet, real_monet) * 10.0\n\n # 总生成器 Loss\n loss_G = loss_id_m + loss_id_p + loss_GAN_m + loss_GAN_p + loss_cycle_photo + loss_cycle_monet\n loss_G.backward()\n optimizer_G.step()\n\n # -------------------------\n # 训练判别器 D_m\n # -------------------------\n optimizer_D_m.zero_grad()\n loss_real_m = criterion_GAN(D_m(real_monet), valid)\n loss_fake_m = criterion_GAN(D_m(fake_monet.detach()), fake) # 注意 detach\n loss_D_m = (loss_real_m + loss_fake_m) / 2\n loss_D_m.backward()\n optimizer_D_m.step()\n\n # -------------------------\n # 训练判别器 D_p\n # -------------------------\n optimizer_D_p.zero_grad()\n loss_real_p = criterion_GAN(D_p(real_photo), valid)\n loss_fake_p = criterion_GAN(D_p(fake_photo.detach()), fake)\n loss_D_p = (loss_real_p + loss_fake_p) / 2\n loss_D_p.backward()\n optimizer_D_p.step()\n\n # 更新进度条日志\n pbar.set_postfix({'G_loss': loss_G.item(), 'Dm_loss': loss_D_m.item(), 'Dp_loss': loss_D_p.item()})\n \n # 训练结束保存模型权重\n torch.save(G_m.state_dict(), 'G_m.pth')\n print(\"模型训练完毕并已保存！\")\n\n# ==========================================\n# 8. Kaggle 竞赛特有：生成 7000 张图片并打包提交\n# ==========================================\ndef generate_and_submit():\n import shutil\n os.makedirs('images', exist_ok=True)\n G_m.eval()\n \n # 推理时去除了数据增强\n test_transform = transforms.Compose([\n transforms.Resize(CFG.img_size, Image.BICUBIC),\n transforms.ToTensor(),\n transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))\n ])\n \n # 只需读取风景照\n photo_paths = glob.glob(os.path.join(CFG.photo_dir, '*.jpg'))\n \n print(\"正在生成莫奈风格图像供提交...\")\n with torch.no_grad():\n for path in tqdm(photo_paths):\n img_name = os.path.basename(path)\n img = Image.open(path).convert('RGB')\n img_tensor = test_transform(img).unsqueeze(0).to(CFG.device)\n \n # 转换为莫奈风格\n gen_monet = G_m(img_tensor).squeeze(0).cpu()\n \n # 将 Tensor [-1, 1] 还原为 [0, 255] 的 PIL Image\n gen_monet = (gen_monet * 0.5 + 0.5).clamp(0, 1)\n gen_monet = transforms.ToPILImage()(gen_monet)\n \n # 保存到 images 文件夹\n gen_monet.save(os.path.join('images', img_name))\n \n # Kaggle 要求必须打包成 images.zip 进行提交\n shutil.make_archive('images', 'zip', 'images')\n print(\"打包完成！你可以直接提交 images.zip 文件。\")\n \ngenerate_and_submit()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-06-02T06:48:45.960417Z","iopub.execute_input":"2026-06-02T06:48:45.961281Z","execution_failed":"2026-06-02T11:07:35.577Z"}},"outputs":[{"name":"stdout","text":"开始训练 CycleGAN...\n","output_type":"stream"},{"output_type":"display_data","data":{"text/plain":"Epoch 1/30: 0%| | 0/7038 [00:00