{"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":"gpu","dataSources":[{"sourceId":113558,"databundleVersionId":14878066,"sourceType":"competition"},{"sourceId":14228426,"sourceType":"datasetVersion","datasetId":9077134},{"sourceId":4534,"sourceType":"modelInstanceVersion","modelInstanceId":3326,"modelId":986}],"dockerImageVersionId":31192,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Submission Notebook - DINOv2 Forgery Detection\n\nThis notebook loads pre-trained weights and generates predictions for the test set.","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom PIL import Image\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom transformers import AutoModel\n\n# ==================== CONFIG ====================\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\nTEST_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test_images\"\nSAMPLE_SUB = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/sample_submission.csv\"\nWEIGHTS_PATH = \"/kaggle/input/recod-1219/best_model_1219.pt\"  # UPDATE THIS\n\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\nIMG_SIZE = 512\nCHANNELS = 4\nUNFREEZE_BLOCKS = 4\nDECODER_DROPOUT = 0.15\n\nIMAGENET_MEAN = (0.485, 0.456, 0.406)\nIMAGENET_STD = (0.229, 0.224, 0.225)\n\n# ==================== MODEL ====================\nclass DinoDecoder(nn.Module):\n    def __init__(self, in_channels=768, out_channels=4, dropout=0.1):\n        super().__init__()\n        self.up1 = self._block(in_channels, 384, dropout)\n        self.up2 = self._block(384, 192, dropout)\n        self.up3 = self._block(192, 96, dropout)\n        self.up4 = self._block(96, 48, dropout)\n        self.final = nn.Conv2d(48, out_channels, kernel_size=1)\n    \n    def _block(self, in_ch, out_ch, dropout):\n        return nn.Sequential(\n            nn.Conv2d(in_ch, out_ch, 3, padding=1), nn.BatchNorm2d(out_ch), nn.ReLU(inplace=True),\n            nn.Dropout2d(dropout), nn.Conv2d(out_ch, out_ch, 3, padding=1), nn.BatchNorm2d(out_ch), nn.ReLU(inplace=True),\n        )\n    \n    def forward(self, features, target_size):\n        x = F.interpolate(features, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up1(x)\n        x = F.interpolate(x, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up2(x)\n        x = F.interpolate(x, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up3(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        x = self.up4(x)\n        return self.final(x)\n\nclass DinoSegmenter(nn.Module):\n    def __init__(self, backbone=\"facebook/dinov2-base\", out_channels=4, unfreeze_blocks=3, decoder_dropout=0.1):\n        super().__init__()\n        self.encoder = AutoModel.from_pretrained(backbone)\n        hidden_size = self.encoder.config.hidden_size\n        self.register_buffer('pixel_mean', torch.tensor(IMAGENET_MEAN).view(1, 3, 1, 1), persistent=False)\n        self.register_buffer('pixel_std', torch.tensor(IMAGENET_STD).view(1, 3, 1, 1), persistent=False)\n        for param in self.encoder.parameters():\n            param.requires_grad = False\n        num_blocks = len(self.encoder.encoder.layer)\n        for i in range(num_blocks - unfreeze_blocks, num_blocks):\n            for param in self.encoder.encoder.layer[i].parameters():\n                param.requires_grad = True\n        for param in self.encoder.layernorm.parameters():\n            param.requires_grad = True\n        self.decoder = DinoDecoder(hidden_size, out_channels, decoder_dropout)\n    \n    def forward(self, x):\n        x = (x - self.pixel_mean) / self.pixel_std\n        feats = self.encoder(pixel_values=x).last_hidden_state\n        B, N, C = feats.shape\n        fmap = feats[:, 1:, :].permute(0, 2, 1).reshape(B, C, int(math.sqrt(N-1)), int(math.sqrt(N-1)))\n        return self.decoder(fmap, (x.shape[2], x.shape[3]))\n\n# ==================== LOAD MODEL ====================\nprint(\"Loading model...\")\nmodel = DinoSegmenter(DINO_PATH, CHANNELS, UNFREEZE_BLOCKS, DECODER_DROPOUT).to(device)\ncheckpoint = torch.load(WEIGHTS_PATH, map_location=device, weights_only=False)\nstate_dict = checkpoint['model_state_dict'] if 'model_state_dict' in checkpoint else checkpoint\nmodel.load_state_dict(state_dict)\nmodel.eval()\nprint(\"✅ Model loaded\")\n\n# ==================== INFERENCE ====================\n@torch.no_grad()\ndef segment_prob_map(pil):\n    img = pil.resize((IMG_SIZE, IMG_SIZE))\n    x = torch.from_numpy(np.array(img, np.float32) / 255.).permute(2, 0, 1)[None].to(device)\n    return torch.sigmoid(model(x))[0].cpu().numpy()\n\ndef enhanced_adaptive_mask(prob, alpha_grad=0.35):\n    gx, gy = cv2.Sobel(prob, cv2.CV_32F, 1, 0, ksize=3), cv2.Sobel(prob, cv2.CV_32F, 0, 1, ksize=3)\n    grad_norm = np.sqrt(gx**2 + gy**2) / (np.sqrt(gx**2 + gy**2).max() + 1e-6)\n    enhanced = cv2.GaussianBlur((1 - alpha_grad) * prob + alpha_grad * grad_norm, (3, 3), 0)\n    thr = np.mean(enhanced) + 0.3 * np.std(enhanced)\n    mask = (enhanced > thr).astype(np.uint8)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))\n    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((3, 3), np.uint8))\n    return mask, thr\n\ndef pipeline_final(pil):\n    probs = segment_prob_map(pil)\n    all_masks = []\n    for ch in range(probs.shape[0]):\n        mask, _ = enhanced_adaptive_mask(probs[ch])\n        mask = cv2.resize(mask, pil.size, interpolation=cv2.INTER_NEAREST)\n        area = int(mask.sum())\n        if area > 0:\n            prob_resized = cv2.resize(probs[ch], pil.size, interpolation=cv2.INTER_LINEAR)\n            mean_inside = float(prob_resized[mask == 1].mean())\n        else:\n            mean_inside = 0.0\n        if area >= 400 and mean_inside >= 0.35:\n            all_masks.append(mask)\n    return all_masks\n\n# ==================== RLE ENCODING ====================\ndef rle_encode_single(mask):\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == 1)[0]\n    if len(dots) == 0:\n        return None\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    return json.dumps([int(x) for x in run_lengths])\n\ndef rle_encode_multi(masks):\n    encoded = [rle_encode_single((m > 0).astype(np.uint8)) for m in masks]\n    encoded = [e for e in encoded if e is not None]\n    return ';'.join(encoded) if encoded else \"authentic\"\n\n# ==================== GENERATE SUBMISSION ====================\nrows = []\ntest_files = sorted(os.listdir(TEST_DIR))\nprint(f\"Processing {len(test_files)} test images...\")\n\nfor f in test_files:\n    pil = Image.open(Path(TEST_DIR) / f).convert(\"RGB\")\n    masks = pipeline_final(pil)\n    annot = rle_encode_multi(masks) if masks else \"authentic\"\n    rows.append({\"case_id\": Path(f).stem, \"annotation\": annot})\n\nsub = pd.DataFrame(rows)\nss = pd.read_csv(SAMPLE_SUB)\nss[\"case_id\"] = ss[\"case_id\"].astype(str)\nsub[\"case_id\"] = sub[\"case_id\"].astype(str)\nfinal = ss[[\"case_id\"]].merge(sub, on=\"case_id\", how=\"left\")\nfinal[\"annotation\"] = final[\"annotation\"].fillna(\"authentic\")\nfinal[[\"case_id\", \"annotation\"]].to_csv(\"/kaggle/working/submission.csv\", index=False)\n\nprint(f\"\\n✅ Saved to /kaggle/working/submission.csv\")\nprint(f\"Total: {len(final)} | Forged: {(final['annotation'] != 'authentic').sum()} | Authentic: {(final['annotation'] == 'authentic').sum()}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-08T03:58:56.216745Z","iopub.execute_input":"2025-12-08T03:58:56.217428Z","iopub.status.idle":"2025-12-08T03:59:22.620715Z","shell.execute_reply.started":"2025-12-08T03:58:56.217396Z","shell.execute_reply":"2025-12-08T03:59:22.619460Z"}},"outputs":[],"execution_count":null}]}