{"metadata":{"kernelspec":{"name":"python3","display_name":"Python 3","language":"python"},"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":"none","dataSources":[{"sourceType":"competition","sourceId":113558,"databundleVersionId":14878066}],"dockerImageVersionId":31234,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Scientific Image Forgery Detection — PyTorch baseline","metadata":{}},{"cell_type":"code","source":" !pip -q install -U albumentations==1.4.20 opencv-python-headless==4.10.0.84 segmentation-models-pytorch==0.3.4\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-12-19T22:47:17.940091Z","iopub.execute_input":"2025-12-19T22:47:17.940551Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport gc\nimport random\nfrom dataclasses import dataclass\nfrom pathlib import Path\n\nimport numpy as np\nimport pandas as pd\n\nimport torch\nimport torch.nn as nn\nfrom torch.utils.data import Dataset, DataLoader\n\nfrom PIL import Image\n\n# Optional deps\ntry:\n    import cv2\nexcept Exception:\n    cv2 = None\n\ntry:\n    import albumentations as A\nexcept Exception:\n    A = None\n\ntry:\n    import segmentation_models_pytorch as smp\nexcept Exception:\n    smp = None\n\ntry:\n    import torchvision\n    from torchvision.models.segmentation import deeplabv3_resnet50\n    try:\n        from torchvision.models.segmentation import DeepLabV3_ResNet50_Weights\n    except Exception:\n        DeepLabV3_ResNet50_Weights = None\nexcept Exception:\n    torchvision = None\n    deeplabv3_resnet50 = None\n    DeepLabV3_ResNet50_Weights = None\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from contextlib import nullcontext\n\ndef get_device() -> str:\n    if torch.cuda.is_available():\n        return 'cuda'\n    # Apple Silicon\n    if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():\n        return 'mps'\n    return 'cpu'\n\n@dataclass\nclass CFG:\n    seed: int = 42\n    device: str = get_device()\n\n    # training\n    epochs: int = 10\n    lr: float = 3e-4\n    weight_decay: float = 1e-4\n    batch_size: int = 8\n    # In notebooks on macOS, num_workers>0 often breaks due to multiprocessing pickling.\n    num_workers: int = 0\n\n    # preprocessing\n    img_size: int = 768\n\n    # inference\n    thr: float = 0.5\n\n    # CV\n    n_splits: int = 5\n    fold: int = 0\n\n    # speed/stability\n    use_amp: bool = True\n\n\ndef seed_everything(seed: int):\n    random.seed(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    if torch.cuda.is_available():\n        torch.cuda.manual_seed_all(seed)\n    torch.backends.cudnn.benchmark = True\n\n\nseed_everything(CFG.seed)\n\n# Kaggle vs local notebook defaults\nRUNNING_ON_KAGGLE = os.environ.get('KAGGLE_URL_BASE') is not None\n# Kaggle supports DataLoader workers; macOS Jupyter often does not (spawn/pickling).\nCFG.num_workers = 2 if RUNNING_ON_KAGGLE else 0\n\nAMP_ENABLED = CFG.use_amp and (CFG.device == 'cuda')\nPIN_MEMORY = (CFG.device == 'cuda')\nprint('device:', CFG.device, 'AMP:', AMP_ENABLED, 'pin_memory:', PIN_MEMORY, 'num_workers:', CFG.num_workers, 'kaggle:', RUNNING_ON_KAGGLE)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Paths (Kaggle + local fallback)\nCANDIDATES = [\n    Path('/kaggle/input/recodai-luc-scientific-image-forgery-detection'),\n    Path('.'),\n]\nDATA_DIR = None\nfor p in CANDIDATES:\n    if (p / 'train_images').exists() and (p / 'test_images').exists():\n        DATA_DIR = p\n        break\nassert DATA_DIR is not None, 'Could not find dataset directory.'\n\nTRAIN_AUTH_DIR = DATA_DIR / 'train_images' / 'authentic'\nTRAIN_FORG_DIR = DATA_DIR / 'train_images' / 'forged'\nTRAIN_MASK_DIR = DATA_DIR / 'train_masks'\n\nSUPP_IMG_DIR = DATA_DIR / 'supplemental_images'\nSUPP_MASK_DIR = DATA_DIR / 'supplemental_masks'\n\nTEST_DIR = DATA_DIR / 'test_images'\n\nprint('DATA_DIR:', DATA_DIR)\nprint('train_auth:', TRAIN_AUTH_DIR.exists(), 'train_forg:', TRAIN_FORG_DIR.exists(), 'test:', TEST_DIR.exists())\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# RLE (OFFICIAL STYLE)\n# This competition encodes INSTANCE masks: a single string contains one or more JSON arrays separated by ';'.\n# For a single predicted mask, pass [mask] into rle_encode([...]).\n\nimport json\n\ntry:\n    import numba\nexcept Exception:\n    numba = None\n\ndef _rle_encode_py(x: np.ndarray, fg_val: int = 1) -> list[int]:\n    # Matches the official behavior: find foreground indices in column-major order.\n    dots = np.where(x.T.flatten() == fg_val)[0]\n    run_lengths: list[int] = []\n    prev = -2\n    for b in dots:\n        if b > prev + 1:\n            run_lengths.extend((int(b) + 1, 0))\n        run_lengths[-1] += 1\n        prev = b\n    return run_lengths\n\nif numba is not None:\n    @numba.jit(nopython=True)\n    def _rle_encode_jit(x: np.ndarray, fg_val: int = 1):\n        dots = np.where(x.T.flatten() == fg_val)[0]\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 run_lengths\n\ndef rle_encode(masks: list[np.ndarray], fg_val: int = 1) -> str:\n    encs = []\n    for x in masks:\n        x = (x == fg_val).astype(np.uint8)\n        if numba is not None:\n            enc = _rle_encode_jit(x, fg_val)\n        else:\n            enc = _rle_encode_py(x, fg_val)\n        encs.append(json.dumps(enc))\n    return ';'.join(encs)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def build_df() -> pd.DataFrame:\n    rows = []\n\n    # authentic\n    for p in sorted(TRAIN_AUTH_DIR.glob('*.png')):\n        cid = int(p.stem)\n        rows.append({'case_id': cid, 'img_path': str(p), 'mask_path': None, 'is_forged': 0})\n\n    # forged\n    for p in sorted(TRAIN_FORG_DIR.glob('*.png')):\n        cid = int(p.stem)\n        m = TRAIN_MASK_DIR / f'{cid}.npy'\n        rows.append({'case_id': cid, 'img_path': str(p), 'mask_path': str(m) if m.exists() else None, 'is_forged': 1})\n\n    # supplemental (treat mask existence as forged)\n    if SUPP_IMG_DIR.exists():\n        for p in sorted(SUPP_IMG_DIR.glob('*.png')):\n            cid = int(p.stem)\n            m = SUPP_MASK_DIR / f'{cid}.npy'\n            rows.append({'case_id': cid, 'img_path': str(p), 'mask_path': str(m) if m.exists() else None, 'is_forged': 1 if m.exists() else 0})\n\n    df = pd.DataFrame(rows).drop_duplicates(subset=['case_id']).reset_index(drop=True)\n    return df\n\ndf = build_df()\ndisplay(df.head())\nprint('df:', df.shape)\nprint(df['is_forged'].value_counts(dropna=False))\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def read_image_rgb(path: str) -> np.ndarray:\n    if cv2 is not None:\n        img = cv2.imread(path, cv2.IMREAD_COLOR)\n        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)\n        return img\n    return np.array(Image.open(path).convert('RGB'))\n\ndef load_union_mask(mask_path, h: int, w: int) -> np.ndarray:\n    if mask_path is None:\n        return np.zeros((h, w), dtype=np.uint8)\n    m = np.load(mask_path)\n    if m.ndim == 2:\n        mm = (m > 0).astype(np.uint8)\n    else:\n        mm = (m > 0).any(axis=0).astype(np.uint8)\n    return mm\n\ndef resize_longest_and_pad(img: np.ndarray, mask, out_size: int):\n    orig_h, orig_w = img.shape[:2]\n    scale = out_size / max(orig_h, orig_w)\n    new_h = int(round(orig_h * scale))\n    new_w = int(round(orig_w * scale))\n\n    if cv2 is not None:\n        img_rs = cv2.resize(img, (new_w, new_h), interpolation=cv2.INTER_LINEAR)\n        mask_rs = None\n        if mask is not None:\n            mask_rs = cv2.resize(mask, (new_w, new_h), interpolation=cv2.INTER_NEAREST)\n    else:\n        img_rs = np.array(Image.fromarray(img).resize((new_w, new_h), resample=Image.BILINEAR))\n        mask_rs = None\n        if mask is not None:\n            mask_rs = np.array(Image.fromarray(mask).resize((new_w, new_h), resample=Image.NEAREST))\n\n    pad_h = out_size - new_h\n    pad_w = out_size - new_w\n    img_pad = np.pad(img_rs, ((0, pad_h), (0, pad_w), (0, 0)), mode='constant', constant_values=0)\n\n    mask_pad = None\n    if mask_rs is not None:\n        mask_pad = np.pad(mask_rs, ((0, pad_h), (0, pad_w)), mode='constant', constant_values=0)\n\n    meta = {\n        'scale': scale,\n        'new_h': new_h,\n        'new_w': new_w,\n        'pad_h': pad_h,\n        'pad_w': pad_w,\n        'orig_h': orig_h,\n        'orig_w': orig_w,\n    }\n    return img_pad, mask_pad, meta\n\ndef unpad_and_resize_back(pred_pad: np.ndarray, meta: dict) -> np.ndarray:\n    new_h, new_w = meta['new_h'], meta['new_w']\n    orig_h, orig_w = meta['orig_h'], meta['orig_w']\n    pred_crop = pred_pad[:new_h, :new_w]\n    if cv2 is not None:\n        pred_back = cv2.resize(pred_crop, (orig_w, orig_h), interpolation=cv2.INTER_NEAREST)\n    else:\n        pred_back = np.array(Image.fromarray(pred_crop).resize((orig_w, orig_h), resample=Image.NEAREST))\n    return pred_back.astype(np.uint8)\n\ndef normalize_img(img: np.ndarray) -> np.ndarray:\n    img = img.astype(np.float32) / 255.0\n    mean = np.array([0.485, 0.456, 0.406], dtype=np.float32)\n    std = np.array([0.229, 0.224, 0.225], dtype=np.float32)\n    return (img - mean) / std\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def get_train_aug():\n    if A is None:\n        return None\n    return A.Compose([\n        A.HorizontalFlip(p=0.5),\n        A.VerticalFlip(p=0.5),\n        A.RandomRotate90(p=0.5),\n        A.RandomBrightnessContrast(p=0.3),\n        A.GaussNoise(p=0.2),\n        A.OneOf([A.MotionBlur(p=1.0), A.GaussianBlur(p=1.0)], p=0.2),\n    ])\n\nclass ForgeryDataset(Dataset):\n    def __init__(self, df: pd.DataFrame, train: bool):\n        self.df = df.reset_index(drop=True)\n        self.train = train\n        self.aug = get_train_aug() if train else None\n\n    def __len__(self):\n        return len(self.df)\n\n    def __getitem__(self, idx: int):\n        row = self.df.iloc[idx]\n        img = read_image_rgb(row.img_path)\n        h, w = img.shape[:2]\n        mask = load_union_mask(row.mask_path, h, w)\n\n        if self.aug is not None:\n            out = self.aug(image=img, mask=mask)\n            img, mask = out['image'], out['mask']\n\n        img_pad, mask_pad, _ = resize_longest_and_pad(img, mask, CFG.img_size)\n        img_pad = normalize_img(img_pad)\n\n        img_t = torch.from_numpy(img_pad).permute(2, 0, 1).float()\n        mask_t = torch.from_numpy(mask_pad).unsqueeze(0).float()\n        return img_t, mask_t\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class DiceLoss(nn.Module):\n    def __init__(self, eps: float = 1e-6):\n        super().__init__()\n        self.eps = eps\n\n    def forward(self, logits, targets):\n        probs = torch.sigmoid(logits)\n        num = 2 * (probs * targets).sum(dim=(2,3))\n        den = (probs + targets).sum(dim=(2,3)).clamp_min(self.eps)\n        dice = num / den\n        return 1 - dice.mean()\n\nclass _TorchvisionDeepLabV3(nn.Module):\n    def __init__(self):\n        super().__init__()\n        if deeplabv3_resnet50 is None:\n            raise RuntimeError('torchvision deeplabv3_resnet50 is not available')\n        weights = None\n        if DeepLabV3_ResNet50_Weights is not None:\n            weights = DeepLabV3_ResNet50_Weights.DEFAULT\n        m = deeplabv3_resnet50(weights=weights)\n        if hasattr(m, 'classifier') and isinstance(m.classifier, nn.Sequential):\n            m.classifier[-1] = nn.Conv2d(m.classifier[-1].in_channels, 1, kernel_size=1)\n        else:\n            m.classifier = nn.Conv2d(256, 1, kernel_size=1)\n        if hasattr(m, 'aux_classifier'):\n            m.aux_classifier = None\n        self.model = m\n\n    def forward(self, x):\n        out = self.model(x)\n        if isinstance(out, dict):\n            return out['out']\n        return out\n\ndef build_model() -> nn.Module:\n    if smp is not None:\n        return smp.Unet(encoder_name='resnet34', encoder_weights='imagenet', in_channels=3, classes=1)\n    if torchvision is not None and deeplabv3_resnet50 is not None:\n        return _TorchvisionDeepLabV3()\n    return nn.Sequential(\n        nn.Conv2d(3, 16, 3, padding=1), nn.ReLU(inplace=True),\n        nn.Conv2d(16, 16, 3, padding=1), nn.ReLU(inplace=True),\n        nn.Conv2d(16, 1, 1),\n    )\n\ndef dice_coef_from_logits(logits: torch.Tensor, targets: torch.Tensor, thr: float = 0.5) -> float:\n    probs = torch.sigmoid(logits)\n    preds = (probs > thr).float()\n    inter = (preds * targets).sum(dim=(2,3))\n    den = (preds + targets).sum(dim=(2,3)).clamp_min(1.0)\n    return (2 * inter / den).mean().item()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from sklearn.model_selection import StratifiedKFold\n\nskf = StratifiedKFold(n_splits=CFG.n_splits, shuffle=True, random_state=CFG.seed)\ndf['fold'] = -1\nfor fold, (_, va_idx) in enumerate(skf.split(df, df['is_forged'])):\n    df.loc[va_idx, 'fold'] = fold\n\ntrain_df = df[df['fold'] != CFG.fold].reset_index(drop=True)\nvalid_df = df[df['fold'] == CFG.fold].reset_index(drop=True)\nprint('train:', train_df.shape, 'valid:', valid_df.shape)\nprint('valid forged rate:', valid_df['is_forged'].mean())\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def train_one_epoch(model, loader, optimizer, scaler, bce, dice_loss):\n    model.train()\n    total_loss = 0.0\n    for imgs, masks in loader:\n        imgs = imgs.to(CFG.device, non_blocking=True)\n        masks = masks.to(CFG.device, non_blocking=True)\n\n        optimizer.zero_grad(set_to_none=True)\n        autocast_ctx = torch.amp.autocast('cuda') if AMP_ENABLED else nullcontext()\n        with autocast_ctx:\n            logits = model(imgs)\n            loss = bce(logits, masks) + dice_loss(logits, masks)\n\n        if scaler is not None:\n            scaler.scale(loss).backward()\n            scaler.step(optimizer)\n            scaler.update()\n        else:\n            loss.backward()\n            optimizer.step()\n\n        total_loss += loss.item() * imgs.size(0)\n    return total_loss / len(loader.dataset)\n\n@torch.no_grad()\ndef valid_one_epoch(model, loader, bce, dice_loss):\n    model.eval()\n    total_loss = 0.0\n    total_dice = 0.0\n    for imgs, masks in loader:\n        imgs = imgs.to(CFG.device, non_blocking=True)\n        masks = masks.to(CFG.device, non_blocking=True)\n        logits = model(imgs)\n        loss = bce(logits, masks) + dice_loss(logits, masks)\n        total_loss += loss.item() * imgs.size(0)\n        total_dice += dice_coef_from_logits(logits, masks, thr=0.5) * imgs.size(0)\n    return total_loss / len(loader.dataset), total_dice / len(loader.dataset)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_ds = ForgeryDataset(train_df, train=True)\nvalid_ds = ForgeryDataset(valid_df, train=False)\n\ntrain_loader = DataLoader(train_ds, batch_size=CFG.batch_size, shuffle=True, num_workers=CFG.num_workers, pin_memory=PIN_MEMORY, drop_last=True)\nvalid_loader = DataLoader(valid_ds, batch_size=CFG.batch_size, shuffle=False, num_workers=CFG.num_workers, pin_memory=PIN_MEMORY)\n\nmodel = build_model().to(CFG.device)\noptimizer = torch.optim.AdamW(model.parameters(), lr=CFG.lr, weight_decay=CFG.weight_decay)\nscheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=CFG.epochs)\nscaler = torch.amp.GradScaler('cuda') if AMP_ENABLED else None\n\nbce = nn.BCEWithLogitsLoss()\ndice_loss = DiceLoss()\n\nbest_dice = -1.0\nbest_path = 'best_model.pth'\n\nfor epoch in range(CFG.epochs):\n    tr_loss = train_one_epoch(model, train_loader, optimizer, scaler, bce, dice_loss)\n    va_loss, va_dice = valid_one_epoch(model, valid_loader, bce, dice_loss)\n    scheduler.step()\n    print(f'Epoch {epoch+1}/{CFG.epochs} | train {tr_loss:.4f} | valid {va_loss:.4f} | dice {va_dice:.4f}')\n    if va_dice > best_dice:\n        best_dice = va_dice\n        torch.save(model.state_dict(), best_path)\n\nprint('best_dice:', best_dice)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"@torch.no_grad()\ndef predict_one(model, img_path: str) -> np.ndarray:\n    img = read_image_rgb(img_path)\n    img_pad, _, meta = resize_longest_and_pad(img, mask=None, out_size=CFG.img_size)\n    img_pad = normalize_img(img_pad)\n\n    x = torch.from_numpy(img_pad).permute(2, 0, 1).float().unsqueeze(0).to(CFG.device)\n    logits = model(x)[0, 0]\n    prob = torch.sigmoid(logits).float().cpu().numpy()\n    pred_pad = (prob > CFG.thr).astype(np.uint8)\n\n    pred = unpad_and_resize_back(pred_pad, meta)\n    return pred\n\nmodel.load_state_dict(torch.load(best_path, map_location=CFG.device))\nmodel.eval()\n\ntest_paths = sorted(TEST_DIR.glob('*.png'))\nprint('num test:', len(test_paths))\n\nrows = []\nfor p in test_paths:\n    case_id = int(p.stem)\n    pred = predict_one(model, str(p))\n    if pred.sum() == 0:\n        ann = 'authentic'\n    else:\n        ann = rle_encode([pred])\n    rows.append({'case_id': case_id, 'annotation': ann})\n\nfrom pathlib import Path\n\nsub = pd.DataFrame(rows).sort_values('case_id').reset_index(drop=True)\nout_csv = '/kaggle/working/submission.csv' if RUNNING_ON_KAGGLE else 'submission.csv'\nsub.to_csv(out_csv, index=False)\ndisplay(sub.head())\np = Path(out_csv)\nprint('wrote', out_csv, 'exists:', p.exists(), 'size:', p.stat().st_size if p.exists() else None)\nassert p.exists(), 'submission.csv was not created'\n\n# cleanup\ntry:\n    del rows\nexcept NameError:\n    pass\ngc.collect()\nif torch.cuda.is_available():\n    torch.cuda.empty_cache()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}