{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":113558,"databundleVersionId":14174843,"sourceType":"competition"},{"sourceId":4534,"sourceType":"modelInstanceVersion","modelInstanceId":3326,"modelId":986}],"dockerImageVersionId":31154,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"\n\nimport 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\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 = 256\nBATCH_SIZE = 1\nEPOCHS_SEG = 1\nLR_SEG = 3e-4\nWEIGHT_DECAY = 1e-4\n\n\nclass ForgerySegDataset(Dataset):\n    def __init__(self, auth_paths, forg_paths, mask_dir, img_size=256):\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)\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.net = nn.Sequential(\n            nn.Conv2d(in_ch,256,3,padding=1), nn.ReLU(),\n            nn.Conv2d(256,64,3,padding=1), nn.ReLU(),\n            nn.Conv2d(64,out_ch,1)\n        )\n    def forward(self, f, size):\n        return self.net(F.interpolate(f, size=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    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(): 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# ----------------------------\n# 🚀 TRAINING\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)\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\nopt_seg = optim.AdamW(model_seg.seg_head.parameters(), lr=LR_SEG, weight_decay=WEIGHT_DECAY)\ncrit_seg = nn.BCEWithLogitsLoss()\n\nfor e in range(EPOCHS_SEG):\n    model_seg.train()\n    total_loss = 0\n    for x,m in tqdm(train_loader, desc=f\"[Segmentation] Epoch {e+1}/{EPOCHS_SEG}\"):\n        x,m = x.to(device),m.to(device)\n        loss = crit_seg(model_seg.forward_seg(x),m)\n        opt_seg.zero_grad(); loss.backward(); opt_seg.step()\n        total_loss += loss.item()\n    print(f\"  → avg_loss={total_loss/len(train_loader):.4f}\")\ntorch.save(model_seg.state_dict(),\"model_seg_final.pt\")\n\n# ----------------------------\n# 🧠 INFERENCE UTILS\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\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    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    # 🔹 condition de filtrage\n    if area < 400 or mean_inside < 0.35:\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\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":"2025-11-10T02:10:10.804698Z","iopub.execute_input":"2025-11-10T02:10:10.805456Z"}},"outputs":[],"execution_count":null},{"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\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n# Example directories for a generic image segmentation task (e.g., medical imaging)\n# Replace with your actual dataset paths\nBASE_DIR = \"/path/to/your/dataset\" \nIMG_DIR = f\"{BASE_DIR}/images\"\nMASK_DIR = f\"{BASE_DIR}/masks\"\nIMG_SIZE = 256\nBATCH_SIZE = 8\nEPOCHS_SEG = 1\nLR_SEG = 3e-4\nWEIGHT_DECAY = 1e-4\n\n# --- Augmentations (Using standard Albumentations) ---\ndef get_train_transforms(img_size):\n    return A.Compose([\n        A.RandomResizedCrop(height=img_size, width=img_size, scale=(0.8, 1.0), ratio=(0.9, 1.1), p=0.6),\n        A.HorizontalFlip(p=0.5),\n        A.VerticalFlip(p=0.2),\n        A.Resize(img_size, img_size), # Final resize ensures consistent output size\n        A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n        ToTensorV2()\n    ])\n\ndef get_valid_transforms(img_size):\n    return A.Compose([\n        A.Resize(img_size, img_size),\n        A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n        ToTensorV2()\n    ])\n\n# --- Dataset Class for Generic Segmentation ---\nclass GenericSegDataset(Dataset):\n    def __init__(self, samples, transforms=None):\n        # samples is a list of tuples (img_path, mask_path)\n        self.samples = samples \n        self.transforms = transforms\n\n    def __len__(self): return len(self.samples)\n\n    def __getitem__(self, idx):\n        img_path, mask_path = self.samples[idx]\n        img = cv2.imread(img_path)\n        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n        \n        mask = cv2.imread(mask_path, cv2.IMREAD_GRAYSCALE) # Assuming single channel mask\n\n        if self.transforms:\n            augmented = self.transforms(image=img, mask=mask)\n            img_t = augmented['image']\n            mask_t = augmented['mask'].unsqueeze(0) # Add channel dimension\n        else:\n            # Fallback (shouldn't be used with this setup)\n            img_t = torch.from_numpy(np.array(img, np.float32)/255.).permute(2,0,1)\n            mask_t = torch.from_numpy(mask[None, ...].astype(np.float32))\n\n        # Ensure mask is binary (0 or 1)\n        mask_t = (mask_t > 0).float() \n\n        return img_t, mask_t\n\n# ----------------------------\n# 🧠 MODEL (Example U-Net like Decoder)\n# ----------------------------\n# This is a simplified decoder example. Replace with a proper segmentation model architecture.\nclass SimpleDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.net = nn.Sequential(\n            nn.Conv2d(in_ch,256,3,padding=1), nn.ReLU(),\n            nn.Conv2d(256,64,3,padding=1), nn.ReLU(),\n            nn.Conv2d(64,out_ch,1)\n        )\n    \n    def forward(self, f, size):\n        x = self.net(f)\n        # Upsample using interpolation to match desired spatial 'size' (H, W)\n        x = F.interpolate(x, size=size, mode='bilinear', align_corners=False)\n        return x\n\n# Example: Using a pre-trained encoder (like a generic vision transformer) and a decoder\n# Replace with an appropriate encoder for your task.\n# This DINO path is for illustration; you'd use a general purpose pre-trained model.\n# processor = AutoImageProcessor.from_pretrained(\"google/vit-base-patch16-224\") \n# encoder = AutoModel.from_pretrained(\"google/vit-base-patch16-224\").eval().to(device)\n\n# ----------------------------\n# Loss Function (Added standard combined loss)\n# ----------------------------\ndef dice_loss(pred, target):\n    smooth = 1e-5\n    intersection = (pred * target).sum()\n    union = pred.sum() + target.sum()\n    dice = (2. * intersection + smooth) / (union + smooth)\n    return 1. - dice\n\nclass CombinedLoss(nn.Module):\n    def __init__(self, bce_weight=0.5, dice_weight=0.5):\n        super().__init__()\n        # Use BCEWithLogitsLoss for numerical stability\n        self.bce = nn.BCEWithLogitsLoss()\n        self.bce_weight = bce_weight\n        self.dice_weight = dice_weight\n\n    def forward(self, pred, target):\n        bce_loss = self.bce(pred, target)\n        # Apply sigmoid before dice calculation for dice loss\n        dice = dice_loss(torch.sigmoid(pred), target) \n        return self.bce_weight * bce_loss + self.dice_weight * dice\n\n# ----------------------------\n# DATA SETUP (Boilerplate)\n# ----------------------------\n\n# ... (You would need code here to generate the `all_samples` list \n#      by pairing image paths from IMG_DIR with corresponding mask paths from MASK_DIR\n#      and then split it into `train_samples` and `valid_samples` using \n#      sklearn's train_test_split) ...\n\n# Example of how you would instantiate the dataset with transforms:\n# # Assume all_samples is a list of (image_path, mask_path) tuples\n# train_samples, valid_samples = train_test_split(all_samples, test_size=0.2, random_state=42)\n# train_dataset = GenericSegDataset(train_samples, transforms=get_train_transforms(IMG_SIZE))\n# valid_dataset = GenericSegDataset(valid_samples, transforms=get_valid_transforms(IMG_SIZE))\n\n# train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, num_workers=4)\n# valid_loader = DataLoader(valid_dataset, batch_size=BATCH_SIZE, shuffle=False, num_workers=4)\n\n# Example of model instantiation (assuming a generic encoder outputting features)\n# Replace with actual encoder output channel size\n# decoder = SimpleDecoder(in_ch=768, out_ch=1).to(device)\n# loss_fn = CombinedLoss().to(device)\n# optimizer = optim.AdamW(decoder.parameters(), lr=LR_SEG, weight_decay=WEIGHT_DECAY)\n\n# ... (Training and validation loop would follow here) ...\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"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 encodeur Kaggle ---\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 = \"submission.csv\"\n\n# --- Inference ---\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)     # ta fonction de décision actuelle\n\n    # Force format masque\n    if mask is not None:\n        mask = np.array(mask, dtype=np.uint8)\n    else:\n        mask = np.zeros(pil.size[::-1], 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))\n\n# --- Visualisation (seulement si forged) ---\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\n    mask = np.array(mask, dtype=np.uint8) if mask is not None else np.zeros(pil.size[::-1], np.uint8)\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); plt.imshow(pil); plt.title(\"Original\"); plt.axis(\"off\")\n        plt.subplot(1,2,2); plt.imshow(pil); plt.imshow(mask, alpha=0.45, cmap=\"Blues\"); \n        plt.title(\"Predicted Mask\"); plt.axis(\"off\")\n        plt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-09T06:51:37.015232Z","iopub.execute_input":"2025-11-09T06:51:37.015485Z","iopub.status.idle":"2025-11-09T06:51:37.646374Z","shell.execute_reply.started":"2025-11-09T06:51:37.015461Z","shell.execute_reply":"2025-11-09T06:51:37.645531Z"}},"outputs":[],"execution_count":null}]}