{"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('<owner>/<dataset-slug>')","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:04:26.529153Z","iopub.execute_input":"2026-07-09T14:04:26.529826Z","iopub.status.idle":"2026-07-09T14:04:26.536179Z","shell.execute_reply.started":"2026-07-09T14:04:26.529789Z","shell.execute_reply":"2026-07-09T14:04:26.535616Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile cnn_train.py\n\"\"\"転移学習を用いて Cats vs Dogs の CNN を学習・評価する。\n\n``train/`` 内の ``cat.0.jpg``、``dog.0.jpg`` のようなファイル名を想定する。\n再現可能な層化 80/20 分割を作成し、ImageNet 事前学習済み CNN を2段階で学習する。\nモデル、評価指標、グラフ、予測 CSV は ``results/`` に保存する。\n\n実行例（Windows PowerShell）：\n    .\\.venv\\Scripts\\python.exe .\\cnn_train.py\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport copy\nimport csv\nimport json\nimport random\nimport time\nfrom dataclasses import dataclass\nfrom pathlib import Path\nfrom typing import Callable, Sequence\n\nimport matplotlib\n\nmatplotlib.use(\"Agg\")\nimport matplotlib.pyplot as plt\nimport torch\nfrom PIL import Image, ImageFile\nfrom torch import nn\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision import models, transforms\n\n\n# 一部破損した JPEG があっても長時間の学習全体を停止させない。\nImageFile.LOAD_TRUNCATED_IMAGES = True\nIMAGENET_MEAN = (0.485, 0.456, 0.406)\nIMAGENET_STD = (0.229, 0.224, 0.225)\nCLASS_NAMES = (\"cat\", \"dog\")\n\n\n@dataclass(frozen=True)\nclass ImageRecord:\n    \"\"\"画像パスと数値ラベルを保持する（cat=0、dog=1）。\"\"\"\n\n    path: Path\n    label: int\n\n\nclass CatsDogsDataset(Dataset):\n    \"\"\"CSV 出力用にファイル名も返す最小構成の Dataset。\"\"\"\n\n    def __init__(\n        self,\n        records: Sequence[ImageRecord],\n        transform: Callable[[Image.Image], torch.Tensor],\n    ) -> None:\n        self.records = list(records)\n        self.transform = transform\n\n    def __len__(self) -> int:\n        return len(self.records)\n\n    def __getitem__(self, index: int) -> tuple[torch.Tensor, int, str]:\n        record = self.records[index]\n        with Image.open(record.path) as image:\n            image_tensor = self.transform(image.convert(\"RGB\"))\n        return image_tensor, record.label, record.path.name\n\n\ndef parse_args() -> argparse.Namespace:\n    parser = argparse.ArgumentParser(\n        description=\"Cats vs Dogs の2段階転移学習。\"\n    )\n    parser.add_argument(\"--data-dir\", type=Path, default=Path(\"train\"))\n    parser.add_argument(\"--results-dir\", type=Path, default=Path(\"results\"))\n    parser.add_argument(\n        \"--model\",\n        choices=(\"resnet50\", \"efficientnet_b0\", \"mobilenet_v2\"),\n        default=\"resnet50\",\n        help=\"バックボーン構造。既定値は ResNet-50。\",\n    )\n    parser.add_argument(\"--epochs-head\", type=int, default=5)\n    parser.add_argument(\"--epochs-finetune\", type=int, default=10)\n    # 第2段階の微調整でも一般的な 4 GB GPU に収まるよう batch size 16 を既定値にする。\n    parser.add_argument(\"--batch-size\", type=int, default=16)\n    parser.add_argument(\"--workers\", type=int, default=4)\n    parser.add_argument(\"--head-lr\", type=float, default=1e-3)\n    parser.add_argument(\"--finetune-lr\", type=float, default=1e-4)\n    parser.add_argument(\"--weight-decay\", type=float, default=1e-4)\n    parser.add_argument(\"--dropout\", type=float, default=0.40)\n    parser.add_argument(\"--patience\", type=int, default=4)\n    parser.add_argument(\"--seed\", type=int, default=42)\n    parser.add_argument(\n        \"--limit-per-class\",\n        type=int,\n        default=None,\n        help=\"動作確認用のクラス別上限。既定では全画像を使用する。\",\n    )\n    parser.add_argument(\n        \"--no-pretrained\",\n        action=\"store_true\",\n        help=\"ImageNet 重みを使わない（オフライン動作確認専用）。\",\n    )\n    return parser.parse_args()\n\n\ndef set_reproducible_seed(seed: int) -> None:\n    \"\"\"CUDA 畳み込みの速度を保ちつつ Python と PyTorch の乱数を固定する。\"\"\"\n\n    random.seed(seed)\n    torch.manual_seed(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed_all(seed)\n    torch.backends.cudnn.benchmark = torch.cuda.is_available()\n\n\ndef discover_images(data_dir: Path) -> list[ImageRecord]:\n    \"\"\"JPG を読み込み、ファイル名の接頭辞だけからラベルを決定する。\"\"\"\n\n    if not data_dir.is_dir():\n        raise FileNotFoundError(\n            f\"Training directory not found: {data_dir.resolve()}\"\n        )\n\n    records: list[ImageRecord] = []\n    ignored = 0\n    for path in sorted(data_dir.glob(\"*.jpg\"), key=lambda p: p.name.lower()):\n        prefix = path.name.split(\".\", maxsplit=1)[0].lower()\n        if prefix == \"cat\":\n            records.append(ImageRecord(path=path, label=0))\n        elif prefix == \"dog\":\n            records.append(ImageRecord(path=path, label=1))\n        else:\n            ignored += 1\n\n    class_counts = [sum(r.label == label for r in records) for label in (0, 1)]\n    if min(class_counts) < 2:\n        raise ValueError(\n            \"At least two cat and two dog images are required for a stratified split.\"\n        )\n    if ignored:\n        print(f\"Ignored {ignored} JPG file(s) without a cat/dog filename prefix.\")\n    print(\n        f\"Discovered {len(records):,} images \"\n        f\"(cat={class_counts[0]:,}, dog={class_counts[1]:,}).\"\n    )\n    return records\n\n\ndef stratified_split(\n    records: Sequence[ImageRecord],\n    validation_fraction: float,\n    seed: int,\n    limit_per_class: int | None = None,\n) -> tuple[list[ImageRecord], list[ImageRecord]]:\n    \"\"\"クラス比を維持するため、cat と dog を別々に層化分割する。\"\"\"\n\n    rng = random.Random(seed)\n    train_records: list[ImageRecord] = []\n    validation_records: list[ImageRecord] = []\n\n    for label in (0, 1):\n        class_records = [record for record in records if record.label == label]\n        rng.shuffle(class_records)\n        if limit_per_class is not None:\n            if limit_per_class < 2:\n                raise ValueError(\"--limit-per-class must be at least 2.\")\n            class_records = class_records[:limit_per_class]\n\n        validation_count = max(1, round(len(class_records) * validation_fraction))\n        validation_records.extend(class_records[:validation_count])\n        train_records.extend(class_records[validation_count:])\n\n    # 第1 epoch からクラスが混在するよう、分割後の両リストをシャッフルする。\n    rng.shuffle(train_records)\n    rng.shuffle(validation_records)\n    return train_records, validation_records\n\n\ndef build_transforms() -> tuple[transforms.Compose, transforms.Compose]:\n    \"\"\"学習用のランダム拡張と、評価用の決定的な変換を作成する。\"\"\"\n\n    train_transform = transforms.Compose(\n        [\n            transforms.RandomResizedCrop(224, scale=(0.75, 1.0)),\n            transforms.RandomHorizontalFlip(),\n            transforms.RandomRotation(degrees=12),\n            transforms.ColorJitter(brightness=0.20, contrast=0.20),\n            transforms.ToTensor(),\n            transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),\n        ]\n    )\n    validation_transform = transforms.Compose(\n        [\n            transforms.Resize(256),\n            transforms.CenterCrop(224),\n            transforms.ToTensor(),\n            transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),\n        ]\n    )\n    return train_transform, validation_transform\n\n\ndef build_model(\n    architecture: str, dropout: float, pretrained: bool\n) -> nn.Module:\n    \"\"\"ImageNet バックボーンを読み込み、分類層を2クラス出力に置換する。\"\"\"\n\n    if architecture == \"resnet50\":\n        weights = models.ResNet50_Weights.DEFAULT if pretrained else None\n        model = models.resnet50(weights=weights)\n        input_features = model.fc.in_features\n        model.fc = nn.Sequential(nn.Dropout(dropout), nn.Linear(input_features, 2))\n    elif architecture == \"efficientnet_b0\":\n        weights = models.EfficientNet_B0_Weights.DEFAULT if pretrained else None\n        model = models.efficientnet_b0(weights=weights)\n        input_features = model.classifier[-1].in_features\n        model.classifier = nn.Sequential(\n            nn.Dropout(dropout), nn.Linear(input_features, 2)\n        )\n    else:\n        weights = models.MobileNet_V2_Weights.DEFAULT if pretrained else None\n        model = models.mobilenet_v2(weights=weights)\n        input_features = model.classifier[-1].in_features\n        model.classifier = nn.Sequential(\n            nn.Dropout(dropout), nn.Linear(input_features, 2)\n        )\n    return model\n\n\ndef classifier_module(model: nn.Module, architecture: str) -> nn.Module:\n    return model.fc if architecture == \"resnet50\" else model.classifier\n\n\ndef configure_trainable_layers(\n    model: nn.Module, architecture: str, phase: str\n) -> None:\n    \"\"\"バックボーンを凍結し、微調整時のみ最後の畳み込みブロックを解凍する。\"\"\"\n\n    for parameter in model.parameters():\n        parameter.requires_grad = False\n\n    for parameter in classifier_module(model, architecture).parameters():\n        parameter.requires_grad = True\n\n    if phase == \"finetune\":\n        if architecture == \"resnet50\":\n            blocks = [model.layer4]\n        elif architecture == \"efficientnet_b0\":\n            blocks = list(model.features[-2:])\n        else:\n            blocks = list(model.features[-3:])\n        for block in blocks:\n            for parameter in block.parameters():\n                parameter.requires_grad = True\n\n\ndef keep_frozen_batch_norm_in_eval_mode(model: nn.Module) -> None:\n    \"\"\"凍結済み BatchNorm の移動統計量が学習中に変化するのを防ぐ。\"\"\"\n\n    for module in model.modules():\n        if isinstance(module, nn.modules.batchnorm._BatchNorm):\n            parameters = list(module.parameters())\n            if parameters and not any(p.requires_grad for p in parameters):\n                module.eval()\n\n\ndef run_epoch(\n    model: nn.Module,\n    loader: DataLoader,\n    criterion: nn.Module,\n    device: torch.device,\n    optimizer: torch.optim.Optimizer | None = None,\n    scaler: torch.amp.GradScaler | None = None,\n) -> tuple[float, float]:\n    \"\"\"学習または検証を1 epoch 実行し、平均 loss と accuracy を返す。\"\"\"\n\n    is_training = optimizer is not None\n    model.train(is_training)\n    if is_training:\n        keep_frozen_batch_norm_in_eval_mode(model)\n\n    total_loss = 0.0\n    total_correct = 0\n    total_examples = 0\n    gradient_context = torch.enable_grad() if is_training else torch.no_grad()\n\n    with gradient_context:\n        for images, labels, _ in loader:\n            images = images.to(device, non_blocking=True)\n            labels = labels.to(device, non_blocking=True)\n\n            if is_training:\n                optimizer.zero_grad(set_to_none=True)\n\n            amp_enabled = device.type == \"cuda\"\n            with torch.autocast(device_type=device.type, enabled=amp_enabled):\n                logits = model(images)\n                loss = criterion(logits, labels)\n\n            if is_training:\n                assert scaler is not None\n                scaler.scale(loss).backward()\n                scaler.step(optimizer)\n                scaler.update()\n\n            batch_size = labels.size(0)\n            total_loss += loss.item() * batch_size\n            total_correct += (logits.argmax(dim=1) == labels).sum().item()\n            total_examples += batch_size\n\n    return total_loss / total_examples, total_correct / total_examples\n\n\ndef save_checkpoint(\n    path: Path,\n    model: nn.Module,\n    architecture: str,\n    dropout: float,\n    epoch: int,\n    validation_accuracy: float,\n) -> None:\n    \"\"\"モデル重みと、分類器を再構築するためのメタデータを保存する。\"\"\"\n\n    checkpoint = {\n        \"architecture\": architecture,\n        \"dropout\": dropout,\n        \"class_to_index\": {\"cat\": 0, \"dog\": 1},\n        \"epoch\": epoch,\n        \"validation_accuracy\": validation_accuracy,\n        \"model_state_dict\": model.state_dict(),\n    }\n    torch.save(checkpoint, path)\n\n\ndef train_phase(\n    *,\n    phase_name: str,\n    model: nn.Module,\n    architecture: str,\n    train_loader: DataLoader,\n    validation_loader: DataLoader,\n    criterion: nn.Module,\n    device: torch.device,\n    epochs: int,\n    learning_rate: float,\n    weight_decay: float,\n    patience: int,\n    checkpoint_path: Path,\n    dropout: float,\n    history: dict[str, list[float]],\n    first_epoch_number: int,\n    best: dict[str, object],\n) -> int:\n    \"\"\"LR scheduler と early stopping を使い、転移学習の1段階を実行する。\"\"\"\n\n    if epochs <= 0:\n        return first_epoch_number\n\n    configure_trainable_layers(model, architecture, phase_name)\n    trainable_parameters = [p for p in model.parameters() if p.requires_grad]\n    optimizer = torch.optim.AdamW(\n        trainable_parameters, lr=learning_rate, weight_decay=weight_decay\n    )\n    scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(\n        optimizer, mode=\"min\", factor=0.3, patience=1, min_lr=1e-7\n    )\n    scaler = torch.amp.GradScaler(\"cuda\", enabled=device.type == \"cuda\")\n    epochs_without_improvement = 0\n    completed_epochs = first_epoch_number\n\n    print(\n        f\"\\nStarting {phase_name} phase: {epochs} epoch(s), \"\n        f\"{sum(p.numel() for p in trainable_parameters):,} trainable parameters.\"\n    )\n    for _ in range(epochs):\n        completed_epochs += 1\n        start_time = time.perf_counter()\n        train_loss, train_accuracy = run_epoch(\n            model,\n            train_loader,\n            criterion,\n            device,\n            optimizer=optimizer,\n            scaler=scaler,\n        )\n        validation_loss, validation_accuracy = run_epoch(\n            model, validation_loader, criterion, device\n        )\n        scheduler.step(validation_loss)\n\n        history[\"epoch\"].append(float(completed_epochs))\n        history[\"train_loss\"].append(train_loss)\n        history[\"train_accuracy\"].append(train_accuracy)\n        history[\"validation_loss\"].append(validation_loss)\n        history[\"validation_accuracy\"].append(validation_accuracy)\n        history[\"learning_rate\"].append(optimizer.param_groups[0][\"lr\"])\n        history[\"phase\"].append(phase_name)\n\n        elapsed = time.perf_counter() - start_time\n        print(\n            f\"Epoch {completed_epochs:02d} [{phase_name:8s}] | \"\n            f\"train loss {train_loss:.4f}, train acc {train_accuracy:.4f} | \"\n            f\"val loss {validation_loss:.4f}, val acc {validation_accuracy:.4f} | \"\n            f\"{elapsed:.1f}s\"\n        )\n\n        best_accuracy = float(best[\"accuracy\"])\n        best_loss = float(best[\"loss\"])\n        improved = validation_accuracy > best_accuracy or (\n            validation_accuracy == best_accuracy and validation_loss < best_loss\n        )\n        if improved:\n            best.update(\n                accuracy=validation_accuracy,\n                loss=validation_loss,\n                epoch=completed_epochs,\n                state=copy.deepcopy(model.state_dict()),\n            )\n            save_checkpoint(\n                checkpoint_path,\n                model,\n                architecture,\n                dropout,\n                completed_epochs,\n                validation_accuracy,\n            )\n            epochs_without_improvement = 0\n            print(f\"  Saved new best model to {checkpoint_path}.\")\n        else:\n            epochs_without_improvement += 1\n            if epochs_without_improvement >= patience:\n                print(f\"  Early stopping: no improvement for {patience} epoch(s).\")\n                break\n\n    return completed_epochs\n\n\ndef predict(\n    model: nn.Module, loader: DataLoader, device: torch.device\n) -> list[dict[str, object]]:\n    \"\"\"検証データの予測クラスと dog 確率を決定的に収集する。\"\"\"\n\n    model.eval()\n    rows: list[dict[str, object]] = []\n    with torch.no_grad():\n        for images, labels, filenames in loader:\n            images = images.to(device, non_blocking=True)\n            logits = model(images)\n            probabilities = torch.softmax(logits, dim=1)[:, 1].cpu().tolist()\n            predictions = logits.argmax(dim=1).cpu().tolist()\n            true_labels = labels.tolist()\n\n            for filename, true_label, predicted_label, dog_probability in zip(\n                filenames, true_labels, predictions, probabilities\n            ):\n                rows.append(\n                    {\n                        \"filename\": filename,\n                        \"true_label\": true_label,\n                        \"predicted_label\": predicted_label,\n                        \"dog_probability\": dog_probability,\n                        \"correct\": true_label == predicted_label,\n                    }\n                )\n    return rows\n\n\ndef confusion_and_metrics(\n    rows: Sequence[dict[str, object]],\n) -> tuple[list[list[int]], dict[str, float]]:\n    \"\"\"dog（class 1）を陽性として二値分類の評価指標を計算する。\"\"\"\n\n    matrix = [[0, 0], [0, 0]]\n    for row in rows:\n        true_label = int(row[\"true_label\"])\n        predicted_label = int(row[\"predicted_label\"])\n        matrix[true_label][predicted_label] += 1\n\n    tn, fp = matrix[0]\n    fn, tp = matrix[1]\n    precision = tp / (tp + fp) if tp + fp else 0.0\n    recall = tp / (tp + fn) if tp + fn else 0.0\n    f1 = (\n        2 * precision * recall / (precision + recall)\n        if precision + recall\n        else 0.0\n    )\n    accuracy = (tp + tn) / sum(sum(row) for row in matrix)\n    return matrix, {\n        \"accuracy\": accuracy,\n        \"precision\": precision,\n        \"recall\": recall,\n        \"f1_score\": f1,\n    }\n\n\ndef write_predictions(\n    path: Path, rows: Sequence[dict[str, object]], misclassified_only: bool = False\n) -> None:\n    \"\"\"PowerShell と Excel で扱いやすい安定した CSV 形式で予測を保存する。\"\"\"\n\n    fieldnames = [\n        \"filename\",\n        \"true_label\",\n        \"predicted_label\",\n        \"dog_probability\",\n        \"correct\",\n    ]\n    with path.open(\"w\", newline=\"\", encoding=\"utf-8-sig\") as output_file:\n        writer = csv.DictWriter(output_file, fieldnames=fieldnames)\n        writer.writeheader()\n        for row in rows:\n            if misclassified_only and bool(row[\"correct\"]):\n                continue\n            formatted = dict(row)\n            formatted[\"dog_probability\"] = f\"{float(row['dog_probability']):.8f}\"\n            writer.writerow(formatted)\n\n\ndef plot_training_curve(history: dict[str, list[float]], path: Path) -> None:\n    \"\"\"学習・検証の loss と accuracy をプレゼン用の1枚の図にまとめる。\"\"\"\n\n    epochs = history[\"epoch\"]\n    figure, axes = plt.subplots(1, 2, figsize=(12, 4.8))\n    axes[0].plot(epochs, history[\"train_loss\"], marker=\"o\", label=\"Train\")\n    axes[0].plot(epochs, history[\"validation_loss\"], marker=\"o\", label=\"Validation\")\n    axes[0].set(title=\"Loss\", xlabel=\"Epoch\", ylabel=\"Cross-entropy loss\")\n    axes[1].plot(epochs, history[\"train_accuracy\"], marker=\"o\", label=\"Train\")\n    axes[1].plot(\n        epochs, history[\"validation_accuracy\"], marker=\"o\", label=\"Validation\"\n    )\n    axes[1].set(title=\"Accuracy\", xlabel=\"Epoch\", ylabel=\"Accuracy\", ylim=(0, 1.02))\n\n    # 分類層のみの学習から fine-tuning へ切り替わる位置を破線で示す。\n    phases = history[\"phase\"]\n    if \"finetune\" in phases and phases.index(\"finetune\") > 0:\n        boundary = epochs[phases.index(\"finetune\")] - 0.5\n        for axis in axes:\n            axis.axvline(boundary, color=\"gray\", linestyle=\"--\", alpha=0.7)\n\n    for axis in axes:\n        axis.grid(alpha=0.25)\n        axis.legend()\n    figure.suptitle(\"Cats vs Dogs Transfer Learning\")\n    figure.tight_layout()\n    figure.savefig(path, dpi=180, bbox_inches=\"tight\")\n    plt.close(figure)\n\n\ndef plot_confusion_matrix(matrix: list[list[int]], path: Path) -> None:\n    \"\"\"追加ライブラリなしで混同行列を画像として保存する。\"\"\"\n\n    figure, axis = plt.subplots(figsize=(6, 5.2))\n    image = axis.imshow(matrix, interpolation=\"nearest\", cmap=\"Blues\")\n    figure.colorbar(image, ax=axis, fraction=0.046, pad=0.04)\n    axis.set(\n        xticks=(0, 1),\n        yticks=(0, 1),\n        xticklabels=CLASS_NAMES,\n        yticklabels=CLASS_NAMES,\n        xlabel=\"Predicted label\",\n        ylabel=\"True label\",\n        title=\"CNN Confusion Matrix\",\n    )\n    threshold = max(max(row) for row in matrix) / 2\n    for true_label in (0, 1):\n        for predicted_label in (0, 1):\n            value = matrix[true_label][predicted_label]\n            axis.text(\n                predicted_label,\n                true_label,\n                f\"{value:,}\",\n                ha=\"center\",\n                va=\"center\",\n                color=\"white\" if value > threshold else \"black\",\n                fontsize=13,\n            )\n    figure.tight_layout()\n    figure.savefig(path, dpi=180, bbox_inches=\"tight\")\n    plt.close(figure)\n\n\ndef main() -> None:\n    args = parse_args()\n    if args.epochs_head < 1:\n        raise ValueError(\"--epochs-head must be at least 1.\")\n    if args.epochs_finetune < 1:\n        raise ValueError(\"--epochs-finetune must be at least 1.\")\n    if args.patience < 1:\n        raise ValueError(\"--patience must be at least 1.\")\n\n    set_reproducible_seed(args.seed)\n    args.results_dir.mkdir(parents=True, exist_ok=True)\n    checkpoint_path = args.results_dir / \"best_cnn_model.pth\"\n\n    records = discover_images(args.data_dir)\n    train_records, validation_records = stratified_split(\n        records,\n        validation_fraction=0.20,\n        seed=args.seed,\n        limit_per_class=args.limit_per_class,\n    )\n    print(\n        f\"Stratified split: train={len(train_records):,} (80%), \"\n        f\"validation={len(validation_records):,} (20%).\"\n    )\n\n    train_transform, validation_transform = build_transforms()\n    train_dataset = CatsDogsDataset(train_records, train_transform)\n    validation_dataset = CatsDogsDataset(validation_records, validation_transform)\n    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    pin_memory = device.type == \"cuda\"\n    generator = torch.Generator().manual_seed(args.seed)\n    common_loader_options = {\n        \"batch_size\": args.batch_size,\n        \"num_workers\": args.workers,\n        \"pin_memory\": pin_memory,\n        \"persistent_workers\": args.workers > 0,\n    }\n    train_loader = DataLoader(\n        train_dataset, shuffle=True, generator=generator, **common_loader_options\n    )\n    validation_loader = DataLoader(\n        validation_dataset, shuffle=False, **common_loader_options\n    )\n\n    print(f\"Using device: {device}.\")\n    if device.type == \"cuda\":\n        print(f\"GPU: {torch.cuda.get_device_name(0)}\")\n    model = build_model(\n        args.model, dropout=args.dropout, pretrained=not args.no_pretrained\n    ).to(device)\n    criterion = nn.CrossEntropyLoss()\n    history: dict[str, list] = {\n        \"epoch\": [],\n        \"train_loss\": [],\n        \"train_accuracy\": [],\n        \"validation_loss\": [],\n        \"validation_accuracy\": [],\n        \"learning_rate\": [],\n        \"phase\": [],\n    }\n    best: dict[str, object] = {\n        \"accuracy\": -1.0,\n        \"loss\": float(\"inf\"),\n        \"epoch\": 0,\n        \"state\": None,\n    }\n\n    completed_epochs = train_phase(\n        phase_name=\"head\",\n        model=model,\n        architecture=args.model,\n        train_loader=train_loader,\n        validation_loader=validation_loader,\n        criterion=criterion,\n        device=device,\n        epochs=args.epochs_head,\n        learning_rate=args.head_lr,\n        weight_decay=args.weight_decay,\n        patience=args.patience,\n        checkpoint_path=checkpoint_path,\n        dropout=args.dropout,\n        history=history,\n        first_epoch_number=0,\n        best=best,\n    )\n\n    # fine-tuning は分類層のみを学習した段階の最良 checkpoint から開始する。\n    assert best[\"state\"] is not None\n    model.load_state_dict(best[\"state\"])\n    train_phase(\n        phase_name=\"finetune\",\n        model=model,\n        architecture=args.model,\n        train_loader=train_loader,\n        validation_loader=validation_loader,\n        criterion=criterion,\n        device=device,\n        epochs=args.epochs_finetune,\n        learning_rate=args.finetune_lr,\n        weight_decay=args.weight_decay,\n        patience=args.patience,\n        checkpoint_path=checkpoint_path,\n        dropout=args.dropout,\n        history=history,\n        first_epoch_number=completed_epochs,\n        best=best,\n    )\n\n    # 最終評価と CSV 出力の前に、全段階を通した最良 epoch を復元する。\n    model.load_state_dict(best[\"state\"])\n    rows = predict(model, validation_loader, device)\n    matrix, metrics = confusion_and_metrics(rows)\n    write_predictions(args.results_dir / \"validation_predictions.csv\", rows)\n    write_predictions(\n        args.results_dir / \"misclassified_images.csv\",\n        rows,\n        misclassified_only=True,\n    )\n    plot_training_curve(history, args.results_dir / \"training_curve.png\")\n    plot_confusion_matrix(matrix, args.results_dir / \"confusion_matrix_cnn.png\")\n\n    metrics_payload = {\n        \"architecture\": args.model,\n        \"best_epoch\": int(best[\"epoch\"]),\n        \"best_validation_accuracy_during_training\": float(best[\"accuracy\"]),\n        **metrics,\n        \"confusion_matrix\": matrix,\n        \"train_size\": len(train_records),\n        \"validation_size\": len(validation_records),\n        \"seed\": args.seed,\n    }\n    with (args.results_dir / \"cnn_metrics.json\").open(\n        \"w\", encoding=\"utf-8\"\n    ) as output_file:\n        json.dump(metrics_payload, output_file, indent=2)\n\n    print(\"\\n=== Final CNN Results ===\")\n    print(f\"Best validation accuracy during training: {float(best['accuracy']):.4f}\")\n    print(f\"Restored-checkpoint validation accuracy: {metrics['accuracy']:.4f}\")\n    print(f\"Precision (dog=positive): {metrics['precision']:.4f}\")\n    print(f\"Recall    (dog=positive): {metrics['recall']:.4f}\")\n    print(f\"F1 score  (dog=positive): {metrics['f1_score']:.4f}\")\n    print(\"Confusion matrix (rows=true, columns=predicted):\")\n    print(f\"[[{matrix[0][0]}, {matrix[0][1]}],\")\n    print(f\" [{matrix[1][0]}, {matrix[1][1]}]]\")\n    print(f\"Artifacts saved to: {args.results_dir.resolve()}\")\n\n\nif __name__ == \"__main__\":\n    # Windows の spawn 方式で DataLoader worker を安全に起動するために必要。\n    torch.multiprocessing.freeze_support()\n    main()\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:04:26.686840Z","iopub.execute_input":"2026-07-09T14:04:26.687526Z","iopub.status.idle":"2026-07-09T14:04:26.704193Z","shell.execute_reply.started":"2026-07-09T14:04:26.687493Z","shell.execute_reply":"2026-07-09T14:04:26.703289Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile baseline_cnn_train.py\n\"\"\"cnn_train.py と公平に比較するため、通常の CNN ベースラインを学習する。\n\nResNet-50 側の改善効果を確認できるよう、このベースラインでは ImageNet 事前学習、\nデータ拡張、Dropout、weight decay、段階的 fine-tuning、LR scheduler を使用しない。\n両実験は同じ層化分割と乱数 seed を使用する。\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport copy\nimport csv\nimport json\nimport time\nfrom pathlib import Path\n\nimport matplotlib\n\nmatplotlib.use(\"Agg\")\nimport matplotlib.pyplot as plt\nimport torch\nfrom torch import nn\nfrom torch.utils.data import DataLoader\nfrom torchvision import transforms\n\nfrom cnn_train import (\n    CatsDogsDataset,\n    confusion_and_metrics,\n    discover_images,\n    predict,\n    set_reproducible_seed,\n    stratified_split,\n    write_predictions,\n)\n\n\nclass PlainCNN(nn.Module):\n    \"\"\"ランダム初期化から学習する一般的な4ブロック CNN。\"\"\"\n\n    def __init__(self) -> None:\n        super().__init__()\n        # 各ブロックで特徴チャネルを増やしながら、空間サイズを半分に縮小する。\n        self.features = nn.Sequential(\n            nn.Conv2d(3, 32, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(2),\n            nn.Conv2d(32, 64, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(2),\n            nn.Conv2d(64, 128, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(2),\n            nn.Conv2d(128, 128, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(2),\n        )\n        self.classifier = nn.Sequential(\n            nn.Flatten(),\n            nn.Linear(128 * 8 * 8, 256),\n            nn.ReLU(inplace=True),\n            nn.Linear(256, 2),\n        )\n\n    def forward(self, images: torch.Tensor) -> torch.Tensor:\n        \"\"\"画像から畳み込み特徴を抽出し、cat/dog の2クラス logits を返す。\"\"\"\n\n        return self.classifier(self.features(images))\n\n\ndef parse_args() -> argparse.Namespace:\n    \"\"\"PowerShell から変更可能な学習パラメータを定義する。\"\"\"\n\n    parser = argparse.ArgumentParser(description=\"通常の CNN ベースラインを学習する。\")\n    parser.add_argument(\"--data-dir\", type=Path, default=Path(\"train\"))\n    parser.add_argument(\"--results-dir\", type=Path, default=Path(\"results\"))\n    parser.add_argument(\"--epochs\", type=int, default=10)\n    parser.add_argument(\"--batch-size\", type=int, default=64)\n    parser.add_argument(\"--workers\", type=int, default=4)\n    parser.add_argument(\"--learning-rate\", type=float, default=1e-3)\n    parser.add_argument(\"--seed\", type=int, default=42)\n    return parser.parse_args()\n\n\ndef run_epoch(\n    model: nn.Module,\n    loader: DataLoader,\n    criterion: nn.Module,\n    device: torch.device,\n    optimizer: torch.optim.Optimizer | None,\n    scaler: torch.amp.GradScaler,\n) -> tuple[float, float]:\n    \"\"\"学習または検証を1 epoch 実行し、平均 loss と accuracy を返す。\"\"\"\n\n    is_training = optimizer is not None\n    model.train(is_training)\n    context = torch.enable_grad() if is_training else torch.no_grad()\n    total_loss = 0.0\n    total_correct = 0\n    total_examples = 0\n\n    with context:\n        for images, labels, _ in loader:\n            images = images.to(device, non_blocking=True)\n            labels = labels.to(device, non_blocking=True)\n            if is_training:\n                optimizer.zero_grad(set_to_none=True)\n            with torch.autocast(device_type=device.type, enabled=device.type == \"cuda\"):\n                logits = model(images)\n                loss = criterion(logits, labels)\n            if is_training:\n                scaler.scale(loss).backward()\n                scaler.step(optimizer)\n                scaler.update()\n\n            batch_size = labels.size(0)\n            total_loss += loss.item() * batch_size\n            total_correct += (logits.argmax(1) == labels).sum().item()\n            total_examples += batch_size\n    return total_loss / total_examples, total_correct / total_examples\n\n\ndef plot_curve(history: dict[str, list[float]], path: Path) -> None:\n    \"\"\"ベースラインの loss/accuracy 曲線を保存する。\"\"\"\n\n    epochs = list(range(1, len(history[\"train_loss\"]) + 1))\n    figure, axes = plt.subplots(1, 2, figsize=(12, 4.8))\n    axes[0].plot(epochs, history[\"train_loss\"], marker=\"o\", label=\"Train\")\n    axes[0].plot(epochs, history[\"validation_loss\"], marker=\"o\", label=\"Validation\")\n    axes[0].set(title=\"Loss\", xlabel=\"Epoch\", ylabel=\"Cross-entropy loss\")\n    axes[1].plot(epochs, history[\"train_accuracy\"], marker=\"o\", label=\"Train\")\n    axes[1].plot(\n        epochs, history[\"validation_accuracy\"], marker=\"o\", label=\"Validation\"\n    )\n    axes[1].set(title=\"Accuracy\", xlabel=\"Epoch\", ylabel=\"Accuracy\", ylim=(0, 1.02))\n    for axis in axes:\n        axis.grid(alpha=0.25)\n        axis.legend()\n    figure.suptitle(\"Plain CNN Baseline\")\n    figure.tight_layout()\n    figure.savefig(path, dpi=180, bbox_inches=\"tight\")\n    plt.close(figure)\n\n\ndef plot_confusion(matrix: list[list[int]], path: Path) -> None:\n    \"\"\"PlainCNN の混同行列を画像として保存する。\"\"\"\n\n    figure, axis = plt.subplots(figsize=(6, 5.2))\n    image = axis.imshow(matrix, cmap=\"Blues\")\n    figure.colorbar(image, ax=axis, fraction=0.046, pad=0.04)\n    axis.set(\n        xticks=(0, 1),\n        yticks=(0, 1),\n        xticklabels=(\"cat\", \"dog\"),\n        yticklabels=(\"cat\", \"dog\"),\n        xlabel=\"Predicted label\",\n        ylabel=\"True label\",\n        title=\"Plain CNN Confusion Matrix\",\n    )\n    threshold = max(max(row) for row in matrix) / 2\n    for true_label in (0, 1):\n        for predicted_label in (0, 1):\n            value = matrix[true_label][predicted_label]\n            axis.text(\n                predicted_label,\n                true_label,\n                f\"{value:,}\",\n                ha=\"center\",\n                va=\"center\",\n                color=\"white\" if value > threshold else \"black\",\n                fontsize=13,\n            )\n    figure.tight_layout()\n    figure.savefig(path, dpi=180, bbox_inches=\"tight\")\n    plt.close(figure)\n\n\ndef save_comparison(\n    results_dir: Path, baseline_metrics: dict[str, object]\n) -> None:\n    \"\"\"ベースラインと改良モデルの差を JSON、CSV、グラフに保存する。\"\"\"\n\n    improved_path = results_dir / \"cnn_metrics.json\"\n    if not improved_path.exists():\n        return\n    with improved_path.open(encoding=\"utf-8\") as input_file:\n        improved_metrics = json.load(input_file)\n\n    baseline_best = float(baseline_metrics[\"best_validation_accuracy_during_training\"])\n    improved_best = float(improved_metrics[\"best_validation_accuracy_during_training\"])\n    comparison = {\n        \"baseline_model\": \"PlainCNN (random initialization)\",\n        \"improved_model\": \"ImageNet-pretrained ResNet-50\",\n        \"baseline_best_validation_accuracy\": baseline_best,\n        \"improved_best_validation_accuracy\": improved_best,\n        \"absolute_improvement_percentage_points\": (improved_best - baseline_best) * 100,\n        \"relative_error_reduction_percent\": (\n            ((1 - baseline_best) - (1 - improved_best)) / (1 - baseline_best) * 100\n            if baseline_best < 1\n            else 0.0\n        ),\n    }\n    with (results_dir / \"model_comparison.json\").open(\n        \"w\", encoding=\"utf-8\"\n    ) as output_file:\n        json.dump(comparison, output_file, indent=2)\n    with (results_dir / \"model_comparison.csv\").open(\n        \"w\", newline=\"\", encoding=\"utf-8-sig\"\n    ) as output_file:\n        writer = csv.writer(output_file)\n        writer.writerow([\"model\", \"best_validation_accuracy\"])\n        writer.writerow([comparison[\"baseline_model\"], f\"{baseline_best:.6f}\"])\n        writer.writerow([comparison[\"improved_model\"], f\"{improved_best:.6f}\"])\n\n    figure, axis = plt.subplots(figsize=(7.2, 5.2))\n    labels = (\"Plain CNN\", \"Pretrained ResNet-50\")\n    values = (baseline_best * 100, improved_best * 100)\n    bars = axis.bar(labels, values, color=(\"#9aa0a6\", \"#1f77b4\"), width=0.58)\n    axis.set(title=\"Validation Accuracy Comparison\", ylabel=\"Accuracy (%)\", ylim=(0, 105))\n    axis.grid(axis=\"y\", alpha=0.25)\n    for bar, value in zip(bars, values):\n        axis.text(\n            bar.get_x() + bar.get_width() / 2,\n            value + 1.0,\n            f\"{value:.2f}%\",\n            ha=\"center\",\n            fontweight=\"bold\",\n        )\n    figure.tight_layout()\n    figure.savefig(results_dir / \"model_comparison.png\", dpi=180, bbox_inches=\"tight\")\n    plt.close(figure)\n\n    print(\"\\n=== Baseline vs Improved Model ===\")\n    print(f\"Plain CNN best validation accuracy: {baseline_best:.4f}\")\n    print(f\"ResNet-50 best validation accuracy:  {improved_best:.4f}\")\n    print(\n        \"Absolute accuracy improvement: \"\n        f\"{comparison['absolute_improvement_percentage_points']:.2f} percentage points\"\n    )\n    print(\n        \"Relative error reduction: \"\n        f\"{comparison['relative_error_reduction_percent']:.2f}%\"\n    )\n\n\ndef main() -> None:\n    \"\"\"データ準備、学習、最良モデル復元、評価、成果物保存を順に実行する。\"\"\"\n\n    args = parse_args()\n    if args.epochs < 1:\n        raise ValueError(\"--epochs must be at least 1.\")\n    set_reproducible_seed(args.seed)\n    args.results_dir.mkdir(parents=True, exist_ok=True)\n\n    records = discover_images(args.data_dir)\n    train_records, validation_records = stratified_split(records, 0.20, args.seed)\n    # 比較用ベースラインなので、Resize 以外のランダムなデータ拡張は使用しない。\n    transform = transforms.Compose(\n        [\n            transforms.Resize((128, 128)),\n            transforms.ToTensor(),\n            transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),\n        ]\n    )\n    train_dataset = CatsDogsDataset(train_records, transform)\n    validation_dataset = CatsDogsDataset(validation_records, transform)\n    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    options = {\n        \"batch_size\": args.batch_size,\n        \"num_workers\": args.workers,\n        \"pin_memory\": device.type == \"cuda\",\n        \"persistent_workers\": args.workers > 0,\n    }\n    generator = torch.Generator().manual_seed(args.seed)\n    train_loader = DataLoader(\n        train_dataset, shuffle=True, generator=generator, **options\n    )\n    validation_loader = DataLoader(validation_dataset, shuffle=False, **options)\n\n    model = PlainCNN().to(device)\n    criterion = nn.CrossEntropyLoss()\n    optimizer = torch.optim.Adam(model.parameters(), lr=args.learning_rate)\n    scaler = torch.amp.GradScaler(\"cuda\", enabled=device.type == \"cuda\")\n    history: dict[str, list[float]] = {\n        \"train_loss\": [],\n        \"train_accuracy\": [],\n        \"validation_loss\": [],\n        \"validation_accuracy\": [],\n    }\n    best_accuracy = -1.0\n    best_loss = float(\"inf\")\n    best_epoch = 0\n    best_state: dict[str, torch.Tensor] | None = None\n\n    print(\n        f\"Plain CNN baseline on {device}: train={len(train_dataset):,}, \"\n        f\"validation={len(validation_dataset):,}.\"\n    )\n    for epoch in range(1, args.epochs + 1):\n        start = time.perf_counter()\n        train_loss, train_accuracy = run_epoch(\n            model, train_loader, criterion, device, optimizer, scaler\n        )\n        validation_loss, validation_accuracy = run_epoch(\n            model, validation_loader, criterion, device, None, scaler\n        )\n        history[\"train_loss\"].append(train_loss)\n        history[\"train_accuracy\"].append(train_accuracy)\n        history[\"validation_loss\"].append(validation_loss)\n        history[\"validation_accuracy\"].append(validation_accuracy)\n        print(\n            f\"Epoch {epoch:02d} | train loss {train_loss:.4f}, \"\n            f\"train acc {train_accuracy:.4f} | val loss {validation_loss:.4f}, \"\n            f\"val acc {validation_accuracy:.4f} | {time.perf_counter() - start:.1f}s\"\n        )\n        if validation_accuracy > best_accuracy or (\n            validation_accuracy == best_accuracy and validation_loss < best_loss\n        ):\n            best_accuracy = validation_accuracy\n            best_loss = validation_loss\n            best_epoch = epoch\n            best_state = copy.deepcopy(model.state_dict())\n            torch.save(\n                {\n                    \"architecture\": \"PlainCNN\",\n                    \"epoch\": epoch,\n                    \"validation_accuracy\": validation_accuracy,\n                    \"class_to_index\": {\"cat\": 0, \"dog\": 1},\n                    \"model_state_dict\": best_state,\n                },\n                args.results_dir / \"baseline_cnn_model.pth\",\n            )\n\n    assert best_state is not None\n    model.load_state_dict(best_state)\n    rows = predict(model, validation_loader, device)\n    matrix, metrics = confusion_and_metrics(rows)\n    write_predictions(args.results_dir / \"baseline_validation_predictions.csv\", rows)\n    plot_curve(history, args.results_dir / \"baseline_training_curve.png\")\n    plot_confusion(matrix, args.results_dir / \"confusion_matrix_baseline.png\")\n\n    payload: dict[str, object] = {\n        \"architecture\": \"PlainCNN\",\n        \"comparison_definition\": (\n            \"Random initialization; no pretraining, augmentation, dropout, \"\n            \"weight decay, scheduler, or staged fine-tuning.\"\n        ),\n        \"best_epoch\": best_epoch,\n        \"best_validation_accuracy_during_training\": best_accuracy,\n        **metrics,\n        \"confusion_matrix\": matrix,\n        \"train_size\": len(train_dataset),\n        \"validation_size\": len(validation_dataset),\n        \"seed\": args.seed,\n    }\n    with (args.results_dir / \"baseline_cnn_metrics.json\").open(\n        \"w\", encoding=\"utf-8\"\n    ) as output_file:\n        json.dump(payload, output_file, indent=2)\n\n    print(\"\\n=== Plain CNN Results ===\")\n    print(f\"Best validation accuracy during training: {best_accuracy:.4f}\")\n    print(f\"Restored-checkpoint accuracy: {metrics['accuracy']:.4f}\")\n    print(f\"Precision: {metrics['precision']:.4f}\")\n    print(f\"Recall: {metrics['recall']:.4f}\")\n    print(f\"F1 score: {metrics['f1_score']:.4f}\")\n    print(f\"Confusion matrix: {matrix}\")\n    save_comparison(args.results_dir, payload)\n\n\nif __name__ == \"__main__\":\n    torch.multiprocessing.freeze_support()\n    main()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:04:27.603030Z","iopub.execute_input":"2026-07-09T14:04:27.603774Z","iopub.status.idle":"2026-07-09T14:04:27.615056Z","shell.execute_reply.started":"2026-07-09T14:04:27.603746Z","shell.execute_reply":"2026-07-09T14:04:27.614224Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile kaggle_predict.py\n\"\"\"Cats vs Dogs の ResNet-50/PlainCNN checkpoint から提出 CSV を作成する。\n\n提出形式は、数字で命名された各テスト画像の ``id`` と dog である確率の2列。\n必要に応じて水平反転による Test-Time Augmentation（TTA）を適用できる。\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport csv\nfrom pathlib import Path\n\nimport torch\nfrom PIL import Image\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision import transforms\n\nfrom cnn_train import IMAGENET_MEAN, IMAGENET_STD, build_model\n\n\nclass KaggleTestDataset(Dataset):\n    \"\"\"数字で命名された Kaggle テスト画像を ID 順に読み込む。\"\"\"\n\n    def __init__(\n        self, test_dir: Path, transform: transforms.Compose\n    ) -> None:\n        self.paths = sorted(\n            test_dir.glob(\"*.jpg\"), key=lambda path: int(path.stem)\n        )\n        if len(self.paths) != 12_500:\n            raise ValueError(\n                f\"Expected 12,500 test images, found {len(self.paths):,}.\"\n            )\n        self.transform = transform\n\n    def __len__(self) -> int:\n        return len(self.paths)\n\n    def __getitem__(self, index: int) -> tuple[torch.Tensor, int]:\n        path = self.paths[index]\n        with Image.open(path) as image:\n            tensor = self.transform(image.convert(\"RGB\"))\n        return tensor, int(path.stem)\n\n\ndef parse_args() -> argparse.Namespace:\n    \"\"\"テスト画像、checkpoint、出力先などのコマンドライン引数を定義する。\"\"\"\n\n    parser = argparse.ArgumentParser(description=\"Kaggle 提出用 CSV を生成する。\")\n    parser.add_argument(\"--test-dir\", type=Path, default=Path(\"test\"))\n    parser.add_argument(\n        \"--checkpoint\", type=Path, default=Path(\"results/best_cnn_model.pth\")\n    )\n    parser.add_argument(\n        \"--output\", type=Path, default=Path(\"results/kaggle_submission.csv\")\n    )\n    parser.add_argument(\"--batch-size\", type=int, default=32)\n    parser.add_argument(\"--workers\", type=int, default=4)\n    parser.add_argument(\n        \"--no-tta\",\n        action=\"store_true\",\n        help=\"水平反転による TTA を無効にする。\",\n    )\n    return parser.parse_args()\n\n\ndef build_model_and_transform(\n    checkpoint: dict[str, object],\n) -> tuple[torch.nn.Module, transforms.Compose]:\n    \"\"\"checkpoint のモデル種別に合うモデルと検証時前処理を復元する。\"\"\"\n\n    architecture = str(checkpoint[\"architecture\"])\n    if architecture == \"PlainCNN\":\n        from baseline_cnn_train import PlainCNN\n\n        model = PlainCNN()\n        transform = transforms.Compose(\n            [\n                transforms.Resize((128, 128)),\n                transforms.ToTensor(),\n                transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),\n            ]\n        )\n        return model, transform\n\n    model = build_model(\n        architecture,\n        float(checkpoint[\"dropout\"]),\n        pretrained=False,\n    )\n    transform = transforms.Compose(\n        [\n            transforms.Resize(256),\n            transforms.CenterCrop(224),\n            transforms.ToTensor(),\n            transforms.Normalize(IMAGENET_MEAN, IMAGENET_STD),\n        ]\n    )\n    return model, transform\n\n\ndef main() -> None:\n    \"\"\"最良 checkpoint を復元し、12,500 枚の dog probability を CSV に保存する。\"\"\"\n\n    args = parse_args()\n    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    checkpoint = torch.load(args.checkpoint, map_location=\"cpu\", weights_only=True)\n    model, transform = build_model_and_transform(checkpoint)\n    model.load_state_dict(checkpoint[\"model_state_dict\"])\n    model.to(device).eval()\n\n    dataset = KaggleTestDataset(args.test_dir, transform)\n    loader = DataLoader(\n        dataset,\n        batch_size=args.batch_size,\n        shuffle=False,\n        num_workers=args.workers,\n        pin_memory=device.type == \"cuda\",\n        persistent_workers=args.workers > 0,\n    )\n    rows: list[tuple[int, float]] = []\n    print(\n        f\"Predicting {len(dataset):,} test images on {device} \"\n        f\"(horizontal-flip TTA={'off' if args.no_tta else 'on'}).\"\n    )\n    with torch.no_grad():\n        for batch_number, (images, image_ids) in enumerate(loader, start=1):\n            images = images.to(device, non_blocking=True)\n            with torch.autocast(device_type=device.type, enabled=device.type == \"cuda\"):\n                probabilities = torch.softmax(model(images), dim=1)[:, 1]\n                if not args.no_tta:\n                    flipped = torch.flip(images, dims=(3,))\n                    flipped_probabilities = torch.softmax(model(flipped), dim=1)[:, 1]\n                    probabilities = (probabilities + flipped_probabilities) / 2\n\n            for image_id, probability in zip(\n                image_ids.tolist(), probabilities.cpu().tolist()\n            ):\n                # Log Loss が発散しないよう、確率が厳密な 0 または 1 になるのを避ける。\n                probability = min(1 - 1e-7, max(1e-7, probability))\n                rows.append((image_id, probability))\n            if batch_number % 50 == 0:\n                print(f\"  Processed {min(len(rows), len(dataset)):,}/{len(dataset):,}\")\n\n    args.output.parent.mkdir(parents=True, exist_ok=True)\n    with args.output.open(\"w\", newline=\"\", encoding=\"utf-8\") as output_file:\n        writer = csv.writer(output_file)\n        writer.writerow((\"id\", \"label\"))\n        writer.writerows((image_id, f\"{probability:.8f}\") for image_id, probability in rows)\n    print(f\"Kaggle submission written to: {args.output.resolve()}\")\n\n\nif __name__ == \"__main__\":\n    torch.multiprocessing.freeze_support()\n    main()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:04:27.759522Z","iopub.execute_input":"2026-07-09T14:04:27.759724Z","iopub.status.idle":"2026-07-09T14:04:27.767049Z","shell.execute_reply.started":"2026-07-09T14:04:27.759706Z","shell.execute_reply":"2026-07-09T14:04:27.766210Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!ls","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:04:27.768326Z","iopub.execute_input":"2026-07-09T14:04:27.768586Z","iopub.status.idle":"2026-07-09T14:04:27.898556Z","shell.execute_reply.started":"2026-07-09T14:04:27.768567Z","shell.execute_reply":"2026-07-09T14:04:27.897848Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from pathlib import Path\nimport zipfile\n\ninput_dir = Path(\"/kaggle/input/competitions/dogs-vs-cats-redux-kernels-edition\")\nwork_dir = Path(\"/kaggle/working\")\n\nwith zipfile.ZipFile(input_dir / \"train.zip\") as z:\n    z.extractall(work_dir)\n\nwith zipfile.ZipFile(input_dir / \"test.zip\") as z:\n    z.extractall(work_dir)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:06:26.837457Z","iopub.execute_input":"2026-07-09T14:06:26.837920Z","iopub.status.idle":"2026-07-09T14:06:39.648120Z","shell.execute_reply.started":"2026-07-09T14:06:26.837890Z","shell.execute_reply":"2026-07-09T14:06:39.647445Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!ls /kaggle/working\n!ls /kaggle/working/train | head\n!ls /kaggle/working/test | head","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:06:49.330831Z","iopub.execute_input":"2026-07-09T14:06:49.331525Z","iopub.status.idle":"2026-07-09T14:06:49.774418Z","shell.execute_reply.started":"2026-07-09T14:06:49.331487Z","shell.execute_reply":"2026-07-09T14:06:49.773432Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!python cnn_train.py \\\n  --data-dir /kaggle/working/train \\\n  --results-dir /kaggle/working/results \\\n  --workers 2","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:07:04.963630Z","iopub.execute_input":"2026-07-09T14:07:04.964617Z","iopub.status.idle":"2026-07-09T14:19:11.577912Z","shell.execute_reply.started":"2026-07-09T14:07:04.964577Z","shell.execute_reply":"2026-07-09T14:19:11.577145Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!python kaggle_predict.py \\\n  --test-dir /kaggle/working/test \\\n  --checkpoint /kaggle/working/results/best_cnn_model.pth \\\n  --output /kaggle/working/submission.csv \\\n  --workers 2","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:22:10.980899Z","iopub.execute_input":"2026-07-09T14:22:10.981206Z","iopub.status.idle":"2026-07-09T14:22:23.936374Z","shell.execute_reply.started":"2026-07-09T14:22:10.981165Z","shell.execute_reply":"2026-07-09T14:22:23.935654Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\n\nsub = pd.read_csv(\"/kaggle/working/submission.csv\")\nsub.head(), sub.shape","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-07-09T14:22:33.664851Z","iopub.execute_input":"2026-07-09T14:22:33.665872Z","iopub.status.idle":"2026-07-09T14:22:33.706808Z","shell.execute_reply.started":"2026-07-09T14:22:33.665830Z","shell.execute_reply":"2026-07-09T14:22:33.705908Z"}},"outputs":[],"execution_count":null}]}