{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.12.12","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":14878066,"sourceType":"competition"},{"sourceId":14407528,"sourceType":"datasetVersion","datasetId":9153851},{"sourceId":4534,"sourceType":"modelInstanceVersion","modelInstanceId":3326,"modelId":986}],"dockerImageVersionId":31236,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true},"papermill":{"default_parameters":{},"duration":63.129007,"end_time":"2025-12-30T04:03:10.929134","environment_variables":{},"exception":null,"input_path":"__notebook__.ipynb","output_path":"__notebook__.ipynb","parameters":{},"start_time":"2025-12-30T04:02:07.800127","version":"2.6.0"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Notebook Overview: CNN-DINOv2 Hybrid\n\nThis notebook demonstrates a hybrid approach for image classification using both Convolutional Neural Networks (CNNs) and DINOv2, a self-supervised vision transformer model. The workflow includes:\n\n- **Data Loading & Preprocessing:** Images are loaded, resized, normalized, and split into training and validation sets.\n- **Feature Extraction:** DINOv2 is used to extract high-level features from images, leveraging its transformer-based architecture for robust representations.\n- **CNN Model Construction:** A custom CNN is built to process image data, learning spatial hierarchies and patterns.\n- **Hybrid Model Integration:** Features from DINOv2 and the CNN are combined, either by concatenation or other fusion techniques, to enhance classification performance.\n- **Training & Evaluation:** The hybrid model is trained on the dataset, with metrics such as accuracy and loss tracked. Validation is performed to assess generalization.\n- **Visualization & Analysis:** Results, including confusion matrices and sample predictions, are visualized to interpret model behavior.\n\nThis approach aims to leverage the strengths of both CNNs (local feature learning) and DINOv2 (global, context-aware representations) for improved image classification results.","metadata":{}},{"cell_type":"markdown","source":"# 释放GPU辅助函数","metadata":{}},{"cell_type":"code","source":"import torch\nimport gc\n\ndef clear():\n    print(\"\\n\" + \"=\"*50)\n    print(\"开始释放GPU资源...\")\n    print(\"=\"*50)\n    \n    # 5. 清空Python垃圾回收\n    gc.collect()\n    print(\"✓ Python garbage collected\")\n    \n    # 6. 清空PyTorch缓存\n    if torch.cuda.is_available():\n        torch.cuda.empty_cache()\n        print(\"✓ CUDA cache cleared\")\n        \n        # 7. 同步CUDA设备\n        torch.cuda.synchronize()\n        print(\"✓ CUDA synchronized\")\n        \n        # 8. 重置CUDA峰值内存统计\n        torch.cuda.reset_peak_memory_stats()\n        torch.cuda.reset_accumulated_memory_stats()\n        print(\"✓ CUDA memory stats reset\")\n        \n        # 9. 显示当前GPU内存使用情况\n        for i in range(torch.cuda.device_count()):\n            allocated = torch.cuda.memory_allocated(i) / 1024**3\n            reserved = torch.cuda.memory_reserved(i) / 1024**3\n            print(f\"  GPU {i}: Allocated: {allocated:.2f} GB, Reserved: {reserved:.2f} GB\")\n    else:\n        print(\"⚠ CUDA not available, skipping CUDA-specific cleanup\")\n    \n    print(\"=\"*50)\n    print(\"GPU资源释放完成!\")\n    print(\"=\"*50)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:47.553014Z","iopub.execute_input":"2026-01-06T14:04:47.553362Z","iopub.status.idle":"2026-01-06T14:04:47.560267Z","shell.execute_reply.started":"2026-01-06T14:04:47.553318Z","shell.execute_reply":"2026-01-06T14:04:47.559514Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#  第一个submission(0.324)","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math, random, torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom sklearn.model_selection import train_test_split\nfrom torch.utils.data import Dataset, DataLoader\nimport torch.nn as nn, torch.nn.functional as F, torch.optim as optim\nfrom transformers import AutoImageProcessor, AutoModel\n\ndef seed_everything(seed=42):\n    random.seed(seed)\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    # This forces CUDA to use deterministic algorithms (slower but consistent)\n    torch.backends.cudnn.deterministic = True \n    torch.backends.cudnn.benchmark = False\n\nseed_everything(42)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nBASE_DIR  = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nAUTH_DIR  = f\"{BASE_DIR}/train_images/authentic\"\nFORG_DIR  = f\"{BASE_DIR}/train_images/forged\"\nMASK_DIR  = f\"{BASE_DIR}/train_masks\"\nTEST_DIR  = f\"{BASE_DIR}/test_images\"\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\n\nIMG_SIZE = 718\nBATCH_SIZE = 2\nMODEL_LOC = '/kaggle/input/cnndinov2-pbd/CNNDINOv2-U52/CNNDINOv2-U52/model_seg_final.pt'  # Set to \"model_seg_final.pt\" to load pretrained weights, None to train from scratch\n# INFERENCE UTILS\nAREA_THR = 200\nMEAN_THR = 0.25\nUSE_TTA = True\n\nclass ForgerySegDataset(Dataset):\n    def __init__(self, auth_paths, forg_paths, mask_dir, img_size=IMG_SIZE):\n        self.samples = []\n        for p in forg_paths:\n            m = os.path.join(mask_dir, Path(p).stem + \".npy\")\n            if os.path.exists(m):\n                self.samples.append((p, m))\n        for p in auth_paths:\n            self.samples.append((p, None))\n        self.img_size = img_size\n    def __len__(self): return len(self.samples)\n    def __getitem__(self, idx):\n        img_path, mask_path = self.samples[idx]\n        img = Image.open(img_path).convert(\"RGB\")\n        w, h = img.size\n        if mask_path is None:\n            mask = np.zeros((h, w), np.uint8)\n        else:\n            m = np.load(mask_path)\n            if m.ndim == 3: m = np.max(m, axis=0)\n            mask = (m > 0).astype(np.uint8)\n        img_r = img.resize((IMG_SIZE, IMG_SIZE))\n        mask_r = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n        img_t = torch.from_numpy(np.array(img_r, np.float32)/255.).permute(2,0,1)\n        mask_t = torch.from_numpy(mask_r[None, ...].astype(np.float32))\n        return img_t, mask_t\n\n\n#  MODEL (DINOv2 + Decoder)\n\nfrom transformers import AutoImageProcessor, AutoModel\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH, local_files_only=True, use_fast=False)\nencoder = AutoModel.from_pretrained(DINO_PATH, local_files_only=True).eval().to(device)\n\nclass DinoTinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        # Block 1: 768 -> 384\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 2: 384 -> 192\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 3: 192 -> 96\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True)\n        )\n        # Final Output: 96 -> 1\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        # f: [B, 768, 37, 37]\n        \n        # Step 1: Up to ~74x74\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        \n        # Step 2: Up to ~148x148\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        \n        # Step 3: Up to ~296x296\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        \n        # Step 4: Final jump to 518x518\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n    \nclass DinoSegmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder, self.processor = encoder, processor\n        for p in self.encoder.parameters(): p.requires_grad = False\n        self.seg_head = DinoTinyDecoder(768,1)\n    def forward_features(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).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    def forward_seg(self,x):\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap,(IMG_SIZE,IMG_SIZE))\n\n\nauth_imgs = sorted([str(Path(AUTH_DIR)/f) for f in os.listdir(AUTH_DIR)])\nforg_imgs = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])\ntrain_auth, val_auth = train_test_split(auth_imgs, test_size=0.2, random_state=42)\ntrain_forg, val_forg = train_test_split(forg_imgs, test_size=0.2, random_state=42)\n\ntrain_loader = DataLoader(ForgerySegDataset(train_auth, train_forg, MASK_DIR),\n                          batch_size=BATCH_SIZE, shuffle=True, num_workers=2)\nval_loader = DataLoader(ForgerySegDataset(val_auth, val_forg, MASK_DIR),\n                        batch_size=BATCH_SIZE, shuffle=False, num_workers=2)\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n\n# Load pretrained weights if MODEL_LOC is specified\nif MODEL_LOC is not None and os.path.exists(MODEL_LOC):\n    model_seg.load_state_dict(torch.load(MODEL_LOC, map_location=device))\n    print(f\"✅ Loaded pretrained model from: {MODEL_LOC}\")\n    model_seg.eval()  # Set model to evaluation mode\n\n@torch.no_grad()\ndef segment_prob_map(pil):\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    prob = torch.sigmoid(model_seg.forward_seg(x))[0,0].cpu().numpy()\n    return prob\n\n@torch.no_grad()\ndef segment_prob_map_with_tta(pil):\n    # 1. Preprocessing: Resize, Normalize, and move to Device\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    \n    predictions = []\n\n    # 2. Original Prediction\n    pred_orig = torch.sigmoid(model_seg.forward_seg(x))\n    predictions.append(pred_orig)\n\n    # 3. Horizontal Flip TTA (dim 3)\n    # Flip input -> Predict -> Flip output back\n    pred_h = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[3])))\n    predictions.append(torch.flip(pred_h, dims=[3]))\n\n    # 4. Vertical Flip TTA (dim 2)\n    # Flip input -> Predict -> Flip output back\n    pred_v = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[2])))\n    predictions.append(torch.flip(pred_v, dims=[2]))\n\n    # 5. Average the predictions and format as numpy\n    # We stack the 3 predictions and take the mean across the stack dimension (0)\n    prob = torch.stack(predictions).mean(0)[0, 0].cpu().numpy()\n\n    return prob\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\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\ndef pipeline_final(pil):\n    if USE_TTA:\n        prob = segment_prob_map_with_tta(pil)\n    else:\n        prob = segment_prob_map(pil)\n    mask, thr = finalize_mask(prob, pil.size)\n    area = int(mask.sum())\n    mean_inside = float(prob[cv2.resize(mask,(IMG_SIZE,IMG_SIZE),interpolation=cv2.INTER_NEAREST)==1].mean()) if area>0 else 0.0\n    if area < AREA_THR or mean_inside < MEAN_THR:\n        return \"authentic\", None, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n    return \"forged\", mask, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n\nfrom sklearn.metrics import f1_score\nval_items = [(p, 1) for p in val_forg[:10]]\nresults = []\nfor p,_ in tqdm(val_items, desc=\"Validation forged-only\"):\n    pil = Image.open(p).convert(\"RGB\")\n    label, m_pred, dbg = pipeline_final(pil)\n    m_gt = np.load(Path(MASK_DIR)/f\"{Path(p).stem}.npy\")\n    if m_gt.ndim==3: m_gt=np.max(m_gt,axis=0)\n    m_gt=(m_gt>0).astype(np.uint8)\n    m_pred=(m_pred>0).astype(np.uint8) if m_pred is not None else np.zeros_like(m_gt)\n    f1 = f1_score(m_gt.flatten(), m_pred.flatten(), zero_division=0)\n    results.append((Path(p).stem, f1, dbg))\nprint(\"\\n F1-score par image falsifiée:\\n\")\nfor cid,f1,dbg in results:\n    print(f\"{cid} — F1={f1:.4f} | area={dbg['area']} mean={dbg['mean_inside']:.3f} thr={dbg['thr']:.3f}\")\nprint(f\"\\n Moyenne F1 (falsifiées) = {np.mean([r[1] for r in results]):.4f}\")","metadata":{"execution":{"iopub.status.busy":"2026-01-06T14:04:47.561586Z","iopub.execute_input":"2026-01-06T14:04:47.561820Z","iopub.status.idle":"2026-01-06T14:04:53.053858Z","shell.execute_reply.started":"2026-01-06T14:04:47.561799Z","shell.execute_reply":"2026-01-06T14:04:53.052936Z"},"papermill":{"duration":51.374583,"end_time":"2025-12-30T04:03:03.449412","exception":false,"start_time":"2025-12-30T04:02:12.074829","status":"completed"},"tags":[],"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os, json, cv2\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\n\n# --- RLE Encoder for Kaggle Submission ---\ndef rle_encode(mask: np.ndarray, fg_val: int = 1) -> str:\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == fg_val)[0]\n    if len(dots) == 0:\n        return \"authentic\"\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# --- 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\"\nOUT_PATH = \"submission1.csv\"\n\nrows = []\nfor f in tqdm(sorted(os.listdir(TEST_DIR)), desc=\"Inference on Test Set\"):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)  # utilise la version améliorée\n\n    # Sécurisation masque\n    if mask is None:\n        mask = np.zeros(pil.size[::-1], np.uint8)\n    else:\n        mask = np.array(mask, dtype=np.uint8)\n\n    # Annotation finale\n    if label == \"authentic\":\n        annot = \"authentic\"\n    else:\n        annot = rle_encode((mask > 0).astype(np.uint8))\n\n    rows.append({\n        \"case_id\": Path(f).stem,\n        \"annotation\": annot,\n        \"area\": int(dbg.get(\"area\", mask.sum())),\n        \"mean\": float(dbg.get(\"mean_inside\", 0.0)),\n        \"thr\": float(dbg.get(\"thr\", 0.0))\n    })\n\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(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Saved submission file: {OUT_PATH}\")\nprint(final.head(10))\n\n\nsample_files = sorted(os.listdir(TEST_DIR))[:5]\nfor f in sample_files:\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)\n    mask = np.array(mask, dtype=np.uint8) if mask is not None else np.zeros(pil.size[::-1], np.uint8)\n\n    print(f\"{'🔴' if label=='forged' else '🟢'} {f}: {label} | area={mask.sum()} mean={dbg.get('mean_inside', 0):.3f}\")\n\n    if label == \"authentic\":\n        plt.figure(figsize=(5,5))\n        plt.imshow(pil)\n        plt.title(f\"{f} — Authentic\")\n        plt.axis(\"off\")\n        plt.show()\n    else:\n        plt.figure(figsize=(10,5))\n        plt.subplot(1,2,1)\n        plt.imshow(pil)\n        plt.title(\"Original Image\")\n        plt.axis(\"off\")\n        plt.subplot(1,2,2)\n        plt.imshow(pil)\n        plt.imshow(mask, alpha=0.45, cmap=\"Reds\")\n        plt.title(f\"Predicted Forged Mask\\nArea={mask.sum()} | Mean={dbg.get('mean_inside', 0):.3f}\")\n        plt.axis(\"off\")\n        plt.show()\n","metadata":{"execution":{"iopub.status.busy":"2026-01-06T14:04:53.055320Z","iopub.execute_input":"2026-01-06T14:04:53.055615Z","iopub.status.idle":"2026-01-06T14:04:53.618806Z","shell.execute_reply.started":"2026-01-06T14:04:53.055590Z","shell.execute_reply":"2026-01-06T14:04:53.617950Z"},"papermill":{"duration":0.706627,"end_time":"2025-12-30T04:03:04.172732","exception":false,"start_time":"2025-12-30T04:03:03.466105","status":"completed"},"tags":[],"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"clear()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:53.619838Z","iopub.execute_input":"2026-01-06T14:04:53.620164Z","iopub.status.idle":"2026-01-06T14:04:54.000287Z","shell.execute_reply.started":"2026-01-06T14:04:53.620131Z","shell.execute_reply":"2026-01-06T14:04:53.999458Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 第二个submission(0.323)","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math, random, torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom sklearn.model_selection import train_test_split\nfrom torch.utils.data import Dataset, DataLoader\nimport torch.nn as nn, torch.nn.functional as F, torch.optim as optim\nfrom transformers import AutoImageProcessor, AutoModel\n\ndef seed_everything(seed=42):\n    random.seed(seed)\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    # This forces CUDA to use deterministic algorithms (slower but consistent)\n    torch.backends.cudnn.deterministic = True \n    torch.backends.cudnn.benchmark = False\n\nseed_everything(42)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nBASE_DIR  = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nAUTH_DIR  = f\"{BASE_DIR}/train_images/authentic\"\nFORG_DIR  = f\"{BASE_DIR}/train_images/forged\"\nMASK_DIR  = f\"{BASE_DIR}/train_masks\"\nTEST_DIR  = f\"{BASE_DIR}/test_images\"\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\n\nIMG_SIZE = 518\nBATCH_SIZE = 2\nMODEL_LOC = '/kaggle/working/best_model_epoch_5.pth'  # Set to \"model_seg_final.pt\" to load pretrained weights, None to train from scratch\n# INFERENCE UTILS\nAREA_THR = 200\nMEAN_THR = 0.25\nUSE_TTA = True\n\nclass ForgerySegDataset(Dataset):\n    def __init__(self, auth_paths, forg_paths, mask_dir, img_size=IMG_SIZE):\n        self.samples = []\n        for p in forg_paths:\n            m = os.path.join(mask_dir, Path(p).stem + \".npy\")\n            if os.path.exists(m):\n                self.samples.append((p, m))\n        for p in auth_paths:\n            self.samples.append((p, None))\n        self.img_size = img_size\n    def __len__(self): return len(self.samples)\n    def __getitem__(self, idx):\n        img_path, mask_path = self.samples[idx]\n        img = Image.open(img_path).convert(\"RGB\")\n        w, h = img.size\n        if mask_path is None:\n            mask = np.zeros((h, w), np.uint8)\n        else:\n            m = np.load(mask_path)\n            if m.ndim == 3: m = np.max(m, axis=0)\n            mask = (m > 0).astype(np.uint8)\n        img_r = img.resize((IMG_SIZE, IMG_SIZE))\n        mask_r = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n        img_t = torch.from_numpy(np.array(img_r, np.float32)/255.).permute(2,0,1)\n        mask_t = torch.from_numpy(mask_r[None, ...].astype(np.float32))\n        return img_t, mask_t\n\n\n#  MODEL (DINOv2 + Decoder)\n\nfrom transformers import AutoImageProcessor, AutoModel\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH, local_files_only=True, use_fast=False)\nencoder = AutoModel.from_pretrained(DINO_PATH, local_files_only=True).eval().to(device)\n\nclass DinoTinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        # Block 1: 768 -> 384\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 2: 384 -> 192\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 3: 192 -> 96\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True)\n        )\n        # Final Output: 96 -> 1\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        # f: [B, 768, 37, 37]\n        \n        # Step 1: Up to ~74x74\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        \n        # Step 2: Up to ~148x148\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        \n        # Step 3: Up to ~296x296\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        \n        # Step 4: Final jump to 518x518\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n    \nclass DinoSegmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder, self.processor = encoder, processor\n        for p in self.encoder.parameters(): p.requires_grad = False\n        self.seg_head = DinoTinyDecoder(768,1)\n    def forward_features(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).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    def forward_seg(self,x):\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap,(IMG_SIZE,IMG_SIZE))\n\n\nauth_imgs = sorted([str(Path(AUTH_DIR)/f) for f in os.listdir(AUTH_DIR)])\nforg_imgs = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])\ntrain_auth, val_auth = train_test_split(auth_imgs, test_size=0.2, random_state=42)\ntrain_forg, val_forg = train_test_split(forg_imgs, test_size=0.2, random_state=42)\n\ntrain_loader = DataLoader(ForgerySegDataset(train_auth, train_forg, MASK_DIR),\n                          batch_size=BATCH_SIZE, shuffle=True, num_workers=2)\nval_loader = DataLoader(ForgerySegDataset(val_auth, val_forg, MASK_DIR),\n                        batch_size=BATCH_SIZE, shuffle=False, num_workers=2)\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n# # ====== 训练配置 ======\n# TRAIN = True                # 本次运行是否训练。训练完推理时改为 False\n# EPOCHS = 5                  # 训练轮数，先设小点（如2）测试流程，成功后再增加\n# LEARNING_RATE = 1e-4\n\n# # ====== Kaggle 路径详解（关键！）=====\n# # Kaggle 将所有“只读”输入数据（如比赛数据、你添加的数据集）都放在 /kaggle/input/ 下\n# # 因此，训练图像和掩码的路径固定如下：\n# TRAIN_IMAGES_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train_images\"\n# TRAIN_MASKS_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train_masks\"\n\n# # Kaggle 规定，所有“生成”的文件（模型、日志、提交文件）必须保存在 /kaggle/working/ 下\n# # 这是唯一有写入权限的目录，文件会永久保存在你的Notebook输出中。\n# MODEL_SAVE_DIR = \"/kaggle/working\"\n\n# # ====== 训练核心组件定义 ======\n# import torch.nn as nn\n# import torch.optim as optim\n# from tqdm import tqdm\n# import os\n\n# # 1. 损失函数：适用于像素级二分类分割\n# criterion = nn.BCEWithLogitsLoss()\n# # 2. 优化器：只优化解码器（seg_head），因为编码器被冻结了\n# optimizer = optim.Adam(model_seg.seg_head.parameters(), lr=LEARNING_RATE, weight_decay=1e-5)\n# # 3. 学习率调度器（可选）\n# scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=2, gamma=0.5)\n\n# print(f\"训练组件初始化完成。\")\n\n# # ====== 训练循环主流程 ======\n# if TRAIN:\n#     print(f\"\\n{'='*60}\")\n#     print(f\"开始训练！共 {EPOCHS} 个周期\")\n#     print(f\"训练集: {len(train_loader.dataset)} 张， 验证集: {len(val_loader.dataset)} 张\")\n#     print(f\"{'='*60}\")\n    \n#     best_val_loss = float('inf')\n    \n#     for epoch in range(EPOCHS):\n#         # ---------- 训练阶段 ----------\n#         model_seg.train()\n#         running_train_loss = 0.0\n#         train_loop = tqdm(train_loader, desc=f'周期 [{epoch+1}/{EPOCHS}] 训练')\n#         for images, masks in train_loop:\n#             images, masks = images.to(device), masks.to(device)\n            \n#             optimizer.zero_grad()\n#             outputs = model_seg.forward_seg(images)\n#             loss = criterion(outputs, masks)\n#             loss.backward()\n#             optimizer.step()\n            \n#             running_train_loss += loss.item() * images.size(0)\n#             train_loop.set_postfix(loss=loss.item())\n        \n#         avg_train_loss = running_train_loss / len(train_loader.dataset)\n        \n#         # ---------- 验证阶段 ----------\n#         model_seg.eval()\n#         running_val_loss = 0.0\n#         with torch.no_grad():\n#             val_loop = tqdm(val_loader, desc=f'周期 [{epoch+1}/{EPOCHS}] 验证')\n#             for images, masks in val_loop:\n#                 images, masks = images.to(device), masks.to(device)\n#                 outputs = model_seg.forward_seg(images)\n#                 loss = criterion(outputs, masks)\n#                 running_val_loss += loss.item() * images.size(0)\n#                 val_loop.set_postfix(loss=loss.item())\n        \n#         avg_val_loss = running_val_loss / len(val_loader.dataset)\n#         scheduler.step()\n        \n#         # ---------- 打印日志 ----------\n#         current_lr = scheduler.get_last_lr()[0]\n#         print(f\"周期 {epoch+1:03d}/{EPOCHS} | 训练损失: {avg_train_loss:.4f} | 验证损失: {avg_val_loss:.4f} | 学习率: {current_lr:.6f}\")\n        \n#         # ---------- 保存最佳模型 ----------\n#         if avg_val_loss < best_val_loss:\n#             best_val_loss = avg_val_loss\n#             best_model_path = os.path.join(MODEL_SAVE_DIR, f'best_model_epoch_{epoch+1}.pth')\n#             torch.save(model_seg.state_dict(), best_model_path)\n#             print(f\"      → 发现更优模型，已保存至: {best_model_path}\")\n    \n#     # 训练结束，保存最终模型\n#     final_model_path = os.path.join(MODEL_SAVE_DIR, 'model_seg_final_trained.pth')\n#     torch.save(model_seg.state_dict(), final_model_path)\n#     print(f\"{'='*60}\")\n#     print(f\"训练完成！最终模型已保存至: {final_model_path}\")\n#     print(f\"{'='*60}\\n\")\n    \n#     # 训练结束后，加载最佳模型进行评估\n# model_seg.load_state_dict(torch.load('/kaggle/working/best_model_epoch_5.pth', map_location=device))\n# model_seg.eval()\n# print(\"已加载训练好的最佳模型，准备进行验证和推理。\")\n\n# Load pretrained weights if MODEL_LOC is specified\nif MODEL_LOC is not None and os.path.exists(MODEL_LOC):\n    model_seg.load_state_dict(torch.load(MODEL_LOC, map_location=device))\n    print(f\"✅ Loaded pretrained model from: {MODEL_LOC}\")\n    model_seg.eval()  # Set model to evaluation mode\n\n@torch.no_grad()\ndef segment_prob_map(pil):\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    prob = torch.sigmoid(model_seg.forward_seg(x))[0,0].cpu().numpy()\n    return prob\n\n@torch.no_grad()\ndef segment_prob_map_with_tta(pil):\n    # 1. Preprocessing: Resize, Normalize, and move to Device\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    \n    predictions = []\n\n    # 2. Original Prediction\n    pred_orig = torch.sigmoid(model_seg.forward_seg(x))\n    predictions.append(pred_orig)\n\n    # 3. Horizontal Flip TTA (dim 3)\n    # Flip input -> Predict -> Flip output back\n    pred_h = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[3])))\n    predictions.append(torch.flip(pred_h, dims=[3]))\n\n    # 4. Vertical Flip TTA (dim 2)\n    # Flip input -> Predict -> Flip output back\n    pred_v = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[2])))\n    predictions.append(torch.flip(pred_v, dims=[2]))\n\n    # 5. Average the predictions and format as numpy\n    # We stack the 3 predictions and take the mean across the stack dimension (0)\n    prob = torch.stack(predictions).mean(0)[0, 0].cpu().numpy()\n\n    return prob\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\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\ndef pipeline_final(pil):\n    if USE_TTA:\n        prob = segment_prob_map_with_tta(pil)\n    else:\n        prob = segment_prob_map(pil)\n    mask, thr = finalize_mask(prob, pil.size)\n    area = int(mask.sum())\n    mean_inside = float(prob[cv2.resize(mask,(IMG_SIZE,IMG_SIZE),interpolation=cv2.INTER_NEAREST)==1].mean()) if area>0 else 0.0\n    if area < AREA_THR or mean_inside < MEAN_THR:\n        return \"authentic\", None, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n    return \"forged\", mask, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n\nfrom sklearn.metrics import f1_score\nval_items = [(p, 1) for p in val_forg[:10]]\nresults = []\nfor p,_ in tqdm(val_items, desc=\"Validation forged-only\"):\n    pil = Image.open(p).convert(\"RGB\")\n    label, m_pred, dbg = pipeline_final(pil)\n    m_gt = np.load(Path(MASK_DIR)/f\"{Path(p).stem}.npy\")\n    if m_gt.ndim==3: m_gt=np.max(m_gt,axis=0)\n    m_gt=(m_gt>0).astype(np.uint8)\n    m_pred=(m_pred>0).astype(np.uint8) if m_pred is not None else np.zeros_like(m_gt)\n    f1 = f1_score(m_gt.flatten(), m_pred.flatten(), zero_division=0)\n    results.append((Path(p).stem, f1, dbg))\nprint(\"\\n F1-score par image falsifiée:\\n\")\nfor cid,f1,dbg in results:\n    print(f\"{cid} — F1={f1:.4f} | area={dbg['area']} mean={dbg['mean_inside']:.3f} thr={dbg['thr']:.3f}\")\nprint(f\"\\n Moyenne F1 (falsifiées) = {np.mean([r[1] for r in results]):.4f}\")\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:54.001948Z","iopub.execute_input":"2026-01-06T14:04:54.002193Z","iopub.status.idle":"2026-01-06T14:04:57.692602Z","shell.execute_reply.started":"2026-01-06T14:04:54.002169Z","shell.execute_reply":"2026-01-06T14:04:57.691539Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\nimport os, json, cv2\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\n\n# --- RLE Encoder for Kaggle Submission ---\ndef rle_encode(mask: np.ndarray, fg_val: int = 1) -> str:\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == fg_val)[0]\n    if len(dots) == 0:\n        return \"authentic\"\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# --- 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\"\nOUT_PATH = \"submission2.csv\"\n\nrows = []\nfor f in tqdm(sorted(os.listdir(TEST_DIR)), desc=\"Inference on Test Set\"):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)  # utilise la version améliorée\n\n    # Sécurisation masque\n    if mask is None:\n        mask = np.zeros(pil.size[::-1], np.uint8)\n    else:\n        mask = np.array(mask, dtype=np.uint8)\n\n    # Annotation finale\n    if label == \"authentic\":\n        annot = \"authentic\"\n    else:\n        annot = rle_encode((mask > 0).astype(np.uint8))\n\n    rows.append({\n        \"case_id\": Path(f).stem,\n        \"annotation\": annot,\n        \"area\": int(dbg.get(\"area\", mask.sum())),\n        \"mean\": float(dbg.get(\"mean_inside\", 0.0)),\n        \"thr\": float(dbg.get(\"thr\", 0.0))\n    })\n\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(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Saved submission file: {OUT_PATH}\")\nprint(final.head(10))\n\n\nsample_files = sorted(os.listdir(TEST_DIR))[:5]\nfor f in sample_files:\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)\n    mask = np.array(mask, dtype=np.uint8) if mask is not None else np.zeros(pil.size[::-1], np.uint8)\n\n    print(f\"{'🔴' if label=='forged' else '🟢'} {f}: {label} | area={mask.sum()} mean={dbg.get('mean_inside', 0):.3f}\")\n\n    if label == \"authentic\":\n        plt.figure(figsize=(5,5))\n        plt.imshow(pil)\n        plt.title(f\"{f} — Authentic\")\n        plt.axis(\"off\")\n        plt.show()\n    else:\n        plt.figure(figsize=(10,5))\n        plt.subplot(1,2,1)\n        plt.imshow(pil)\n        plt.title(\"Original Image\")\n        plt.axis(\"off\")\n        plt.subplot(1,2,2)\n        plt.imshow(pil)\n        plt.imshow(mask, alpha=0.45, cmap=\"Reds\")\n        plt.title(f\"Predicted Forged Mask\\nArea={mask.sum()} | Mean={dbg.get('mean_inside', 0):.3f}\")\n        plt.axis(\"off\")\n        plt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:57.693952Z","iopub.execute_input":"2026-01-06T14:04:57.694387Z","iopub.status.idle":"2026-01-06T14:04:58.863958Z","shell.execute_reply.started":"2026-01-06T14:04:57.694334Z","shell.execute_reply":"2026-01-06T14:04:58.863158Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"clear()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:58.864970Z","iopub.execute_input":"2026-01-06T14:04:58.865319Z","iopub.status.idle":"2026-01-06T14:04:59.219649Z","shell.execute_reply.started":"2026-01-06T14:04:58.865295Z","shell.execute_reply":"2026-01-06T14:04:59.218918Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 第三个submission(0.322)","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math, random, torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport torch.nn as nn, torch.nn.functional as F\nfrom transformers import AutoImageProcessor, AutoModel\n\n# --- 1. Global Setup & Seeding ---\ndef seed_everything(seed=42):\n    random.seed(seed)\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    torch.backends.cudnn.deterministic = True \n    torch.backends.cudnn.benchmark = False\n\nseed_everything(42)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nBASE_DIR  = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\nMODEL_LOC = '/kaggle/input/cnndinov2-pbd/CNNDINOv2-R69/model_seg_final.pt' \n\n# --- 2. Precision Hyperparameters (Target: 0.323) ---\nIMG_SIZE = 518\nAREA_THR = 150        # Bumped back up slightly to reduce noise (was 120)\nMEAN_THR = 0.18       # Keep sensitive\nSTD_COEFF = 0.25      # Keep tight\n\n# --- 3. Model Architecture (Must match saved weights) ---\nclass DinoTinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.block1 = nn.Sequential(nn.Conv2d(in_ch, 384, 3, 1, 1), nn.ReLU(True), nn.Dropout2d(0.1))\n        self.block2 = nn.Sequential(nn.Conv2d(384, 192, 3, 1, 1), nn.ReLU(True), nn.Dropout2d(0.1))\n        self.block3 = nn.Sequential(nn.Conv2d(192, 96, 3, 1, 1), nn.ReLU(True))\n        self.conv_out = nn.Conv2d(96, out_ch, 1)\n\n    def forward(self, f, target_size):\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        x = self.conv_out(x)\n        return F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n\nclass DinoSegmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder, self.processor = encoder, processor\n        for p in self.encoder.parameters(): p.requires_grad = False\n        self.seg_head = DinoTinyDecoder(768, 1)\n        \n    def forward_features(self, x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        feats = self.encoder(**inputs).last_hidden_state\n        B, N, C = feats.shape\n        s = int(math.sqrt(N-1))\n        return feats[:,1:,:].permute(0,2,1).reshape(B, C, s, s)\n\n    def forward_seg(self, x):\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap, (IMG_SIZE, IMG_SIZE))\n\n# --- 4. Load Model ---\nprint(\"⏳ Loading Model...\")\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH, local_files_only=True, use_fast=False)\nencoder = AutoModel.from_pretrained(DINO_PATH, local_files_only=True).eval().to(device)\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n\nif os.path.exists(MODEL_LOC):\n    model_seg.load_state_dict(torch.load(MODEL_LOC, map_location=device))\n    print(f\"✅ Loaded weights from: {MODEL_LOC}\")\nmodel_seg.eval()\n\n# --- 5. Optimized Multi-Scale TTA Pipeline ---\n@torch.no_grad()\ndef segment_prob_map_multiscale(pil):\n    \"\"\"Run inference at 1.0x (Original & Flip) and 0.75x (Zoom Out).\"\"\"\n    # Standard Input\n    img_tensor = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    \n    # 1. Scale 1.0x Original\n    prob_1 = torch.sigmoid(model_seg.forward_seg(img_tensor))[0,0]\n    \n    # 2. Scale 1.0x Flip\n    img_flip = torch.flip(img_tensor, [3])\n    prob_1_flip = torch.sigmoid(model_seg.forward_seg(img_flip))[0,0]\n    prob_1_flip = torch.flip(prob_1_flip, [1]) \n    \n    # 3. Scale 0.75x (Zoom Out) -> helps detect larger context anomalies\n    size_small = (392, 392) # approx 518 * 0.75\n    img_small = F.interpolate(img_tensor, size=size_small, mode='bilinear', align_corners=False)\n    prob_small = torch.sigmoid(model_seg.forward_seg(img_small))\n    prob_small = F.interpolate(prob_small, size=(IMG_SIZE, IMG_SIZE), mode='bilinear', align_corners=False)[0,0]\n\n    # --- KEY CHANGE: WEIGHT TUNING ---\n    # Old: 0.5 / 0.25 / 0.25\n    # New: 0.6 / 0.20 / 0.20 -> Trust the sharp original image more!\n    prob_avg = (prob_1 * 0.60) + (prob_1_flip * 0.20) + (prob_small * 0.20)\n    \n    return prob_avg.cpu().numpy()\n\ndef enhanced_adaptive_mask(prob, alpha_grad=0.35):\n    # Edge enhancement\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_norm = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_norm / (grad_norm.max() + 1e-6)\n    \n    enhanced = (1 - alpha_grad) * prob + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (3,3), 0)\n    \n    # Dynamic Thresholding\n    thr = np.mean(enhanced) + STD_COEFF * np.std(enhanced)\n    mask = (enhanced > thr).astype(np.uint8)\n    \n    # Only CLOSE gaps, do NOT Open (prevents losing small dots)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5,5), np.uint8))\n    return mask, thr\n\ndef pipeline_final(pil):\n    prob = segment_prob_map_multiscale(pil)\n    mask, thr = enhanced_adaptive_mask(prob)\n    mask = cv2.resize(mask, pil.size, interpolation=cv2.INTER_NEAREST)\n    \n    area = int(mask.sum())\n    \n    if area > 0:\n        # Resize mask for calc\n        mask_small = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n        if mask_small.sum() > 0:\n            mean_inside = float(prob[mask_small==1].mean())\n        else:\n            mean_inside = 0.0\n        max_conf = float(prob.max())\n    else:\n        mean_inside = 0.0\n        max_conf = 0.0\n\n    # --- KEY CHANGE: \"SMALL BUT STRONG\" FILTER ---\n    # If the area is small (< 300 pixels), we require HIGH confidence (> 0.85).\n    # This deletes \"weak noise\" but keeps \"strong dots\".\n    if area < 300 and max_conf < 0.85:\n        return \"authentic\", None, {\"area\": area, \"mean\": mean_inside, \"thr\": thr}\n\n    # Logic Override: If model is super confident (>95%), ignore low mean\n    if max_conf > 0.95 and area > 50:\n        return \"forged\", mask, {\"area\": area, \"mean\": mean_inside, \"thr\": thr}\n\n    # Standard Filter\n    if area < AREA_THR or mean_inside < MEAN_THR:\n        return \"authentic\", None, {\"area\": area, \"mean\": mean_inside, \"thr\": thr}\n    \n    return \"forged\", mask, {\"area\": area, \"mean\": mean_inside, \"thr\": thr}\n\n# --- 6. RLE Encoder & Submission ---\ndef rle_encode(mask):\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == 1)[0]\n    if len(dots) == 0: return \"authentic\"\n    run_lengths, prev = [], -2\n    for b in dots:\n        if b > prev + 1: 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\nTEST_DIR = f\"{BASE_DIR}/test_images\"\nSAMPLE_SUB = f\"{BASE_DIR}/sample_submission.csv\"\nOUT_PATH = \"submission3.csv\"\n\nrows = []\ntest_files = sorted(os.listdir(TEST_DIR))\nprint(f\"🚀 Precision Run on {len(test_files)} images...\")\n\nfor f in tqdm(test_files):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)\n    \n    if label == \"authentic\" or mask is None:\n        annot = \"authentic\"\n    else:\n        annot = rle_encode((mask > 0).astype(np.uint8))\n        \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(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Precision Submission Saved: {OUT_PATH}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:04:59.220728Z","iopub.execute_input":"2026-01-06T14:04:59.221037Z","iopub.status.idle":"2026-01-06T14:05:00.040690Z","shell.execute_reply.started":"2026-01-06T14:04:59.221003Z","shell.execute_reply":"2026-01-06T14:05:00.039788Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"clear()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:05:00.041801Z","iopub.execute_input":"2026-01-06T14:05:00.042126Z","iopub.status.idle":"2026-01-06T14:05:00.409930Z","shell.execute_reply.started":"2026-01-06T14:05:00.042092Z","shell.execute_reply":"2026-01-06T14:05:00.409117Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 第四个submission(0.320)","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math, random, torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom sklearn.model_selection import train_test_split\nfrom torch.utils.data import Dataset, DataLoader\nimport torch.nn as nn, torch.nn.functional as F, torch.optim as optim\nfrom transformers import AutoImageProcessor, AutoModel\n\ndef seed_everything(seed=42):\n    random.seed(seed)\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    # This forces CUDA to use deterministic algorithms (slower but consistent)\n    torch.backends.cudnn.deterministic = True \n    torch.backends.cudnn.benchmark = False\n\nseed_everything(42)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nBASE_DIR  = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nAUTH_DIR  = f\"{BASE_DIR}/train_images/authentic\"\nFORG_DIR  = f\"{BASE_DIR}/train_images/forged\"\nMASK_DIR  = f\"{BASE_DIR}/train_masks\"\nTEST_DIR  = f\"{BASE_DIR}/test_images\"\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\n\nIMG_SIZE = 518\nBATCH_SIZE = 2\n# MODEL_LOC = '/kaggle/input/cnndinov2-pbd/CNNDINOv2-U54/CNNDINOv2-U54/model_seg_final.pt'  # 0.310\n# MODEL_LOC = '/kaggle/input/cnndinov2-pbd/CNNDINOv2-U52/CNNDINOv2-U52/model_seg_final.pt'  # 0.321\nMODEL_LOC = '/kaggle/input/cnndinov2-pbd/CNNDINOv2-U52/CNNDINOv2-U52/model_seg_final.pt'  # 0.321\n\n# INFERENCE UTILS\nAREA_THR = 200\nMEAN_THR = 0.22\nUSE_TTA = False\nGRID_SEARCH = False\n\nclass ForgerySegDataset(Dataset):\n    def __init__(self, auth_paths, forg_paths, mask_dir, img_size=IMG_SIZE):\n        self.samples = []\n        for p in forg_paths:\n            m = os.path.join(mask_dir, Path(p).stem + \".npy\")\n            if os.path.exists(m):\n                self.samples.append((p, m))\n        for p in auth_paths:\n            self.samples.append((p, None))\n        self.img_size = img_size\n    def __len__(self): return len(self.samples)\n    def __getitem__(self, idx):\n        img_path, mask_path = self.samples[idx]\n        img = Image.open(img_path).convert(\"RGB\")\n        w, h = img.size\n        if mask_path is None:\n            mask = np.zeros((h, w), np.uint8)\n        else:\n            m = np.load(mask_path)\n            if m.ndim == 3: m = np.max(m, axis=0)\n            mask = (m > 0).astype(np.uint8)\n        img_r = img.resize((IMG_SIZE, IMG_SIZE))\n        mask_r = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n        img_t = torch.from_numpy(np.array(img_r, np.float32)/255.).permute(2,0,1)\n        mask_t = torch.from_numpy(mask_r[None, ...].astype(np.float32))\n        return img_t, mask_t\n\n\n#  MODEL (DINOv2 + Decoder)\n\nfrom transformers import AutoImageProcessor, AutoModel\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH, local_files_only=True, use_fast=False)\nencoder = AutoModel.from_pretrained(DINO_PATH, local_files_only=True).eval().to(device)\n\nclass DinoTinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        # Block 1: 768 -> 384\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 2: 384 -> 192\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1)\n        )\n        # Block 3: 192 -> 96\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True)\n        )\n        # Final Output: 96 -> 1\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        # f: [B, 768, 37, 37]\n        \n        # Step 1: Up to ~74x74\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        \n        # Step 2: Up to ~148x148\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        \n        # Step 3: Up to ~296x296\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        \n        # Step 4: Final jump to 518x518\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n    \nclass DinoSegmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder, self.processor = encoder, processor\n        for p in self.encoder.parameters(): p.requires_grad = False\n        self.seg_head = DinoTinyDecoder(768,1)\n    def forward_features(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).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    def forward_seg(self,x):\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap,(IMG_SIZE,IMG_SIZE))\n\n\nauth_imgs = sorted([str(Path(AUTH_DIR)/f) for f in os.listdir(AUTH_DIR)])\nforg_imgs = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])\ntrain_auth, val_auth = train_test_split(auth_imgs, test_size=0.2, random_state=42)\ntrain_forg, val_forg = train_test_split(forg_imgs, test_size=0.2, random_state=42)\n\ntrain_loader = DataLoader(ForgerySegDataset(train_auth, train_forg, MASK_DIR),\n                          batch_size=BATCH_SIZE, shuffle=True, num_workers=2)\nval_loader = DataLoader(ForgerySegDataset(val_auth, val_forg, MASK_DIR),\n                        batch_size=BATCH_SIZE, shuffle=False, num_workers=2)\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n\n# Load pretrained weights if MODEL_LOC is specified\nif MODEL_LOC is not None and os.path.exists(MODEL_LOC):\n    model_seg.load_state_dict(torch.load(MODEL_LOC, map_location=device))\n    print(f\"✅ Loaded pretrained model from: {MODEL_LOC}\")\n    model_seg.eval()  # Set model to evaluation mode\n\n@torch.no_grad()\ndef segment_prob_map(pil):\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    prob = torch.sigmoid(model_seg.forward_seg(x))[0,0].cpu().numpy()\n    return prob\n\n@torch.no_grad()\ndef segment_prob_map_with_tta(pil):\n    # 1. Preprocessing: Resize, Normalize, and move to Device\n    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    \n    predictions = []\n\n    # 2. Original Prediction\n    pred_orig = torch.sigmoid(model_seg.forward_seg(x))\n    predictions.append(pred_orig)\n\n    # 3. Horizontal Flip TTA (dim 3)\n    # Flip input -> Predict -> Flip output back\n    pred_h = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[3])))\n    predictions.append(torch.flip(pred_h, dims=[3]))\n\n    # 4. Vertical Flip TTA (dim 2)\n    # Flip input -> Predict -> Flip output back\n    pred_v = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[2])))\n    predictions.append(torch.flip(pred_v, dims=[2]))\n\n    # 5. Average the predictions and format as numpy\n    # We stack the 3 predictions and take the mean across the stack dimension (0)\n    prob = torch.stack(predictions).mean(0)[0, 0].cpu().numpy()\n\n    return prob\n    \ndef enhanced_adaptive_mask(prob, alpha_grad=0.45):\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\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\ndef pipeline_final(pil):\n    if USE_TTA:\n        prob = segment_prob_map_with_tta(pil)\n    else:\n        prob = segment_prob_map(pil)\n    mask, thr = finalize_mask(prob, pil.size)\n    area = int(mask.sum())\n    mean_inside = float(prob[cv2.resize(mask,(IMG_SIZE,IMG_SIZE),interpolation=cv2.INTER_NEAREST)==1].mean()) if area>0 else 0.0\n    if area < AREA_THR or mean_inside < MEAN_THR:\n        return \"authentic\", None, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n    return \"forged\", mask, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n\nimport itertools\nfrom sklearn.metrics import f1_score\n\ndef grid_search_area_mean(forg_paths, auth_paths, mask_dir):\n    mean_range = [round(x, 2) for x in np.arange(0.20, 0.291, 0.01)]\n    area_range = [200]\n    # 1. Use ALL images from both paths to maximize robustness\n    val_set = [(p, \"forged\") for p in forg_paths] + [(p, \"authentic\") for p in auth_paths]\n    \n    print(f\"🚀 Step 1: Caching probability maps for ALL {len(val_set)} images...\")\n    cache = []\n    for p, label in tqdm(val_set):\n        pil = Image.open(p).convert(\"RGB\")\n        w, h = pil.size\n        \n        # Get raw probability map\n        prob = segment_prob_map_with_tta(pil) if USE_TTA else segment_prob_map(pil)\n        \n        # USE OLD MASK LOGIC: mean + 0.3*std\n        mask_raw, _ = enhanced_adaptive_mask(prob) # Your function using np.mean + 0.3*np.std\n        mask_resized = cv2.resize(mask_raw, (w, h), interpolation=cv2.INTER_NEAREST)\n        \n        # Handle Ground Truth\n        if label == \"forged\":\n            m_gt = np.load(Path(mask_dir)/f\"{Path(p).stem}.npy\")\n            if m_gt.ndim == 3: m_gt = np.max(m_gt, axis=0)\n            m_gt = (m_gt > 0).astype(np.uint8)\n        else:\n            m_gt = np.zeros((h, w), np.uint8) # Authentic = blank GT\n            \n        cache.append({\"prob\": prob, \"mask\": mask_resized, \"gt\": m_gt, \"label\": label})\n\n    # 2. Sweep thresholds\n    best_f1 = -1\n    best_params = {}\n    combinations = list(itertools.product(area_range, mean_range))\n    \n    for a_thr, m_thr in combinations:\n        current_f1s = []\n        for item in cache:\n            mask = item[\"mask\"]\n            area = int(mask.sum()) # OLD AREA LOGIC\n            \n            # OLD MEAN LOGIC\n            mask_small = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n            mean_in = float(item[\"prob\"][mask_small == 1].mean()) if area > 0 else 0.0\n            \n            # Pipeline decision\n            is_forged = (area >= a_thr and mean_in >= m_thr)\n            m_pred = (mask > 0).astype(np.uint8) if is_forged else np.zeros_like(item[\"gt\"])\n            \n            # F1 Calculation (Authentic silence = 1.0, noisy prediction = 0.0)\n            f1 = f1_score(item[\"gt\"].flatten(), m_pred.flatten(), \n                          zero_division=1 if item[\"label\"] == \"authentic\" else 0)\n            current_f1s.append(f1)\n            \n        avg_f1 = np.mean(current_f1s)\n        if avg_f1 > best_f1:\n            best_f1 = avg_f1\n            best_params = {\"AREA_THR\": a_thr, \"MEAN_THR\": m_thr}\n            print(f\"⭐ New Best F1: {best_f1:.4f} -> AREA: {a_thr}, MEAN: {m_thr}\")\n\n    return best_params\n\nif GRID_SEARCH:\n    best_cfg = grid_search_area_mean(val_forg, val_auth, MASK_DIR)\n    AREA_THR = best_cfg['AREA_THR']\n    MEAN_THR = best_cfg['MEAN_THR']\n\n\nfrom sklearn.metrics import f1_score\nval_items = [(p, 1) for p in val_forg[:10]]\nresults = []\nfor p,_ in tqdm(val_items, desc=\"Validation forged-only\"):\n    pil = Image.open(p).convert(\"RGB\")\n    label, m_pred, dbg = pipeline_final(pil)\n    m_gt = np.load(Path(MASK_DIR)/f\"{Path(p).stem}.npy\")\n    if m_gt.ndim==3: m_gt=np.max(m_gt,axis=0)\n    m_gt=(m_gt>0).astype(np.uint8)\n    m_pred=(m_pred>0).astype(np.uint8) if m_pred is not None else np.zeros_like(m_gt)\n    f1 = f1_score(m_gt.flatten(), m_pred.flatten(), zero_division=0)\n    results.append((Path(p).stem, f1, dbg))\nprint(\"\\n F1-score par image falsifiée:\\n\")\nfor cid,f1,dbg in results:\n    print(f\"{cid} — F1={f1:.4f} | area={dbg['area']} mean={dbg['mean_inside']:.3f} thr={dbg['thr']:.3f}\")\nprint(f\"\\n Moyenne F1 (falsifiées) = {np.mean([r[1] for r in results]):.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:05:00.411133Z","iopub.execute_input":"2026-01-06T14:05:00.411406Z","iopub.status.idle":"2026-01-06T14:05:03.762811Z","shell.execute_reply.started":"2026-01-06T14:05:00.411378Z","shell.execute_reply":"2026-01-06T14:05:03.761841Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\nimport os, json, cv2\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\n\n# --- RLE Encoder for Kaggle Submission ---\ndef rle_encode(mask: np.ndarray, fg_val: int = 1) -> str:\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == fg_val)[0]\n    if len(dots) == 0:\n        return \"authentic\"\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# --- 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\"\nOUT_PATH = \"submission4.csv\"\n\nrows = []\nfor f in tqdm(sorted(os.listdir(TEST_DIR)), desc=\"Inference on Test Set\"):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)  # utilise la version améliorée\n\n    # Sécurisation masque\n    if mask is None:\n        mask = np.zeros(pil.size[::-1], np.uint8)\n    else:\n        mask = np.array(mask, dtype=np.uint8)\n\n    # Annotation finale\n    if label == \"authentic\":\n        annot = \"authentic\"\n    else:\n        annot = rle_encode((mask > 0).astype(np.uint8))\n\n    rows.append({\n        \"case_id\": Path(f).stem,\n        \"annotation\": annot,\n        \"area\": int(dbg.get(\"area\", mask.sum())),\n        \"mean\": float(dbg.get(\"mean_inside\", 0.0)),\n        \"thr\": float(dbg.get(\"thr\", 0.0))\n    })\n\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(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Saved submission file: {OUT_PATH}\")\nprint(final.head(10))\n\n\nsample_files = sorted(os.listdir(TEST_DIR))[:5]\nfor f in sample_files:\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)\n    mask = np.array(mask, dtype=np.uint8) if mask is not None else np.zeros(pil.size[::-1], np.uint8)\n\n    print(f\"{'🔴' if label=='forged' else '🟢'} {f}: {label} | area={mask.sum()} mean={dbg.get('mean_inside', 0):.3f}\")\n\n    if label == \"authentic\":\n        plt.figure(figsize=(5,5))\n        plt.imshow(pil)\n        plt.title(f\"{f} — Authentic\")\n        plt.axis(\"off\")\n        plt.show()\n    else:\n        plt.figure(figsize=(10,5))\n        plt.subplot(1,2,1)\n        plt.imshow(pil)\n        plt.title(\"Original Image\")\n        plt.axis(\"off\")\n        plt.subplot(1,2,2)\n        plt.imshow(pil)\n        plt.imshow(mask, alpha=0.45, cmap=\"Reds\")\n        plt.title(f\"Predicted Forged Mask\\nArea={mask.sum()} | Mean={dbg.get('mean_inside', 0):.3f}\")\n        plt.axis(\"off\")\n        plt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:05:03.764967Z","iopub.execute_input":"2026-01-06T14:05:03.765248Z","iopub.status.idle":"2026-01-06T14:05:04.203746Z","shell.execute_reply.started":"2026-01-06T14:05:03.765217Z","shell.execute_reply":"2026-01-06T14:05:04.203009Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 集成","metadata":{}},{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport json\nfrom collections import Counter\n\ndef ensemble_submissions(file_score_dict, method='weighted', threshold=0.5, \n                        use_pixel_level=False, image_sizes=None):\n    \"\"\"\n    改进版集成函数\n    \n    参数:\n        file_score_dict: {\"文件路径\": 得分}\n        method: 'weighted', 'majority', 'soft_weighted', 'union', 'intersection'\n        threshold: 阈值\n        use_pixel_level: 是否尝试像素级集成（需要额外信息）\n        image_sizes: 字典，{case_id: (height, width)}，用于像素级集成\n    \"\"\"\n    \n    # 读取所有提交文件\n    submissions = {}\n    for file_path, score in file_score_dict.items():\n        df = pd.read_csv(file_path)\n        df['case_id'] = df['case_id'].astype(str)\n        submissions[file_path] = df\n    \n    # 检查所有文件case_id是否一致\n    all_case_ids = set()\n    for df in submissions.values():\n        all_case_ids.update(df['case_id'].tolist())\n    \n    # 按case_id排序，确保顺序一致\n    all_case_ids = sorted(list(all_case_ids))\n    \n    # 方法1: 加权投票（改进版）\n    if method == 'weighted':\n        # 基于排名分配权重（更好地处理微小差异）\n        scores = list(file_score_dict.values())\n        scores_arr = np.array(scores)\n        \n        # 方法A: 使用排名（第一名权重最高）\n        ranks = len(scores) - np.argsort(np.argsort(scores_arr))  # 1=最好\n        weights = ranks / ranks.sum()\n        \n        # 方法B: 或者使用softmax，但调整温度参数\n        # temperature = 0.1  # 更小的温度放大差异\n        # weights = np.exp((scores_arr - max(scores_arr)) / temperature)\n        # weights = weights / weights.sum()\n        \n        print(\"权重分配:\")\n        for (file, score), weight in zip(file_score_dict.items(), weights):\n            print(f\"  {file} (得分: {score:.4f}): {weight:.4f}\")\n        \n        ensemble_annotations = []\n        \n        for case_id in all_case_ids:\n            votes = []\n            vote_weights = []\n            forged_annotations = []  # 收集所有伪造标注\n            \n            for file_path, df in submissions.items():\n                row = df[df['case_id'] == case_id]\n                if row.empty:\n                    continue\n                    \n                annot = row.iloc[0]['annotation']\n                weight = weights[list(file_score_dict.keys()).index(file_path)]\n                \n                if annot == 'authentic':\n                    votes.append(0)\n                    vote_weights.append(weight)\n                else:\n                    votes.append(1)\n                    vote_weights.append(weight)\n                    forged_annotations.append((annot, weight))\n            \n            if not votes:  # 没有该case_id的数据\n                ensemble_annotations.append('authentic')\n                continue\n                \n            # 计算加权投票结果\n            weighted_vote = np.average(votes, weights=vote_weights)\n            \n            if weighted_vote >= threshold and forged_annotations:\n                # 改进：选择最优的伪造标注\n                # 方案1: 选择权重最高的\n                best_annot = max(forged_annotations, key=lambda x: x[1])[0]\n                # 方案2: 选择面积适中的（避免极端）\n                # best_annot = select_best_by_area(forged_annotations)\n                ensemble_annotations.append(best_annot)\n            else:\n                ensemble_annotations.append('authentic')\n    \n    # 方法2: 改进的多数投票\n    elif method == 'majority':\n        ensemble_annotations = []\n        \n        for case_id in all_case_ids:\n            annotations = []\n            for df in submissions.values():\n                row = df[df['case_id'] == case_id]\n                if not row.empty:\n                    annotations.append(row.iloc[0]['annotation'])\n            \n            if not annotations:\n                ensemble_annotations.append('authentic')\n                continue\n                \n            # 统计伪造数量\n            forged_count = sum(1 for annot in annotations if annot != 'authentic')\n            \n            # 需要严格多数（>50%）\n            if forged_count > len(annotations) / 2:\n                # 从伪造标注中选择（可以基于附加信息）\n                forged_annots = [annot for annot in annotations if annot != 'authentic']\n                \n                if forged_annots:\n                    # 选择策略1: 第一个\n                    # best_annot = forged_annots[0]\n                    \n                    # 选择策略2: 基于RLE面积\n                    best_annot = select_by_rle_area(forged_annots, strategy='median')\n                    \n                    # 选择策略3: 投票给最佳模型\n                    # best_model = max(file_score_dict, key=file_score_dict.get)\n                    # best_annot = submissions[best_model][submissions[best_model]['case_id']==case_id].iloc[0]['annotation']\n                    \n                    ensemble_annotations.append(best_annot)\n                else:\n                    ensemble_annotations.append('authentic')\n            else:\n                ensemble_annotations.append('authentic')\n    \n    # 方法3: 并集（任何模型认为伪造就是伪造）\n    elif method == 'union':\n        ensemble_annotations = []\n        \n        for case_id in all_case_ids:\n            forged_annotations = []\n            for file_path, df in submissions.items():\n                row = df[df['case_id'] == case_id]\n                if not row.empty and row.iloc[0]['annotation'] != 'authentic':\n                    forged_annotations.append(row.iloc[0]['annotation'])\n            \n            if forged_annotations:\n                # 选择面积最大的（并集思想）\n                best_annot = max(forged_annotations, key=lambda x: calculate_rle_area(x))\n                ensemble_annotations.append(best_annot)\n            else:\n                ensemble_annotations.append('authentic')\n    \n    # 方法4: 交集（所有模型都认为是伪造）\n    elif method == 'intersection':\n        ensemble_annotations = []\n        \n        for case_id in all_case_ids:\n            all_annotations = []\n            for df in submissions.values():\n                row = df[df['case_id'] == case_id]\n                if not row.empty:\n                    all_annotations.append(row.iloc[0]['annotation'])\n            \n            if not all_annotations:\n                ensemble_annotations.append('authentic')\n                continue\n                \n            # 所有模型都认为是伪造\n            if all(annot != 'authentic' for annot in all_annotations):\n                # 选择面积最小的（交集思想）\n                best_annot = min(all_annotations, key=lambda x: calculate_rle_area(x))\n                ensemble_annotations.append(best_annot)\n            else:\n                ensemble_annotations.append('authentic')\n    \n    # 创建集成后的DataFrame\n    ensemble_df = pd.DataFrame({\n        'case_id': all_case_ids,\n        'annotation': ensemble_annotations\n    })\n    \n    return ensemble_df\n\n\ndef calculate_rle_area(rle_string):\n    \"\"\"计算RLE编码的面积\"\"\"\n    if rle_string == 'authentic':\n        return 0\n    try:\n        runs = json.loads(rle_string)\n        # RLE格式: [start1, length1, start2, length2, ...]\n        area = sum(runs[i] for i in range(1, len(runs), 2))\n        return area\n    except:\n        return 0\n\n\ndef select_by_rle_area(rle_list, strategy='median'):\n    \"\"\"根据RLE面积选择最佳标注\"\"\"\n    if not rle_list:\n        return 'authentic'\n    \n    areas = [calculate_rle_area(rle) for rle in rle_list]\n    \n    if strategy == 'min':\n        idx = np.argmin(areas)\n    elif strategy == 'max':\n        idx = np.argmax(areas)\n    elif strategy == 'median':\n        median_area = np.median(areas)\n        idx = np.argmin(np.abs(areas - median_area))\n    elif strategy == 'mean':\n        mean_area = np.mean(areas)\n        idx = np.argmin(np.abs(areas - mean_area))\n    else:\n        idx = 0\n    \n    return rle_list[idx]\n\n\ndef analyze_ensemble_quality(original_subs, ensemble_df):\n    \"\"\"分析集成质量\"\"\"\n    print(\"\\n📊 集成分析报告\")\n    print(\"=\"*50)\n    \n    # 统计authentic数量变化\n    for file_path, df in original_subs.items():\n        auth_count = (df['annotation'] == 'authentic').sum()\n        print(f\"{file_path}: {auth_count}个authentic\")\n    \n    ensemble_auth = (ensemble_df['annotation'] == 'authentic').sum()\n    print(f\"集成结果: {ensemble_auth}个authentic\")\n    \n    # 检查一致性\n    print(f\"\\n集成结果与各模型差异:\")\n    for file_path, df in original_subs.items():\n        same_count = sum(ensemble_df['annotation'] == df['annotation'])\n        total_count = len(ensemble_df)\n        print(f\"  与{file_path}一致率: {same_count/total_count:.1%}\")\n    \n    return ensemble_auth\n\n\n# 使用示例\nif __name__ == \"__main__\":\n    # 定义文件和得分\n    file_scores = {\n        \"submission1.csv\": 0.324,\n        #\"submission2.csv\": 0.323,\n        #\"submission3.csv\": 0.322,\n        \"submission4.csv\": 0.325\n    }\n    \n    # 读取原始提交文件\n    original_submissions = {}\n    for file_path in file_scores.keys():\n        original_submissions[file_path] = pd.read_csv(file_path)\n    \n    # 测试不同方法\n    # methods = ['weighted', 'majority', 'union', 'intersection']\n    methods = ['majority']\n    \n    for method in methods:\n        print(f\"\\n🔧 使用方法: {method}\")\n        ensemble_df = ensemble_submissions(file_scores, method=method, threshold=0.5)\n        \n        # 分析质量\n        auth_count = analyze_ensemble_quality(original_submissions, ensemble_df)\n        \n        # 保存结果\n        output_file = f\"submission.csv\"\n        ensemble_df.to_csv(output_file, index=False)\n        print(f\"✅ 已保存: {output_file}\")\n        \n        # 显示前几个结果\n        print(\"前5个预测:\")\n        print(ensemble_df.head())\n        print()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-06T14:05:04.204927Z","iopub.execute_input":"2026-01-06T14:05:04.205361Z","iopub.status.idle":"2026-01-06T14:05:04.239704Z","shell.execute_reply.started":"2026-01-06T14:05:04.205307Z","shell.execute_reply":"2026-01-06T14:05:04.238861Z"}},"outputs":[],"execution_count":null}]}