{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.11.13"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":14148141,"sourceType":"datasetVersion","datasetId":9016861},{"sourceId":4537,"sourceType":"modelInstanceVersion","modelInstanceId":3329,"modelId":986}],"dockerImageVersionId":31192,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os, cv2, json, math\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom transformers import AutoImageProcessor, AutoModel","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== CONFIG ====================\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# Kaggle paths\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\"\n\n# Path to your uploaded model weights dataset\n# Change this to match your Kaggle dataset name!\nWEIGHTS_PATH = \"/kaggle/input/dino-giant-1/model_seg_best.pt\"  # UPDATE THIS!\n\n# Model config (must match training)\n# Use giant model from Kaggle models hub\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/giant/1\"  # Changed to giant!\nIMG_SIZE = 512\nCHANNELS = 4\n\nOUT_PATH = \"submission.csv\"\n\nprint(f\"Device: {device}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== MODEL DEFINITION ====================\n\nclass DinoDecoder(nn.Module):\n    \"\"\"Progressive upsampling decoder with regularization - supports any DINOv2 variant\"\"\"\n    def __init__(self, in_ch=768, out_ch=CHANNELS, dropout=0.1):\n        super().__init__()\n        \n        # Scale intermediate channels based on input (works for base=768, giant=1536, etc.)\n        ch1 = in_ch // 2      # 768->384, 1536->768\n        ch2 = ch1 // 2        # 384->192, 768->384\n        ch3 = ch2 // 2        # 192->96,  384->192\n        ch4 = ch3 // 2        # 96->48,   192->96\n        \n        # Conv blocks for each upsampling stage\n        self.up1 = self._block(in_ch, ch1, dropout)\n        self.up2 = self._block(ch1, ch2, dropout)\n        self.up3 = self._block(ch2, ch3, dropout)\n        self.up4 = self._block(ch3, ch4, dropout)\n        \n        self.final = nn.Conv2d(ch4, out_ch, 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),\n            nn.BatchNorm2d(out_ch),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(dropout),\n            nn.Conv2d(out_ch, out_ch, 3, padding=1),\n            nn.BatchNorm2d(out_ch),\n            nn.ReLU(inplace=True),\n        )\n    \n    def forward(self, f, size):\n        # Progressive upsampling\n        x = F.interpolate(f, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up1(x)\n        \n        x = F.interpolate(x, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up2(x)\n        \n        x = F.interpolate(x, scale_factor=2, mode='bilinear', align_corners=False)\n        x = self.up3(x)\n        \n        # Chunked interpolation to avoid INT_MAX limit with large batches\n        B = x.shape[0]\n        max_chunk = 64\n        if B > max_chunk:\n            chunks = []\n            for i in range(0, B, max_chunk):\n                chunk = x[i:i+max_chunk]\n                chunk = F.interpolate(chunk, size=size, mode='bilinear', align_corners=False)\n                chunk = self.up4(chunk)\n                chunks.append(self.final(chunk))\n            return torch.cat(chunks, dim=0)\n        else:\n            x = F.interpolate(x, size=size, mode='bilinear', align_corners=False)\n            x = self.up4(x)\n            return self.final(x)\n\n\nclass DinoSegmenter(nn.Module):\n    def __init__(self, encoder, processor, unfreeze_blocks=3):\n        super().__init__()\n        self.encoder, self.processor = encoder, processor\n        \n        # Freeze all parameters\n        for p in self.encoder.parameters():\n            p.requires_grad = False\n        \n        # Unfreeze last N blocks\n        num_blocks = len(self.encoder.encoder.layer)\n        for i in range(num_blocks - unfreeze_blocks, num_blocks):\n            for p in self.encoder.encoder.layer[i].parameters():\n                p.requires_grad = True\n        \n        for p in self.encoder.layernorm.parameters():\n            p.requires_grad = True\n        \n        # Get hidden size from encoder config (768 for base, 1536 for giant, etc.)\n        hidden_size = self.encoder.config.hidden_size\n        self.seg_head = DinoDecoder(hidden_size, CHANNELS)\n\n    def forward_features(self, x):\n        # GPU-native preprocessing (bypasses slow CPU processor)\n        if not hasattr(self, '_mean'):\n            # Cache the normalization tensors (from processor or ImageNet defaults)\n            mean = getattr(self.processor, 'image_mean', [0.485, 0.456, 0.406])\n            std = getattr(self.processor, 'image_std', [0.229, 0.224, 0.225])\n            self.register_buffer('_mean', torch.tensor(mean).view(1, 3, 1, 1))\n            self.register_buffer('_std', torch.tensor(std).view(1, 3, 1, 1))\n            # Get target size from processor (DINOv2 uses 518)\n            self._target_size = getattr(self.processor, 'size', {}).get('shortest_edge', 518)\n        \n        x_norm = (x - self._mean.to(x.device)) / self._std.to(x.device)\n        \n        # Resize to processor's expected size\n        if x_norm.shape[-1] != self._target_size:\n            x_norm = F.interpolate(x_norm, size=(self._target_size, self._target_size), mode='bilinear', align_corners=False)\n        \n        feats = self.encoder(pixel_values=x_norm).last_hidden_state\n        B, N, C = feats.shape\n        fmap = feats[:, 1:, :].permute(0, 2, 1)\n        s = int(math.sqrt(N-1))\n        fmap = fmap.reshape(B, C, s, s)\n        return fmap\n\n    def forward(self, x):\n        \"\"\"Main forward pass\"\"\"\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap, (IMG_SIZE, IMG_SIZE))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== LOAD MODEL ====================\n\nprint(\"Loading DINOv2-Giant encoder...\")\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH)\nencoder = AutoModel.from_pretrained(DINO_PATH).eval().to(device)\n\nprint(f\"Encoder hidden size: {encoder.config.hidden_size}\")\n\nprint(\"Building model...\")\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n\nprint(f\"Loading weights from {WEIGHTS_PATH}...\")\nmodel_seg.load_state_dict(torch.load(WEIGHTS_PATH, map_location=device))\nmodel_seg.eval()\n\nprint(\"Model loaded successfully!\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== INFERENCE FUNCTIONS ====================\n\n@torch.no_grad()\ndef segment_prob_map_all_channels(pil):\n    \"\"\"Returns probability maps for ALL channels.\"\"\"\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    return torch.sigmoid(model_seg(x))[0].cpu().numpy()  # Uses forward() now\n\n\ndef enhanced_adaptive_mask(prob, alpha_grad=0.35):\n    gx = cv2.Sobel(prob, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(prob, cv2.CV_32F, 0, 1, ksize=3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_mag / (grad_mag.max() + 1e-6)\n    enhanced = (1 - alpha_grad) * prob + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (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\n\ndef finalize_mask(prob, orig_size):\n    mask, thr = enhanced_adaptive_mask(prob)\n    mask = cv2.resize(mask, orig_size, interpolation=cv2.INTER_NEAREST)\n    return mask, thr\n\n\ndef pipeline_final(pil):\n    \"\"\"Returns a LIST of masks (one per detected forged region).\"\"\"\n    probs = segment_prob_map_all_channels(pil)\n    \n    all_masks = []\n    all_areas = []\n    all_means = []\n    all_thrs = []\n    \n    for ch in range(probs.shape[0]):\n        prob = probs[ch]\n        mask, thr = finalize_mask(prob, pil.size)\n        area = int(mask.sum())\n        \n        if area > 0:\n            prob_resized = cv2.resize(prob, pil.size, interpolation=cv2.INTER_LINEAR)\n            mean_inside = float(prob_resized[mask == 1].mean())\n        else:\n            mean_inside = 0.0\n        \n        # Filter out small/weak detections\n        if area >= 400 and mean_inside >= 0.35:\n            all_masks.append(mask)\n            all_areas.append(area)\n            all_means.append(mean_inside)\n            all_thrs.append(thr)\n    \n    if len(all_masks) == 0:\n        return \"authentic\", [], {\"area\": 0, \"mean_inside\": 0.0, \"thr\": 0.0}\n    \n    total_area = sum(all_areas)\n    avg_mean = sum(all_means) / len(all_means)\n    avg_thr = sum(all_thrs) / len(all_thrs)\n    \n    return \"forged\", all_masks, {\"area\": total_area, \"mean_inside\": avg_mean, \"thr\": avg_thr, \"num_masks\": len(all_masks)}","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== RLE ENCODING ====================\n\ndef rle_encode_single(mask: np.ndarray, fg_val: int = 1) -> str:\n    \"\"\"Encode a single 2D mask to RLE JSON string.\"\"\"\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == fg_val)[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\n\ndef rle_encode_multi(masks: list, fg_val: int = 1) -> str:\n    \"\"\"Encode multiple masks, joining with semicolons.\"\"\"\n    encoded = []\n    for m in masks:\n        enc = rle_encode_single((m > 0).astype(np.uint8), fg_val)\n        if enc is not None:\n            encoded.append(enc)\n    return ';'.join(encoded) if encoded else \"authentic\"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ==================== GENERATE SUBMISSION ====================\n\nrows = []\ntest_files = sorted(os.listdir(TEST_DIR))\nprint(f\"Processing {len(test_files)} test images...\")\n\nfor f in tqdm(test_files, desc=\"Inference\"):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, masks, dbg = pipeline_final(pil)\n\n    if label == \"authentic\" or len(masks) == 0:\n        annot = \"authentic\"\n    else:\n        annot = rle_encode_multi(masks)\n\n    rows.append({\n        \"case_id\": Path(f).stem,\n        \"annotation\": annot,\n    })\n\n# Create submission DataFrame\nsub = pd.DataFrame(rows)\n\n# Merge with sample submission to ensure correct order\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\")\n\n# Save\nfinal[[\"case_id\", \"annotation\"]].to_csv(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Saved submission to: {OUT_PATH}\")\nprint(f\"Total rows: {len(final)}\")\nprint(f\"Forged: {(final['annotation'] != 'authentic').sum()}\")\nprint(f\"Authentic: {(final['annotation'] == 'authentic').sum()}\")\nprint(\"\\nFirst 10 rows:\")\nprint(final.head(10))","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}