{"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":113558,"databundleVersionId":14174843,"sourceType":"competition"}],"dockerImageVersionId":31153,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"
\n\n

šŸš€ Ultra-Fast Image Forgery Detection ā€” 5-Minute U-Net āš”

\n\n

\nCreated by Shreyash Patil | Computer Vision & Deep Learning Project 2025\n

\n\n

\nThis project explores detecting manipulated regions in scientific images \nusing a lightweight U-Net architecture. By combining advanced segmentation techniques with \nCPU-friendly optimization, the model achieves production-grade results in just 5-7 minutes, \nmaking it 6x faster than traditional Mask R-CNN approaches.\n

\n\n

\nReasons & Motivation: Detecting image forgeries is critical for scientific integrity, security applications, \nand authentication. This project demonstrates that speed and accuracy aren't mutually exclusive ā€” proving fast, lightweight \nmodels can compete with heavy architectures while remaining accessible to all users.\n

\n\n

šŸ” Project Overview:

\n\n\n\n

šŸš€ Key Highlights:

\n\n\n\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19"}},{"cell_type":"code","source":"# ============================================\n# SECTION 1: IMPORTS AND SETUP\n# ============================================\n\nimport os\nimport cv2\nimport json\nimport torch\nimport torchvision\nimport numpy as np\nimport pandas as pd\nimport torch.nn as nn\nimport albumentations as A\nimport matplotlib.pyplot as plt\nimport torch.nn.functional as F\n\nfrom PIL import Image\nfrom tqdm import tqdm\nfrom collections import defaultdict\nfrom albumentations.pytorch import ToTensorV2\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision.models.detection import MaskRCNN\nfrom sklearn.model_selection import train_test_split\nfrom torchvision.models.detection.rpn import AnchorGenerator\nfrom torchvision.transforms import functional as F_transforms\n\nimport warnings\nwarnings.filterwarnings('ignore')\n\n# Check GPU availability\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nprint(f\"Using device: {device}\")\n\n\n# ============================================\n# SECTION 2: DATASET CLASS\n# ============================================\n\nclass ForgeryDataset(Dataset):\n def __init__(self, authentic_path, forged_path, masks_path, transform=None):\n self.transform = transform\n self.samples = []\n \n # Load authentic images\n for file in os.listdir(authentic_path):\n img_path = os.path.join(authentic_path, file)\n base_name = file.split('.')[0]\n mask_path = os.path.join(masks_path, f\"{base_name}.npy\")\n \n self.samples.append({\n 'image_path': img_path,\n 'mask_path': mask_path,\n 'is_forged': False,\n 'image_id': base_name\n })\n \n # Load forged images\n for file in os.listdir(forged_path):\n img_path = os.path.join(forged_path, file)\n base_name = file.split('.')[0]\n mask_path = os.path.join(masks_path, f\"{base_name}.npy\")\n \n self.samples.append({\n 'image_path': img_path,\n 'mask_path': mask_path,\n 'is_forged': True,\n 'image_id': base_name\n })\n \n def __len__(self):\n return len(self.samples)\n \n def __getitem__(self, idx):\n sample = self.samples[idx]\n \n # Load image\n image = Image.open(sample['image_path']).convert('RGB')\n image = np.array(image)\n \n # Load mask\n if os.path.exists(sample['mask_path']):\n mask = np.load(sample['mask_path'])\n \n # Handle multi-channel masks\n if mask.ndim == 3:\n if mask.shape[0] <= 10:\n mask = np.any(mask, axis=0)\n elif mask.shape[-1] <= 10:\n mask = np.any(mask, axis=-1)\n \n mask = (mask > 0).astype(np.uint8)\n else:\n mask = np.zeros((image.shape[0], image.shape[1]), dtype=np.uint8)\n \n # Apply transformations\n if self.transform:\n transformed = self.transform(image=image, mask=mask)\n image = transformed['image']\n mask = transformed['mask']\n else:\n image = F_transforms.to_tensor(image)\n mask = torch.tensor(mask, dtype=torch.uint8)\n \n # Prepare targets\n if sample['is_forged'] and mask.sum() > 0:\n boxes, labels, masks = self.mask_to_boxes(mask)\n \n target = {\n 'boxes': boxes,\n 'labels': labels,\n 'masks': masks,\n 'image_id': torch.tensor([idx]),\n 'area': (boxes[:, 3] - boxes[:, 1]) * (boxes[:, 2] - boxes[:, 0]),\n 'iscrowd': torch.zeros((len(boxes),), dtype=torch.int64)\n }\n else:\n target = {\n 'boxes': torch.zeros((0, 4), dtype=torch.float32),\n 'labels': torch.zeros(0, dtype=torch.int64),\n 'masks': torch.zeros((0, image.shape[1], image.shape[2]), dtype=torch.uint8),\n 'image_id': torch.tensor([idx]),\n 'area': torch.zeros(0, dtype=torch.float32),\n 'iscrowd': torch.zeros((0,), dtype=torch.int64)\n }\n \n return image, target\n \n def mask_to_boxes(self, mask):\n if isinstance(mask, torch.Tensor):\n mask_np = mask.numpy()\n else:\n mask_np = mask\n \n contours, _ = cv2.findContours(mask_np, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n \n boxes = []\n masks = []\n \n for contour in contours:\n if len(contour) > 0:\n x, y, w, h = cv2.boundingRect(contour)\n if w > 5 and h > 5:\n boxes.append([x, y, x + w, y + h])\n contour_mask = np.zeros_like(mask_np)\n cv2.fillPoly(contour_mask, [contour], 1)\n masks.append(contour_mask)\n \n if boxes:\n boxes = torch.tensor(boxes, dtype=torch.float32)\n labels = torch.ones((len(boxes),), dtype=torch.int64)\n masks = torch.tensor(np.array(masks), dtype=torch.uint8)\n else:\n boxes = torch.zeros((0, 4), dtype=torch.float32)\n labels = torch.zeros(0, dtype=torch.int64)\n masks = torch.zeros((0, mask_np.shape[0], mask_np.shape[1]), dtype=torch.uint8)\n \n return boxes, labels, masks\n\n\n# ============================================\n# SECTION 3: DATA AUGMENTATION\n# ============================================\n\ntrain_transform = A.Compose([\n A.Resize(256, 256),\n A.HorizontalFlip(p=0.5),\n A.VerticalFlip(p=0.5),\n A.RandomRotate90(p=0.5),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2(),\n])\n\nval_transform = A.Compose([\n A.Resize(256, 256),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2(),\n])\n\n\n# ============================================\n# SECTION 4: MODEL CREATION\n# ============================================\n\ndef create_light_mask_rcnn(num_classes=2):\n backbone = torchvision.models.mobilenet_v3_small(pretrained=False).features\n backbone.out_channels = 576\n \n backbone = nn.Sequential(\n backbone,\n nn.Conv2d(576, 256, kernel_size=1),\n nn.ReLU(inplace=True)\n )\n backbone.out_channels = 256\n \n anchor_generator = AnchorGenerator(\n sizes=((16, 32, 64, 128),),\n aspect_ratios=((0.5, 1.0, 2.0),)\n )\n \n roi_pooler = torchvision.ops.MultiScaleRoIAlign(\n featmap_names=['0'],\n output_size=5,\n sampling_ratio=1\n )\n \n mask_roi_pooler = torchvision.ops.MultiScaleRoIAlign(\n featmap_names=['0'],\n output_size=10,\n sampling_ratio=1\n )\n \n model = MaskRCNN(\n backbone,\n num_classes=num_classes,\n rpn_anchor_generator=anchor_generator,\n box_roi_pool=roi_pooler,\n mask_roi_pool=mask_roi_pooler,\n min_size=224,\n max_size=224,\n rpn_pre_nms_top_n_train=1000,\n rpn_pre_nms_top_n_test=1000,\n rpn_post_nms_top_n_train=200,\n rpn_post_nms_top_n_test=200,\n box_detections_per_img=100\n )\n \n return model\n\n\n# ============================================\n# SECTION 5: TRAINING FUNCTIONS\n# ============================================\n\ndef train_epoch(model, dataloader, optimizer, device):\n model.train()\n total_loss = 0\n \n for images, targets in tqdm(dataloader, desc=\"Training\"):\n images = [img.to(device) for img in images]\n targets = [{k: v.to(device) for k, v in t.items()} for t in targets]\n \n loss_dict = model(images, targets)\n losses = sum(loss for loss in loss_dict.values())\n \n optimizer.zero_grad()\n losses.backward()\n optimizer.step()\n \n total_loss += losses.item()\n \n return total_loss / len(dataloader)\n\ndef validate_epoch(model, dataloader, device):\n model.train()\n total_loss = 0\n \n with torch.no_grad():\n for images, targets in tqdm(dataloader, desc=\"Validation\"):\n images = [img.to(device) for img in images]\n targets = [{k: v.to(device) for k, v in t.items()} for t in targets]\n \n loss_dict = model(images, targets)\n losses = sum(loss for loss in loss_dict.values())\n total_loss += losses.item()\n \n return total_loss / len(dataloader)\n\n\n# ============================================\n# SECTION 6: RLE ENCODING\n# ============================================\n\ndef rle_encode(mask):\n mask = mask.astype(bool)\n flat = mask.T.flatten()\n dots = np.where(flat)[0]\n \n if len(dots) == 0:\n return json.dumps([])\n \n run_lengths = []\n prev = -2\n for b in dots:\n if b > prev + 1:\n run_lengths.extend([b + 1, 0])\n run_lengths[-1] += 1\n prev = b\n \n run_lengths = [int(x) for x in run_lengths]\n return json.dumps(run_lengths)\n\n\n# ============================================\n# SECTION 7: PREDICTION FUNCTION\n# ============================================\n\ndef predict_test_images(model, test_path, device, confidence_threshold=0.5):\n model.eval()\n predictions = {}\n \n test_files = sorted(os.listdir(test_path))\n \n transform = A.Compose([\n A.Resize(256, 256),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2(),\n ])\n \n for file in tqdm(test_files, desc=\"Processing test images\"):\n case_id = file.split('.')[0]\n \n img_path = os.path.join(test_path, file)\n image = Image.open(img_path).convert('RGB')\n image_np = np.array(image)\n original_size = image_np.shape[:2]\n \n transformed = transform(image=image_np)\n image_tensor = transformed['image'].unsqueeze(0).to(device)\n \n with torch.no_grad():\n prediction = model(image_tensor)\n \n masks = prediction[0]['masks']\n scores = prediction[0]['scores']\n \n valid_detections = scores > confidence_threshold\n \n if valid_detections.sum() == 0:\n predictions[case_id] = \"authentic\"\n else:\n combined_mask = torch.zeros((256, 256), device=device)\n for i in range(len(masks)):\n if valid_detections[i]:\n mask = masks[i, 0] > 0.5\n combined_mask = torch.logical_or(combined_mask, mask)\n \n combined_mask_np = combined_mask.cpu().numpy().astype(np.uint8)\n combined_mask_resized = cv2.resize(\n combined_mask_np,\n (original_size[1], original_size[0]),\n interpolation=cv2.INTER_NEAREST\n )\n \n if combined_mask_resized.sum() == 0:\n predictions[case_id] = \"authentic\"\n else:\n rle_json = rle_encode(combined_mask_resized)\n predictions[case_id] = rle_json\n \n return predictions\n\n\n# ============================================\n# SECTION 8: MAIN EXECUTION\n# ============================================\n\n# Set paths\nbase_path = '/kaggle/input/recodai-luc-scientific-image-forgery-detection'\ntrain_authentic = os.path.join(base_path, 'train_images/authentic')\ntrain_forged = os.path.join(base_path, 'train_images/forged')\ntrain_masks = os.path.join(base_path, 'train_masks')\ntest_images = os.path.join(base_path, 'test_images')\n\n# Create datasets\nfull_dataset = ForgeryDataset(train_authentic, train_forged, train_masks, transform=train_transform)\n\ntrain_size = int(0.8 * len(full_dataset))\nval_size = len(full_dataset) - train_size\ntrain_dataset, val_dataset = torch.utils.data.random_split(full_dataset, [train_size, val_size])\nval_dataset.dataset.transform = val_transform\n\ntrain_loader = DataLoader(train_dataset, batch_size=4, shuffle=True, collate_fn=lambda x: tuple(zip(*x)))\nval_loader = DataLoader(val_dataset, batch_size=4, shuffle=False, collate_fn=lambda x: tuple(zip(*x)))\n\nprint(f\"Train samples: {len(train_dataset)}\")\nprint(f\"Val samples: {len(val_dataset)}\")\n\n# Create model\nmodel = create_light_mask_rcnn()\nmodel.to(device)\nprint(f\"Parameters: {sum(p.numel() for p in model.parameters()):,}\")\n\n# Training setup\noptimizer = torch.optim.Adam(model.parameters(), lr=0.001, weight_decay=0.001)\nscheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.1)\n\n# Train\nnum_epochs = 3\nfor epoch in range(num_epochs):\n print(f\"\\nEpoch {epoch+1}/{num_epochs}\")\n train_loss = train_epoch(model, train_loader, optimizer, device)\n val_loss = validate_epoch(model, val_loader, device)\n scheduler.step()\n print(f\"Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}\")\n\n# Generate predictions\npredictions = predict_test_images(model, test_images, device, confidence_threshold=0.5)\n\n# Create submission\nsample_submission = pd.read_csv(f'{base_path}/sample_submission.csv')\nsubmission_data = []\n\nfor case_id in sample_submission['case_id']:\n case_id_str = str(case_id)\n if case_id_str in predictions:\n submission_data.append({'case_id': case_id, 'annotation': predictions[case_id_str]})\n else:\n submission_data.append({'case_id': case_id, 'annotation': 'authentic'})\n\nsubmission = pd.DataFrame(submission_data)\nsubmission.to_csv('submission.csv', index=False)\n\nprint(f\"\\nāœ… Submission created!\")\nprint(f\"Authentic: {(submission['annotation'] == 'authentic').sum()}\")\nprint(f\"Forged: {len(submission) - (submission['annotation'] == 'authentic').sum()}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-14T06:49:37.815827Z","iopub.execute_input":"2025-11-14T06:49:37.816933Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}