{"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"}],"dockerImageVersionId":31193,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"from pathlib import Path\nimport re\nfrom collections import defaultdict\nimport csv\nimport sys\nimport math\nfrom typing import List, Tuple, Optional\n\n# Change paths if necessary\nROOT = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nDIR_TRAIN_AUTH = Path(f\"{ROOT}/train_images/authentic\")\nDIR_TRAIN_FORG = Path(f\"{ROOT}/train_images/forged\")\nDIR_TEST       = Path(f\"{ROOT}/test_images\")  # not used, kept for compatibility\nOUTPUT_CSV     = Path(\"/kaggle/working/pairs.csv\")\n\n# Method selection\nUSE_PHASH = True           # Use pHash for unmatched images by name?\nPHASH_THRESHOLD = 10       # Hamming distance threshold (lower is better)\n\nUSE_SSIM = False           # (Optional) Use SSIM for remaining unmatched -after pHash?\nSSIM_THRESHOLD = 0.70      # SSIM threshold (higher is better)\nSSIM_MAX_SIDE = 512        # resize for speed\n\n# If the dataset is too large, you can limit the number of test samples (None = no limit)\nDEBUG_LIMIT_FORGED = None   # e.g. 200\n\n# ------------------- Dependencies -------------------\n# Install required packages (allowed in Kaggle). If already installed, this section will just pass.\ntry:\n    import imagehash  # type: ignore\n    from PIL import Image  # type: ignore\nexcept Exception:\n    !pip -q install imagehash\n    import imagehash\n    from PIL import Image\n\nif USE_SSIM:\n    try:\n        import cv2  # type: ignore\n        from skimage.metrics import structural_similarity as ssim  # type: ignore\n    except Exception:\n        !pip -q install scikit-image opencv-python-headless\n        import cv2\n        from skimage.metrics import structural_similarity as ssim\n\n# ------------------- Utilities -------------------\nIMG_EXTS = {\".png\", \".jpg\", \".jpeg\", \".tif\", \".tiff\", \".bmp\"}\n\ndef list_images(root: Path) -> List[Path]:\n    return sorted([p for p in root.rglob(\"*\") if p.suffix.lower() in IMG_EXTS])\n\ndef normalize_name(p: Path) -> str:\n    s = p.stem.lower()\n    s = s.replace(\"-\", \"_\").replace(\" \", \"_\")\n    # Remove common ending tags\n    s = re.sub(r\"(authentic|auth|original|orig|clean|real|gt)$\", \"\", s)\n    s = re.sub(r\"(forg(ed)?|fake|tampered|edit(ed)?|manipulated?)$\", \"\", s)\n    s = re.sub(r\"(__+|_+$)\", \"\", s)\n    return s\n\ndef safe_open_image(path: Path) -> Optional[Image.Image]:\n    try:\n        return Image.open(path).convert(\"RGB\")\n    except Exception:\n        return None\n\ndef phash_value(path: Path):\n    im = safe_open_image(path)\n    if im is None:\n        return None\n    try:\n        return imagehash.phash(im)\n    except Exception:\n        return None\n\ndef hamming_distance(h1, h2) -> int:\n    return abs(h1 - h2)\n\ndef load_gray_resized_cv2(path: Path, max_side=512):\n    import numpy as np\n    img = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE)\n    if img is None:\n        return None\n    h, w = img.shape[:2]\n    scale = min(1.0, max_side / max(h, w))\n    if scale < 1.0:\n        img = cv2.resize(img, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA)\n    return img\n\ndef best_match_by_phash(gf: Path, auth_hashes: List[Tuple[Path, object]]) -> Tuple[Optional[Path], Optional[int]]:\n    gh = phash_value(gf)\n    if gh is None:\n        return (None, None)\n    best_p, best_d = None, None\n    for ap, ah in auth_hashes:\n        if ah is None:\n            continue\n        d = hamming_distance(gh, ah)\n        if best_d is None or d < best_d:\n            best_p, best_d = ap, d\n    return (best_p, best_d)\n\ndef best_match_by_ssim(gf: Path, auth_imgs: List[Tuple[Path, 'np.ndarray']], max_side=512) -> Tuple[Optional[Path], Optional[float]]:\n    import numpy as np\n    gi = load_gray_resized_cv2(gf, max_side=max_side)\n    if gi is None:\n        return (None, None)\n    best_p, best_sc = None, None\n    gh, gw = gi.shape[:2]\n    for ap, ai in auth_imgs:\n        if ai is None:\n            continue\n        ah, aw = ai.shape[:2]\n        # Simple alignment if dimensions differ\n        if (ah, aw) != (gh, gw):\n            ai_r = cv2.resize(ai, (gw, gh), interpolation=cv2.INTER_AREA)\n        else:\n            ai_r = ai\n        try:\n            sc = ssim(gi, ai_r)\n        except Exception:\n            continue\n        if best_sc is None or sc > best_sc:\n            best_p, best_sc = ap, sc\n    return (best_p, best_sc)\n\n# ------------------- Main Flow -------------------\ndef main():\n    # 1) List image files\n    auth_files = list_images(DIR_TRAIN_AUTH)\n    forg_files = list_images(DIR_TRAIN_FORG)\n    if DEBUG_LIMIT_FORGED is not None:\n        forg_files = forg_files[:DEBUG_LIMIT_FORGED]\n\n    print(f\"#auth = {len(auth_files)}, #forg = {len(forg_files)}\")\n\n    # 2) Pairing based on file names\n    auth_map = defaultdict(list)\n    for p in auth_files:\n        auth_map[normalize_name(p)].append(p)\n\n    pairs: List[Tuple[Path, Path, str, float]] = []\n    unmatched_forg: List[Path] = []\n\n    for gf in forg_files:\n        key = normalize_name(gf)\n        if key in auth_map and len(auth_map[key]) > 0:\n            # If multiple candidates exist, take the first for now\n            ap = auth_map[key][0]\n            pairs.append((ap, gf, \"name\", 0.0))\n        else:\n            unmatched_forg.append(gf)\n\n    print(f\"Name-matched pairs: {len(pairs)}\")\n    print(f\"Unmatched forged by name: {len(unmatched_forg)}\")\n\n    # 3) pHash matching for unmatched images\n    still_unmatched: List[Path] = unmatched_forg\n    if USE_PHASH and len(still_unmatched) > 0:\n        print(\"Computing pHash for authentic images...\")\n        auth_hashes = []\n        for ap in auth_files:\n            try:\n                ah = phash_value(ap)\n            except Exception:\n                ah = None\n            auth_hashes.append((ap, ah))\n\n        print(\"Matching unmatched forged by pHash...\")\n        newly_paired = 0\n        next_unmatched = []\n        for gf in still_unmatched:\n            ap, dist = best_match_by_phash(gf, auth_hashes)\n            if ap is not None and dist is not None and dist <= PHASH_THRESHOLD:\n                pairs.append((ap, gf, \"phash\", float(dist)))\n                newly_paired += 1\n            else:\n                next_unmatched.append(gf)\n        still_unmatched = next_unmatched\n        print(f\"pHash new pairs (<= {PHASH_THRESHOLD}): {newly_paired}\")\n        print(f\"Remaining unmatched after pHash: {len(still_unmatched)}\")\n\n    # 4) Optional SSIM matching\n    if USE_SSIM and len(still_unmatched) > 0:\n        print(\"Preloading grayscale resized authentic images for SSIM...\")\n        auth_imgs = []\n        for ap in auth_files:\n            try:\n                ai = load_gray_resized_cv2(ap, max_side=SSIM_MAX_SIDE)\n            except Exception:\n                ai = None\n            auth_imgs.append((ap, ai))\n\n        print(\"Matching unmatched forged by SSIM...\")\n        newly_paired = 0\n        next_unmatched = []\n        for gf in still_unmatched:\n            try:\n                ap, sc = best_match_by_ssim(gf, auth_imgs, max_side=SSIM_MAX_SIDE)\n            except Exception:\n                ap, sc = (None, None)\n            if ap is not None and sc is not None and sc >= SSIM_THRESHOLD:\n                pairs.append((ap, gf, \"ssim\", float(sc)))\n                newly_paired += 1\n            else:\n                next_unmatched.append(gf)\n        still_unmatched = next_unmatched\n        print(f\"SSIM new pairs (>= {SSIM_THRESHOLD}): {newly_paired}\")\n        print(f\"Remaining unmatched after SSIM: {len(still_unmatched)}\")\n\n    # 5) Save output CSV\n    OUTPUT_CSV.parent.mkdir(parents=True, exist_ok=True)\n    with open(OUTPUT_CSV, \"w\", newline=\"\", encoding=\"utf-8\") as f:\n        writer = csv.writer(f)\n        writer.writerow([\"auth_path\", \"forg_path\", \"method\", \"score\"])\n        for ap, gf, m, s in pairs:\n            writer.writerow([str(ap), str(gf), m, s])\n\n    print(f\"\\nSaved pairs: {len(pairs)} -> {OUTPUT_CSV}\")\n    if len(still_unmatched) > 0:\n        print(\"Sample unmatched forged (up to 10):\")\n        for g in still_unmatched[:10]:\n            print(\" -\", g)\n\nif __name__ == \"__main__\":\n    main()","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-11-08T21:35:32.083088Z","iopub.execute_input":"2025-11-08T21:35:32.083318Z","iopub.status.idle":"2025-11-08T21:37:48.360181Z","shell.execute_reply.started":"2025-11-08T21:35:32.083294Z","shell.execute_reply":"2025-11-08T21:37:48.359518Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"markdown","source":"# ReCoDAI | Mask R-CNN + ELA (RGB+ELA 4ch)","metadata":{}},{"cell_type":"code","source":"###############################################\n# ReCoDAI | Mask R-CNN + ELA (RGB+ELA 4ch)\n###############################################\n\nimport os, gc, json, random, warnings\nwarnings.filterwarnings(\"ignore\")\nfrom pathlib import Path\nfrom typing import List, Tuple\n\nos.environ.setdefault(\"CUDA_LAUNCH_BLOCKING\", \"1\")\n\nimport numpy as np\nimport pandas as pd\nfrom PIL import Image\nfrom tqdm import tqdm\nimport cv2\nimport scipy.optimize\n\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nimport torchvision\nimport torchvision.transforms as T\nfrom torch.utils.data import Dataset, DataLoader, ConcatDataset, Subset\n\n# =============== Seeding & CuDNN ===============\ndef seed_all(s=42):\n    random.seed(s); np.random.seed(s); torch.manual_seed(s)\nGLOBAL_SEED = 42\nseed_all(GLOBAL_SEED)\n\nimport torch.backends.cudnn as cudnn\ncudnn.benchmark = False\ncudnn.deterministic = True\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(\"Device:\", device)\n\n# =============== Paths ===============\nROOT         = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nAUTH_DIR     = f\"{ROOT}/train_images/authentic\"\nFORG_DIR     = f\"{ROOT}/train_images/forged\"\nMASK_DIR     = f\"{ROOT}/train_masks\"\nTEST_DIR     = f\"{ROOT}/test_images\"\nSAMPLE_SUB   = f\"{ROOT}/sample_submission.csv\"\nPAIRS_CSV    = \"/kaggle/working/pairs.csv\"\n\n# =============== Hyperparams ===============\nIMG_SIZE            = 128\nMIN_INSTANCE_AREA   = 16   \n\nEPOCHS              = 4  \nBATCH_SIZE          = 2\nNUM_WORKERS         = max(2, (os.cpu_count() or 4)//2)\n\n\nUSE_AMP             = False\n\n\nBASE_LR             = 4e-4\nWARMUP_ITERS        = 500\nWARMUP_START_LR     = 1e-5\nWEIGHT_DECAY        = 1e-4\nMOMENTUM            = 0.9\nMAX_NORM            = 1.0     # grad clip\n\n# Inference thresholds \nSCORE_THR           = 0.15\nMASK_THR            = 0.30\nNMS_THR             = 0.30\n\nELA_JPEG_Q          = 95\n\n# =============== Official metric ===============\nimport numba\nimport numpy.typing as npt\n\nclass ParticipantVisibleError(Exception): pass\n\n@numba.jit(nopython=True)\ndef _rle_encode_jit(x: npt.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[npt.NDArray], fg_val: int = 1) -> str:\n    return ';'.join([json.dumps(_rle_encode_jit(x, fg_val)) for x in masks])\n\n@numba.njit\ndef _rle_decode_jit(mask_rle: npt.NDArray, height: int, width: int) -> npt.NDArray:\n    if len(mask_rle) % 2 != 0:\n        raise ValueError('One or more rows has an odd number of values.')\n    starts, lengths = mask_rle[0::2], mask_rle[1::2]\n    starts -= 1\n    ends = starts + lengths\n    for i in range(len(starts) - 1):\n        if ends[i] > starts[i + 1]:\n            raise ValueError('Pixels must not be overlapping.')\n    img = np.zeros(height * width, dtype=np.bool_)\n    for lo, hi in zip(starts, ends):\n        img[lo:hi] = 1\n    return img\n\ndef rle_decode(mask_rle: str, shape: Tuple[int,int]) -> npt.NDArray:\n    mask_rle = json.loads(mask_rle)\n    mask_rle = np.asarray(mask_rle, dtype=np.int32)\n    starts = mask_rle[0::2]\n    if sorted(starts) != list(starts):\n        raise ParticipantVisibleError('Submitted values must be in ascending order.')\n    try:\n        return _rle_decode_jit(mask_rle, shape[0], shape[1]).reshape(shape, order='F')\n    except ValueError as e:\n        raise ParticipantVisibleError(str(e)) from e\n\ndef calculate_f1_score(pred_mask: npt.NDArray, gt_mask: npt.NDArray):\n    pred_flat = pred_mask.flatten()\n    gt_flat = gt_mask.flatten()\n    tp = np.sum((pred_flat == 1) & (gt_flat == 1))\n    fp = np.sum((pred_flat == 1) & (gt_flat == 0))\n    fn = np.sum((pred_flat == 0) & (gt_flat == 1))\n    precision = tp / (tp + fp) if (tp + fp) > 0 else 0\n    recall    = tp / (tp + fn) if (tp + fn) > 0 else 0\n    return 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0\n\ndef calculate_f1_matrix(pred_masks: List[npt.NDArray], gt_masks: List[npt.NDArray]):\n    num_instances_pred = len(pred_masks)\n    num_instances_gt = len(gt_masks)\n    f1_matrix = np.zeros((num_instances_pred, num_instances_gt))\n    for i in range(num_instances_pred):\n        for j in range(num_instances_gt):\n            pred_flat = pred_masks[i].flatten()\n            gt_flat = gt_masks[j].flatten()\n            f1_matrix[i, j] = calculate_f1_score(pred_flat, gt_flat)\n    if f1_matrix.shape[0] < len(gt_masks):\n        f1_matrix = np.vstack((f1_matrix, np.zeros((len(gt_masks) - len(f1_matrix), num_instances_gt))))\n    return f1_matrix\n\ndef oF1_score(pred_masks: List[npt.NDArray], gt_masks: List[npt.NDArray]):\n    f1_matrix = calculate_f1_matrix(pred_masks, gt_masks)\n    row_ind, col_ind = scipy.optimize.linear_sum_assignment(-f1_matrix)\n    if len(row_ind)==0: return 0.0\n    excess_predictions_penalty = len(gt_masks) / max(len(pred_masks), len(gt_masks)) if max(len(pred_masks), len(gt_masks))>0 else 0.0\n    return np.mean(f1_matrix[row_ind, col_ind]) * excess_predictions_penalty\n\ndef evaluate_single_image(label_rles: str, prediction_rles: str, shape_str: str) -> float:\n    shape = json.loads(shape_str)\n    label_rles = [rle_decode(x, shape=tuple(shape)) for x in label_rles.split(';')]\n    prediction_rles = [rle_decode(x, shape=tuple(shape)) for x in prediction_rles.split(';')]\n    return oF1_score(prediction_rles, label_rles)\n\ndef score(solution: pd.DataFrame, submission: pd.DataFrame, row_id_column_name: str) -> float:\n    df = solution.copy().rename(columns={'annotation': 'label'})\n    df['prediction'] = submission['annotation']\n    authentic = (df['label'] == 'authentic') | (df['prediction'] == 'authentic')\n    df['image_score'] = ((df['label'] == df['prediction']) & authentic).astype(float)\n    df.loc[~authentic, 'image_score'] = df.loc[~authentic].apply(\n        lambda row: evaluate_single_image(row['label'], row['prediction'], row['shape']), axis=1\n    )\n    return float(np.mean(df['image_score']))\n\n# =============== Data & Splits ===============\ndef list_images(root: str) -> List[str]:\n    exts = (\".png\",\".jpg\",\".jpeg\",\".tif\",\".tiff\",\".bmp\")\n    return sorted([str(p) for p in Path(root).rglob(\"*\") if p.suffix.lower() in exts])\n\npairs_df = pd.read_csv(PAIRS_CSV)\nval_df = pairs_df.sample(frac=0.2, random_state=GLOBAL_SEED)\ntrain_df = pairs_df.drop(val_df.index)\n\nauth_all = list_images(AUTH_DIR)\nauth_val_count = max(1, int(0.2*len(auth_all)))\nrng = np.random.default_rng(GLOBAL_SEED)\nperm = rng.permutation(len(auth_all))\nauth_val_idx = set(perm[:auth_val_count].tolist())\nauth_train = [auth_all[i] for i in range(len(auth_all)) if i not in auth_val_idx]\nauth_val   = [auth_all[i] for i in range(len(auth_all)) if i in auth_val_idx]\n\nprint(f\"Train forged: {len(train_df)} | Val forged: {len(val_df)}\")\nprint(f\"Train authentic (info): {len(auth_train)} | Val authentic: {len(auth_val)}\")\n\n# =============== ELA helper ===============\ndef ela_map_gray_uint8(img_np_bgr, q=ELA_JPEG_Q):\n    _, enc = cv2.imencode('.jpg', img_np_bgr, [int(cv2.IMWRITE_JPEG_QUALITY), q])\n    dec = cv2.imdecode(enc, cv2.IMREAD_COLOR)\n    if dec is None: dec = img_np_bgr.copy()\n    diff = cv2.absdiff(img_np_bgr, dec).astype(np.float32)\n    m = float(np.percentile(diff, 99)); m = max(m, 1.0)\n    ela = np.clip(diff / m, 0, 1.0)\n    ela_gray = cv2.cvtColor((ela*255).astype(np.uint8), cv2.COLOR_BGR2GRAY).astype(np.float32)/255.0\n    return ela_gray\n\n# =============== Box helpers (fix) ===============\nfrom torchvision.ops import masks_to_boxes, nms\n\ndef _fix_boxes_inplace(boxes: torch.Tensor, H: int, W: int) -> torch.Tensor:\n    if boxes.numel() == 0: return boxes\n    boxes[:, 0] = boxes[:, 0].clamp(0, W - 1)  # x1\n    boxes[:, 2] = boxes[:, 2].clamp(0, W - 1)  # x2\n    boxes[:, 1] = boxes[:, 1].clamp(0, H - 1)  # y1\n    boxes[:, 3] = boxes[:, 3].clamp(0, H - 1)  # y2\n    bad_w = boxes[:, 2] <= boxes[:, 0]\n    if bad_w.any():\n        boxes[bad_w, 0] = (boxes[bad_w, 0] - 1).clamp(0, W - 2)\n        boxes[bad_w, 2] = (boxes[bad_w, 0] + 1).clamp(1, W - 1)\n    bad_h = boxes[:, 3] <= boxes[:, 1]\n    if bad_h.any():\n        boxes[bad_h, 1] = (boxes[bad_h, 1] - 1).clamp(0, H - 2)\n        boxes[bad_h, 3] = (boxes[bad_h, 1] + 1).clamp(1, H - 1)\n    return boxes\n\ndef _instances_to_valid_boxes(masks_np: np.ndarray, H: int, W: int) -> torch.Tensor:\n    if masks_np.size == 0:\n        return torch.zeros((0,4), dtype=torch.float32)\n    m = torch.from_numpy(masks_np.astype(np.uint8))\n    boxes = masks_to_boxes(m)  # [N,4] float32\n    boxes = _fix_boxes_inplace(boxes, H, W)\n    return boxes\n\n# =============== Datasets (4ch: RGB+ELA) ===============\nclass ForgeryInstancesDataset(Dataset):\n    def __init__(self, df_pairs: pd.DataFrame, mask_dir: str, train: bool):\n        self.df = df_pairs.reset_index(drop=True)\n        self.mask_dir = mask_dir\n        self.train = train\n        self.tf_resize = T.Resize((IMG_SIZE, IMG_SIZE))\n        self.to_tensor = T.ToTensor()\n\n    def __len__(self): return len(self.df)\n\n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        pf = row['forg_path']  # تصویر دستکاری‌شده\n        img_pil = Image.open(pf).convert(\"RGB\")\n\n        # GT mask\n        stem = Path(pf).stem\n        mp = os.path.join(self.mask_dir, stem + \".npy\")\n        if os.path.exists(mp):\n            m = np.load(mp)\n            if m.ndim==3: m = np.max(m, axis=0)\n        else:\n            m = np.zeros(img_pil.size[::-1], dtype=np.uint8)  # (H,W)\n        m_bin = (m > 0).astype(np.uint8)\n\n        # ELA\n        img_bgr = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)\n        ela = ela_map_gray_uint8(img_bgr, q=ELA_JPEG_Q)\n\n        # Resize \n        img_pil = self.tf_resize(img_pil)\n        ela_rs  = cv2.resize(ela, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_LINEAR)\n        mask_rs = cv2.resize(m_bin, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n\n        # 4ch \n        rgb   = self.to_tensor(img_pil)                         # [3,H,W] 0..1\n        ela_t = torch.from_numpy(ela_rs).float().unsqueeze(0)   # [1,H,W]\n        img4  = torch.cat([rgb, ela_t], dim=0)                  # [4,H,W]\n\n        # \n        num, labels = cv2.connectedComponents(mask_rs, connectivity=8)\n        insts = []\n        for k in range(1, num):\n            inst = (labels == k).astype(np.uint8)\n            if int(inst.sum()) >= MIN_INSTANCE_AREA:\n                insts.append(inst)\n\n        if len(insts) == 0:\n            target = {\n                \"boxes\": torch.zeros((0,4), dtype=torch.float32),\n                \"labels\": torch.zeros((0,), dtype=torch.int64),\n                \"masks\": torch.zeros((0, IMG_SIZE, IMG_SIZE), dtype=torch.uint8),\n                \"image_id\": torch.tensor([idx]),\n                \"area\": torch.zeros((0,), dtype=torch.float32),\n                \"iscrowd\": torch.zeros((0,), dtype=torch.int64),\n            }\n        else:\n            masks_t = torch.from_numpy(np.stack(insts, 0)).to(torch.uint8)  # [N,H,W]\n            boxes_t = _instances_to_valid_boxes(masks_t.numpy(), IMG_SIZE, IMG_SIZE)  # [N,4]\n            keep = (boxes_t[:,2] > boxes_t[:,0]) & (boxes_t[:,3] > boxes_t[:,1])\n            if keep.sum().item() < boxes_t.size(0):\n                masks_t = masks_t[keep]\n                boxes_t = boxes_t[keep]\n            labels_t= torch.ones((masks_t.size(0),), dtype=torch.int64)   # forgery=1\n            area_t  = masks_t.sum(dim=(1,2)).float()\n            iscrowd = torch.zeros((masks_t.size(0),), dtype=torch.int64)\n            target = {\n                \"boxes\": boxes_t, \"labels\": labels_t, \"masks\": masks_t,\n                \"image_id\": torch.tensor([idx]), \"area\": area_t, \"iscrowd\": iscrowd\n            }\n\n        return img4, target\n\nclass AuthenticEmptyDataset(Dataset):\n    \n    def __init__(self, paths: List[str]):\n        self.paths = paths\n        self.tf_resize = T.Resize((IMG_SIZE, IMG_SIZE))\n        self.to_tensor = T.ToTensor()\n    def __len__(self): return len(self.paths)\n    def __getitem__(self, idx):\n        p = self.paths[idx]\n        img = Image.open(p).convert(\"RGB\")\n        ela = ela_map_gray_uint8(cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR), q=ELA_JPEG_Q)\n        img  = self.tf_resize(img)\n        ela  = cv2.resize(ela, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_LINEAR)\n        x4   = torch.cat([self.to_tensor(img), torch.from_numpy(ela).float().unsqueeze(0)], dim=0)\n        target = {\n            \"boxes\": torch.zeros((0,4), dtype=torch.float32),\n            \"labels\": torch.zeros((0,), dtype=torch.int64),\n            \"masks\": torch.zeros((0, IMG_SIZE, IMG_SIZE), dtype=torch.uint8),\n            \"image_id\": torch.tensor([idx]),\n            \"area\": torch.zeros((0,), dtype=torch.float32),\n            \"iscrowd\": torch.zeros((0,), dtype=torch.int64),\n        }\n        return x4, target\n\ndef collate_fn(batch):\n    imgs, tgts = list(zip(*batch))\n    return list(imgs), list(tgts)\n\ntrain_ds_pos = ForgeryInstancesDataset(train_df, MASK_DIR, train=True)\nval_ds_pos   = ForgeryInstancesDataset(val_df,   MASK_DIR, train=False)\nneg_train    = AuthenticEmptyDataset(auth_train)\nneg_val      = AuthenticEmptyDataset(auth_val)\n\n\nk = min(len(train_ds_pos), len(neg_train))\nrng = np.random.default_rng(GLOBAL_SEED)\nidxs = rng.choice(len(neg_train), k, replace=False)\nneg_train_small = Subset(neg_train, idxs)\n\ntrain_mix = ConcatDataset([train_ds_pos, neg_train_small])\nval_mix   = ConcatDataset([val_ds_pos, neg_val])\n\ntrain_loader = DataLoader(train_mix, batch_size=BATCH_SIZE, shuffle=True,\n                          num_workers=NUM_WORKERS, pin_memory=True,\n                          collate_fn=collate_fn, persistent_workers=False)\nval_loader   = DataLoader(val_mix, batch_size=BATCH_SIZE, shuffle=False,\n                          num_workers=NUM_WORKERS, pin_memory=True,\n                          collate_fn=collate_fn, persistent_workers=False)\n\nprint(f\"[RCNN] Train samples (pos+neg): {len(train_mix)} | Val samples (pos+neg): {len(val_mix)}\")\n\n# =============== Model (4ch, from scratch) ===============\nfrom torchvision.models.detection import maskrcnn_resnet50_fpn\nfrom torchvision.models.detection.faster_rcnn import FastRCNNPredictor\nfrom torchvision.models.detection.mask_rcnn import MaskRCNNPredictor\n\ndef maskrcnn_resnet50_fpn_4ch_from_scratch(num_classes=2):\n    model = maskrcnn_resnet50_fpn(weights=None, weights_backbone=None)\n    # conv1 چهارکاناله\n    old_conv1 = model.backbone.body.conv1\n    new_conv1 = nn.Conv2d(4, old_conv1.out_channels, kernel_size=old_conv1.kernel_size,\n                          stride=old_conv1.stride, padding=old_conv1.padding, bias=False)\n    with torch.no_grad():\n        nn.init.kaiming_normal_(new_conv1.weight, mode='fan_out', nonlinearity='relu')\n    model.backbone.body.conv1 = new_conv1\n\n    # heads\n    in_features = model.roi_heads.box_predictor.cls_score.in_features\n    model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)\n    in_features_mask = model.roi_heads.mask_predictor.conv5_mask.in_channels\n    hidden_layer = 256\n    model.roi_heads.mask_predictor = MaskRCNNPredictor(in_features_mask, hidden_layer, num_classes)\n\n    # init heads\n    for m in [model.roi_heads.box_predictor.cls_score,\n              model.roi_heads.box_predictor.bbox_pred,\n              model.roi_heads.mask_predictor.conv5_mask,\n              model.roi_heads.mask_predictor.mask_fcn_logits]:\n        if hasattr(m, 'weight') and m.weight is not None:\n            nn.init.normal_(m.weight, std=0.01)\n        if hasattr(m, 'bias') and m.bias is not None:\n            nn.init.constant_(m.bias, 0)\n\n    # Normalize \n    model.transform.image_mean = [0.0, 0.0, 0.0, 0.0]\n    model.transform.image_std  = [1.0, 1.0, 1.0, 1.0]\n    model.transform.min_size = (IMG_SIZE,)\n    model.transform.max_size = IMG_SIZE\n    return model\n\n# build\ntry:\n    del model\n    torch.cuda.empty_cache()\nexcept:\n    pass\n\nmodel = maskrcnn_resnet50_fpn_4ch_from_scratch(num_classes=2).to(device)\n\nMULTI_GPU = (torch.cuda.is_available() and torch.cuda.device_count() > 1)\n\n# =============== Optimizer + Warmup ===============\noptimizer = optim.SGD([p for p in model.parameters() if p.requires_grad],\n                      lr=WARMUP_START_LR, momentum=MOMENTUM, weight_decay=WEIGHT_DECAY)\n\nglobal_iter = 0\ndef warmup_lr(it):\n    if it >= WARMUP_ITERS:\n        return BASE_LR\n    \n    return WARMUP_START_LR + (BASE_LR - WARMUP_START_LR) * (it / max(1, WARMUP_ITERS))\n\n# =============== Utilities ===============\ndef is_finite_tensor(x: torch.Tensor) -> bool:\n    return torch.isfinite(x).all().item()\n\n@torch.no_grad()\ndef quick_val_stats(model, loader):\n    model.eval()\n    n_imgs, n_preds, n_auth_preds = 0, 0, 0\n    for imgs, _ in loader:\n        outs = model([im.to(device) for im in imgs])\n        for out in outs:\n            n_imgs += 1\n            scores = out.get(\"scores\", torch.tensor([]))\n            k = int((scores.detach().cpu() >= SCORE_THR).sum())\n            n_preds += k\n            if k == 0: n_auth_preds += 1\n    print(f\"[VAL] images={n_imgs} | avg_preds_per_image={n_preds/max(1,n_imgs):.2f} | authentic_preds%={100*n_auth_preds/max(1,n_imgs):.1f}%\")\n\n# RLE sanity test\nm = (np.random.rand(64,64) > 0.7).astype(np.uint8)\nenc = rle_encode([m])\n_dec = rle_decode(enc.split(';')[0], (64,64))\nassert np.all(m == _dec), \"RLE encode/decode mismatch!\"\n\n# =============== Train (NaN-safe) ===============\nfor epoch in range(1, EPOCHS+1):\n    model.train()\n    loss_hist = []\n    pbar = tqdm(train_loader, desc=f\"RCNN Train {epoch}/{EPOCHS}\")\n    for imgs, tgts in pbar:\n        # warmup lr\n        lr_now = warmup_lr(global_iter)\n        for g in optimizer.param_groups: g['lr'] = lr_now\n        global_iter += 1\n\n        imgs = [im.to(device) for im in imgs]\n        tgts = [{k: (v.to(device) if torch.is_tensor(v) else v) for k,v in t.items()} for t in tgts]\n\n        \n        loss_dict = model(imgs, tgts)   # dict of losses\n        \n        bad = False\n        total = 0.0\n        for k, v in loss_dict.items():\n            if not is_finite_tensor(v):\n                bad = True; break\n            total += float(v.detach().cpu().item())\n        if bad or not np.isfinite(total):\n            optimizer.zero_grad(set_to_none=True)\n            continue\n\n        loss = sum(loss_dict.values())\n        if not is_finite_tensor(loss):\n            optimizer.zero_grad(set_to_none=True)\n            continue\n\n        optimizer.zero_grad(set_to_none=True)\n        loss.backward()\n        torch.nn.utils.clip_grad_norm_(model.parameters(), MAX_NORM)\n        optimizer.step()\n\n        lv = float(loss.detach().cpu().item())\n        loss_hist.append(lv)\n        pbar.set_postfix(loss=np.mean(loss_hist) if len(loss_hist) else lv)\n\n    torch.cuda.empty_cache(); gc.collect()\n    \n    quick_val_stats(model, val_loader)\n\n# =============== Validation (official metric) ===============\n@torch.no_grad()\ndef build_val_gt_and_pred_rcnn() -> Tuple[pd.DataFrame, pd.DataFrame]:\n    model.eval()\n    rows_gt, rows_sub = [], []\n    for imgs, tgts in tqdm(val_loader, desc=\"RCNN Val\"):\n        imgs = [im.to(device) for im in imgs]\n        outs = model(imgs)\n\n        for im, tgt, out in zip(imgs, tgts, outs):\n            H, W = im.shape[-2], im.shape[-1]\n            masks_gt = tgt[\"masks\"].numpy().astype(np.uint8) if tgt[\"masks\"].numel()>0 else np.zeros((0,H,W), np.uint8)\n            if masks_gt.shape[0]==0:\n                rows_gt.append({\"row_id\": f\"val_{len(rows_gt)}\", \"annotation\": \"authentic\", \"shape\": \"authentic\"})\n            else:\n                rows_gt.append({\"row_id\": f\"val_{len(rows_gt)}\", \"annotation\": rle_encode([m for m in masks_gt]), \"shape\": json.dumps([H,W])})\n\n            if (\"scores\" not in out) or (len(out[\"scores\"])==0):\n                pred = \"authentic\"\n            else:\n                boxes = out[\"boxes\"].detach().cpu()\n                scores= out[\"scores\"].detach().cpu()\n                keep = nms(boxes, scores, NMS_THR)\n                boxes, scores = boxes[keep], scores[keep]\n                keep2 = scores >= SCORE_THR\n                boxes, scores = boxes[keep2], scores[keep2]\n\n                insts = []\n                if \"masks\" in out and len(out[\"masks\"])>0 and len(keep)>0:\n                    masks = out[\"masks\"].detach().cpu()[keep][:,0]    # [K,H,W]\n                    masks = masks[keep2]\n                    for m in masks:\n                        mb = (m.numpy() >= MASK_THR).astype(np.uint8)\n                        if mb.sum() >= MIN_INSTANCE_AREA:\n                            insts.append(mb)\n                pred = rle_encode(insts) if len(insts)>0 else \"authentic\"\n\n            rows_sub.append({\"row_id\": rows_gt[-1][\"row_id\"], \"annotation\": pred})\n\n    gt_df = pd.DataFrame(rows_gt)\n    sub_df= pd.DataFrame(rows_sub).reindex(range(len(rows_gt)))\n    return gt_df, sub_df\n\ngt_df_val, sub_df_val = build_val_gt_and_pred_rcnn()\noverall_val = score(gt_df_val.copy(), sub_df_val.copy(), row_id_column_name=\"row_id\")\nprint(\"\\n====== RCNN Validation (OFFICIAL METRIC) ======\")\nprint(f\"Overall: {overall_val:.6f}\")\nprint(\"===============================================\\n\")\n\n# =============== Test + submission ===============\n@torch.no_grad()\ndef predict_one_test_rcnn(image_path: str) -> str:\n    img = Image.open(image_path).convert(\"RGB\")\n    rgb = T.Resize((IMG_SIZE, IMG_SIZE))(img)\n    rgb_t = T.ToTensor()(rgb)\n    ela = ela_map_gray_uint8(cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR), q=ELA_JPEG_Q)\n    ela_rs = cv2.resize(ela, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_LINEAR)\n    ela_t = torch.from_numpy(ela_rs).float().unsqueeze(0)\n    x4 = torch.cat([rgb_t, ela_t], dim=0).unsqueeze(0).to(device)\n\n    out = model(x4)[0]\n    if (\"scores\" not in out) or (len(out[\"scores\"])==0):\n        return \"authentic\"\n    boxes = out[\"boxes\"].detach().cpu()\n    scores= out[\"scores\"].detach().cpu()\n    keep = nms(boxes, scores, NMS_THR)\n    boxes, scores = boxes[keep], scores[keep]\n    keep2 = scores >= SCORE_THR\n    if keep2.sum()==0: return \"authentic\"\n    masks = out[\"masks\"].detach().cpu()[keep][:,0][keep2]  # [K,H,W]\n    insts = []\n    for m in masks:\n        mb = (m.numpy() >= MASK_THR).astype(np.uint8)\n        if mb.sum() >= MIN_INSTANCE_AREA:\n            insts.append(mb)\n    return rle_encode(insts) if len(insts)>0 else \"authentic\"\n\ndef build_submission_rcnn(test_dir: str, sample_csv_path: str, out_csv_path: str = \"submission_rcnn.csv\"):\n    if os.path.exists(sample_csv_path):\n        df_sub = pd.read_csv(sample_csv_path)\n        id_col, target_col = df_sub.columns[0], df_sub.columns[1]\n        test_files = [str(p) for p in Path(test_dir).glob(\"*\")]\n        by_stem = {Path(p).stem: p for p in test_files}\n        by_name = {Path(p).name: p for p in test_files}\n        preds = []\n        for rid in tqdm(df_sub[id_col].astype(str), desc=\"Predicting test (RCNN)\"):\n            cand = by_name.get(rid) or by_stem.get(rid)\n            if cand is None:\n                cand = next((by_name.get(rid+ext) for ext in (\".png\",\".jpg\",\".jpeg\",\".tif\",\".tiff\",\".bmp\") if by_name.get(rid+ext)), None)\n            preds.append(\"authentic\" if cand is None else predict_one_test_rcnn(cand))\n        out = df_sub.copy()\n        out[target_col] = preds\n        out.to_csv(out_csv_path, index=False)\n    else:\n        test_files = sorted([str(p) for p in Path(test_dir).glob(\"*\") if p.lower().endswith((\".png\",\".jpg\",\".jpeg\",\".tif\",\".tiff\",\".bmp\"))])\n        preds = [predict_one_test_rcnn(p) for p in tqdm(test_files, desc=\"Predicting test (RCNN)\")]\n        out = pd.DataFrame({\"case_id\":[Path(p).stem for p in test_files], \"annotation\":preds})\n        out.to_csv(out_csv_path, index=False)\n    print(f\"Saved submission to {out_csv_path}\")\n    return out_csv_path\n\nif os.path.exists(TEST_DIR):\n    build_submission_rcnn(TEST_DIR, SAMPLE_SUB, out_csv_path=\"submission.csv\")\n\ntorch.cuda.empty_cache(); gc.collect()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T21:37:48.361689Z","iopub.execute_input":"2025-11-08T21:37:48.362021Z","iopub.status.idle":"2025-11-08T22:12:15.529758Z","shell.execute_reply.started":"2025-11-08T21:37:48.362001Z","shell.execute_reply":"2025-11-08T22:12:15.528960Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ============================================================\n# ReCoDAI | Visualization for Mask R-CNN (RGB+ELA, 4ch)\n# Shows Authentic | Forged | Predicted (and GT if exists)\n# ============================================================\nimport os, cv2, torch, numpy as np, matplotlib.pyplot as plt\nfrom pathlib import Path\nfrom PIL import Image\nimport torchvision.transforms as T\nfrom torchvision.ops import nms\n\nVIS_SCORE_THR = 0.05     \nVIS_MASK_THR  = 0.20     \nVIS_NMS_THR   = 0.50\nTOPK          = 5        \n\n\ndef make_4ch_tensor(pil_img, img_size=IMG_SIZE, ela_q=ELA_JPEG_Q, device=device):\n    rgb = T.Resize((img_size, img_size))(pil_img)\n    rgb_t = T.ToTensor()(rgb)  # [3,H,W] in [0,1]\n    # ELA\n    ela = ela_map_gray_uint8(cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR), q=ela_q)\n    ela_rs = cv2.resize(ela, (img_size, img_size), interpolation=cv2.INTER_LINEAR)\n    ela_t = torch.from_numpy(ela_rs).float().unsqueeze(0)  # [1,H,W]\n    x4 = torch.cat([rgb_t, ela_t], dim=0)  # [4,H,W]\n    return x4.to(device)\n\ndef to_rgb_disp(x4_tensor):\n    \n    x = x4_tensor.detach().float().cpu().numpy()\n    x = np.transpose(x[:3], (1,2,0))\n    return np.clip(x*255, 0, 255).astype(np.uint8)\n\n@torch.no_grad()\ndef rcnn_predict_masks(pil_img,\n                       score_thr=VIS_SCORE_THR,\n                       mask_thr=VIS_MASK_THR,\n                       nms_thr=VIS_NMS_THR,\n                       topk=TOPK):\n    \n    assert 'model' in globals(), \"Model not found. Make sure 'model' is defined.\"\n    model.eval()\n    x = make_4ch_tensor(pil_img).unsqueeze(0)\n    out = model(x)[0]\n\n    if (\"scores\" not in out) or (len(out[\"scores\"]) == 0):\n        return []\n\n    boxes  = out[\"boxes\"].detach().cpu()\n    scores = out[\"scores\"].detach().cpu()\n\n    keep = nms(boxes, scores, nms_thr)\n    boxes, scores = boxes[keep], scores[keep]\n\n    order = torch.argsort(scores, descending=True)\n    boxes, scores = boxes[order], scores[order]\n    if len(scores) > topk:\n        boxes, scores = boxes[:topk], scores[:topk]\n\n    masks = []\n    if \"masks\" in out and len(out[\"masks\"]) > 0:\n        m_all = out[\"masks\"].detach().cpu()[keep][:,0][order]\n        if len(m_all) > topk: m_all = m_all[:topk]\n        for m, sc in zip(m_all, scores):\n            sc = float(sc.item())\n            if sc < score_thr: \n                continue\n            mb = (m.numpy() >= mask_thr).astype(np.uint8)\n            if int(mb.sum()) >= MIN_INSTANCE_AREA:\n                masks.append((mb, sc))\n    return masks\n\ndef resize_mask_to(img_size_wh, mask_small):\n    \n    W, H = img_size_wh\n    return cv2.resize(mask_small.astype(np.uint8), (W, H), interpolation=cv2.INTER_NEAREST)\n\ndef overlay_colored(base_img_bgr, masks_scores, alpha=0.45):\n    \n    out = base_img_bgr.copy()\n    palette = [\n        (0, 0, 255),   # قرمز (BGR)\n        (0, 255, 0),   # سبز\n        (255, 0, 0),   # آبی\n        (0, 255, 255), # زرد\n        (255, 0, 255), # ارغوانی\n        (255, 255, 0), # فیروزه‌ای\n        (0, 128, 255), # نارنجی کم‌رنگ\n        (255, 0, 128), # صورتی\n    ]\n    for idx, (m, sc) in enumerate(masks_scores):\n        color = palette[idx % len(palette)]\n        color_img = np.zeros_like(out, dtype=np.uint8)\n        color_img[:] = color\n        m3 = m[..., None].astype(bool)\n        blended = (alpha * color_img + (1 - alpha) * out).astype(np.uint8)\n        out = np.where(m3, blended, out)\n\n        # کانتور + نمره\n        cnts, _ = cv2.findContours(m.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n        cv2.drawContours(out, cnts, -1, (255, 255, 255), 1)\n        if len(cnts):\n            x,y,w,h = cv2.boundingRect(np.vstack(cnts))\n            cv2.putText(out, f\"{sc:.2f}\", (x, max(12,y-4)),\n                        cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255,255,255), 1, cv2.LINE_AA)\n    return out\n\ndef load_gt_mask_if_exists(forged_path, target_size=None):\n    \n    stem = Path(forged_path).stem\n    mp = os.path.join(MASK_DIR, stem + \".npy\")\n    if os.path.exists(mp):\n        m = np.load(mp)\n        if m.ndim == 3: m = np.max(m, axis=0)\n        m = (m > 0).astype(np.uint8)\n        if target_size is not None:\n            W, H = target_size\n            m = cv2.resize(m, (W, H), interpolation=cv2.INTER_NEAREST)\n        return m\n    return None\n\ndef pil_to_bgr(img_pil):\n    return cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)\n\ndef bgr_to_rgb(img_bgr):\n    return cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)\n\n\n@torch.no_grad()\ndef visualize_pairs_with_masks(df=None, n_samples=8,\n                               upscale_to_original=True,\n                               show_gt=True,\n                               random_state=123,\n                               save_path=None):\n\n    assert 'model' in globals(), \"Model not found. Make sure 'model' is defined.\"\n    use_df = df if df is not None else val_df\n    samp = use_df.sample(n=min(n_samples, len(use_df)), random_state=random_state).reset_index(drop=True)\n\n    rows = len(samp)\n    cols = 4 if show_gt else 3\n    plt.figure(figsize=(14, 3.2*rows))\n\n    plot_idx = 1\n    for _, row in samp.iterrows():\n        ap, fp = row['auth_path'], row['forg_path']\n        A = Image.open(ap).convert(\"RGB\")\n        F = Image.open(fp).convert(\"RGB\")\n\n        # نمایش نسخه‌ی 128×128 (برای ستون Authentic/Forged)\n        a_disp_small = to_rgb_disp(make_4ch_tensor(A).unsqueeze(0)[0])\n        f_disp_small = to_rgb_disp(make_4ch_tensor(F).unsqueeze(0)[0])\n\n        \n        masks_scores_small = rcnn_predict_masks(F)  \n\n        \n        if upscale_to_original:\n            \n            W0, H0 = F.size\n            masks_scores = [(resize_mask_to((W0, H0), m), sc) for (m, sc) in masks_scores_small]\n            base_bgr = pil_to_bgr(F)\n        else:\n            masks_scores = masks_scores_small\n            base_bgr = cv2.resize(pil_to_bgr(F), (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_LINEAR)\n\n        overlay_pred_bgr = overlay_colored(base_bgr, masks_scores) if len(masks_scores) else base_bgr.copy()\n\n        \n        overlay_gt_rgb = None\n        if show_gt:\n            if upscale_to_original:\n                gt = load_gt_mask_if_exists(fp, target_size=F.size)\n                base_for_gt = np.array(F)  # RGB\n            else:\n                gt = load_gt_mask_if_exists(fp, target_size=(IMG_SIZE, IMG_SIZE))\n                base_for_gt = cv2.resize(np.array(F), (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_LINEAR)\n\n            if gt is not None:\n                green = np.zeros_like(base_for_gt)\n                green[...,1] = 255\n                overlay_gt = np.where(gt[...,None].astype(bool),\n                                      (0.45*green + 0.55*base_for_gt).astype(np.uint8),\n                                      base_for_gt)\n                cnts, _ = cv2.findContours(gt.astype(np.uint8), cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n                cv2.drawContours(overlay_gt, cnts, -1, (255,255,255), 1)\n                overlay_gt_rgb = overlay_gt\n\n        \n        ax = plt.subplot(rows, cols, plot_idx); plot_idx += 1\n        ax.imshow(a_disp_small); ax.set_title(\"Authentic\"); ax.axis('off')\n\n        ax = plt.subplot(rows, cols, plot_idx); plot_idx += 1\n        ax.imshow(f_disp_small); ax.set_title(\"Forged\"); ax.axis('off')\n\n        ax = plt.subplot(rows, cols, plot_idx); plot_idx += 1\n        ax.imshow(bgr_to_rgb(overlay_pred_bgr))\n        ttl = \"Predicted\"\n        if len(masks_scores):\n            ttl += f\" (n={len(masks_scores)})\"\n        ax.set_title(ttl); ax.axis('off')\n\n        if show_gt:\n            ax = plt.subplot(rows, cols, plot_idx); plot_idx += 1\n            if overlay_gt_rgb is not None:\n                ax.imshow(overlay_gt_rgb)\n                ax.set_title(\"GT mask\")\n            else:\n                ax.imshow(np.zeros_like(a_disp_small))\n                ax.set_title(\"GT mask (N/A)\")\n            ax.axis('off')\n\n    plt.tight_layout()\n    if save_path:\n        plt.savefig(save_path, dpi=150, bbox_inches='tight')\n        print(f\"Saved figure to: {save_path}\")\n    plt.show()\n\n\nvisualize_pairs_with_masks(df=val_df,\n                           n_samples=8,\n                           upscale_to_original=True,\n                           show_gt=True,\n                           random_state=123,\n                           save_path=None)            \n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T22:12:15.530848Z","iopub.execute_input":"2025-11-08T22:12:15.531377Z","iopub.status.idle":"2025-11-08T22:12:21.125476Z","shell.execute_reply.started":"2025-11-08T22:12:15.531356Z","shell.execute_reply":"2025-11-08T22:12:21.124320Z"}},"outputs":[],"execution_count":null}]}