{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","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":31193,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Scientific Image Forgery Detection - Copy-Move Forgery Detection\n\nThis notebook implements EDA and model training for detecting copy-move forgeries in biomedical images.\n","metadata":{}},{"cell_type":"code","source":"# Install required packages\n# pip install torch torchvision torchaudio\n# pip install numpy pandas matplotlib seaborn opencv-python tqdm jupyter ipykernel\n# pip install pillow scikit-image","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:24.871667Z","iopub.execute_input":"2025-11-10T10:06:24.871938Z","iopub.status.idle":"2025-11-10T10:06:24.875490Z","shell.execute_reply.started":"2025-11-10T10:06:24.871917Z","shell.execute_reply":"2025-11-10T10:06:24.874673Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### Required Packages:\n- `torch` - PyTorch deep learning framework\n- `torchvision` - Computer vision utilities for PyTorch\n- `numpy` - Numerical computing\n- `pandas` - Data manipulation and analysis\n- `matplotlib` - Plotting and visualization\n- `seaborn` - Statistical data visualization\n- `opencv-python` - Computer vision library\n- `tqdm` - Progress bars\n- `jupyter` - Jupyter notebook environment\n- `ipykernel` - IPython kernel for Jupyter\n- `Pillow` - Image processing\n- `scikit-image` - Image processing algorithms\n\n**Note:** For GPU support, install PyTorch with CUDA. Visit [pytorch.org](https://pytorch.org/get-started/locally/) for the correct installation command for your system.\n","metadata":{}},{"cell_type":"code","source":"# Import necessary libraries\nimport os\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom pathlib import Path\nimport cv2\nfrom tqdm import tqdm\nimport warnings\nwarnings.filterwarnings('ignore')\n\n# Deep learning libraries\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as transforms\nfrom torchvision.transforms import functional as F\n\n# Set random seeds for reproducibility\nnp.random.seed(42)\ntorch.manual_seed(42)\nif torch.cuda.is_available():\n    torch.cuda.manual_seed(42)\n\n# Check for GPU\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nprint(f\"Using device: {device}\")\nif torch.cuda.is_available():\n    print(f\"GPU: {torch.cuda.get_device_name(0)}\")\n    print(f\"CUDA Version: {torch.version.cuda}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:25.464281Z","iopub.execute_input":"2025-11-10T10:06:25.465005Z","iopub.status.idle":"2025-11-10T10:06:25.472941Z","shell.execute_reply.started":"2025-11-10T10:06:25.464982Z","shell.execute_reply":"2025-11-10T10:06:25.472221Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 1. Exploratory Data Analysis (EDA)\n","metadata":{}},{"cell_type":"code","source":"# Define data paths for Kaggle\n# Input data is in /kaggle/input/ - adjust the dataset name as needed\n# Example: if your dataset is named 'copy-move-forgery-detection', use:\n# input_dir = Path('/kaggle/input/copy-move-forgery-detection')\n# Or if data is directly in input, use:\ninput_dir = Path('/kaggle/input/recodai-luc-scientific-image-forgery-detection')\nbase_dir = input_dir  # Change this to your specific dataset path if needed\n\ntrain_authentic_dir = base_dir / 'train_images' / 'authentic'\ntrain_forged_dir = base_dir / 'train_images' / 'forged'\ntrain_masks_dir = base_dir / 'train_masks'\ntest_images_dir = base_dir / 'test_images'\n\n# Working directory for outputs (models, submissions, etc.)\nworking_dir = Path('/kaggle/working')\nmodel_path = working_dir / 'best_model.pth'\nsubmission_path = working_dir / 'submission.csv'\n\n# Count files\ndef count_files(directory, extension='.png'):\n    if not directory.exists():\n        return 0\n    return len(list(directory.glob(f'*{extension}')))\n\nauthentic_count = count_files(train_authentic_dir)\nforged_count = count_files(train_forged_dir)\nmask_count = count_files(train_masks_dir, '.npy')\ntest_count = count_files(test_images_dir)\n\nprint(f\"\\nFile counts:\")\nprint(f\"Authentic images: {authentic_count}\")\nprint(f\"Forged images: {forged_count}\")\nprint(f\"Mask files: {mask_count}\")\nprint(f\"Test images: {test_count}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:26.024862Z","iopub.execute_input":"2025-11-10T10:06:26.025543Z","iopub.status.idle":"2025-11-10T10:06:26.049839Z","shell.execute_reply.started":"2025-11-10T10:06:26.025518Z","shell.execute_reply":"2025-11-10T10:06:26.049296Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Load sample images and analyze\ndef load_sample_images(authentic_dir, forged_dir, masks_dir, n_samples=5):\n    \"\"\"Load sample images for analysis\"\"\"\n    authentic_files = list(authentic_dir.glob('*.png'))[:n_samples]\n    forged_files = list(forged_dir.glob('*.png'))[:n_samples]\n    \n    authentic_images = []\n    forged_images = []\n    masks = []\n    \n    for img_path in authentic_files:\n        img = cv2.imread(str(img_path))\n        if img is not None:\n            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n            authentic_images.append((img_path.name, img))\n    \n    for img_path in forged_files:\n        img = cv2.imread(str(img_path))\n        if img is not None:\n            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n            forged_images.append((img_path.name, img))\n            \n            # Try to load corresponding mask\n            mask_path = masks_dir / img_path.name.replace('.png', '.npy')\n            if mask_path.exists():\n                mask = np.load(mask_path)\n                masks.append((img_path.name, mask))\n    \n    return authentic_images, forged_images, masks\n\nif train_authentic_dir.exists() and train_forged_dir.exists():\n    authentic_samples, forged_samples, mask_samples = load_sample_images(\n        train_authentic_dir, train_forged_dir, train_masks_dir, n_samples=3\n    )\n    print(f\"Loaded {len(authentic_samples)} authentic samples\")\n    print(f\"Loaded {len(forged_samples)} forged samples\")\n    print(f\"Loaded {len(mask_samples)} mask samples\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:26.305116Z","iopub.execute_input":"2025-11-10T10:06:26.305370Z","iopub.status.idle":"2025-11-10T10:06:26.394592Z","shell.execute_reply.started":"2025-11-10T10:06:26.305350Z","shell.execute_reply":"2025-11-10T10:06:26.393962Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Analyze image statistics\ndef analyze_image_statistics(image_dir, label=''):\n    \"\"\"Analyze statistics of images in a directory\"\"\"\n    image_files = list(image_dir.glob('*.png'))\n    if len(image_files) == 0:\n        return None\n    \n    heights = []\n    widths = []\n    channels = []\n    \n    for img_path in tqdm(image_files[:100], desc=f\"Analyzing {label}\"):  # Sample first 100\n        img = cv2.imread(str(img_path))\n        if img is not None:\n            h, w, c = img.shape\n            heights.append(h)\n            widths.append(w)\n            channels.append(c)\n    \n    if len(heights) == 0:\n        return None\n    \n    stats = {\n        'label': label,\n        'count': len(image_files),\n        'height_mean': np.mean(heights),\n        'height_std': np.std(heights),\n        'width_mean': np.mean(widths),\n        'width_std': np.std(widths),\n        'height_min': np.min(heights),\n        'height_max': np.max(heights),\n        'width_min': np.min(widths),\n        'width_max': np.max(widths),\n    }\n    return stats\n\n# Analyze authentic and forged images\nif train_authentic_dir.exists():\n    authentic_stats = analyze_image_statistics(train_authentic_dir, 'Authentic')\n    if authentic_stats:\n        print(\"\\nAuthentic Images Statistics:\")\n        for key, value in authentic_stats.items():\n            print(f\"  {key}: {value}\")\n\nif train_forged_dir.exists():\n    forged_stats = analyze_image_statistics(train_forged_dir, 'Forged')\n    if forged_stats:\n        print(\"\\nForged Images Statistics:\")\n        for key, value in forged_stats.items():\n            print(f\"  {key}: {value}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:26.534430Z","iopub.execute_input":"2025-11-10T10:06:26.534627Z","iopub.status.idle":"2025-11-10T10:06:32.757262Z","shell.execute_reply.started":"2025-11-10T10:06:26.534611Z","shell.execute_reply":"2025-11-10T10:06:32.756324Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import numpy as np\nimport matplotlib.pyplot as plt\n\ndef _to_rgb(img):\n    \"\"\"Ensure image is displayable by plt.imshow (H,W) or (H,W,3/4).\"\"\"\n    arr = np.asarray(img)\n    # If CHW, move to HWC\n    if arr.ndim == 3 and arr.shape[0] in (1,3,4) and arr.shape[-1] not in (3,4):\n        arr = np.moveaxis(arr, 0, -1)  # CHW -> HWC\n    return arr\n\ndef _to_2d_mask(mask):\n    \"\"\"Make mask 2D (H,W) and normalized to [0,1].\"\"\"\n    m = np.asarray(mask)\n\n    # If CHW format with multiple channels -> take first channel or combine\n    if m.ndim == 3 and m.shape[0] in (1, 2, 3, 4) and m.shape[-1] not in (1, 2, 3, 4):\n        # (C, H, W) format where C is small\n        if m.shape[0] == 1:\n            m = np.squeeze(m, axis=0)  # (1,H,W) -> (H,W)\n        elif m.shape[0] == 2:\n            # For 2-channel masks, take the maximum (union) or first channel\n            # Using maximum to combine both channels\n            m = np.max(m, axis=0)  # (2,H,W) -> (H,W)\n        else:\n            # For 3+ channels, take first channel\n            m = m[0]  # (C,H,W) -> (H,W)\n\n    # If HWC with a single channel -> squeeze last\n    elif m.ndim == 3 and m.shape[-1] == 1:\n        m = np.squeeze(m, axis=-1)  # (H,W,1) -> (H,W)\n\n    # If still 3D (e.g., RGB mask), convert to grayscale\n    elif m.ndim == 3 and m.shape[-1] in (3,4):\n        m = m[..., :3].mean(axis=-1)\n\n    # Now expect 2D\n    if m.ndim != 2:\n        raise ValueError(f\"Mask must be 2D after processing, got shape {m.shape}\")\n\n    # Normalize to [0,1]\n    m = m.astype(np.float32)\n    m_min, m_max = np.min(m), np.max(m)\n    if m_max > m_min:\n        m = (m - m_min) / (m_max - m_min)\n    else:\n        m = np.zeros_like(m, dtype=np.float32)\n\n    return m\n\ndef visualize_samples(authentic_samples, forged_samples, mask_samples):\n    \"\"\"Visualize sample images and masks.\"\"\"\n    # Build quick lookup for forged images by name\n    forged_dict = {name: img for name, img in forged_samples}\n\n    n = 3\n    n_auth = min(n, len(authentic_samples))\n    n_forg = min(n, len(forged_samples))\n    n_mask = min(n, len(mask_samples))\n\n    fig, axes = plt.subplots(3, 3, figsize=(15, 15))\n\n    # Row 0: authentic\n    for idx in range(n_auth):\n        name, img = authentic_samples[idx]\n        axes[0, idx].imshow(_to_rgb(img))\n        axes[0, idx].set_title(f'Authentic: {name}', fontsize=10)\n        axes[0, idx].axis('off')\n    for idx in range(n_auth, 3):\n        axes[0, idx].axis('off')\n\n    # Row 1: forged\n    for idx in range(n_forg):\n        name, img = forged_samples[idx]\n        axes[1, idx].imshow(_to_rgb(img))\n        axes[1, idx].set_title(f'Forged: {name}', fontsize=10)\n        axes[1, idx].axis('off')\n    for idx in range(n_forg, 3):\n        axes[1, idx].axis('off')\n\n    # Row 2: forged + mask overlay\n    for idx in range(n_mask):\n        name, mask = mask_samples[idx]\n        forged_img = forged_dict.get(name, None)\n        if forged_img is None:\n            axes[2, idx].set_title(f'No forged image for: {name}', fontsize=10)\n            axes[2, idx].axis('off')\n            continue\n\n        img_rgb = _to_rgb(forged_img)\n        m2d = _to_2d_mask(mask)\n\n        # If mask and image sizes differ, center-crop or pad is an option;\n        # here we do a quick resize with numpy if shapes mismatch (nearest-neighbor).\n        if m2d.shape[:2] != img_rgb.shape[:2]:\n            # Simple nearest-neighbor resize without external deps\n            ih, iw = img_rgb.shape[:2]\n            mh, mw = m2d.shape\n            ys = (np.linspace(0, mh-1, ih)).astype(int)\n            xs = (np.linspace(0, mw-1, iw)).astype(int)\n            m2d = m2d[ys][:, xs]\n\n        axes[2, idx].imshow(img_rgb)\n        axes[2, idx].imshow(m2d, alpha=0.5, cmap='Reds')  # mask is now (H,W) in [0,1]\n        axes[2, idx].set_title(f'Forged + Mask: {name}', fontsize=10)\n        axes[2, idx].axis('off')\n\n    for idx in range(n_mask, 3):\n        axes[2, idx].axis('off')\n\n    plt.tight_layout()\n    plt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:32.758492Z","iopub.execute_input":"2025-11-10T10:06:32.758748Z","iopub.status.idle":"2025-11-10T10:06:32.772378Z","shell.execute_reply.started":"2025-11-10T10:06:32.758730Z","shell.execute_reply":"2025-11-10T10:06:32.771692Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(\"auth sample shape:\", np.asarray(authentic_samples[0][1]).shape)\nprint(\"forged sample shape:\", np.asarray(forged_samples[0][1]).shape)\nprint(\"mask sample shape:\", np.asarray(mask_samples[0][1]).shape)\n\n# Visualize sample images and masks\nif 'authentic_samples' in locals() and 'forged_samples' in locals() and 'mask_samples' in locals():\n    visualize_samples(authentic_samples, forged_samples, mask_samples)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-10T10:06:32.772967Z","iopub.execute_input":"2025-11-10T10:06:32.773176Z","iopub.status.idle":"2025-11-10T10:06:34.594189Z","shell.execute_reply.started":"2025-11-10T10:06:32.773154Z","shell.execute_reply":"2025-11-10T10:06:34.593137Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 2. Data Loading and Preprocessing\n","metadata":{}},{"cell_type":"code","source":"# Data augmentation functions\ndef augment_image(image, mask=None):\n    \"\"\"Apply random augmentations to image and mask\"\"\"\n    # Random horizontal flip\n    if np.random.random() > 0.5:\n        image = cv2.flip(image, 1)\n        if mask is not None:\n            mask = cv2.flip(mask, 1)\n    \n    # Random vertical flip\n    if np.random.random() > 0.5:\n        image = cv2.flip(image, 0)\n        if mask is not None:\n            mask = cv2.flip(mask, 0)\n    \n    # Random rotation\n    angle = np.random.uniform(-15, 15)\n    h, w = image.shape[:2]\n    center = (w // 2, h // 2)\n    M = cv2.getRotationMatrix2D(center, angle, 1.0)\n    image = cv2.warpAffine(image, M, (w, h), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REFLECT)\n    if mask is not None:\n        mask = cv2.warpAffine(mask, M, (w, h), flags=cv2.INTER_NEAREST, borderMode=cv2.BORDER_REFLECT)\n    \n    # Random brightness adjustment\n    if np.random.random() > 0.5:\n        brightness = np.random.uniform(0.8, 1.2)\n        image = np.clip(image * brightness, 0, 255).astype(np.uint8)\n    \n    return image, mask\n\ndef preprocess_image(image, target_size=(512, 512)):\n    \"\"\"Preprocess image for model input\"\"\"\n    # Resize\n    image = cv2.resize(image, target_size, interpolation=cv2.INTER_LINEAR)\n    # Normalize to [0, 1]\n    image = image.astype(np.float32) / 255.0\n    # Convert to tensor format (H, W, C) -> (C, H, W)\n    image = np.transpose(image, (2, 0, 1))\n    return image\n\ndef preprocess_mask(mask, target_size=(512, 512)):\n    \"\"\"Preprocess mask for model input\"\"\"\n    # Resize\n    mask = cv2.resize(mask, target_size, interpolation=cv2.INTER_NEAREST)\n    # Ensure binary mask\n    mask = (mask > 0.5).astype(np.float32)\n    return mask\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Dataset class (using functional approach without class definition)\ndef create_dataset(image_dir, mask_dir=None, is_train=True, target_size=(512, 512)):\n    \"\"\"Create dataset from directory\"\"\"\n    image_files = list(image_dir.glob('*.png'))\n    dataset = []\n    \n    for img_path in image_files:\n        item = {\n            'image_path': str(img_path),\n            'image_id': img_path.stem\n        }\n        \n        if mask_dir is not None:\n            mask_path = mask_dir / f\"{img_path.stem}.npy\"\n            if mask_path.exists():\n                item['mask_path'] = str(mask_path)\n            else:\n                item['mask_path'] = None\n        else:\n            item['mask_path'] = None\n        \n        dataset.append(item)\n    \n    return dataset\n\ndef load_image_and_mask(item, is_train=True, target_size=(512, 512)):\n    \"\"\"Load and preprocess image and mask\"\"\"\n    # Load image\n    image = cv2.imread(item['image_path'])\n    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n    \n    # Load mask if available\n    mask = None\n    if item['mask_path'] is not None and os.path.exists(item['mask_path']):\n        mask = np.load(item['mask_path'])\n        # Handle different mask formats\n        if len(mask.shape) == 3:\n            # Check if it's (C, H, W) or (H, W, C) format\n            # Priority: squeeze dimension of size 1 first\n            if mask.shape[0] == 1:\n                # (1, H, W) format - squeeze first dimension\n                mask = mask.squeeze(0)\n            elif mask.shape[-1] == 1:\n                # (H, W, 1) format - squeeze last dimension\n                mask = mask.squeeze(-1)\n            elif mask.shape[0] < mask.shape[-1]:\n                # Likely (C, H, W) format where C < H - take first channel\n                mask = mask[0]\n            elif mask.shape[-1] < mask.shape[0]:\n                # Likely (H, W, C) format where C < H - take first channel from last dimension\n                mask = mask[:, :, 0]\n            else:\n                # Ambiguous case - take first channel from last dimension\n                mask = mask[:, :, 0]\n        mask = mask.astype(np.float32)\n    \n    # Apply augmentation during training\n    if is_train and mask is not None:\n        image, mask = augment_image(image, mask)\n    \n    # Preprocess\n    image = preprocess_image(image, target_size)\n    if mask is not None:\n        mask = preprocess_mask(mask, target_size)\n    else:\n        # Create empty mask for authentic images\n        mask = np.zeros(target_size, dtype=np.float32)\n    \n    return image, mask\n\n# Create datasets\nif train_authentic_dir.exists() and train_forged_dir.exists():\n    authentic_dataset = create_dataset(train_authentic_dir, None, is_train=True)\n    forged_dataset = create_dataset(train_forged_dir, train_masks_dir, is_train=True)\n    \n    print(f\"Authentic dataset size: {len(authentic_dataset)}\")\n    print(f\"Forged dataset size: {len(forged_dataset)}\")\n    \n    # Combine datasets\n    full_dataset = authentic_dataset + forged_dataset\n    print(f\"Total dataset size: {len(full_dataset)}\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 3. Model Architecture\n","metadata":{}},{"cell_type":"code","source":"# U-Net architecture components (functional approach)\ndef double_conv(in_channels, out_channels):\n    \"\"\"Double convolution block\"\"\"\n    return nn.Sequential(\n        nn.Conv2d(in_channels, out_channels, 3, padding=1),\n        nn.BatchNorm2d(out_channels),\n        nn.ReLU(inplace=True),\n        nn.Conv2d(out_channels, out_channels, 3, padding=1),\n        nn.BatchNorm2d(out_channels),\n        nn.ReLU(inplace=True)\n    )\n\ndef create_unet_model(in_channels=3, n_classes=1):\n    \"\"\"Create U-Net model for segmentation\"\"\"\n    # Encoder (downsampling path)\n    enc1 = double_conv(in_channels, 64)\n    enc2 = double_conv(64, 128)\n    enc3 = double_conv(128, 256)\n    enc4 = double_conv(256, 512)\n    \n    # Bottleneck\n    bottleneck = double_conv(512, 1024)\n    \n    # Decoder (upsampling path)\n    up4 = nn.ConvTranspose2d(1024, 512, 2, stride=2)\n    dec4 = double_conv(1024, 512)\n    \n    up3 = nn.ConvTranspose2d(512, 256, 2, stride=2)\n    dec3 = double_conv(512, 256)\n    \n    up2 = nn.ConvTranspose2d(256, 128, 2, stride=2)\n    dec2 = double_conv(256, 128)\n    \n    up1 = nn.ConvTranspose2d(128, 64, 2, stride=2)\n    dec1 = double_conv(128, 64)\n    \n    # Final layer\n    final_conv = nn.Conv2d(64, n_classes, 1)\n    \n    return {\n        'enc1': enc1, 'enc2': enc2, 'enc3': enc3, 'enc4': enc4,\n        'bottleneck': bottleneck,\n        'up4': up4, 'dec4': dec4,\n        'up3': up3, 'dec3': dec3,\n        'up2': up2, 'dec2': dec2,\n        'up1': up1, 'dec1': dec1,\n        'final_conv': final_conv\n    }\n\ndef forward_unet(model_dict, x):\n    \"\"\"Forward pass through U-Net\"\"\"\n    # Encoder\n    enc1 = model_dict['enc1'](x)\n    x1 = nn.MaxPool2d(2)(enc1)\n    \n    enc2 = model_dict['enc2'](x1)\n    x2 = nn.MaxPool2d(2)(enc2)\n    \n    enc3 = model_dict['enc3'](x2)\n    x3 = nn.MaxPool2d(2)(enc3)\n    \n    enc4 = model_dict['enc4'](x3)\n    x4 = nn.MaxPool2d(2)(enc4)\n    \n    # Bottleneck\n    bottleneck = model_dict['bottleneck'](x4)\n    \n    # Decoder\n    up4 = model_dict['up4'](bottleneck)\n    up4 = torch.cat([up4, enc4], dim=1)\n    dec4 = model_dict['dec4'](up4)\n    \n    up3 = model_dict['up3'](dec4)\n    up3 = torch.cat([up3, enc3], dim=1)\n    dec3 = model_dict['dec3'](up3)\n    \n    up2 = model_dict['up2'](dec3)\n    up2 = torch.cat([up2, enc2], dim=1)\n    dec2 = model_dict['dec2'](up2)\n    \n    up1 = model_dict['up1'](dec2)\n    up1 = torch.cat([up1, enc1], dim=1)\n    dec1 = model_dict['dec1'](up1)\n    \n    # Final output\n    output = model_dict['final_conv'](dec1)\n    return torch.sigmoid(output)\n\n# Create model wrapper class for PyTorch compatibility\nclass UNet(nn.Module):\n    def __init__(self, in_channels=3, n_classes=1):\n        super(UNet, self).__init__()\n        model_dict = create_unet_model(in_channels, n_classes)\n        for key, value in model_dict.items():\n            setattr(self, key, value)\n        self.model_dict = model_dict\n    \n    def forward(self, x):\n        return forward_unet(self.model_dict, x)\n\n# Initialize model\nmodel = UNet(in_channels=3, n_classes=1)\nmodel = model.to(device)\nprint(f\"Model created and moved to {device}\")\nprint(f\"Total parameters: {sum(p.numel() for p in model.parameters()):,}\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 4. Dataset and Loss Functions\n","metadata":{}},{"cell_type":"code","source":"# Custom Dataset class for PyTorch DataLoader\nclass ForgeryDataset(Dataset):\n    def __init__(self, dataset, is_train=True, target_size=(512, 512)):\n        self.dataset = dataset\n        self.is_train = is_train\n        self.target_size = target_size\n    \n    def __len__(self):\n        return len(self.dataset)\n    \n    def __getitem__(self, idx):\n        item = self.dataset[idx]\n        image, mask = load_image_and_mask(item, self.is_train, self.target_size)\n        \n        # Convert to tensors\n        image_tensor = torch.from_numpy(image).float()\n        mask_tensor = torch.from_numpy(mask).float().unsqueeze(0)  # Add channel dimension\n        \n        return image_tensor, mask_tensor\n\n# Loss function - Dice Loss + BCE Loss\ndef dice_loss(pred, target, smooth=1e-6):\n    \"\"\"Dice loss for segmentation\"\"\"\n    pred_flat = pred.view(-1)\n    target_flat = target.view(-1)\n    intersection = (pred_flat * target_flat).sum()\n    dice = (2. * intersection + smooth) / (pred_flat.sum() + target_flat.sum() + smooth)\n    return 1 - dice\n\ndef combined_loss(pred, target, bce_weight=0.5):\n    \"\"\"Combined BCE and Dice loss\"\"\"\n    bce = nn.functional.binary_cross_entropy(pred, target)\n    dice = dice_loss(pred, target)\n    return bce_weight * bce + (1 - bce_weight) * dice\n\n# Metrics\ndef calculate_iou(pred, target, threshold=0.5):\n    \"\"\"Calculate IoU (Intersection over Union)\"\"\"\n    pred_binary = (pred > threshold).float()\n    target_binary = target.float()\n    \n    intersection = (pred_binary * target_binary).sum()\n    union = pred_binary.sum() + target_binary.sum() - intersection\n    \n    if union == 0:\n        return 1.0\n    \n    iou = intersection / union\n    return iou.item()\n\ndef calculate_dice_score(pred, target, threshold=0.5):\n    \"\"\"Calculate Dice score\"\"\"\n    pred_binary = (pred > threshold).float()\n    target_binary = target.float()\n    \n    intersection = (pred_binary * target_binary).sum()\n    dice = (2. * intersection) / (pred_binary.sum() + target_binary.sum() + 1e-6)\n    return dice.item()\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Split dataset into train and validation\ndef split_dataset(dataset, val_ratio=0.2):\n    \"\"\"Split dataset into train and validation sets\"\"\"\n    np.random.shuffle(dataset)\n    split_idx = int(len(dataset) * (1 - val_ratio))\n    train_dataset = dataset[:split_idx]\n    val_dataset = dataset[split_idx:]\n    return train_dataset, val_dataset\n\nif 'full_dataset' in locals() and len(full_dataset) > 0:\n    train_data, val_data = split_dataset(full_dataset.copy(), val_ratio=0.2)\n    \n    # Create data loaders\n    train_dataset = ForgeryDataset(train_data, is_train=True, target_size=(512, 512))\n    val_dataset = ForgeryDataset(val_data, is_train=False, target_size=(512, 512))\n    \n    batch_size = 8 if device.type == 'cuda' else 4\n    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=0)\n    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=0)\n    \n    print(f\"Train samples: {len(train_data)}\")\n    print(f\"Validation samples: {len(val_data)}\")\n    print(f\"Batch size: {batch_size}\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 5. Training Loop\n","metadata":{}},{"cell_type":"code","source":"# Training function\ndef train_epoch(model, train_loader, optimizer, criterion, device):\n    \"\"\"Train for one epoch\"\"\"\n    model.train()\n    total_loss = 0.0\n    total_iou = 0.0\n    total_dice = 0.0\n    num_batches = 0\n    \n    for images, masks in tqdm(train_loader, desc=\"Training\"):\n        images = images.to(device)\n        masks = masks.to(device)\n        \n        # Forward pass\n        optimizer.zero_grad()\n        outputs = model(images)\n        \n        # Calculate loss\n        loss = criterion(outputs, masks)\n        \n        # Backward pass\n        loss.backward()\n        optimizer.step()\n        \n        # Calculate metrics\n        with torch.no_grad():\n            iou = calculate_iou(outputs, masks)\n            dice = calculate_dice_score(outputs, masks)\n        \n        total_loss += loss.item()\n        total_iou += iou\n        total_dice += dice\n        num_batches += 1\n    \n    avg_loss = total_loss / num_batches\n    avg_iou = total_iou / num_batches\n    avg_dice = total_dice / num_batches\n    \n    return avg_loss, avg_iou, avg_dice\n\n# Validation function\ndef validate_epoch(model, val_loader, criterion, device):\n    \"\"\"Validate for one epoch\"\"\"\n    model.eval()\n    total_loss = 0.0\n    total_iou = 0.0\n    total_dice = 0.0\n    num_batches = 0\n    \n    with torch.no_grad():\n        for images, masks in tqdm(val_loader, desc=\"Validating\"):\n            images = images.to(device)\n            masks = masks.to(device)\n            \n            # Forward pass\n            outputs = model(images)\n            \n            # Calculate loss\n            loss = criterion(outputs, masks)\n            \n            # Calculate metrics\n            iou = calculate_iou(outputs, masks)\n            dice = calculate_dice_score(outputs, masks)\n            \n            total_loss += loss.item()\n            total_iou += iou\n            total_dice += dice\n            num_batches += 1\n    \n    avg_loss = total_loss / num_batches\n    avg_iou = total_iou / num_batches\n    avg_dice = total_dice / num_batches\n    \n    return avg_loss, avg_iou, avg_dice\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Main training loop\ndef train_model(model, train_loader, val_loader, num_epochs=10, learning_rate=1e-4):\n    \"\"\"Main training function\"\"\"\n    # Setup optimizer and loss\n    optimizer = optim.Adam(model.parameters(), lr=learning_rate)\n    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=3)\n    criterion = combined_loss\n    \n    # Training history\n    history = {\n        'train_loss': [],\n        'train_iou': [],\n        'train_dice': [],\n        'val_loss': [],\n        'val_iou': [],\n        'val_dice': []\n    }\n    \n    best_val_loss = float('inf')\n    current_lr = learning_rate\n    \n    print(f\"Starting training for {num_epochs} epochs...\")\n    print(f\"Device: {device}\")\n    print(f\"Initial learning rate: {learning_rate}\")\n    print(\"-\" * 50)\n    \n    for epoch in range(num_epochs):\n        print(f\"\\nEpoch {epoch+1}/{num_epochs}\")\n        \n        # Train\n        train_loss, train_iou, train_dice = train_epoch(model, train_loader, optimizer, criterion, device)\n        \n        # Validate\n        val_loss, val_iou, val_dice = validate_epoch(model, val_loader, criterion, device)\n        \n        # Update learning rate\n        old_lr = current_lr\n        scheduler.step(val_loss)\n        current_lr = optimizer.param_groups[0]['lr']\n        if current_lr < old_lr:\n            print(f\"Learning rate reduced from {old_lr:.2e} to {current_lr:.2e}\")\n        \n        # Save history\n        history['train_loss'].append(train_loss)\n        history['train_iou'].append(train_iou)\n        history['train_dice'].append(train_dice)\n        history['val_loss'].append(val_loss)\n        history['val_iou'].append(val_iou)\n        history['val_dice'].append(val_dice)\n        \n        # Print metrics\n        print(f\"Train Loss: {train_loss:.4f}, Train IoU: {train_iou:.4f}, Train Dice: {train_dice:.4f}\")\n        print(f\"Val Loss: {val_loss:.4f}, Val IoU: {val_iou:.4f}, Val Dice: {val_dice:.4f}\")\n        \n        # Save best model\n        if val_loss < best_val_loss:\n            best_val_loss = val_loss\n            torch.save(model.state_dict(), str(model_path))\n            print(f\"Saved best model (Val Loss: {val_loss:.4f})\")\n    \n    return history\n\n# Start training\nif 'train_loader' in locals() and 'val_loader' in locals():\n    print(\"\\n\" + \"=\"*50)\n    print(\"Starting Model Training\")\n    print(\"=\"*50)\n    history = train_model(model, train_loader, val_loader, num_epochs=20, learning_rate=1e-4)\n    print(\"\\nTraining completed!\")\nelse:\n    print(\"Data loaders not available. Please run the previous cells first.\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 6. Visualization and Evaluation\n","metadata":{}},{"cell_type":"code","source":"# Plot training history\ndef plot_training_history(history):\n    \"\"\"Plot training curves\"\"\"\n    fig, axes = plt.subplots(1, 3, figsize=(15, 4))\n    \n    # Loss\n    axes[0].plot(history['train_loss'], label='Train Loss')\n    axes[0].plot(history['val_loss'], label='Val Loss')\n    axes[0].set_title('Loss')\n    axes[0].set_xlabel('Epoch')\n    axes[0].set_ylabel('Loss')\n    axes[0].legend()\n    axes[0].grid(True)\n    \n    # IoU\n    axes[1].plot(history['train_iou'], label='Train IoU')\n    axes[1].plot(history['val_iou'], label='Val IoU')\n    axes[1].set_title('IoU')\n    axes[1].set_xlabel('Epoch')\n    axes[1].set_ylabel('IoU')\n    axes[1].legend()\n    axes[1].grid(True)\n    \n    # Dice\n    axes[2].plot(history['train_dice'], label='Train Dice')\n    axes[2].plot(history['val_dice'], label='Val Dice')\n    axes[2].set_title('Dice Score')\n    axes[2].set_xlabel('Epoch')\n    axes[2].set_ylabel('Dice')\n    axes[2].legend()\n    axes[2].grid(True)\n    \n    plt.tight_layout()\n    plt.show()\n\n# Plot training history after training\nif 'history' in locals():\n    plot_training_history(history)\nelse:\n    print(\"Training history not available. Please train the model first.\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Visualize predictions\ndef visualize_predictions(model, val_loader, device, num_samples=5):\n    \"\"\"Visualize model predictions on validation set\"\"\"\n    model.eval()\n    \n    fig, axes = plt.subplots(num_samples, 4, figsize=(16, 4*num_samples))\n    \n    with torch.no_grad():\n        for idx, (images, masks) in enumerate(val_loader):\n            if idx >= num_samples:\n                break\n            \n            images = images.to(device)\n            masks = masks.to(device)\n            \n            # Get predictions\n            outputs = model(images)\n            pred_masks = (outputs > 0.5).float()\n            \n            # Convert to numpy for visualization\n            img = images[0].cpu().numpy().transpose(1, 2, 0)\n            true_mask = masks[0, 0].cpu().numpy()\n            pred_mask = pred_masks[0, 0].cpu().numpy()\n            prob_mask = outputs[0, 0].cpu().numpy()\n            \n            # Plot\n            axes[idx, 0].imshow(img)\n            axes[idx, 0].set_title('Original Image')\n            axes[idx, 0].axis('off')\n            \n            axes[idx, 1].imshow(true_mask, cmap='gray')\n            axes[idx, 1].set_title('True Mask')\n            axes[idx, 1].axis('off')\n            \n            axes[idx, 2].imshow(prob_mask, cmap='hot')\n            axes[idx, 2].set_title('Predicted Probability')\n            axes[idx, 2].axis('off')\n            \n            axes[idx, 3].imshow(pred_mask, cmap='gray')\n            axes[idx, 3].set_title('Predicted Mask')\n            axes[idx, 3].axis('off')\n    \n    plt.tight_layout()\n    plt.show()\n\n# Visualize predictions on validation set\nif 'val_loader' in locals() and 'model' in locals():\n    # Load best model if available\n    if os.path.exists(str(model_path)):\n        model.load_state_dict(torch.load(str(model_path)))\n        print(\"Loaded best model for visualization\")\n    visualize_predictions(model, val_loader, device, num_samples=5)\nelse:\n    print(\"Model or validation loader not available.\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 7. Model Evaluation and Inference\n","metadata":{}},{"cell_type":"code","source":"# Run-Length Encoding (RLE) for mask encoding\ndef rle_encode(mask):\n    \"\"\"Encode binary mask to RLE string\"\"\"\n    pixels = mask.flatten()\n    pixels = np.concatenate([[0], pixels, [0]])\n    runs = np.where(pixels[1:] != pixels[:-1])[0] + 1\n    runs[1::2] -= runs[::2]\n    return ' '.join(str(x) for x in runs)\n\ndef rle_decode(rle_str, shape):\n    \"\"\"Decode RLE string to binary mask\"\"\"\n    s = rle_str.split()\n    starts, lengths = [np.asarray(x, dtype=int) for x in (s[0:][::2], s[1:][::2])]\n    starts -= 1\n    ends = starts + lengths\n    img = np.zeros(shape[0] * shape[1], dtype=np.uint8)\n    for lo, hi in zip(starts, ends):\n        img[lo:hi] = 1\n    return img.reshape(shape)\n\n# Inference function for test images\ndef predict_test_image(model, image_path, device, target_size=(512, 512), threshold=0.5):\n    \"\"\"Predict forgery mask for a test image\"\"\"\n    model.eval()\n    \n    # Load and preprocess image\n    image = cv2.imread(str(image_path))\n    if image is None:\n        raise ValueError(f\"Could not load image: {image_path}\")\n    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n    original_size = image.shape[:2]\n    \n    # Preprocess\n    processed_image = preprocess_image(image, target_size)\n    image_tensor = torch.from_numpy(processed_image).float().unsqueeze(0).to(device)\n    \n    # Predict\n    with torch.no_grad():\n        output = model(image_tensor)\n        mask = output[0, 0].cpu().numpy()\n    \n    # Resize mask back to original size\n    mask_resized = cv2.resize(mask, (original_size[1], original_size[0]), interpolation=cv2.INTER_LINEAR)\n    binary_mask = (mask_resized > threshold).astype(np.uint8)\n    \n    return mask_resized, binary_mask\n\ndef process_test_set(model, test_dir, device, output_file='submission.csv', threshold=0.5):\n    \"\"\"Process all test images and create submission file with RLE encoded masks\"\"\"\n    test_files = list(test_dir.glob('*.png'))\n    results = []\n    \n    if len(test_files) == 0:\n        print(f\"No test images found in {test_dir}\")\n        return pd.DataFrame()\n    \n    print(f\"Processing {len(test_files)} test images...\")\n    \n    for test_file in tqdm(test_files, desc=\"Processing test images\"):\n        case_id = test_file.stem\n        try:\n            mask_prob, binary_mask = predict_test_image(model, test_file, device, threshold=threshold)\n            \n            # Check if image is authentic (no significant forgery detected)\n            forgery_ratio = binary_mask.sum() / (binary_mask.shape[0] * binary_mask.shape[1])\n            \n            if forgery_ratio < 0.01:  # Less than 1% of image is forged\n                annotation = 'authentic'\n                rle = ''  # No mask for authentic images\n            else:\n                annotation = 'forged'\n                # Encode mask using RLE\n                rle = rle_encode(binary_mask)\n            \n            results.append({\n                'case_id': case_id,\n                'annotation': annotation,\n                'rle': rle\n            })\n        except Exception as e:\n            print(f\"Error processing {test_file}: {e}\")\n            results.append({\n                'case_id': case_id,\n                'annotation': 'authentic',  # Default to authentic on error\n                'rle': ''\n            })\n    \n    # Create submission DataFrame\n    submission_df = pd.DataFrame(results)\n    submission_df.to_csv(output_file, index=False)\n    print(f\"\\nSubmission file saved to {output_file}\")\n    print(f\"Summary:\")\n    print(f\"  Authentic: {len(submission_df[submission_df['annotation'] == 'authentic'])}\")\n    print(f\"  Forged: {len(submission_df[submission_df['annotation'] == 'forged'])}\")\n    \n    return submission_df\n\n# Process test set if model is trained\nif 'model' in locals() and test_images_dir.exists():\n    # Load best model if available\n    if os.path.exists(str(model_path)):\n        model.load_state_dict(torch.load(str(model_path), map_location=device))\n        print(\"Loaded best model for inference\")\n        submission = process_test_set(model, test_images_dir, device, output_file=str(submission_path))\n        if not submission.empty:\n            print(\"\\nSubmission preview:\")\n            print(submission.head(10))\n    else:\n        print(\"Best model not found. Please train the model first.\")\nelse:\n    print(\"Model or test directory not available.\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Model evaluation utility\ndef evaluate_model(model, val_loader, device, model_path='best_model.pth'):\n    \"\"\"Evaluate model on validation set\"\"\"\n    if model_path and os.path.exists(model_path):\n        model.load_state_dict(torch.load(model_path, map_location=device))\n        print(f\"Loaded model from {model_path}\")\n    elif model_path:\n        print(f\"Model file {model_path} not found. Using current model state.\")\n    \n    model.eval()\n    criterion = combined_loss\n    \n    val_loss, val_iou, val_dice = validate_epoch(model, val_loader, criterion, device)\n    \n    print(\"\\n\" + \"=\"*50)\n    print(\"Model Evaluation Results\")\n    print(\"=\"*50)\n    print(f\"Validation Loss: {val_loss:.4f}\")\n    print(f\"Validation IoU:  {val_iou:.4f}\")\n    print(f\"Validation Dice: {val_dice:.4f}\")\n    print(\"=\"*50)\n    \n    return val_loss, val_iou, val_dice\n\n# Evaluate model if available\nif 'val_loader' in locals() and 'model' in locals():\n    if os.path.exists(str(model_path)):\n        evaluate_model(model, val_loader, device, model_path=str(model_path))\n    else:\n        print(\"Best model not found. Evaluating current model state...\")\n        evaluate_model(model, val_loader, device, model_path=None)\nelse:\n    print(\"Model or validation loader not available for evaluation.\")\n","metadata":{},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{},"outputs":[],"execution_count":null}]}