{"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":14375458,"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},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"#  第一个submission(0.325)","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},"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 = \"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":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"clear()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# 第二个submission(0.325)","metadata":{}},{"cell_type":"code","source":"import os, cv2, json, math, random, torch, io\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    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/input/cnndinov2-pbd/CNNDINOv2-U52/CNNDINOv2-U52/model_seg_final.pt'\n\n# INFERENCE UTILS\nAREA_THR = 200\nMEAN_THR = 0.22\nUSE_TTA = False\nUSE_ELA = True  # NEW: Enable ELA\nELA_WEIGHT = 0.3  # Weight for ELA contribution\nELA_QUALITY = 90  # JPEG quality for ELA\n\nprint(f'Device: {device}')\nprint(f'ELA Enabled: {USE_ELA}')","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def compute_ela(pil_image, quality=90):\n    \"\"\"Compute Error Level Analysis for an image.\n    \n    Like adding forensic powder to reveal hidden fingerprints,\n    ELA reveals hidden manipulations by analyzing compression artifacts.\n    \n    Args:\n        pil_image: PIL Image in RGB format\n        quality: JPEG quality for re-compression (default 90)\n    \n    Returns:\n        PIL Image showing ELA result\n    \"\"\"\n    # Re-compress at specified quality\n    buffer = io.BytesIO()\n    pil_image.save(buffer, 'JPEG', quality=quality)\n    buffer.seek(0)\n    recompressed = Image.open(buffer).convert('RGB')\n    \n    # Compute difference\n    original = np.array(pil_image, dtype=np.float32)\n    compressed = np.array(recompressed, dtype=np.float32)\n    ela = np.abs(original - compressed)\n    \n    # Scale for visibility (amplify differences)\n    scale = 255.0 / (ela.max() + 1e-6)\n    ela = np.clip(ela * scale, 0, 255).astype(np.uint8)\n    \n    return Image.fromarray(ela)\n\n\ndef compute_ela_heatmap(pil_image, quality=90):\n    \"\"\"Compute ELA and return as grayscale heatmap.\"\"\"\n    ela_img = compute_ela(pil_image, quality)\n    ela_arr = np.array(ela_img, dtype=np.float32)\n    # Convert to grayscale by taking max channel\n    heatmap = ela_arr.max(axis=2)\n    # Normalize\n    heatmap = heatmap / (heatmap.max() + 1e-6)\n    return heatmap\n\n\ndef visualize_ela(pil_image, quality=90):\n    \"\"\"Visualize original image and its ELA side by side.\"\"\"\n    ela = compute_ela(pil_image, quality)\n    \n    fig, axes = plt.subplots(1, 3, figsize=(15, 5))\n    \n    axes[0].imshow(pil_image)\n    axes[0].set_title('Original Image')\n    axes[0].axis('off')\n    \n    axes[1].imshow(ela)\n    axes[1].set_title(f'ELA (Quality={quality})')\n    axes[1].axis('off')\n    \n    # Heatmap version\n    heatmap = compute_ela_heatmap(pil_image, quality)\n    axes[2].imshow(heatmap, cmap='hot')\n    axes[2].set_title('ELA Heatmap')\n    axes[2].axis('off')\n    \n    plt.tight_layout()\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def detect_hidden_jpeg(pil_image):\n    \"\"\"Detect if a PNG/lossless image was likely originally a JPEG.\n    \n    Checks for 8x8 block artifacts using a simple horizontal/vertical gradient check.\n    \"\"\"\n    arr = np.array(pil_image.convert('L'), dtype=np.float32)\n    h, w = arr.shape\n    if h < 16 or w < 16: return False, 0.0\n    \n    # Compute gradients\n    grad_x = np.abs(arr[:, 1:] - arr[:, :-1])\n    grad_y = np.abs(arr[1:, :] - arr[:-1, :])\n    \n    # Check for 8x8 periodic peaks in gradient sums (block boundaries)\n    def check_periodicity(grads, axis=0):\n        sums = grads.sum(axis=axis)\n        if len(sums) < 16: return 0.0\n        # Look at the 8-cycle variance vs local mean\n        peaks = [sums[i::8].mean() for i in range(8)]\n        return np.max(peaks) / (np.mean(peaks) + 1e-6)\n    \n    score_x = check_periodicity(grad_x, axis=0)\n    score_y = check_periodicity(grad_y, axis=1)\n    \n    final_score = (score_x + score_y) / 2.0\n    is_hjpeg = final_score > 1.2 # Empirical threshold\n    \n    return is_hjpeg, final_score\n\ndef visualize_ela_suitability(pil_image):\n    is_hjt, score = detect_hidden_jpeg(pil_image)\n    ela = compute_ela(pil_image, quality=ELA_QUALITY)\n    \n    plt.figure(figsize=(10, 4))\n    plt.subplot(1, 2, 1)\n    plt.imshow(pil_image)\n    plt.title(f\"Original (H-JPEG: {is_hjt}, Score: {score:.2f})\")\n    plt.axis('off')\n    \n    plt.subplot(1, 2, 2)\n    plt.imshow(ela)\n    plt.title(\"ELA Transformation\")\n    plt.axis('off')\n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# MODEL (DINOv2 + Decoder)\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        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        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        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True)\n        )\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=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        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\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        \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        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_seg(self, x):\n        fmap = self.forward_features(x)\n        return self.seg_head(fmap, (IMG_SIZE, IMG_SIZE))\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\n\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()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"@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    x = torch.from_numpy(np.array(pil.resize((IMG_SIZE, IMG_SIZE)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    predictions = []\n    pred_orig = torch.sigmoid(model_seg.forward_seg(x))\n    predictions.append(pred_orig)\n    pred_h = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[3])))\n    predictions.append(torch.flip(pred_h, dims=[3]))\n    pred_v = torch.sigmoid(model_seg.forward_seg(torch.flip(x, dims=[2])))\n    predictions.append(torch.flip(pred_v, dims=[2]))\n    prob = torch.stack(predictions).mean(0)[0, 0].cpu().numpy()\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","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def pipeline_final_with_ela(pil, use_ela=True, ela_weight=0.3):\n    \"\"\"Enhanced pipeline with ELA integration.\n    \n    Like adding forensic powder to paper, ELA reveals hidden manipulations\n    by highlighting regions with different compression histories.\n    \"\"\"\n    # Get RGB prediction\n    if USE_TTA:\n        prob_rgb = segment_prob_map_with_tta(pil)\n    else:\n        prob_rgb = segment_prob_map(pil)\n    \n    prob_combined = prob_rgb.copy()\n    ela_info = {}\n    \n    if use_ela:\n        # Compute ELA image\n        ela_img = compute_ela(pil, quality=ELA_QUALITY)\n        \n        # Get model prediction on ELA image\n        if USE_TTA:\n            prob_ela = segment_prob_map_with_tta(ela_img)\n        else:\n            prob_ela = segment_prob_map(ela_img)\n        \n        # Get direct ELA heatmap (resized to model output size)\n        ela_heatmap = compute_ela_heatmap(pil, quality=ELA_QUALITY)\n        ela_heatmap = cv2.resize(ela_heatmap, (IMG_SIZE, IMG_SIZE))\n        \n        # Combine: RGB prob + weighted ELA prediction + small ELA heatmap boost\n        prob_combined = (\n            (1 - ela_weight) * prob_rgb + \n            ela_weight * 0.7 * prob_ela + \n            ela_weight * 0.3 * ela_heatmap  # Direct ELA signal\n        )\n        \n        ela_info = {\n            'ela_max': float(ela_heatmap.max()),\n            'ela_mean': float(ela_heatmap.mean()),\n        }\n    \n    mask, thr = finalize_mask(prob_combined, pil.size)\n    area = int(mask.sum())\n    mean_inside = float(prob_combined[cv2.resize(mask,(IMG_SIZE,IMG_SIZE),interpolation=cv2.INTER_NEAREST)==1].mean()) if area>0 else 0.0\n    \n    if area < AREA_THR or mean_inside < MEAN_THR:\n        return \"authentic\", None, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr, **ela_info}\n    return \"forged\", mask, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr, **ela_info}\n\n\ndef pipeline_final(pil):\n    \"\"\"Wrapper to use ELA-enhanced pipeline if enabled.\"\"\"\n    return pipeline_final_with_ela(pil, use_ela=USE_ELA, ela_weight=ELA_WEIGHT)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Test ELA visualization on forged images\nif os.path.exists(FORG_DIR):\n    forg_imgs = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])[:3]\n    \n    for img_path in forg_imgs:\n        print(f\"\\n📸 {Path(img_path).name}\")\n        pil = Image.open(img_path).convert(\"RGB\")\n        visualize_ela_suitability(pil)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"if os.path.exists(FORG_DIR):\n    test_p = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])[5]\n    pil = Image.open(test_p).convert(\"RGB\")\n    \n    # 1. No ELA\n    l1, m1, d1 = pipeline_final_with_ela(pil, use_ela=False)\n    \n    # 2. With ELA\n    l2, m2, d2 = pipeline_final_with_ela(pil, use_ela=True)\n    \n    fig, axes = plt.subplots(1, 3, figsize=(15, 5))\n    axes[0].imshow(pil)\n    axes[0].set_title(\"Original\")\n    \n    if m1 is not None: axes[1].imshow(m1)\n    axes[1].set_title(f\"No ELA (Area: {d1['area']})\")\n    \n    if m2 is not None: axes[2].imshow(m2)\n    axes[2].set_title(f\"With ELA (Area: {d2['area']})\")\n    \n    plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def 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\nSAMPLE_SUB = f\"{BASE_DIR}/sample_submission.csv\"\nOUT_PATH = \"submission2.csv\"\n\nrows = []\nfor f in tqdm(sorted(os.listdir(TEST_DIR)), desc=\"Inference on Test Set with ELA\"):\n    pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n    label, mask, dbg = pipeline_final(pil)\n    \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    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\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))","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"clear()","metadata":{"trusted":true},"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.325,\n        \"submission2.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},"outputs":[],"execution_count":null}]}