{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceType":"competition","sourceId":13836,"databundleVersionId":1718836}],"dockerImageVersionId":31287,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"\n# ============================================================\n# Cassava Leaf Disease Classification - 1차 베이스라인\n# 환경: Kaggle Notebook / Google Colab\n# 작성: PyTorch + timm + albumentations 기반\n# ============================================================\n\n# ──────────────────────────────────────────────────────────────\n# 0. 라이브러리 설치 (Colab/Kaggle 환경에서 필요 시 주석 해제)\n# ──────────────────────────────────────────────────────────────\n# !pip install timm albumentations -q\n\n# ──────────────────────────────────────────────────────────────\n# 1. 라이브러리 임포트\n# ──────────────────────────────────────────────────────────────\nimport os\nimport random\nimport math\nimport numpy as np\nimport pandas as pd\nfrom PIL import Image\nfrom pathlib import Path\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nfrom torch.optim import AdamW\nfrom torch.optim.lr_scheduler import ReduceLROnPlateau\n# AMP (Automatic Mixed Precision): float16 연산으로 학습 속도 1.5~2배 향상\nfrom torch.amp import GradScaler, autocast\n\nimport timm                          # 사전학습 모델 라이브러리\nimport albumentations as A           # 이미지 데이터 증강\nfrom albumentations.pytorch import ToTensorV2\n\nfrom sklearn.model_selection import StratifiedKFold\nfrom sklearn.metrics import accuracy_score, confusion_matrix\n\nimport matplotlib.pyplot as plt\nimport matplotlib.patches as mpatches\nimport seaborn as sns\n\n# Kaggle/Colab 인라인 출력 설정\nplt.rcParams[\"figure.dpi\"] = 100\nplt.rcParams[\"font.size\"] = 12\n\n# ──────────────────────────────────────────────────────────────\n# 2. 전역 설정 (Config)\n# ──────────────────────────────────────────────────────────────\nclass CFG:\n    # 경로 설정 (Kaggle 기본 경로 기준; 로컬이면 수정)\n    DATA_DIR   = Path(\"/kaggle/input/competitions/cassava-leaf-disease-classification\")\n    TRAIN_IMG  = DATA_DIR / \"train_images\"\n    TRAIN_CSV  = DATA_DIR / \"train.csv\"\n\n    # 모델 파라미터\n    MODEL_NAME = \"tf_efficientnet_b4\"\n    NUM_CLASSES = 5\n    IMG_SIZE   = 384        # 입력 이미지 크기\n\n    # 학습 하이퍼파라미터\n    SEED       = 42\n    FOLD       = 0          # 학습할 Fold 번호 (0번 Fold만 우선 실행)\n    N_SPLITS   = 5          # StratifiedKFold 분할 수\n    EPOCHS     = 20\n    BATCH_SIZE = 16\n    LR         = 1e-4\n    WEIGHT_DECAY = 1e-6\n\n    # Label Smoothing 계수\n    SMOOTHING  = 0.1\n\n    # EarlyStopping 인내 횟수 (patience)\n    PATIENCE   = 3\n\n    # 디바이스 설정 (GPU 우선)\n    DEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n    # 데이터로더 성능 설정\n    # ⚡ 속도 최적화: NUM_WORKERS 4, BATCH_SIZE 32 (AMP로 메모리 여유 생김)\n    NUM_WORKERS = 4\n    BATCH_SIZE  = 32   # AMP 적용 후 2배 증가 가능\n\n\n# ──────────────────────────────────────────────────────────────\n# 3. 시드 고정 (재현성 확보)\n# ──────────────────────────────────────────────────────────────\ndef seed_everything(seed: int = 42):\n    \"\"\"모든 난수 시드를 고정하여 실험 재현성을 보장합니다.\"\"\"\n    random.seed(seed)\n    np.random.seed(seed)\n    os.environ[\"PYTHONHASHSEED\"] = str(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    torch.cuda.manual_seed_all(seed)\n    # ⚡ 속도 최적화: benchmark=True로 cuDNN이 하드웨어에 업진 가장 빠른 커널 자동 선택\n    # (단, 매 에포크마다 입력 크기가 변하면 안 됨)\n    torch.backends.cudnn.deterministic = False\n    torch.backends.cudnn.benchmark     = True\n\nseed_everything(CFG.SEED)\n\n\n# ──────────────────────────────────────────────────────────────\n# 4. 데이터 증강 (Albumentations)\n# ──────────────────────────────────────────────────────────────\ndef get_transforms(data: str = \"train\") -> A.Compose:\n    \"\"\"\n    학습/검증 단계별 albumentations 증강 파이프라인을 반환합니다.\n\n    Args:\n        data: 'train' 또는 'valid'\n    Returns:\n        albumentations Compose 객체\n    \"\"\"\n    # ImageNet 사전학습 모델에서 사용하는 정규화 통계값\n    IMAGENET_MEAN = [0.485, 0.456, 0.406]\n    IMAGENET_STD  = [0.229, 0.224, 0.225]\n\n    if data == \"train\":\n        return A.Compose([\n            # 무작위 영역을 잘라 지정 크기로 리사이즈 (다양한 스케일/비율 학습)\n            # albumentations 2.x부터 size=(H, W) 형식으로 변경됨\n            A.RandomResizedCrop(size=(CFG.IMG_SIZE, CFG.IMG_SIZE),\n                                scale=(0.7, 1.0), ratio=(0.75, 1.33), p=1.0),\n            # 좌우 대칭 반전\n            A.HorizontalFlip(p=0.5),\n            # 상하 대칭 반전 (잎 이미지는 방향 불변이므로 유효)\n            A.VerticalFlip(p=0.5),\n            # 이동(Shift), 스케일(Scale), 회전(Rotate) 동시 적용\n            # albumentations 2.x에서 ShiftScaleRotate가 deprecated → Affine 사용\n            A.Affine(translate_percent={\"x\": (-0.1, 0.1), \"y\": (-0.1, 0.1)},\n                     scale=(0.8, 1.2), rotate=(-30, 30),\n                     mode=0, p=0.5),\n            # 밝기/대비/채도/색조 변환 (ColorJitter 역할)\n            A.ColorJitter(brightness=0.2, contrast=0.2,\n                          saturation=0.2, hue=0.1, p=0.5),\n            # 가우시안 블러 (노이즈 견고성 향상)\n            A.GaussianBlur(blur_limit=(3, 7), p=0.2),\n            # 픽셀값 정규화 (ImageNet 통계 기준)\n            A.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD),\n            # numpy → PyTorch Tensor 변환 (HWC → CHW)\n            ToTensorV2(),\n        ])\n    else:  # valid / test\n        return A.Compose([\n            # 검증 시에는 단순 리사이즈만 수행 (증강 없음)\n            A.Resize(height=CFG.IMG_SIZE, width=CFG.IMG_SIZE),\n            A.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD),\n            ToTensorV2(),\n        ])\n\n\n# ──────────────────────────────────────────────────────────────\n# 5. 커스텀 Dataset 클래스\n# ──────────────────────────────────────────────────────────────\nclass CassavaDataset(Dataset):\n    \"\"\"\n    Cassava 잎 질병 이미지 데이터셋 클래스.\n\n    Args:\n        df         : image_id, label 컬럼을 포함한 DataFrame\n        img_dir    : 이미지 파일이 저장된 디렉터리 경로\n        transform  : albumentations 변환 객체\n        is_test    : 테스트 모드 여부 (label 없음)\n    \"\"\"\n    def __init__(self, df: pd.DataFrame, img_dir: Path,\n                 transform=None, is_test: bool = False):\n        self.df        = df.reset_index(drop=True)\n        self.img_dir   = img_dir\n        self.transform = transform\n        self.is_test   = is_test\n\n    def __len__(self) -> int:\n        return len(self.df)\n\n    def __getitem__(self, idx: int):\n        row      = self.df.iloc[idx]\n        img_path = self.img_dir / row[\"image_id\"]\n\n        # PIL로 이미지 로드 후 RGB 변환 (EXIF 회전 정보 포함)\n        image = np.array(Image.open(img_path).convert(\"RGB\"))\n\n        if self.transform:\n            image = self.transform(image=image)[\"image\"]\n\n        if self.is_test:\n            return image\n\n        label = torch.tensor(row[\"label\"], dtype=torch.long)\n        return image, label\n\n\n# ──────────────────────────────────────────────────────────────\n# 6. 손실 함수: Label Smoothing CrossEntropy\n# ──────────────────────────────────────────────────────────────\nclass LabelSmoothingCrossEntropy(nn.Module):\n    \"\"\"\n    Label Smoothing을 적용한 CrossEntropy Loss.\n\n    라벨 노이즈(Noisy Label)와 과적합(Overfitting)을 완화합니다.\n    smoothing=0.0이면 일반 CrossEntropyLoss와 동일하게 동작합니다.\n\n    Args:\n        smoothing: 스무딩 팩터 (0.0 ~ 1.0), 기본값 0.1\n    \"\"\"\n    def __init__(self, smoothing: float = 0.1):\n        super().__init__()\n        self.smoothing = smoothing\n\n    def forward(self, pred: torch.Tensor, target: torch.Tensor) -> torch.Tensor:\n        n_classes = pred.size(-1)\n        # 로그 소프트맥스 적용\n        log_prob = F.log_softmax(pred, dim=-1)\n\n        # 원-핫 벡터에 smoothing 값을 균등하게 분배\n        # (1 - smoothing)은 정답 클래스에, smoothing/(C-1)은 나머지에 분배\n        with torch.no_grad():\n            smooth_target = torch.full_like(log_prob,\n                                            fill_value=self.smoothing / (n_classes - 1))\n            smooth_target.scatter_(1, target.unsqueeze(1), 1.0 - self.smoothing)\n\n        # NLL Loss 계산 (합산 후 배치 평균)\n        loss = -(smooth_target * log_prob).sum(dim=-1).mean()\n        return loss\n\n\n# ──────────────────────────────────────────────────────────────\n# 7. 모델 정의 (timm EfficientNet-B4)\n# ──────────────────────────────────────────────────────────────\nclass CassavaModel(nn.Module):\n    \"\"\"\n    timm의 tf_efficientnet_b4 사전학습 모델을 불러와\n    마지막 분류 헤드(Classifier)를 5개 클래스에 맞게 교체한 모델.\n\n    Args:\n        model_name : timm 모델 이름\n        num_classes: 분류 클래스 개수 (기본 5)\n        pretrained : ImageNet 사전 가중치 사용 여부\n    \"\"\"\n    def __init__(self, model_name: str = CFG.MODEL_NAME,\n                 num_classes: int = CFG.NUM_CLASSES,\n                 pretrained: bool = True):\n        super().__init__()\n        # timm으로 모델 로드 (num_classes=0 → 원본 헤드 제거)\n        self.backbone = timm.create_model(\n            model_name,\n            pretrained=pretrained,\n            num_classes=0,          # 글로벌 풀링까지만 유지\n        )\n        # 백본 출력 피처 차원 자동 추출\n        in_features = self.backbone.num_features\n\n        # 커스텀 분류 헤드: Dropout → Linear\n        self.head = nn.Sequential(\n            nn.Dropout(p=0.3),\n            nn.Linear(in_features, num_classes),\n        )\n\n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n        features = self.backbone(x)    # (B, in_features)\n        output   = self.head(features) # (B, num_classes)\n        return output\n\n\n# ──────────────────────────────────────────────────────────────\n# 8. EarlyStopping 헬퍼 클래스\n# ──────────────────────────────────────────────────────────────\nclass EarlyStopping:\n    \"\"\"\n    검증 손실(Val Loss)이 지정한 patience 횟수 동안 개선되지 않으면\n    학습을 조기 종료하고 최적 모델 가중치를 저장합니다.\n\n    Args:\n        patience : 개선 없이 허용할 최대 에포크 수\n        delta    : 개선으로 인정하는 최소 손실 감소량\n        path     : 최적 모델 저장 경로\n        verbose  : 상태 메시지 출력 여부\n    \"\"\"\n    def __init__(self, patience: int = 3, delta: float = 0.0,\n                 path: str = \"best_model.pth\", verbose: bool = True):\n        self.patience   = patience\n        self.delta      = delta\n        self.path       = path\n        self.verbose    = verbose\n        self.counter    = 0\n        self.best_score = None\n        self.early_stop = False\n        self.best_loss  = np.inf\n\n    def __call__(self, val_loss: float, model: nn.Module):\n        score = -val_loss  # Loss가 낮을수록 좋으므로 부호 반전\n\n        if self.best_score is None:\n            # 첫 번째 에포크: 현재 값을 기준으로 설정\n            self.best_score = score\n            self._save_checkpoint(val_loss, model)\n\n        elif score < self.best_score + self.delta:\n            # 개선이 없을 때: 카운터 증가\n            self.counter += 1\n            if self.verbose:\n                print(f\"  [EarlyStopping] 개선 없음: {self.counter}/{self.patience}\")\n            if self.counter >= self.patience:\n                self.early_stop = True\n\n        else:\n            # 개선 시: 카운터 초기화 및 모델 저장\n            self.best_score = score\n            self._save_checkpoint(val_loss, model)\n            self.counter = 0\n\n    def _save_checkpoint(self, val_loss: float, model: nn.Module):\n        \"\"\"검증 손실이 개선될 때 모델 가중치를 저장합니다.\"\"\"\n        if self.verbose:\n            print(f\"  [EarlyStopping] Val Loss 개선: {self.best_loss:.4f} → {val_loss:.4f}. 모델 저장 중...\")\n        torch.save(model.state_dict(), self.path)\n        self.best_loss = val_loss\n\n\n# ──────────────────────────────────────────────────────────────\n# 9. 에포크 단위 학습 / 검증 함수\n# ──────────────────────────────────────────────────────────────\ndef train_one_epoch(model, loader, criterion, optimizer, device,\n                    scaler: GradScaler = None) -> tuple[float, float]:\n    \"\"\"\n    한 에포크 동안 학습을 수행하고 평균 손실과 정확도를 반환합니다.\n\n    Args:\n        scaler : GradScaler 객체. None이면 일반 FP32 학습 수행.\n                 GPU가 있으면 AMP(자동 혼합 정밀도)를 활성화합니다.\n    Returns:\n        (avg_loss, avg_accuracy) 튜플\n    \"\"\"\n    model.train()\n    running_loss  = 0.0\n    all_preds     = []\n    all_labels    = []\n    use_amp = (scaler is not None)  # GPU 환경에서만 AMP 활성화\n\n    for step, (images, labels) in enumerate(loader):\n        images = images.to(device, non_blocking=True)\n        labels = labels.to(device, non_blocking=True)\n\n        optimizer.zero_grad()\n\n        if use_amp:\n            # ── AMP 모드: float16으로 순전파 연산 (속도 ↑, 메모리 ↓) ──\n            with autocast(device_type=\"cuda\"):\n                outputs = model(images)\n                loss    = criterion(outputs, labels)\n            # scaler: 역전파 전 loss를 스케일링해 언더플로우 방지\n            scaler.scale(loss).backward()\n            scaler.step(optimizer)   # 스케일 해제 후 가중치 업데이트\n            scaler.update()          # 다음 스텝을 위한 스케일 값 갱신\n        else:\n            # ── 일반 FP32 모드 (CPU 환경 또는 AMP 미사용 시) ──\n            outputs = model(images)\n            loss    = criterion(outputs, labels)\n            loss.backward()\n            optimizer.step()\n\n        running_loss += loss.item() * images.size(0)\n\n        # 예측 클래스 추출 (argmax)\n        preds = outputs.argmax(dim=1).detach().cpu().numpy()\n        all_preds.extend(preds)\n        all_labels.extend(labels.detach().cpu().numpy())\n\n        # 100 스텝마다 중간 결과 출력\n        if (step + 1) % 100 == 0:\n            interim_acc = accuracy_score(all_labels, all_preds)\n            print(f\"    Step [{step+1}/{len(loader)}] \"\n                  f\"Loss: {loss.item():.4f} | Acc: {interim_acc:.4f}\")\n\n    avg_loss = running_loss / len(loader.dataset)\n    avg_acc  = accuracy_score(all_labels, all_preds)\n    return avg_loss, avg_acc\n\n\n@torch.no_grad()\ndef valid_one_epoch(model, loader, criterion, device) -> tuple[float, float]:\n    \"\"\"\n    한 에포크 동안 검증을 수행하고 평균 손실과 정확도를 반환합니다.\n    torch.no_grad()로 그래디언트 계산을 비활성화합니다.\n\n    Returns:\n        (avg_loss, avg_accuracy) 튜플\n    \"\"\"\n    model.eval()\n    running_loss = 0.0\n    all_preds    = []\n    all_labels   = []\n\n    for images, labels in loader:\n        images = images.to(device, non_blocking=True)\n        labels = labels.to(device, non_blocking=True)\n\n        outputs = model(images)\n        loss    = criterion(outputs, labels)\n\n        running_loss += loss.item() * images.size(0)\n\n        preds = outputs.argmax(dim=1).cpu().numpy()\n        all_preds.extend(preds)\n        all_labels.extend(labels.cpu().numpy())\n\n    avg_loss = running_loss / len(loader.dataset)\n    avg_acc  = accuracy_score(all_labels, all_preds)\n    return avg_loss, avg_acc\n\n\n# ──────────────────────────────────────────────────────────────\n# 10. 시각화 함수 모음\n# ──────────────────────────────────────────────────────────────\n\n# 클래스 이름 매핑 (대회 공식 라벨)\nCLASS_NAMES = [\n    \"Bacterial Blight\",       # 0: 세균성 잎마름병\n    \"Brown Streak Disease\",   # 1: 갈색 줄무늬병\n    \"Green Mottle\",           # 2: 녹색 반점병\n    \"Cassava Mosaic Disease\", # 3: 모자이크병 (가장 많음)\n    \"Healthy\",                # 4: 정상\n]\n\n\ndef visualize_class_distribution(df: pd.DataFrame) -> None:\n    \"\"\"\n    [시각화 1] 클래스별 샘플 수 막대 그래프\n\n    클래스 불균형(Class Imbalance)을 한눈에 확인할 수 있습니다.\n    라벨 3번(Cassava Mosaic Disease)이 압도적으로 많음을 볼 수 있습니다.\n    \"\"\"\n    counts = df[\"label\"].value_counts().sort_index()\n    colors = [\"#e74c3c\", \"#e67e22\", \"#2ecc71\", \"#3498db\", \"#9b59b6\"]\n\n    fig, ax = plt.subplots(figsize=(10, 5))\n    bars = ax.bar(CLASS_NAMES, counts.values, color=colors, edgecolor=\"white\", linewidth=1.5)\n\n    # 각 막대 위에 샘플 수 표시\n    for bar, count in zip(bars, counts.values):\n        ax.text(\n            bar.get_x() + bar.get_width() / 2,\n            bar.get_height() + 100,\n            f\"{count:,}\\n({count/len(df)*100:.1f}%)\",\n            ha=\"center\", va=\"bottom\", fontsize=10, fontweight=\"bold\"\n        )\n\n    ax.set_title(\"클래스별 샘플 수 분포 (Class Distribution)\", fontsize=15, fontweight=\"bold\", pad=15)\n    ax.set_ylabel(\"샘플 수\")\n    ax.set_ylim(0, counts.max() * 1.2)\n    ax.tick_params(axis=\"x\", rotation=15)\n    plt.tight_layout()\n    plt.savefig(\"class_distribution.png\", bbox_inches=\"tight\")\n    plt.show()\n    print(\"[시각화 1] 클래스 분포 저장 완료: class_distribution.png\")\n\n\ndef visualize_sample_images(df: pd.DataFrame, img_dir: Path, n_per_class: int = 4) -> None:\n    \"\"\"\n    [시각화 2] 클래스별 샘플 이미지 그리드\n\n    각 클래스에서 n_per_class장 이미지를 무작위로 선택해 보여줍니다.\n    실제 잎 이미지가 어떻게 생겼는지 확인할 수 있습니다.\n\n    Args:\n        df          : 전체 학습 DataFrame\n        img_dir     : 이미지 폴더 경로\n        n_per_class : 클래스당 표시할 이미지 수\n    \"\"\"\n    num_classes = len(CLASS_NAMES)\n    fig, axes = plt.subplots(num_classes, n_per_class,\n                             figsize=(n_per_class * 3, num_classes * 3))\n    fig.suptitle(\"클래스별 샘플 이미지 (Sample Images per Class)\",\n                 fontsize=15, fontweight=\"bold\", y=1.01)\n\n    for cls_idx, cls_name in enumerate(CLASS_NAMES):\n        # 해당 클래스의 이미지 목록에서 무작위 샘플 추출\n        cls_df = df[df[\"label\"] == cls_idx].sample(\n            n=min(n_per_class, len(df[df[\"label\"] == cls_idx])),\n            random_state=CFG.SEED\n        )\n        for col_idx, (_, row) in enumerate(cls_df.iterrows()):\n            ax = axes[cls_idx][col_idx]\n            img = Image.open(img_dir / row[\"image_id\"]).convert(\"RGB\")\n            img = img.resize((256, 256))  # 빠른 출력을 위해 리사이즈\n            ax.imshow(img)\n            ax.axis(\"off\")\n            if col_idx == 0:\n                # 첫 번째 열에 클래스 이름 표시\n                ax.set_title(f\"[{cls_idx}] {cls_name}\", fontsize=9,\n                             fontweight=\"bold\", loc=\"left\", pad=3)\n\n    plt.tight_layout()\n    plt.savefig(\"sample_images.png\", bbox_inches=\"tight\")\n    plt.show()\n    print(\"[시각화 2] 샘플 이미지 저장 완료: sample_images.png\")\n\n\ndef plot_training_history(history: dict) -> None:\n    \"\"\"\n    [시각화 3] 학습 곡선 (Training Curves)\n\n    에포크별 Train/Val Loss와 Accuracy를 나란히 그립니다.\n    - Loss 곡선: 두 곡선이 함께 내려가야 정상\n    - Accuracy 곡선: 두 곡선이 함께 올라가야 정상\n    - Train은 좋은데 Val이 나쁘면 → 과적합(Overfitting) 의심\n\n    Args:\n        history: train_loss, val_loss, train_acc, val_acc 리스트를 담은 딕셔너리\n    \"\"\"\n    epochs = range(1, len(history[\"train_loss\"]) + 1)\n\n    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))\n    fig.suptitle(f\"학습 곡선 (Fold {CFG.FOLD})\", fontsize=15, fontweight=\"bold\")\n\n    # ── Loss 그래프 ──────────────────────────────────────────\n    ax1.plot(epochs, history[\"train_loss\"], \"b-o\", label=\"Train Loss\",\n             markersize=5, linewidth=2)\n    ax1.plot(epochs, history[\"val_loss\"],   \"r-o\", label=\"Val Loss\",\n             markersize=5, linewidth=2)\n    ax1.set_title(\"Loss 곡선\")\n    ax1.set_xlabel(\"Epoch\")\n    ax1.set_ylabel(\"Loss\")\n    ax1.legend()\n    ax1.grid(True, alpha=0.3)\n    # 최저 Val Loss 위치에 별표\n    best_epoch = history[\"val_loss\"].index(min(history[\"val_loss\"])) + 1\n    ax1.axvline(x=best_epoch, color=\"gray\", linestyle=\"--\", alpha=0.6,\n                label=f\"Best Epoch ({best_epoch})\")\n\n    # ── Accuracy 그래프 ──────────────────────────────────────\n    ax2.plot(epochs, history[\"train_acc\"], \"b-o\", label=\"Train Acc\",\n             markersize=5, linewidth=2)\n    ax2.plot(epochs, history[\"val_acc\"],   \"r-o\", label=\"Val Acc\",\n             markersize=5, linewidth=2)\n    ax2.set_title(\"Accuracy 곡선\")\n    ax2.set_xlabel(\"Epoch\")\n    ax2.set_ylabel(\"Accuracy\")\n    ax2.legend()\n    ax2.grid(True, alpha=0.3)\n    # 최고 Val Accuracy 값 강조\n    best_val_acc = max(history[\"val_acc\"])\n    ax2.set_ylim(0, 1.05)\n    ax2.text(0.98, 0.05, f\"Best Val Acc: {best_val_acc:.4f}\",\n             transform=ax2.transAxes, ha=\"right\", fontsize=11,\n             color=\"red\", fontweight=\"bold\",\n             bbox=dict(boxstyle=\"round\", fc=\"white\", ec=\"red\", alpha=0.8))\n\n    plt.tight_layout()\n    plt.savefig(\"training_curves.png\", bbox_inches=\"tight\")\n    plt.show()\n    print(\"[시각화 3] 학습 곡선 저장 완료: training_curves.png\")\n\n\n@torch.no_grad()\ndef plot_confusion_matrix(model: nn.Module, loader: DataLoader, device) -> None:\n    \"\"\"\n    [시각화 4] 혼동 행렬 (Confusion Matrix)\n\n    모델이 어떤 클래스를 어떤 클래스로 잘못 분류하는지 보여줍니다.\n    - 대각선: 정확히 맞춘 예측 (많을수록 좋음)\n    - 비대각선: 오분류 (잘못 예측한 경우)\n\n    Args:\n        model  : 학습된 모델 (best weights 로드 권장)\n        loader : 검증 DataLoader\n        device : GPU/CPU\n    \"\"\"\n    model.eval()\n    all_preds  = []\n    all_labels = []\n\n    for images, labels in loader:\n        images = images.to(device)\n        outputs = model(images)\n        preds   = outputs.argmax(dim=1).cpu().numpy()\n        all_preds.extend(preds)\n        all_labels.extend(labels.numpy())\n\n    # confusion_matrix: 행=실제 라벨, 열=예측 라벨\n    cm = confusion_matrix(all_labels, all_preds)\n    # 행별로 정규화 (각 실제 클래스의 비율로)\n    cm_norm = cm.astype(float) / cm.sum(axis=1, keepdims=True)\n\n    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 6))\n    fig.suptitle(\"혼동 행렬 (Confusion Matrix)\", fontsize=15, fontweight=\"bold\")\n\n    # ── 원본 숫자 ───────────────────────────────────────────\n    sns.heatmap(cm, annot=True, fmt=\"d\", cmap=\"Blues\",\n                xticklabels=CLASS_NAMES, yticklabels=CLASS_NAMES,\n                ax=ax1, linewidths=0.5)\n    ax1.set_title(\"샘플 수 (Raw Counts)\")\n    ax1.set_ylabel(\"실제 라벨 (True)\")\n    ax1.set_xlabel(\"예측 라벨 (Predicted)\")\n    ax1.tick_params(axis=\"x\", rotation=30)\n    ax1.tick_params(axis=\"y\", rotation=0)\n\n    # ── 정규화된 비율 ────────────────────────────────────────\n    sns.heatmap(cm_norm, annot=True, fmt=\".2f\", cmap=\"Greens\",\n                xticklabels=CLASS_NAMES, yticklabels=CLASS_NAMES,\n                ax=ax2, linewidths=0.5, vmin=0, vmax=1)\n    ax2.set_title(\"비율 (Normalized, 행 기준)\")\n    ax2.set_ylabel(\"실제 라벨 (True)\")\n    ax2.set_xlabel(\"예측 라벨 (Predicted)\")\n    ax2.tick_params(axis=\"x\", rotation=30)\n    ax2.tick_params(axis=\"y\", rotation=0)\n\n    plt.tight_layout()\n    plt.savefig(\"confusion_matrix.png\", bbox_inches=\"tight\")\n    plt.show()\n\n    # 클래스별 정확도 출력\n    print(\"\\n[클래스별 정확도]\")\n    for i, name in enumerate(CLASS_NAMES):\n        print(f\"  [{i}] {name:<25}: {cm_norm[i, i]:.4f} ({cm[i, i]}/{cm[i].sum()})\")\n    print(\"[시각화 4] 혼동 행렬 저장 완료: confusion_matrix.png\")\n\n\n# ──────────────────────────────────────────────────────────────\n# 11. 메인 학습 루프 (StratifiedKFold, Fold 0 기준)\n# ──────────────────────────────────────────────────────────────\ndef run_training():\n    \"\"\"\n    StratifiedKFold를 사용하여 데이터를 분할하고,\n    CFG.FOLD번 Fold에 대해서만 모델을 학습합니다.\n\n    - 학습 루프에서 EarlyStopping과 ReduceLROnPlateau를 적용합니다.\n    - 각 에포크마다 Train/Val Loss 및 Accuracy를 출력합니다.\n    \"\"\"\n    print(\"=\" * 60)\n    print(f\"디바이스: {CFG.DEVICE}\")\n    print(f\"모델: {CFG.MODEL_NAME} | 이미지 크기: {CFG.IMG_SIZE}\")\n    print(\"=\" * 60)\n\n    # ── 11-1. 데이터 로드 ──────────────────────────────────────\n    df = pd.read_csv(CFG.TRAIN_CSV)\n    print(f\"\\n전체 데이터 수: {len(df)}\")\n    print(f\"클래스 분포:\\n{df['label'].value_counts().sort_index()}\\n\")\n\n    # 클래스 분포 & 샘플 이미지 시각화 (학습 전 데이터 탐색)\n    visualize_class_distribution(df)\n    visualize_sample_images(df, CFG.TRAIN_IMG)\n\n    # ── 10-2. StratifiedKFold 분할 ────────────────────────────\n    skf = StratifiedKFold(n_splits=CFG.N_SPLITS, shuffle=True,\n                          random_state=CFG.SEED)\n\n    # fold_splits: 각 Fold의 (train_idx, val_idx) 튜플 리스트\n    fold_splits = list(skf.split(df[\"image_id\"], df[\"label\"]))\n\n    train_idx, val_idx = fold_splits[CFG.FOLD]\n    train_df = df.iloc[train_idx]\n    val_df   = df.iloc[val_idx]\n\n    print(f\"[Fold {CFG.FOLD}] Train: {len(train_df)} | Val: {len(val_df)}\")\n\n    # ── 10-3. Dataset & DataLoader ────────────────────────────\n    train_dataset = CassavaDataset(\n        df=train_df,\n        img_dir=CFG.TRAIN_IMG,\n        transform=get_transforms(\"train\"),\n    )\n    val_dataset = CassavaDataset(\n        df=val_df,\n        img_dir=CFG.TRAIN_IMG,\n        transform=get_transforms(\"valid\"),\n    )\n\n    train_loader = DataLoader(\n        train_dataset,\n        batch_size=CFG.BATCH_SIZE,\n        shuffle=True,\n        num_workers=CFG.NUM_WORKERS,\n        pin_memory=True,\n        drop_last=True,\n        # ⚡ 에포크마다 워커 프로세스를 재시작하지 않아 오버헤드 ↓\n        persistent_workers=True,\n    )\n    val_loader = DataLoader(\n        val_dataset,\n        batch_size=CFG.BATCH_SIZE * 2,\n        shuffle=False,\n        num_workers=CFG.NUM_WORKERS,\n        pin_memory=True,\n        persistent_workers=True,\n    )\n\n    # ── 10-4. 모델 초기화 ──────────────────────────────────────\n    model = CassavaModel(\n        model_name=CFG.MODEL_NAME,\n        num_classes=CFG.NUM_CLASSES,\n        pretrained=True,\n    ).to(CFG.DEVICE)\n    print(f\"\\n모델 '{CFG.MODEL_NAME}' 로드 완료 (파라미터 수: \"\n          f\"{sum(p.numel() for p in model.parameters() if p.requires_grad):,})\")\n\n    # ⚡ torch.compile: 모델 커널을 퍼집(Fuse)하여 연산 최적화 (PyTorch 2.0+ 기능)\n    # 첫 배치는 컴파일 시간으로 느리지만 이후 에포크부터 10~30% 추가 가속\n    if CFG.DEVICE.type == \"cuda\":\n        print(\"  torch.compile() 적용 중... (첫 배치 컴파일에 30초등 소요, 이후 복수 에포크에서 효과)\")\n        model = torch.compile(model, mode=\"reduce-overhead\")\n    print()\n\n    # ── 10-5. 손실 함수, 옵티마이저, 스케줄러 ─────────────────\n    # 라벨 노이즈 대응을 위한 Label Smoothing CrossEntropy\n    criterion = LabelSmoothingCrossEntropy(smoothing=CFG.SMOOTHING)\n\n    # AdamW: weight_decay로 L2 정규화 효과\n    optimizer = AdamW(\n        model.parameters(),\n        lr=CFG.LR,\n        weight_decay=CFG.WEIGHT_DECAY,\n    )\n\n    # 검증 손실이 plateau에 도달하면 학습률을 0.5배로 감소\n    # verbose 파라미터는 PyTorch 2.2+에서 제거됨 → 직접 LR 출력으로 대체\n    scheduler = ReduceLROnPlateau(\n        optimizer,\n        mode=\"min\",       # 손실 최소화 기준\n        factor=0.5,       # LR 감소 비율\n        patience=2,       # 2 에포크 미개선 시 적용\n    )\n\n    # ── 10-6. EarlyStopping 초기화 ────────────────────────────\n    early_stopping = EarlyStopping(\n        patience=CFG.PATIENCE,\n        path=f\"best_model_fold{CFG.FOLD}.pth\",\n        verbose=True,\n    )\n\n    # ── AMP GradScaler 초기화 ─────────────────────────────────\n    # GPU가 있을 때만 AMP 사용 (CPU는 float16 미지원)\n    scaler = GradScaler(\"cuda\") if CFG.DEVICE.type == \"cuda\" else None\n    print(f\"  AMP(혼합 정밀도) 학습: {'✅ 활성화' if scaler else '❌ CPU 모드 (비활성화)'}\")\n\n    # ── 10-7. 에포크 학습 루프 ────────────────────────────────\n    best_val_acc = 0.0\n    history = {\"train_loss\": [], \"train_acc\": [],\n               \"val_loss\":   [], \"val_acc\":   []}\n\n    for epoch in range(1, CFG.EPOCHS + 1):\n        current_lr = optimizer.param_groups[0][\"lr\"]\n        print(f\"\\n{'─'*60}\")\n        print(f\"Epoch {epoch}/{CFG.EPOCHS}  |  LR: {current_lr:.6f}\")\n        print(f\"{'─'*60}\")\n\n        # 학습 (scaler를 전달하면 AMP 자동 활성화)\n        train_loss, train_acc = train_one_epoch(\n            model, train_loader, criterion, optimizer, CFG.DEVICE,\n            scaler=scaler,\n        )\n\n        # 검증\n        val_loss, val_acc = valid_one_epoch(\n            model, val_loader, criterion, CFG.DEVICE\n        )\n\n        # 히스토리 기록\n        history[\"train_loss\"].append(train_loss)\n        history[\"train_acc\"].append(train_acc)\n        history[\"val_loss\"].append(val_loss)\n        history[\"val_acc\"].append(val_acc)\n\n        # 에포크 결과 출력\n        print(f\"\\n  Train Loss: {train_loss:.4f} | Train Acc: {train_acc:.4f}\")\n        print(f\"  Val   Loss: {val_loss:.4f} | Val   Acc: {val_acc:.4f}\")\n\n        if val_acc > best_val_acc:\n            best_val_acc = val_acc\n            print(f\"  ★ Best Val Accuracy 갱신: {best_val_acc:.4f}\")\n\n        # 스케줄러 업데이트 (검증 손실 기준)\n        scheduler.step(val_loss)\n\n        # EarlyStopping 체크 (검증 손실 기준)\n        early_stopping(val_loss, model)\n        if early_stopping.early_stop:\n            print(f\"\\n[조기 종료] {CFG.PATIENCE} 에포크 동안 개선 없음. 학습 중단.\")\n            break\n\n    # ── 11-8. 최종 결과 출력 ──────────────────────────────────\n    print(f\"\\n{'='*60}\")\n    print(f\"[Fold {CFG.FOLD}] 학습 완료!\")\n    print(f\"  최고 Val Accuracy: {best_val_acc:.4f}\")\n    print(f\"  최적 모델 저장 위치: best_model_fold{CFG.FOLD}.pth\")\n    print(f\"{'='*60}\")\n\n    # ── 11-9. 학습 후 시각화 ──────────────────────────────────\n    # 학습 곡선 (Loss & Accuracy 변화)\n    plot_training_history(history)\n\n    # 혼동 행렬: 최적 모델 가중치 로드 후 검증 세트에서 계산\n    print(\"\\n최적 모델 가중치 로드 중...\")\n    model.load_state_dict(torch.load(f\"best_model_fold{CFG.FOLD}.pth\",\n                                     map_location=CFG.DEVICE))\n    plot_confusion_matrix(model, val_loader, CFG.DEVICE)\n\n    return history\n\n\n# ──────────────────────────────────────────────────────────────\n# 12. 엔트리 포인트\n# ──────────────────────────────────────────────────────────────\nif __name__ == \"__main__\":\n    history = run_training()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-03-09T15:39:22.731153Z","iopub.execute_input":"2026-03-09T15:39:22.731717Z"}},"outputs":[],"execution_count":null}]}