{"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":"nvidiaTeslaT4","dataSources":[{"sourceId":113558,"databundleVersionId":14174843,"sourceType":"competition"}],"dockerImageVersionId":31154,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Install Required Segmentation Packages","metadata":{}},{"cell_type":"code","source":"!pip install -q segmentation-models-pytorch","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import sys\nsys.path.append(\"/kaggle/input/smp-lib-zip/\") \n\nimport warnings\nwarnings.filterwarnings(\"ignore\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom PIL import Image\nimport cv2\nfrom tqdm import tqdm\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\nimport segmentation_models_pytorch as smp\nimport warnings\nimport time\nfrom collections import defaultdict\nimport copy\nimport random","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"BASE_PATH = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nTRAIN_AUTHENTIC_PATH = os.path.join(BASE_PATH, \"train_images/authentic\")\nTRAIN_FORGED_PATH = os.path.join(BASE_PATH, \"train_images/forged\")\nTRAIN_MASKS_PATH = os.path.join(BASE_PATH, \"train_masks\")\nTEST_IMAGES_PATH = os.path.join(BASE_PATH, \"test_images\")\n\n\nprint(\"=\"*60)\nprint(\" DATA LOADING AND BASIC STATISTICS\")\nprint(\"=\"*60)\n\nauthentic_images = sorted([f for f in os.listdir(TRAIN_AUTHENTIC_PATH) if f.endswith('.png')])\nforged_images = sorted([f for f in os.listdir(TRAIN_FORGED_PATH) if f.endswith('.png')])\nmask_files = sorted([f for f in os.listdir(TRAIN_MASKS_PATH) if f.endswith('.npy')])\ntest_images = sorted([f for f in os.listdir(TEST_IMAGES_PATH) if f.endswith('.png')])\n\nprint(f\"Number of AUTHENTIC images: {len(authentic_images)}\")\nprint(f\"Number of FORGED images: {len(forged_images)}\")\nprint(f\"Number of MASK files: {len(mask_files)}\")\nprint(f\"Number of TEST images: {len(test_images)}\")\nprint(f\"\\nTotal TRAIN images: {len(authentic_images) + len(forged_images)}\")\nprint(f\"Forged/Authentic ratio: {len(forged_images)/len(authentic_images):.2f}\")\nprint(\"\\n\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Data Visualization","metadata":{}},{"cell_type":"code","source":"def visualize_forgery_overlay(forged_list, masks_path, n_samples=4): \n    selected_images = random.sample(forged_list, min(n_samples, len(forged_list)))\n    \n    fig, axes = plt.subplots(len(selected_images), 3, figsize=(18, 6*len(selected_images)))\n    fig.suptitle('Forged Images with Mask Overlay', fontsize=16, fontweight='bold')\n    \n    for i, forged_file in enumerate(selected_images):\n        img_path = os.path.join(TRAIN_FORGED_PATH, forged_file)\n        img = np.array(Image.open(img_path))\n        mask_name = forged_file.replace('.png', '.npy')\n        mask_path = os.path.join(masks_path, mask_name)\n        \n        if os.path.exists(mask_path):\n            mask = np.load(mask_path)\n            if len(mask.shape) == 3:\n                combined_mask = np.max(mask, axis=0)\n                n_regions = mask.shape[0]\n            else:\n                combined_mask = mask\n                n_regions = 1\n            \n            axes[i, 0].imshow(img)\n            axes[i, 0].set_title(f'Original Image\\n{forged_file}', fontsize=10)\n            axes[i, 0].axis('off')\n\n            axes[i, 1].imshow(combined_mask, cmap='hot')\n            axes[i, 1].set_title(f'Forgery Mask\\n{n_regions} region(s)', fontsize=10)\n            axes[i, 1].axis('off')\n            \n            overlay = img.copy()\n            if len(img.shape) == 2:  \n                overlay = cv2.cvtColor(overlay, cv2.COLOR_GRAY2RGB)\n            \n            red_mask = np.zeros_like(overlay)\n            red_mask[:,:,0] = combined_mask * 255  \n            alpha = 0.4\n            blended = cv2.addWeighted(overlay, 1-alpha, red_mask, alpha, 0)\n            \n            axes[i, 2].imshow(blended)\n            axes[i, 2].set_title('Overlay (Red = Forged)', fontsize=10)\n            axes[i, 2].axis('off')\n        else:\n            for j in range(3):\n                axes[i, j].text(0.5, 0.5, 'MASK NOT FOUND', \n                               ha='center', va='center', fontsize=12)\n                axes[i, j].axis('off')\n    \n    plt.tight_layout()\n    plt.show()\n\nvisualize_forgery_overlay(forged_images, TRAIN_MASKS_PATH, n_samples=8)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Data Preparation","metadata":{}},{"cell_type":"code","source":"image_names = []\nimage_paths = []\nmask_paths = []\nlabels = []  \ncategories = []\n\nfor img_name in authentic_images:\n    image_names.append(img_name)\n    image_paths.append(os.path.join(TRAIN_AUTHENTIC_PATH, img_name))\n    mask_paths.append(None)  \n    labels.append(0)\n    categories.append('authentic')\n\n\nfor img_name in forged_images:\n    image_names.append(img_name)\n    image_paths.append(os.path.join(TRAIN_FORGED_PATH, img_name))\n    \n    mask_name = img_name.replace('.png', '.npy')\n    mask_path = os.path.join(TRAIN_MASKS_PATH, mask_name)\n    mask_paths.append(mask_path if os.path.exists(mask_path) else None)\n    \n    labels.append(1)\n    categories.append('forged')\n\ntrain_df = pd.DataFrame({\n    'image_name': image_names,\n    'image_path': image_paths,\n    'mask_path': mask_paths,\n    'label': labels,\n    'category': categories\n})\n\nprint(f\"\\nTraining DataFrame created!\")\nprint(f\"Total samples: {len(train_df)}\")\nprint(f\"\\nClass distribution:\")\nprint(train_df['category'].value_counts())\nprint(f\"\\nFirst few rows:\")\nprint(train_df.head())\n\nforged_df = train_df[train_df['category'] == 'forged']\nmissing_masks = forged_df['mask_path'].isna().sum()\nprint(f\"\\nForged images without masks: {missing_masks}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from sklearn.model_selection import train_test_split\n\ntrain_data, val_data = train_test_split(\n    train_df, \n    test_size=0.2, \n    random_state=42,\n    stratify=train_df['label']\n)\n\ntrain_data = train_data.reset_index(drop=True)\nval_data = val_data.reset_index(drop=True)\n\nprint(f\"\\nTrain set size: {len(train_data)}\")\nprint(f\"Validation set size: {len(val_data)}\")\n\nprint(f\"\\nTrain set distribution:\")\nprint(train_data['category'].value_counts())\nprint(f\"  Forged percentage: {(train_data['label'].sum() / len(train_data)) * 100:.2f}%\")\n\nprint(f\"\\nValidation set distribution:\")\nprint(val_data['category'].value_counts())\nprint(f\"  Forged percentage: {(val_data['label'].sum() / len(val_data)) * 100:.2f}%\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Data Loader Helper Function","metadata":{}},{"cell_type":"code","source":"def load_image(image_path, target_size=(512, 512)):\n    img = Image.open(image_path).convert('RGB')\n    img = img.resize(target_size, Image.BILINEAR)\n    img_array = np.array(img, dtype=np.float32) / 255.0  # Normalize to [0, 1]\n    return img_array\n\ndef load_mask(mask_path, target_size=(512, 512)):\n    if mask_path is None or not os.path.exists(mask_path):\n        \n        return np.zeros(target_size[::-1], dtype=np.float32)\n    \n   \n    mask = np.load(mask_path)\n    \n\n    if len(mask.shape) == 3:\n        mask = np.max(mask, axis=0)\n    \n    mask_resized = cv2.resize(mask.astype(np.float32), target_size, interpolation=cv2.INTER_NEAREST)\n    mask_binary = (mask_resized > 0.5).astype(np.float32)\n    \n    return mask_binary\n\ndef load_sample(df, idx, target_size=(512, 512)):\n    row = df.iloc[idx]\n    \n    image = load_image(row['image_path'], target_size)\n    mask = load_mask(row['mask_path'], target_size)\n    label = row['label']\n    \n    return image, mask, label","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"TARGET_SIZE = (512, 512)\n\nprint(f\"\\nTesting data loading with target size: {TARGET_SIZE}\")\nprint(\"Loading 3 samples from training set...\\n\")\n\nforged_indices = train_data[train_data['category'] == 'forged'].index.tolist()\nrandom_indices = random.sample(forged_indices, 5)\n\nfig, axes = plt.subplots(5, 3, figsize=(15, 12))\n\nfor i, forged_idx in enumerate(random_indices):\n    image, mask, label = load_sample(train_data, forged_idx, TARGET_SIZE)\n    \n    # Original image\n    axes[i, 0].imshow(image)\n    axes[i, 0].set_title(f'Image (Label: {label})', fontsize=10)\n    axes[i, 0].axis('off')\n    \n    # Mask\n    axes[i, 1].imshow(mask, cmap='hot')\n    axes[i, 1].set_title('Mask', fontsize=10)\n    axes[i, 1].axis('off')\n    \n    # Overlay\n    overlay = (image * 255).astype(np.uint8)\n    red_overlay = np.zeros_like(overlay)\n    red_overlay[:, :, 0] = (mask * 255).astype(np.uint8)\n    blended = cv2.addWeighted(overlay, 0.6, red_overlay, 0.4, 0)\n    \n    axes[i, 2].imshow(blended)\n    axes[i, 2].set_title('Overlay', fontsize=10)\n    axes[i, 2].axis('off')\n\nplt.tight_layout()\nplt.show()\n\nprint(\"\\nData loading test completed successfully!\")\nprint(f\"Image shape: {image.shape}\")\nprint(f\"Mask shape: {mask.shape}\")\nprint(f\"Image value range: [{image.min():.3f}, {image.max():.3f}]\")\nprint(f\"Mask unique values: {np.unique(mask)}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Data Augumentation","metadata":{}},{"cell_type":"code","source":"print(f\"\\nPyTorch version: {torch.__version__}\")\nprint(f\"CUDA available: {torch.cuda.is_available()}\")\nif torch.cuda.is_available():\n    print(f\"CUDA device: {torch.cuda.get_device_name(0)}\")\nprint(\"\\n\")\n\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nprint(f\"Using device: {device}\\n\")\n\n\nprint(\"=\"*50)\nprint(\"CREATING AUGMENTATION PIPELINES\")\nprint(\"=\"*50)\n\ntrain_transform = A.Compose([\n    A.HorizontalFlip(p=0.5),\n    A.VerticalFlip(p=0.5),\n    A.Rotate(limit=30, p=0.5, border_mode=cv2.BORDER_CONSTANT, value=0),\n    A.ShiftScaleRotate(\n        shift_limit=0.1,\n        scale_limit=0.15,\n        rotate_limit=15,\n        border_mode=cv2.BORDER_CONSTANT,\n        value=0,\n        p=0.5\n    ),\n    \n    A.OneOf([\n        A.OpticalDistortion(distort_limit=0.1, p=1.0),\n        A.GridDistortion(num_steps=5, distort_limit=0.1, p=1.0),\n        A.ElasticTransform(alpha=1, sigma=50, p=1.0),\n    ], p=0.3),\n    \n    A.OneOf([\n        A.RandomBrightnessContrast(\n            brightness_limit=0.2,\n            contrast_limit=0.2,\n            p=1.0\n        ),\n        A.HueSaturationValue(\n            hue_shift_limit=10,\n            sat_shift_limit=20,\n            val_shift_limit=20,\n            p=1.0\n        ),\n        A.RandomGamma(gamma_limit=(80, 120), p=1.0),\n    ], p=0.5),\n    \n    A.OneOf([\n        A.GaussNoise(var_limit=(10.0, 50.0), p=1.0),\n        A.GaussianBlur(blur_limit=(3, 5), p=1.0),\n        A.MotionBlur(blur_limit=5, p=1.0),\n    ], p=0.3),\n    \n    A.CLAHE(clip_limit=2.0, p=0.3),\n    A.RandomShadow(\n        shadow_roi=(0, 0.5, 1, 1),\n        num_shadows_lower=1,\n        num_shadows_upper=2,\n        shadow_dimension=5,\n        p=0.2\n    ),\n    \n    A.Normalize(\n        mean=[0.485, 0.456, 0.406],  \n        std=[0.229, 0.224, 0.225],\n        max_pixel_value=1.0,\n    ),\n    ToTensorV2(),\n])\n\nval_transform = A.Compose([\n    A.Normalize(\n        mean=[0.485, 0.456, 0.406],\n        std=[0.229, 0.224, 0.225],\n        max_pixel_value=1.0,\n    ),\n    ToTensorV2(),\n])","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# PyTorch Dataset Generator","metadata":{}},{"cell_type":"code","source":"class ForgeryDetectionDataset(Dataset):\n    \n    def __init__(self, dataframe, transform=None, target_size=(512, 512)):\n    \n        self.df = dataframe.reset_index(drop=True)\n        self.transform = transform\n        self.target_size = target_size\n        \n    def __len__(self):\n        return len(self.df)\n    \n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        image_path = row['image_path']\n        mask_path = row['mask_path']\n        label = row['label']\n        \n        image = Image.open(image_path).convert('RGB')\n        image = image.resize(self.target_size, Image.BILINEAR)\n        image = np.array(image, dtype=np.float32) / 255.0\n        \n    \n        if mask_path is not None and os.path.exists(mask_path):\n            mask = np.load(mask_path)\n            \n            if len(mask.shape) == 3:\n                mask = np.max(mask, axis=0)\n            \n            mask = cv2.resize(\n                mask.astype(np.float32),\n                self.target_size,\n                interpolation=cv2.INTER_NEAREST\n            )\n            \n        \n            mask = (mask > 0.5).astype(np.float32)\n        else:\n        \n            mask = np.zeros(self.target_size, dtype=np.float32)\n        \n      \n        if self.transform:\n            augmented = self.transform(image=image, mask=mask)\n            image = augmented['image']\n            mask = augmented['mask']\n        else:\n           \n            image = torch.from_numpy(image).permute(2, 0, 1) \n            mask = torch.from_numpy(mask).unsqueeze(0) \n        \n        if len(mask.shape) == 2:\n            mask = mask.unsqueeze(0)\n        \n        return {\n            'image': image,\n            'mask': mask,\n            'label': torch.tensor(label, dtype=torch.float32)\n        }","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"BATCH_SIZE = 6\nNUM_WORKERS = 2\nTARGET_SIZE = (512, 512)\n\ntrain_dataset = ForgeryDetectionDataset(\n    dataframe=train_data,\n    transform=train_transform,\n    target_size=TARGET_SIZE\n)\n\nval_dataset = ForgeryDetectionDataset(\n    dataframe=val_data,\n    transform=val_transform,\n    target_size=TARGET_SIZE\n)\n\ntrain_loader = DataLoader(\n    train_dataset,\n    batch_size=BATCH_SIZE,\n    shuffle=True,\n    num_workers=NUM_WORKERS,\n    pin_memory=True if torch.cuda.is_available() else False,\n    drop_last=True\n)\n\nval_loader = DataLoader(\n    val_dataset,\n    batch_size=BATCH_SIZE,\n    shuffle=False,\n    num_workers=NUM_WORKERS,\n    pin_memory=True if torch.cuda.is_available() else False\n)\n\nprint(f\"✓ Dataloaders created\")\nprint(f\"  - Batch size: {BATCH_SIZE}\")\nprint(f\"  - Training batches: {len(train_loader)}\")\nprint(f\"  - Validation batches: {len(val_loader)}\")\nprint(f\"  - Training samples: {len(train_dataset)}\")\nprint(f\"  - Validation samples: {len(val_dataset)}\")\nprint(\"\\n\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Validate DataLoader","metadata":{}},{"cell_type":"code","source":"sample_batch = next(iter(train_loader))\n\nprint(f\"Sample batch shapes:\")\nprint(f\"  - Images: {sample_batch['image'].shape}\")\nprint(f\"  - Masks: {sample_batch['mask'].shape}\")\nprint(f\"  - Labels: {sample_batch['label'].shape}\")\n\nprint(f\"\\nImage tensor stats:\")\nprint(f\"  - Min: {sample_batch['image'].min():.3f}\")\nprint(f\"  - Max: {sample_batch['image'].max():.3f}\")\nprint(f\"  - Mean: {sample_batch['image'].mean():.3f}\")\n\nprint(f\"\\nMask tensor stats:\")\nprint(f\"  - Unique values: {torch.unique(sample_batch['mask'])}\")\nprint(f\"  - Forged pixels ratio: {sample_batch['mask'].mean():.4f}\")\n\n\ndef denormalize(image):\n    mean = torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1)\n    std = torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1)\n    image = image * std + mean\n    return torch.clamp(image, 0, 1)\n\nfig, axes = plt.subplots(3, 3, figsize=(15, 15))\nfig.suptitle('Augmented Training Samples', fontsize=16, fontweight='bold')\n\nfor i in range(3):\n    # Get sample\n    img = denormalize(sample_batch['image'][i]).permute(1, 2, 0).cpu().numpy()\n    mask = sample_batch['mask'][i, 0].cpu().numpy()\n    label = sample_batch['label'][i].item()\n    \n    # Plot\n    axes[i, 0].imshow(img)\n    axes[i, 0].set_title(f'Augmented Image (Label: {label:.0f})')\n    axes[i, 0].axis('off')\n    \n    axes[i, 1].imshow(mask, cmap='hot')\n    axes[i, 1].set_title('Augmented Mask')\n    axes[i, 1].axis('off')\n    \n    # Overlay\n    overlay = (img * 255).astype(np.uint8)\n    red_overlay = np.zeros_like(overlay)\n    red_overlay[:,:,0] = (mask * 255).astype(np.uint8)\n    blended = cv2.addWeighted(overlay, 0.6, red_overlay, 0.4, 0)\n    \n    axes[i, 2].imshow(blended)\n    axes[i, 2].set_title('Overlay')\n    axes[i, 2].axis('off')\n\nplt.tight_layout()\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Model Aricturaccture","metadata":{}},{"cell_type":"code","source":"class SegformerForgeryDetector(nn.Module):\n    def __init__(\n        self,\n        encoder_name='mit_b2',  # MiT Transformer backbone\n        encoder_weights='imagenet',  # pretrained weights\n        in_channels=3,\n        classes=1,\n        activation=None\n    ):\n        super(SegformerForgeryDetector, self).__init__()\n        \n        self.model = smp.Segformer(\n            encoder_name=encoder_name,\n            encoder_weights=encoder_weights,\n            in_channels=in_channels,\n            classes=classes,\n            activation=activation\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\nmodel = SegformerForgeryDetector(\n    encoder_name='mit_b2',\n    encoder_weights='imagenet',\n    in_channels=3,\n    classes=1,\n    activation=None\n).to(device)\n\nmodel = model.to(device)\n\nprint(f\"✓ SegFormer model created\")\nprint(f\"  - Encoder: SegFormer \")\nprint(f\"  - Pre-trained weights: MiT-B2\")\nprint(f\"  - Input channels: 3 (RGB)\")\nprint(f\"  - Output classes: 1 (binary segmentation)\")\nprint(f\"  - Device: {device}\")\nprint(\"\\n\")\n\n\ndef count_parameters(model):\n    total = sum(p.numel() for p in model.parameters())\n    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)\n    return total, trainable\n\ntotal_params, trainable_params = count_parameters(model)\n\nprint(f\"Model Parameters:\")\nprint(f\"  - Total: {total_params:,}\")\nprint(f\"  - Trainable: {trainable_params:,}\")\nprint(f\"  - Size: ~{total_params * 4 / (1024**2):.2f} MB (FP32)\")\n\n\nprint(\"\\nTesting forward pass...\")\nwith torch.no_grad():\n    dummy_input = torch.randn(2, 3, 512, 512).to(device)\n    dummy_output = model(dummy_input)\n    print(f\"  - Input shape: {dummy_input.shape}\")\n    print(f\"  - Output shape: {dummy_output.shape}\")\n    print(\"✓ Forward pass successful!\")\nprint(\"\\n\")\n\n\n\nprint(\"=\"*60)\nprint(\" DEFINING LOSS FUNCTIONS\")\nprint(\"=\"*60)\n\nclass DiceLoss(nn.Module):  \n    def __init__(self, smooth=1.0):\n        super(DiceLoss, self).__init__()\n        self.smooth = smooth\n    \n    def forward(self, predictions, targets):\n        predictions = torch.sigmoid(predictions)\n        \n        predictions = predictions.view(-1)\n        targets = targets.view(-1)\n        \n        intersection = (predictions * targets).sum()\n        dice = (2. * intersection + self.smooth) / (\n            predictions.sum() + targets.sum() + self.smooth\n        )\n        \n        return 1 - dice\n\nclass CombinedLoss(nn.Module):\n    \n    def __init__(self, bce_weight=0.5, dice_weight=0.5):\n        super(CombinedLoss, self).__init__()\n        self.bce_weight = bce_weight\n        self.dice_weight = dice_weight\n        self.bce = nn.BCEWithLogitsLoss()\n        self.dice = DiceLoss()\n    \n    def forward(self, predictions, targets):\n        bce_loss = self.bce(predictions, targets)\n        dice_loss = self.dice(predictions, targets)\n        \n        combined = self.bce_weight * bce_loss + self.dice_weight * dice_loss\n        \n        return combined, bce_loss, dice_loss\n\ncriterion = CombinedLoss(bce_weight=0.5, dice_weight=0.5)\n\nprint(\"✓ Loss functions created\")\nprint(\"  - BCE Loss: Binary Cross-Entropy with Logits\")\nprint(\"  - Dice Loss: Custom implementation\")\nprint(\"  - Combined Loss: 0.5 * BCE + 0.5 * Dice\")\nprint(\"\\n\")\n\n\nprint(\"=\"*60)\nprint(\" CONFIGURING OPTIMIZER AND SCHEDULER\")\nprint(\"=\"*60)\n\nLEARNING_RATE = 3e-4\nWEIGHT_DECAY = 1e-5\n\noptimizer = torch.optim.AdamW(\n    model.parameters(),\n    lr=LEARNING_RATE,\n    weight_decay=WEIGHT_DECAY\n)\n\n\nscheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(\n    optimizer,\n    mode='min',\n    factor=0.5,\n    patience=5,\n    verbose=True,\n    min_lr=1e-7\n)\n\nprint(f\"✓ Optimizer: AdamW\")\nprint(f\"  - Learning rate: {LEARNING_RATE}\")\nprint(f\"  - Weight decay: {WEIGHT_DECAY}\")\n\nprint(f\"\\n✓ Scheduler: ReduceLROnPlateau\")\nprint(f\"  - Mode: min (reduce on plateau)\")\nprint(f\"  - Factor: 0.5 (halve LR)\")\nprint(f\"  - Patience: 5 epochs\")\nprint(f\"  - Min LR: 1e-7\")\nprint(\"\\n\")\n\nprint(\"=\"*60)\nprint(\"DEFINING EVALUATION METRICS\")\nprint(\"=\"*60)\n\ndef dice_coefficient(predictions, targets, threshold=0.5, smooth=1.0):\n\n    predictions = (torch.sigmoid(predictions) > threshold).float()\n    predictions = predictions.view(-1)\n    targets = targets.view(-1)\n    \n    intersection = (predictions * targets).sum()\n    dice = (2. * intersection + smooth) / (\n        predictions.sum() + targets.sum() + smooth\n    )\n    \n    return dice.item()\n\ndef iou_score(predictions, targets, threshold=0.5, smooth=1.0):\n    predictions = (torch.sigmoid(predictions) > threshold).float()\n    predictions = predictions.view(-1)\n    targets = targets.view(-1)\n    \n    intersection = (predictions * targets).sum()\n    union = predictions.sum() + targets.sum() - intersection\n    \n    iou = (intersection + smooth) / (union + smooth)\n    \n    return iou.item()\n\ndef pixel_accuracy(predictions, targets, threshold=0.5):\n   \n    predictions = (torch.sigmoid(predictions) > threshold).float()\n    correct = (predictions == targets).float().sum()\n    total = targets.numel()\n    \n    return (correct / total).item()\n\nprint(\"✓ Evaluation metrics defined\")\nprint(\"  - Dice Coefficient\")\nprint(\"  - IoU (Intersection over Union)\")\nprint(\"  - Pixel Accuracy\")\nprint(\"\\n\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Configuration","metadata":{}},{"cell_type":"code","source":"CONFIG = {\n    'epochs': 10,\n    'early_stopping_patience': 7,\n    'best_model_metric': 'val_dice',  \n    'checkpoint_dir': 'checkpoints',\n    'save_best_only': True,\n}\n\nprint(f\"\"\"\nTraining Configuration:\n  - Epochs: {CONFIG['epochs']}\n  - Early stopping patience: {CONFIG['early_stopping_patience']}\n  - Best model metric: {CONFIG['best_model_metric']}\n  - Checkpoint directory: {CONFIG['checkpoint_dir']}\n  - Save best only: {CONFIG['save_best_only']}\n\"\"\")\n\nos.makedirs(CONFIG['checkpoint_dir'], exist_ok=True)\n\n\nclass AverageMeter:\n\n    def __init__(self):\n        self.reset()\n    \n    def reset(self):\n        self.val = 0\n        self.avg = 0\n        self.sum = 0\n        self.count = 0\n    \n    def update(self, val, n=1):\n        self.val = val\n        self.sum += val * n\n        self.count += n\n        self.avg = self.sum / self.count\n\nclass EarlyStopping:\n    def __init__(self, patience=7, mode='min', delta=0.0001):\n        self.patience = patience\n        self.mode = mode\n        self.delta = delta\n        self.counter = 0\n        self.best_score = None\n        self.early_stop = False\n        \n    def __call__(self, score):\n        if self.best_score is None:\n            self.best_score = score\n            return False\n        \n        if self.mode == 'min':\n            improved = score < (self.best_score - self.delta)\n        else:\n            improved = score > (self.best_score + self.delta)\n        \n        if improved:\n            self.best_score = score\n            self.counter = 0\n        else:\n            self.counter += 1\n            if self.counter >= self.patience:\n                self.early_stop = True\n        \n        return self.early_stop\n\nprint(\"✓ Utility classes created:\")\nprint(\"  - AverageMeter: Track running averages\")\nprint(\"  - EarlyStopping: Stop training early if no improvement\")\nprint(\"\\n\")\n\n\ndef train_one_epoch(model, dataloader, criterion, optimizer, device, epoch):\n    model.train()\n    \n    loss_meter = AverageMeter()\n    bce_meter = AverageMeter()\n    dice_loss_meter = AverageMeter()\n    dice_score_meter = AverageMeter()\n    iou_meter = AverageMeter()\n    \n    pbar = tqdm(dataloader, desc=f'Epoch {epoch} [TRAIN]')\n    \n    for batch_idx, batch in enumerate(pbar):\n        \n        images = batch['image'].to(device)\n        masks = batch['mask'].to(device)\n        \n        outputs = model(images)\n        \n        loss, bce_loss, dice_loss = criterion(outputs, masks)\n        \n        \n        optimizer.zero_grad()\n        loss.backward()\n        optimizer.step()\n        \n        with torch.no_grad():\n            dice_score = dice_coefficient(outputs, masks)\n            iou = iou_score(outputs, masks)\n        \n        batch_size = images.size(0)\n        loss_meter.update(loss.item(), batch_size)\n        bce_meter.update(bce_loss.item(), batch_size)\n        dice_loss_meter.update(dice_loss.item(), batch_size)\n        dice_score_meter.update(dice_score, batch_size)\n        iou_meter.update(iou, batch_size)\n        \n        pbar.set_postfix({\n            'loss': f'{loss_meter.avg:.4f}',\n            'dice': f'{dice_score_meter.avg:.4f}',\n            'iou': f'{iou_meter.avg:.4f}'\n        })\n    \n    return {\n        'loss': loss_meter.avg,\n        'bce_loss': bce_meter.avg,\n        'dice_loss': dice_loss_meter.avg,\n        'dice_score': dice_score_meter.avg,\n        'iou': iou_meter.avg\n    }\n\ndef validate(model, dataloader, criterion, device, epoch):\n    model.eval()\n    \n    loss_meter = AverageMeter()\n    bce_meter = AverageMeter()\n    dice_loss_meter = AverageMeter()\n    dice_score_meter = AverageMeter()\n    iou_meter = AverageMeter()\n    \n    pbar = tqdm(dataloader, desc=f'Epoch {epoch} [VALID]')\n    \n    with torch.no_grad():\n        for batch_idx, batch in enumerate(pbar):\n            \n            images = batch['image'].to(device)\n            masks = batch['mask'].to(device)\n            \n            outputs = model(images)\n            \n            loss, bce_loss, dice_loss = criterion(outputs, masks)\n            \n            dice_score = dice_coefficient(outputs, masks)\n            iou = iou_score(outputs, masks)\n            \n            batch_size = images.size(0)\n            loss_meter.update(loss.item(), batch_size)\n            bce_meter.update(bce_loss.item(), batch_size)\n            dice_loss_meter.update(dice_loss.item(), batch_size)\n            dice_score_meter.update(dice_score, batch_size)\n            iou_meter.update(iou, batch_size)\n            \n            pbar.set_postfix({\n                'loss': f'{loss_meter.avg:.4f}',\n                'dice': f'{dice_score_meter.avg:.4f}',\n                'iou': f'{iou_meter.avg:.4f}'\n            })\n    \n    return {\n        'loss': loss_meter.avg,\n        'bce_loss': bce_meter.avg,\n        'dice_loss': dice_loss_meter.avg,\n        'dice_score': dice_score_meter.avg,\n        'iou': iou_meter.avg\n    }","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Model Training Pipeline","metadata":{}},{"cell_type":"code","source":"history = defaultdict(list)\nbest_dice = 0.0\nbest_model_wts = copy.deepcopy(model.state_dict())\n\nearly_stopping = EarlyStopping(\n    patience=CONFIG['early_stopping_patience'],\n    mode='max',  \n    delta=0.001\n)\n\ntraining_start_time = time.time()\n\nprint(f\"Starting training for {CONFIG['epochs']} epochs...\")\nprint(f\"Training samples: {len(train_dataset)}\")\nprint(f\"Validation samples: {len(val_dataset)}\")\nprint(f\"Device: {device}\")\nprint(\"=\"*80)\nprint(\"\\n\")\n\n\nfor epoch in range(1, CONFIG['epochs'] + 1):\n    epoch_start_time = time.time()\n    \n    print(f\"\\n{'='*80}\")\n    print(f\"EPOCH {epoch}/{CONFIG['epochs']}\")\n    print(f\"{'='*80}\")\n    \n    train_metrics = train_one_epoch(\n        model=model,\n        dataloader=train_loader,\n        criterion=criterion,\n        optimizer=optimizer,\n        device=device,\n        epoch=epoch\n    )\n    \n    val_metrics = validate(\n        model=model,\n        dataloader=val_loader,\n        criterion=criterion,\n        device=device,\n        epoch=epoch\n    )\n    \n    scheduler.step(val_metrics['loss'])\n    \n    current_lr = optimizer.param_groups[0]['lr']\n    \n    epoch_time = time.time() - epoch_start_time\n    \n    print(f\"\\n{'='*80}\")\n    print(f\"EPOCH {epoch} SUMMARY\")\n    print(f\"{'='*80}\")\n    print(f\"Time: {epoch_time:.2f}s | LR: {current_lr:.2e}\")\n    print(f\"\\nTrain Metrics:\")\n    print(f\"  Loss: {train_metrics['loss']:.4f} | BCE: {train_metrics['bce_loss']:.4f} | Dice Loss: {train_metrics['dice_loss']:.4f}\")\n    print(f\"  Dice Score: {train_metrics['dice_score']:.4f} | IoU: {train_metrics['iou']:.4f}\")\n    print(f\"\\nValidation Metrics:\")\n    print(f\"  Loss: {val_metrics['loss']:.4f} | BCE: {val_metrics['bce_loss']:.4f} | Dice Loss: {val_metrics['dice_loss']:.4f}\")\n    print(f\"  Dice Score: {val_metrics['dice_score']:.4f} | IoU: {val_metrics['iou']:.4f}\")\n    \n    history['epoch'].append(epoch)\n    history['lr'].append(current_lr)\n    history['train_loss'].append(train_metrics['loss'])\n    history['train_dice'].append(train_metrics['dice_score'])\n    history['train_iou'].append(train_metrics['iou'])\n    history['val_loss'].append(val_metrics['loss'])\n    history['val_dice'].append(val_metrics['dice_score'])\n    history['val_iou'].append(val_metrics['iou'])\n    \n    if val_metrics['dice_score'] > best_dice:\n        best_dice = val_metrics['dice_score']\n        best_model_wts = copy.deepcopy(model.state_dict())\n        \n        checkpoint_path = os.path.join(\n            CONFIG['checkpoint_dir'],\n            f'best_model_epoch{epoch}_dice{best_dice:.4f}.pth'\n        )\n        torch.save({\n            'epoch': epoch,\n            'model_state_dict': model.state_dict(),\n            'optimizer_state_dict': optimizer.state_dict(),\n            'scheduler_state_dict': scheduler.state_dict(),\n            'best_dice': best_dice,\n            'val_metrics': val_metrics,\n        }, checkpoint_path)\n        \n        print(f\"\\n🎯 New best model saved! Dice: {best_dice:.4f}\")\n        print(f\"   Saved to: {checkpoint_path}\")\n    \n    if early_stopping(val_metrics['dice_score']):\n        print(f\"\\n  Early stopping triggered after {epoch} epochs\")\n        print(f\"   No improvement for {CONFIG['early_stopping_patience']} epochs\")\n        break\n    \n    print(f\"{'='*80}\\n\")\n\n\n\ntraining_time = time.time() - training_start_time\n\nprint(\"\\n\")\nprint(\"=\"*80)\nprint(\"TRAINING COMPLETED!\")\nprint(\"=\"*80)\nprint(f\"Total training time: {training_time/60:.2f} minutes\")\nprint(f\"Best validation Dice score: {best_dice:.4f}\")\nprint(f\"Total epochs trained: {len(history['epoch'])}\")\nprint(\"=\"*80)\nprint(\"\\n\")\n\n\nmodel.load_state_dict(best_model_wts)\nprint(\"✓ Best model weights loaded into model\")\nprint(\"\\n\")\n\n\nhistory_df = pd.DataFrame(history)\n\nfig, axes = plt.subplots(2, 3, figsize=(20, 12))\nfig.suptitle('Training History', fontsize=16, fontweight='bold')\n\naxes[0, 0].plot(history_df['epoch'], history_df['train_loss'], \n                label='Train Loss', marker='o', linewidth=2)\naxes[0, 0].plot(history_df['epoch'], history_df['val_loss'], \n                label='Val Loss', marker='s', linewidth=2)\naxes[0, 0].set_xlabel('Epoch')\naxes[0, 0].set_ylabel('Loss')\naxes[0, 0].set_title('Loss Curve')\naxes[0, 0].legend()\naxes[0, 0].grid(True, alpha=0.3)\n\naxes[0, 1].plot(history_df['epoch'], history_df['train_dice'], \n                label='Train Dice', marker='o', linewidth=2, color='green')\naxes[0, 1].plot(history_df['epoch'], history_df['val_dice'], \n                label='Val Dice', marker='s', linewidth=2, color='orange')\naxes[0, 1].set_xlabel('Epoch')\naxes[0, 1].set_ylabel('Dice Score')\naxes[0, 1].set_title('Dice Score Curve')\naxes[0, 1].legend()\naxes[0, 1].grid(True, alpha=0.3)\n\naxes[0, 2].plot(history_df['epoch'], history_df['train_iou'], \n                label='Train IoU', marker='o', linewidth=2, color='purple')\naxes[0, 2].plot(history_df['epoch'], history_df['val_iou'], \n                label='Val IoU', marker='s', linewidth=2, color='brown')\naxes[0, 2].set_xlabel('Epoch')\naxes[0, 2].set_ylabel('IoU Score')\naxes[0, 2].set_title('IoU Score Curve')\naxes[0, 2].legend()\naxes[0, 2].grid(True, alpha=0.3)\n\naxes[1, 0].plot(history_df['epoch'], history_df['lr'], \n                marker='o', linewidth=2, color='red')\naxes[1, 0].set_xlabel('Epoch')\naxes[1, 0].set_ylabel('Learning Rate')\naxes[1, 0].set_title('Learning Rate Schedule')\naxes[1, 0].set_yscale('log')\naxes[1, 0].grid(True, alpha=0.3)\n\n\naxes[1, 1].scatter(history_df['train_loss'], history_df['val_loss'], \n                   c=history_df['epoch'], cmap='viridis', s=100, alpha=0.6)\naxes[1, 1].plot([history_df['train_loss'].min(), history_df['train_loss'].max()],\n                [history_df['train_loss'].min(), history_df['train_loss'].max()],\n                'r--', linewidth=2, label='Perfect fit')\naxes[1, 1].set_xlabel('Train Loss')\naxes[1, 1].set_ylabel('Val Loss')\naxes[1, 1].set_title('Train vs Val Loss')\naxes[1, 1].legend()\naxes[1, 1].grid(True, alpha=0.3)\n\naxes[1, 2].scatter(history_df['train_dice'], history_df['val_dice'], \n                   c=history_df['epoch'], cmap='viridis', s=100, alpha=0.6)\naxes[1, 2].plot([history_df['train_dice'].min(), history_df['train_dice'].max()],\n                [history_df['train_dice'].min(), history_df['train_dice'].max()],\n                'r--', linewidth=2, label='Perfect fit')\naxes[1, 2].set_xlabel('Train Dice')\naxes[1, 2].set_ylabel('Val Dice')\naxes[1, 2].set_title('Train vs Val Dice')\naxes[1, 2].legend()\naxes[1, 2].grid(True, alpha=0.3)\n\nplt.tight_layout()\nplt.show()\n\nprint(\"✓ Training curves plotted\")\nprint(\"\\n\")\n\n\n\nbest_epoch_idx = history_df['val_dice'].idxmax()\nbest_epoch_data = history_df.iloc[best_epoch_idx]\n\nprint(f\"\\nBest Epoch: {int(best_epoch_data['epoch'])}\")\nprint(f\"\\nMetrics at Best Epoch:\")\nprint(f\"  Train Loss: {best_epoch_data['train_loss']:.4f}\")\nprint(f\"  Val Loss: {best_epoch_data['val_loss']:.4f}\")\nprint(f\"  Train Dice: {best_epoch_data['train_dice']:.4f}\")\nprint(f\"  Val Dice: {best_epoch_data['val_dice']:.4f}\")\nprint(f\"  Train IoU: {best_epoch_data['train_iou']:.4f}\")\nprint(f\"  Val IoU: {best_epoch_data['val_iou']:.4f}\")\nprint(f\"  Learning Rate: {best_epoch_data['lr']:.2e}\")\n\n# Check for overfitting\ndice_gap = best_epoch_data['train_dice'] - best_epoch_data['val_dice']\nloss_gap = best_epoch_data['val_loss'] - best_epoch_data['train_loss']\n\nprint(f\"\\nOverfitting Analysis:\")\nprint(f\"  Dice gap (Train - Val): {dice_gap:.4f}\")\nprint(f\"  Loss gap (Val - Train): {loss_gap:.4f}\")\n\nif dice_gap > 0.1:\n    print(\"  ⚠️  Warning: Significant overfitting detected (Dice gap > 0.1)\")\nelif dice_gap > 0.05:\n    print(\"  ⚡ Moderate overfitting (Dice gap > 0.05)\")\nelse:\n    print(\"  ✓ Good generalization (Dice gap < 0.05)\")\n\nprint(\"\\n\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"---","metadata":{}},{"cell_type":"code","source":"def predict_batch(model, images, device, threshold=0.5):\n    model.eval()\n    with torch.no_grad():\n        outputs = model(images.to(device))\n        predictions = torch.sigmoid(outputs)\n        binary_predictions = (predictions > threshold).float()\n    return predictions, binary_predictions\n\ndef visualize_predictions(model, dataloader, device, n_samples=6, threshold=0.5):\n  \n    model.eval()\n    \n    fig, axes = plt.subplots(n_samples, 4, figsize=(20, 5*n_samples))\n    fig.suptitle(f'Validation Predictions (Threshold: {threshold})', \n                 fontsize=16, fontweight='bold')\n    \n    batch = next(iter(dataloader))\n    images = batch['image'][:n_samples].to(device)\n    masks = batch['mask'][:n_samples]\n    labels = batch['label'][:n_samples]\n    \n    with torch.no_grad():\n        outputs = model(images)\n        predictions = torch.sigmoid(outputs)\n        binary_preds = (predictions > threshold).float()\n    \n    def denormalize(img):\n        mean = torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1).to(img.device)\n        std = torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1).to(img.device)\n        return torch.clamp(img * std + mean, 0, 1)\n    \n    for i in range(n_samples):\n        img = denormalize(images[i]).cpu().permute(1, 2, 0).numpy()\n        axes[i, 0].imshow(img)\n        axes[i, 0].set_title(f'Image (Label: {labels[i].item():.0f})')\n        axes[i, 0].axis('off')\n        \n        gt_mask = masks[i, 0].cpu().numpy()\n        axes[i, 1].imshow(gt_mask, cmap='hot', vmin=0, vmax=1)\n        axes[i, 1].set_title('Ground Truth Mask')\n        axes[i, 1].axis('off')\n        \n        pred_prob = predictions[i, 0].cpu().numpy()\n        axes[i, 2].imshow(pred_prob, cmap='hot', vmin=0, vmax=1)\n        axes[i, 2].set_title(f'Predicted (Prob)\\nMax: {pred_prob.max():.3f}')\n        axes[i, 2].axis('off')\n        \n        pred_binary = binary_preds[i, 0].cpu().numpy()\n        axes[i, 3].imshow(pred_binary, cmap='hot', vmin=0, vmax=1)\n        \n        # Calculate metrics for this sample\n        dice = dice_coefficient(\n            outputs[i:i+1], \n            masks[i:i+1].to(device), \n            threshold=threshold\n        )\n        axes[i, 3].set_title(f'Binary Prediction\\nDice: {dice:.3f}')\n        axes[i, 3].axis('off')\n    \n    plt.tight_layout()\n    plt.show()\n\n# Visualize predictions\nprint(\"\\nVisualizing predictions on validation set...\")\nvisualize_predictions(\n    model=model,\n    dataloader=val_loader,\n    device=device,\n    n_samples=6,\n    threshold=0.5\n)\n\nprint(\"✓ Validation predictions visualized\")\nprint(\"\\n\")\n\n\nprint(\"=\"*80)\nprint(\"STEP 7.1: OPTIMIZING PREDICTION THRESHOLD\")\nprint(\"=\"*80)\n\ndef evaluate_threshold(model, dataloader, device, threshold):\n    model.eval()\n    dice_scores = []\n    iou_scores = []\n    \n    with torch.no_grad():\n        for batch in dataloader:\n            images = batch['image'].to(device)\n            masks = batch['mask'].to(device)\n            \n            outputs = model(images)\n            \n            dice = dice_coefficient(outputs, masks, threshold=threshold)\n            iou = iou_score(outputs, masks, threshold=threshold)\n            \n            dice_scores.append(dice)\n            iou_scores.append(iou)\n    \n    return np.mean(dice_scores), np.mean(iou_scores)\n\nprint(\"Testing different thresholds...\")\nthresholds = [0.3,  0.7]\nthreshold_results = []\n\nfor thresh in tqdm(thresholds, desc=\"Testing thresholds\"):\n    dice, iou = evaluate_threshold(model, val_loader, device, thresh)\n    threshold_results.append({\n        'threshold': thresh,\n        'dice': dice,\n        'iou': iou\n    })\n    print(f\"  Threshold {thresh:.2f}: Dice={dice:.4f}, IoU={iou:.4f}\")\n\nthreshold_df = pd.DataFrame(threshold_results)\nbest_threshold_idx = threshold_df['dice'].idxmax()\nbest_threshold = threshold_df.iloc[best_threshold_idx]['threshold']\nbest_dice = threshold_df.iloc[best_threshold_idx]['dice']\n\nprint(f\"\\n✓ Best threshold: {best_threshold}\")\nprint(f\"  Dice score: {best_dice:.4f}\")\n\nfig, axes = plt.subplots(1, 2, figsize=(15, 5))\nfig.suptitle('Threshold Optimization', fontsize=16, fontweight='bold')\n\naxes[0].plot(threshold_df['threshold'], threshold_df['dice'], \n             marker='o', linewidth=2, markersize=8, color='green')\naxes[0].axvline(best_threshold, color='red', linestyle='--', \n                label=f'Best: {best_threshold}')\naxes[0].set_xlabel('Threshold')\naxes[0].set_ylabel('Dice Score')\naxes[0].set_title('Dice Score vs Threshold')\naxes[0].legend()\naxes[0].grid(True, alpha=0.3)\n\naxes[1].plot(threshold_df['threshold'], threshold_df['iou'], \n             marker='s', linewidth=2, markersize=8, color='purple')\naxes[1].axvline(best_threshold, color='red', linestyle='--', \n                label=f'Best: {best_threshold}')\naxes[1].set_xlabel('Threshold')\naxes[1].set_ylabel('IoU Score')\naxes[1].set_title('IoU Score vs Threshold')\naxes[1].legend()\naxes[1].grid(True, alpha=0.3)\n\nplt.tight_layout()\nplt.show()\n\nprint(\"\\n\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}