{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":14774,"databundleVersionId":875431,"sourceType":"competition"},{"sourceId":13983054,"sourceType":"datasetVersion","datasetId":8912910},{"sourceId":673496,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":510398,"modelId":525070}],"dockerImageVersionId":31192,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport cv2\nimport numpy as np\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader, WeightedRandomSampler\nfrom torchvision import transforms\n# Note: Using train_test_split twice to achieve a three-way split (Train/Val/Test)\nfrom sklearn.model_selection import train_test_split \nfrom sklearn.metrics import precision_recall_fscore_support, confusion_matrix\nimport timm \nfrom tqdm import tqdm\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nimport warnings\nwarnings.filterwarnings(\"ignore\")\n\n\n# Configuration\n\nCONFIG = {\n    \"seed\": 42,\n    \"img_size\": 224,\n    \"batch_size\": 32, \n    \"epochs\": 15,            \n    \"lr\": 1e-4,              \n    \"num_classes\": 3,        \n    \"model_name\": \"swin_tiny_patch4_window7_224\", \n    \"device\": torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\"),\n    \"csv_path\": \"/kaggle/input/aptos2019-blindness-detection/train.csv\",\n    \"image_dir\": \"/kaggle/input/preprocessed-images-224\"\n}\n\n# Label Mapping: 0 -> 0 (Healthy); 1,2 -> 1 (Mild/Mod); 3,4 -> 2 (Sev/Prolif)\nLABEL_MAP = {0: 0, 1: 1, 2: 1, 3: 2, 4: 2}\n\ndef seed_everything(seed):\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed(seed)\n        torch.backends.cudnn.deterministic = True\n\nseed_everything(CONFIG['seed'])\n\n# Dataset\nclass APTOSFastDataset(Dataset):\n    def __init__(self, df, img_dir, transform=None):\n        self.df = df\n        self.img_dir = img_dir\n        self.transform = transform\n\n    def __len__(self):\n        return len(self.df)\n\n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        img_name = row['id_code']\n        img_path = os.path.join(self.img_dir, img_name + \".png\")\n        \n        image = cv2.imread(img_path)\n        if image is None: \n            # Return a black image if loading fails\n            image = np.zeros((CONFIG['img_size'], CONFIG['img_size'], 3), dtype=np.uint8)\n        else:\n            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n\n        if self.transform: \n            image = self.transform(image)\n            \n        new_label = LABEL_MAP[row['diagnosis']]\n        return image, torch.tensor(new_label, dtype=torch.long)\n\n# Data Augmentation Transforms\n\ntrain_tf = transforms.Compose([\n    transforms.ToPILImage(),\n    transforms.RandomHorizontalFlip(),\n    transforms.RandomVerticalFlip(),\n    transforms.RandomRotation(45),\n    transforms.RandomAffine(degrees=0, translate=(0.1, 0.1), scale=(0.9, 1.1)),\n    transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.1),\n    transforms.ToTensor(), \n    transforms.Normalize([0.5, 0.5, 0.5], [0.5, 0.5, 0.5])\n])\n\nval_tf = transforms.Compose([\n    transforms.ToPILImage(),\n    transforms.ToTensor(),\n    transforms.Normalize([0.5, 0.5, 0.5], [0.5, 0.5, 0.5])\n])\n\n# Use val_tf for the test set as well\ntest_tf = val_tf\n\n# ====================================================\n# Data Preparation: Three-way Split (Train/Validation/Test) & Sampler Calculation\n# ====================================================\ndf = pd.read_csv(CONFIG['csv_path'])\ndf['new_label'] = df['diagnosis'].map(LABEL_MAP)\n\n# 1. First split: Separate 20% for the Test Set\ntrain_val_df, test_df = train_test_split(\n    df, \n    test_size=0.2, # 20% for final independent testing\n    random_state=CONFIG['seed'], \n    stratify=df['new_label']\n)\n\n# 2. Second split: Separate Validation Set from the remaining 80%\n# We use 0.25 of the remaining data (0.25 * 0.8 = 0.2 total) to ensure Validation is also 20% of total\ntrain_df, val_df = train_test_split(\n    train_val_df, \n    test_size=0.25, \n    random_state=CONFIG['seed'], \n    stratify=train_val_df['new_label']\n)\n\nprint(f\"Dataset Split: Train={len(train_df)}, Validation={len(val_df)}, Test={len(test_df)}\")\nprint(\"-\" * 65)\n\n# --- Calculate WeightedRandomSampler weights (Train set only) ---\ntargets = train_df['new_label'].values\nclass_counts = np.bincount(targets)\nclass_weights = 1. / class_counts\nsample_weights = torch.FloatTensor([class_weights[t] for t in targets])\nsampler = WeightedRandomSampler(weights=sample_weights, num_samples=len(sample_weights), replacement=True)\n\nprint(f\"✅ Sampler Ready. Class Counts in Train: {class_counts}\")\n\n# Loaders\ntrain_loader = DataLoader(\n    APTOSFastDataset(train_df, CONFIG['image_dir'], train_tf), \n    batch_size=CONFIG['batch_size'], \n    sampler=sampler, \n    shuffle=False,    \n    num_workers=2\n)\n\nval_loader = DataLoader(\n    APTOSFastDataset(val_df, CONFIG['image_dir'], val_tf), \n    batch_size=CONFIG['batch_size'], \n    shuffle=False, \n    num_workers=2\n)\n\n# New Test Loader\ntest_loader = DataLoader(\n    APTOSFastDataset(test_df, CONFIG['image_dir'], test_tf), \n    batch_size=CONFIG['batch_size'], \n    shuffle=False, \n    num_workers=2\n)\n\n# Model Training\n\nsave_dir = \"checkpoints\"\nos.makedirs(save_dir, exist_ok=True)\n\n# Using Swin Transformer (Tiny)\nmodel = timm.create_model(CONFIG['model_name'], pretrained=True, num_classes=3).to(CONFIG['device'])\ncriterion = nn.CrossEntropyLoss(label_smoothing=0.1)\noptimizer = optim.AdamW(model.parameters(), lr=CONFIG['lr'])\nscheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='max', factor=0.5, patience=2, verbose=True)\n\nbest_sick_f1 = -1.0 \nprint(f\"\\n🚀 Starting Training with WeightedSampler & Augmentation...\")\n\nfor epoch in range(CONFIG['epochs']):\n    # --------------------------- Train ---------------------------\n    model.train()\n    train_loss = 0\n    for imgs, lbls in tqdm(train_loader, desc=f\"Epoch {epoch+1} (Train)\"):\n        imgs, lbls = imgs.to(CONFIG['device']), lbls.to(CONFIG['device'])\n        \n        optimizer.zero_grad()\n        outputs = model(imgs)\n        loss = criterion(outputs, lbls)\n        loss.backward()\n        optimizer.step()\n        train_loss += loss.item()\n\n    # --------------------------- Validation ---------------------------\n    model.eval()\n    val_preds, val_labels = [], []\n    with torch.no_grad():\n        for imgs, lbls in val_loader:\n            imgs = imgs.to(CONFIG['device'])\n            outputs = model(imgs)\n            _, p = torch.max(outputs, 1)\n            val_preds.extend(p.cpu().numpy())\n            val_labels.extend(lbls.numpy())\n\n    # Metrics\n    metrics = precision_recall_fscore_support(val_labels, val_preds, labels=[0, 1, 2], zero_division=0)\n    precision, recall, f1_scores = metrics[0], metrics[1], metrics[2]\n    avg_sick_f1 = (f1_scores[1] + f1_scores[2]) / 2.0\n    overall_acc = np.mean(np.array(val_preds) == np.array(val_labels))\n\n    # Log\n    print(f\"\\n✅ Epoch {epoch+1} Summary:\")\n    print(f\"Train Loss: {train_loss/len(train_loader):.4f} | Val Overall Acc: {overall_acc:.4f}\")\n    print(\"-\" * 65)\n    print(f\"{'Class':<20} | {'Recall':<10} | {'Precision':<10} | {'F1-Score':<10}\")\n    print(\"-\" * 65)\n    print(f\"{'0 (Healthy)':<20} | {recall[0]:.4f}      | {precision[0]:.4f}         | {f1_scores[0]:.4f}\")\n    print(f\"{'1 (Mild/Mod)':<20} | {recall[1]:.4f}      | {precision[1]:.4f}         | {f1_scores[1]:.4f}\")\n    print(f\"{'2 (Sev/Prolif)':<20} | {recall[2]:.4f}      | {precision[2]:.4f}         | {f1_scores[2]:.4f}\")\n    print(\"-\" * 65)\n    print(f\"🎯 Val Avg Sick F1: {avg_sick_f1:.4f} (Best: {best_sick_f1:.4f})\")\n\n    # Save\n    if avg_sick_f1 > best_sick_f1:\n        best_sick_f1 = avg_sick_f1\n        # Save model to the new save_dir path\n        torch.save(model.state_dict(), os.path.join(save_dir, \"best_sick_f1_model.pth\"))\n        print(\"🏆 Best Model Updated!\")\n\n    scheduler.step(avg_sick_f1)\n    print(\"\\n\")\n\n# Final Evaluation on Independent Test Set\nprint(\"\\n\" + \"=\"*70)\nprint(\"🏆 FINAL EVALUATION ON INDEPENDENT TEST SET\")\nprint(\"=\"*70)\n\n# 1. Load best model\nmodel.load_state_dict(torch.load(os.path.join(save_dir, \"best_sick_f1_model.pth\")))\nmodel.eval()\n\n# 2. Inference on Test Set\ntest_preds, test_labels = [], []\nwith torch.no_grad():\n    for imgs, lbls in tqdm(test_loader, desc=\"Final Test\"): # Use test_loader\n        imgs = imgs.to(CONFIG['device'])\n        outputs = model(imgs)\n        _, p = torch.max(outputs, 1)\n        test_preds.extend(p.cpu().numpy())\n        test_labels.extend(lbls.numpy())\n\n# 3. Calculate Final Metrics\ntest_metrics = precision_recall_fscore_support(test_labels, test_preds, labels=[0, 1, 2], zero_division=0)\ntest_precision, test_recall, test_f1_scores = test_metrics[0], test_metrics[1], test_metrics[2]\ntest_avg_sick_f1 = (test_f1_scores[1] + test_f1_scores[2]) / 2.0\ntest_overall_acc = np.mean(np.array(test_preds) == np.array(test_labels))\n\n# 4. Print Report\nprint(\"\\n\" + \"#\"*65)\nprint(f\"✨ Test Overall Acc: {test_overall_acc:.4f}\")\nprint(\"-\" * 65)\nprint(f\"{'Class':<20} | {'Recall':<10} | {'Precision':<10} | {'F1-Score':<10}\")\nprint(\"-\" * 65)\nprint(f\"{'0 (Healthy)':<20} | {test_recall[0]:.4f}      | {test_precision[0]:.4f}         | {test_f1_scores[0]:.4f}\")\nprint(f\"{'1 (Mild/Mod)':<20} | {test_recall[1]:.4f}      | {test_precision[1]:.4f}         | {test_f1_scores[1]:.4f}\")\nprint(f\"{'2 (Sev/Prolif)':<20} | {test_recall[2]:.4f}      | {test_precision[2]:.4f}         | {test_f1_scores[2]:.4f}\")\nprint(\"-\" * 65)\nprint(f\"🎉 Final Test Avg Sick F1: {test_avg_sick_f1:.4f} (Val Best F1: {best_sick_f1:.4f})\")\nprint(\"#\"*65)\n\n# 5. Plot Confusion Matrix\ncm = confusion_matrix(test_labels, test_preds)\nplt.figure(figsize=(8, 6))\n# sns.heatmap(cm, annot=True, fmt='d', cmap='Reds', xticklabels=['0', '1', '2'], yticklabels=['0', '1', '2'])\nplt.title(f'Confusion Matrix on **Independent Test Set**\\nTest Avg Sick F1: {test_avg_sick_f1:.4f}')\nplt.xlabel('Predicted Label'); plt.ylabel('True Label')\nplt.show()\nprint(\"🏁 Training and Final Evaluation Complete.\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-12-05T15:59:37.636205Z","iopub.execute_input":"2025-12-05T15:59:37.636464Z","iopub.status.idle":"2025-12-05T16:05:06.406942Z","shell.execute_reply.started":"2025-12-05T15:59:37.636443Z","shell.execute_reply":"2025-12-05T16:05:06.406098Z"}},"outputs":[],"execution_count":null}]}