{"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"}],"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":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T14:59:02.754597Z","iopub.execute_input":"2025-11-06T14:59:02.754902Z","iopub.status.idle":"2025-11-06T15:01:07.508224Z","shell.execute_reply.started":"2025-11-06T14:59:02.754879Z","shell.execute_reply":"2025-11-06T15:01:07.507417Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"###############################################\n# ReCoDAI | Paired EfficientNet-B0 — OFFICIAL METRIC + Improvements + Retrieval\n# - Official metric only (score/evaluate_single_image/oF1_score)\n# - IMG_SIZE=256, preserve masks (dilate before resize)\n# - Seg + Cls; background loss for empty masks\n# - Edge channel (Sobel) + amplified diff\n# - TTA (horizontal flip) for seg & cls\n# - Morphological postproc + grid search on seg thresholds (validation)\n# - Retrieval (name → pHash → SSIM) to build realistic diff at TEST (and optional VALIDATION)\n# - Submission: case_id,annotation | \"authentic\" or JSON-RLE via rle_encode\n###############################################\n\nimport os, gc, json, random, re\nfrom pathlib import Path\nfrom typing import List, Tuple, Optional\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 torch.nn.functional as F\nimport torchvision\nimport torchvision.transforms as T\nfrom torch.utils.data import Dataset, DataLoader\nfrom sklearn.metrics import f1_score\n\n# -------------------------\n# Repro & Device\n# -------------------------\ndef seed_everything(s=42):\n    random.seed(s); np.random.seed(s)\n    torch.manual_seed(s); torch.cuda.manual_seed_all(s)\nseed_everything(42)\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(\"Device:\", device)\n\n# -------------------------\n# Paths\n# -------------------------\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# -------------------------\n# Hyperparams\n# -------------------------\nIMG_SIZE          = 256\nBATCH_FORGED      = 2\nBATCH_AUTH_CLS    = 8\nEPOCHS            = 8\nLR                = 1e-4\nWEIGHT_DECAY      = 1e-4\nALPHA_CLS         = 0.5\nNEG_BG_WEIGHT     = 0.25  # background loss weight when mask is empty\n\n# Postproc (initial; tuned later on val)\nTHRESH_SEG        = 0.20\nMIN_INSTANCE_AREA = 4\nTOPK_INSTANCES    = 5\n\n# Detection threshold (tuned later)\nBEST_CLS_THR      = 0.5\n\n# Retrieval flags (for building realistic diff at TEST; optional at VAL)\nUSE_RETR_IN_TEST  = True\nUSE_RETR_IN_VAL   = False  # اگر True کنی، در ولیدیشن هم diff با retrieval ساخته می‌شود (Dataset باید مسیرها را بدهد)\nUSE_RETR_PHASH    = True\nPHASH_THRESHOLD   = 10\nUSE_RETR_SSIM     = False     # در صورت نیاز True شود (کندتر)\nSSIM_THRESHOLD    = 0.70\nSSIM_MAX_SIDE     = 512\n\n# -------------------------\n# Utilities\n# -------------------------\ndef list_images(root: str) -> List[str]:\n    IMG_EXTS = (\".png\",\".jpg\",\".jpeg\",\".tif\",\".tiff\",\".bmp\")\n    return sorted([str(p) for p in Path(root).rglob(\"*\") if p.suffix.lower() in IMG_EXTS])\n\n# =========================================================\n# OFFICIAL METRIC (EXACTLY as provided)\n# =========================================================\nimport numba\nimport numpy.typing as npt\n\nclass ParticipantVisibleError(Exception):\n    pass\n\n@numba.jit(nopython=True)\ndef _rle_encode_jit(x: npt.NDArray, fg_val: int = 1) -> list[int]:\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    excess_predictions_penalty = len(gt_masks) / max(len(pred_masks), len(gt_masks))\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=shape) for x in label_rles.split(';')]\n    prediction_rles = [rle_decode(x, shape=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\n    df = df.rename(columns={'annotation': 'label'})\n    df['prediction'] = submission['annotation']  # assumes same order\n    authentic_indices = (df['label'] == 'authentic') | (df['prediction'] == 'authentic')\n    df['image_score'] = ((df['label'] == df['prediction']) & authentic_indices).astype(float)\n    df.loc[~authentic_indices, 'image_score'] = df.loc[~authentic_indices].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\n# -------------------------\n# Transforms\n# -------------------------\ntrain_tf = T.Compose([\n    T.Resize((IMG_SIZE, IMG_SIZE)),\n    T.RandomHorizontalFlip(),\n    T.RandomRotation(5),\n    T.ColorJitter(0.05,0.05,0.05,0.02),\n    T.ToTensor(),\n])\nval_tf = T.Compose([\n    T.Resize((IMG_SIZE, IMG_SIZE)),\n    T.ToTensor(),\n])\n\n# -------------------------\n# Datasets\n# -------------------------\nclass PairedForgeryDataset(Dataset):\n    \"\"\"Forged paired with authentic. Seg+Cls (label=1). Keeps mask=0; adds bg loss for empty masks.\n       Optionally returns file paths for retrieval in validation.\"\"\"\n    def __init__(self, df_pairs: pd.DataFrame, mask_dir: str, train: bool, return_paths: bool=False):\n        self.df = df_pairs.reset_index(drop=True)\n        self.mask_dir = mask_dir\n        self.train = train\n        self.return_paths = return_paths\n\n    def __len__(self): return len(self.df)\n\n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        pa, pf = row['auth_path'], row['forg_path']\n        A = Image.open(pa).convert(\"RGB\")\n        F = Image.open(pf).convert(\"RGB\")\n        xa = train_tf(A) if self.train else val_tf(A)\n        xf = train_tf(F) if self.train else val_tf(F)\n\n        # difference channel (paired authentic vs forged)\n        diff = torch.abs(xf - xa)\n\n        # load mask and preserve small positives when downsampling\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((A.height, A.width), dtype=np.uint8)\n\n        H, W = m.shape\n        scale = max(H/IMG_SIZE, W/IMG_SIZE)\n        m_bin = (m > 0).astype(np.uint8)\n        if scale > 1:\n            k = int(np.ceil(scale))\n            kernel = np.ones((k, k), np.uint8)\n            m_bin = cv2.dilate(m_bin, kernel, iterations=1)\n        m_res = cv2.resize(m_bin, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n        mask = torch.from_numpy(m_res.astype(np.float32)).unsqueeze(0)\n\n        if self.return_paths:\n            return xf, diff, mask, torch.tensor(1, dtype=torch.long), pf, pa\n        return xf, diff, mask, torch.tensor(1, dtype=torch.long)\n\nclass AuthenticOnlyClsDataset(Dataset):\n    \"\"\"Authentic images as negatives for classification (label=0).\"\"\"\n    def __init__(self, paths: List[str], train: bool):\n        self.paths = paths; self.train = train\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        x = train_tf(img) if self.train else val_tf(img)\n        return x, torch.tensor(0, dtype=torch.long)\n\n# -------------------------\n# Build loaders\n# -------------------------\npairs_df = pd.read_csv(PAIRS_CSV)\nval_df = pairs_df.sample(frac=0.2, random_state=42)\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(42)\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\ntrain_forged_ds = PairedForgeryDataset(train_df, MASK_DIR, train=True, return_paths=False)\nval_forged_ds   = PairedForgeryDataset(val_df,   MASK_DIR, train=False, return_paths=USE_RETR_IN_VAL)\ntrain_auth_ds   = AuthenticOnlyClsDataset(auth_train, train=True)\nval_auth_ds     = AuthenticOnlyClsDataset(auth_val,   train=False)\n\ntrain_forged_loader = DataLoader(train_forged_ds, batch_size=BATCH_FORGED, shuffle=True,  num_workers=2, pin_memory=True)\nval_forged_loader   = DataLoader(val_forged_ds,   batch_size=BATCH_FORGED, shuffle=False, num_workers=2, pin_memory=True)\ntrain_auth_loader   = DataLoader(train_auth_ds,   batch_size=BATCH_AUTH_CLS, shuffle=True,  num_workers=2, pin_memory=True)\nval_auth_loader     = DataLoader(val_auth_ds,     batch_size=BATCH_AUTH_CLS, shuffle=False, num_workers=2, pin_memory=True)\n\nprint(f\"Train forged: {len(train_forged_ds)} | Val forged: {len(val_forged_ds)}\")\nprint(f\"Train authentic (cls neg): {len(train_auth_ds)} | Val authentic (cls neg): {len(val_auth_ds)}\")\n\n# -------------------------\n# Model (EfficientNet-B0) + heads\n# -------------------------\nclass EffNetB0Multi(nn.Module):\n    def __init__(self):\n        super().__init__()\n        self.backbone = torchvision.models.efficientnet_b0(weights=None)\n        self.backbone.classifier = nn.Identity()\n        # 1280 backbone + 3(diff RGB) + 1(edge)\n        self.seg_head = nn.Sequential(\n            nn.Conv2d(1280+4, 256, 3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(256, 64, 3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(64, 1, 1)\n        )\n        self.cls_head = nn.Linear(1280, 2)\n\n    def forward(self, x_forg, diff):\n        feats = self.backbone.features(x_forg)   # [B,1280,h,w]\n        pooled = F.adaptive_avg_pool2d(feats,1).flatten(1)\n        logits_cls = self.cls_head(pooled)\n        f_up = F.interpolate(feats, size=x_forg.shape[2:], mode=\"bilinear\", align_corners=False)\n\n        # edges (Sobel)\n        with torch.no_grad():\n            x_gray = x_forg.mean(dim=1, keepdim=True)\n            kx = torch.tensor([[[[-1,0,1],[-2,0,2],[-1,0,1]]]], device=x_forg.device, dtype=x_forg.dtype)\n            ky = torch.tensor([[[[-1,-2,-1],[0,0,0],[1,2,1]]]], device=x_forg.device, dtype=x_forg.dtype)\n            sobel_x = F.conv2d(x_gray, kx, padding=1)\n            sobel_y = F.conv2d(x_gray, ky, padding=1)\n            edges = torch.clamp(torch.abs(sobel_x)+torch.abs(sobel_y), 0, 1)\n\n        diff_amp = diff * 2.0\n        seg_in = torch.cat([f_up, diff_amp, edges], dim=1)  # 1280 + 3 + 1\n\n        logits_seg = self.seg_head(seg_in)\n        return logits_seg, logits_cls\n\nmodel = EffNetB0Multi().to(device)\n\n# -------------------------\n# Loss & Optim\n# -------------------------\ncrit_cls = nn.CrossEntropyLoss()\noptimizer = optim.AdamW(model.parameters(), lr=LR, weight_decay=WEIGHT_DECAY)\n\n# -------------------------\n# TTA helper (flip) for seg & cls\n# -------------------------\ndef seg_with_tta(x, diff):\n    model.eval()\n    with torch.no_grad():\n        log_seg1, log_cls1 = model(x, diff)\n        x2 = torch.flip(x, dims=[3]); d2 = torch.flip(diff, dims=[3])\n        log_seg2, log_cls2 = model(x2, d2)\n    p1 = torch.sigmoid(log_seg1)\n    p2 = torch.flip(torch.sigmoid(log_seg2), dims=[3])\n    p = (p1 + p2) / 2.0\n    logits = torch.log(p.clamp(1e-6, 1-1e-6) / (1 - p.clamp(1e-6, 1-1e-6)))\n    cls_logits = (log_cls1 + log_cls2) / 2.0\n    return logits, cls_logits\n\n# -------------------------\n# Post-processing helpers\n# -------------------------\ndef masks_from_probs(prob: np.ndarray, thr: float, min_area: int, use_morph: bool=True) -> List[np.ndarray]:\n    bm = (prob > thr).astype(np.uint8)\n    if use_morph:\n        bm = cv2.morphologyEx(bm, cv2.MORPH_CLOSE, np.ones((3,3),np.uint8))\n        bm = cv2.morphologyEx(bm, cv2.MORPH_OPEN,  np.ones((3,3),np.uint8))\n    num, labels = cv2.connectedComponents(bm, connectivity=8)\n    insts, scores = [], []\n    for k in range(1, num):\n        inst = (labels==k).astype(np.uint8)\n        area = int(inst.sum())\n        if area >= min_area:\n            insts.append(inst)\n            scores.append(float(prob[inst==1].mean()))\n    if TOPK_INSTANCES is not None and len(insts) > TOPK_INSTANCES:\n        order = np.argsort(scores)[::-1][:TOPK_INSTANCES]\n        insts = [insts[i] for i in order]\n    return insts\n\ndef logits_to_instances_rle(logits: torch.Tensor,\n                            thr: float,\n                            min_area: int,\n                            use_morph: bool=True) -> str:\n    if logits.ndim == 4:\n        prob = torch.sigmoid(logits)[0,0].detach().cpu().numpy()\n    elif logits.ndim == 3:\n        prob = torch.sigmoid(logits)[0].detach().cpu().numpy()\n    else:\n        raise ValueError(\"logits must be [B,1,H,W] or [1,H,W]\")\n    insts = masks_from_probs(prob, thr=thr, min_area=min_area, use_morph=use_morph)\n    return rle_encode(insts) if insts else \"[]\"\n\n# -------------------------\n# TRAIN LOOP (with bg loss for empty masks)\n# -------------------------\nfor epoch in range(1, EPOCHS+1):\n    model.train()\n    tr_loss = 0.0\n    cls_true, cls_pred = [], []\n\n    forged_iter = iter(train_forged_loader)\n    auth_iter   = iter(train_auth_loader)\n    steps = max(len(train_forged_loader), len(train_auth_loader))\n\n    for _ in range(steps):\n        total_loss = 0.0\n\n        # forged step (seg+cls)\n        try:\n            xf, diff, mask, y_f = next(forged_iter)\n            xf, diff, mask, y_f = xf.to(device), diff.to(device), mask.to(device), y_f.to(device)\n            seg_logits, cls_logits = model(xf, diff)\n\n            probs = torch.sigmoid(seg_logits)\n            has_pos = (mask.view(mask.size(0), -1).sum(dim=1) > 0).float()\n            pos_idx = (has_pos == 1)\n            neg_idx = (has_pos == 0)\n\n            seg_loss_pos = torch.tensor(0.0, device=device)\n            if pos_idx.any():\n                bce_pos = nn.BCEWithLogitsLoss(reduction=\"none\")(seg_logits[pos_idx], mask[pos_idx]) \\\n                          .view(pos_idx.sum(), -1).mean(dim=1)\n                inter = (probs[pos_idx]*mask[pos_idx]).view(pos_idx.sum(), -1).sum(dim=1)\n                den   = (probs[pos_idx].view(pos_idx.sum(), -1).sum(dim=1) +\n                         mask[pos_idx].view(pos_idx.sum(), -1).sum(dim=1) + 1e-6)\n                dice_pos = 1 - (2*inter)/den\n                seg_loss_pos = (bce_pos + dice_pos).mean()\n\n            seg_loss_neg = torch.tensor(0.0, device=device)\n            if neg_idx.any():\n                zero_target = torch.zeros_like(seg_logits[neg_idx])\n                bce_neg = nn.BCEWithLogitsLoss(reduction=\"mean\")(seg_logits[neg_idx], zero_target)\n                seg_loss_neg = NEG_BG_WEIGHT * bce_neg\n\n            seg_loss = seg_loss_pos + seg_loss_neg\n            cls_loss = crit_cls(cls_logits, y_f)\n            loss_f = seg_loss + ALPHA_CLS*cls_loss\n            total_loss += loss_f\n\n            cls_true.extend(y_f.cpu().tolist())\n            cls_pred.extend(cls_logits.argmax(1).detach().cpu().tolist())\n        except StopIteration:\n            pass\n\n        # authentic-only step (cls)\n        try:\n            xa, y_a = next(auth_iter)\n            xa, y_a = xa.to(device), y_a.to(device)\n            feats = model.backbone.features(xa)\n            pooled = F.adaptive_avg_pool2d(feats,1).flatten(1)\n            cls_logits_a = model.cls_head(pooled)\n            loss_a = ALPHA_CLS*crit_cls(cls_logits_a, y_a)\n            total_loss += loss_a\n\n            cls_true.extend(y_a.cpu().tolist())\n            cls_pred.extend(cls_logits_a.argmax(1).detach().cpu().tolist())\n        except StopIteration:\n            pass\n\n        if isinstance(total_loss, torch.Tensor):\n            optimizer.zero_grad()\n            total_loss.backward()\n            optimizer.step()\n            tr_loss += total_loss.item()\n\n    clsF1 = f1_score(cls_true, cls_pred) if len(set(cls_true))>1 else 0.0\n    print(f\"[Epoch {epoch}/{EPOCHS}] train_loss={tr_loss/steps:.4f} | ClsF1={clsF1:.3f}\")\n    torch.cuda.empty_cache(); gc.collect()\n\n# -------------------------\n# Validation — tune detection threshold (cls) with TRUE labels\n# -------------------------\ndef cls_probs_on_val_true_labels() -> Tuple[np.ndarray, np.ndarray]:\n    model.eval()\n    y_true, y_prob = [], []\n    with torch.no_grad():\n        # forged → 1\n        for batch in val_forged_loader:\n            if USE_RETR_IN_VAL:\n                xf, diff, _, _, _, _ = batch\n            else:\n                xf, diff, _, _ = batch\n            xf, diff = xf.to(device), diff.to(device)\n            _, logit = model(xf, diff)\n            prob = torch.softmax(logit, 1)[:, 1].cpu().numpy()\n            y_prob.extend(prob.tolist()); y_true.extend([1]*len(prob))\n        # authentic → 0\n        for xa, y in val_auth_loader:\n            xa = xa.to(device)\n            feats = model.backbone.features(xa)\n            pooled = F.adaptive_avg_pool2d(feats,1).flatten(1)\n            logit = model.cls_head(pooled)\n            prob = torch.softmax(logit, 1)[:, 1].cpu().numpy()\n            y_prob.extend(prob.tolist()); y_true.extend([0]*len(prob))\n    return np.array(y_true), np.array(y_prob)\n\ndef best_det_threshold_for_f1(y_true, y_prob):\n    thrs = np.linspace(0.05, 0.95, 37)\n    best_thr, best_f1 = 0.5, 0.0\n    for t in thrs:\n        y_pred = (y_prob >= t).astype(int)\n        f1 = f1_score(y_true, y_pred)\n        if f1 > best_f1:\n            best_f1, best_thr = float(f1), float(t)\n    return best_thr, best_f1\n\ny_true_det, y_prob_det = cls_probs_on_val_true_labels()\nBEST_CLS_THR, best_detF1_val = best_det_threshold_for_f1(y_true_det, y_prob_det)\nprint(f\"[Validation] Best detection threshold = {BEST_CLS_THR:.3f} | DetF1={best_detF1_val:.4f}\")\n\n# -------------------------\n# Retrieval index (for TEST and optional VAL)\n# -------------------------\nIMG_EXTS = {\".png\", \".jpg\", \".jpeg\", \".tif\", \".tiff\", \".bmp\"}\n\ndef _normalize_name(p: Path) -> str:\n    s = p.stem.lower().replace(\"-\", \"_\").replace(\" \", \"_\")\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\n# imports for retrieval\ntry:\n    import imagehash\nexcept Exception:\n    # \n    pass\n\nfrom PIL import Image as _PIL_Image\n\nAUTH_LIST = list_images(AUTH_DIR)\nPHASH_DB = []\nif USE_RETR_PHASH and len(AUTH_LIST):\n    print(\"[Retrieval] Building pHash index for authentic...\")\n    for ap in AUTH_LIST:\n        try:\n            h = imagehash.phash(_PIL_Image.open(ap).convert(\"RGB\"))\n        except Exception:\n            h = None\n        PHASH_DB.append((ap, h))\n\nSSIM_DB = []\nif USE_RETR_SSIM:\n    try:\n        from skimage.metrics import structural_similarity as ssim\n    except Exception:\n        pass\n    def _load_gray_resized(path, max_side=SSIM_MAX_SIDE):\n        img = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE)\n        if img is None: return None\n        h, w = img.shape[:2]\n        sc = min(1.0, max_side / max(h, w))\n        if sc < 1: img = cv2.resize(img, (int(w*sc), int(h*sc)), interpolation=cv2.INTER_AREA)\n        return img\n    if len(AUTH_LIST):\n        print(\"[Retrieval] Preloading grayscale authentic for SSIM...\")\n        for ap in AUTH_LIST:\n            try:\n                ai = _load_gray_resized(ap, SSIM_MAX_SIDE)\n            except Exception:\n                ai = None\n            SSIM_DB.append((ap, ai))\n\ndef _best_by_name(test_path):\n    stem = _normalize_name(Path(test_path))\n    for ap in AUTH_LIST:\n        if _normalize_name(Path(ap)) == stem:\n            return ap\n    return None\n\ndef _best_by_phash(test_path):\n    if not USE_RETR_PHASH or len(PHASH_DB)==0: return (None, None)\n    try:\n        h = imagehash.phash(_PIL_Image.open(test_path).convert(\"RGB\"))\n    except Exception:\n        return (None, None)\n    best_p, best_d = None, None\n    for ap, ah in PHASH_DB:\n        if ah is None: continue\n        d = abs(h - 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_by_ssim(test_path):\n    if not USE_RETR_SSIM or len(SSIM_DB)==0: return (None, None)\n    from skimage.metrics import structural_similarity as ssim\n    def _load_gray_resized(path, max_side=SSIM_MAX_SIDE):\n        img = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE)\n        if img is None: return None\n        h, w = img.shape[:2]\n        sc = min(1.0, max_side / max(h, w))\n        if sc < 1: img = cv2.resize(img, (int(w*sc), int(h*sc)), interpolation=cv2.INTER_AREA)\n        return img\n    gi = _load_gray_resized(test_path, SSIM_MAX_SIDE)\n    if gi is None: return (None, None)\n    gh, gw = gi.shape[:2]\n    best_p, best_sc = None, None\n    for ap, ai in SSIM_DB:\n        if ai is None: continue\n        ah, aw = ai.shape[:2]\n        ai_r = cv2.resize(ai, (gw, gh), interpolation=cv2.INTER_AREA) if (ah, aw)!=(gh, gw) else ai\n        try:\n            sc = ssim(gi, ai_r)\n        except:\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\ndef find_best_authentic_for_path(test_path: str) -> Optional[str]:\n    ap = _best_by_name(test_path)\n    if ap is not None: return ap\n    ap, d = _best_by_phash(test_path)\n    if ap is not None and d is not None and d <= PHASH_THRESHOLD:\n        return ap\n    ap, sc = _best_by_ssim(test_path)\n    if ap is not None and sc is not None and sc >= SSIM_THRESHOLD:\n        return ap\n    return None\n\ndef build_diff_from_retrieved(test_img_path: str) -> Optional[torch.Tensor]:\n    \"\"\"Return xa tensor [1,3,H,W] for retrieved authentic (val_tf applied), None if not found.\"\"\"\n    ap = find_best_authentic_for_path(test_img_path)\n    if ap is None:\n        return None\n    try:\n        A = Image.open(ap).convert(\"RGB\")\n    except:\n        return None\n    xa = val_tf(A).unsqueeze(0).to(device)\n    return xa\n\n# -------------------------\n# Build GT & PRED dataframes for validation (order-aligned)\n# -------------------------\n@torch.no_grad()\ndef build_val_gt_and_pred_for_forged(thr_seg: float, min_area: int) -> Tuple[pd.DataFrame, pd.DataFrame]:\n    rows_gt, rows_sub = [], []\n    for batch in tqdm(val_forged_loader, desc=\"Val forged eval\"):\n        if USE_RETR_IN_VAL:\n            xf, diff, mask, _, forg_path_batch, _ = batch\n        else:\n            xf, diff, mask, _ = batch; forg_path_batch = None\n        xf, diff = xf.to(device), diff.to(device)\n\n        # اگر retrieval را در val بخواهیم، diff را جایگزین کنیم (شبیه تست)\n        if USE_RETR_IN_VAL and forg_path_batch is not None:\n            diffs = []\n            for i in range(xf.size(0)):\n                xa = build_diff_from_retrieved(forg_path_batch[i])\n                if xa is None:\n                    diffs.append(torch.zeros_like(xf[i:i+1]))\n                else:\n                    diffs.append(torch.abs(xf[i:i+1] - xa))\n            diff = torch.cat(diffs, dim=0)\n\n        log_seg, log_cls = seg_with_tta(xf, diff)  # TTA\n        probs = torch.softmax(log_cls, 1)[:, 1].cpu().numpy()\n        B = xf.size(0)\n        for i in range(B):\n            shape = (IMG_SIZE, IMG_SIZE)\n            gt_m = (mask[i,0].cpu().numpy() > 0.5).astype(np.uint8)\n\n            # GT row\n            if gt_m.sum() == 0:\n                rows_gt.append({\"row_id\": f\"val_{len(rows_gt)}\",\n                                \"annotation\": \"authentic\",\n                                \"shape\": \"authentic\"})\n            else:\n                num, labels = cv2.connectedComponents(gt_m, connectivity=8)\n                insts = [(labels==k).astype(np.uint8) for k in range(1, num) if (labels==k).sum()>0]\n                rle_gt = rle_encode(insts) if len(insts)>0 else \"authentic\"\n                rows_gt.append({\"row_id\": f\"val_{len(rows_gt)}\",\n                                \"annotation\": rle_gt,\n                                \"shape\": json.dumps(list(shape))})\n\n            # PRED row\n            if probs[i] < BEST_CLS_THR:\n                pred = \"authentic\"\n            else:\n                pred = logits_to_instances_rle(log_seg[i:i+1], thr=thr_seg, min_area=min_area, use_morph=True)\n                pred = \"authentic\" if pred==\"[]\" else pred\n            rows_sub.append({\"row_id\": rows_gt[-1][\"row_id\"], \"annotation\": pred})\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\n# -------------------------\n# Small grid-search for seg thresholds on validation (official score)\n# -------------------------\nbest_official, best_thr_seg, best_min_area = -1.0, THRESH_SEG, MIN_INSTANCE_AREA\nfor thr in [0.10, 0.15, 0.20, 0.25, 0.30]:\n    for area in [1, 2, 4, 8]:\n        gt_tmp, sub_tmp = build_val_gt_and_pred_for_forged(thr_seg=thr, min_area=area)\n        s = score(gt_tmp.copy(), sub_tmp.copy(), row_id_column_name=\"row_id\")\n        if s > best_official:\n            best_official, best_thr_seg, best_min_area = s, thr, area\n        del gt_tmp, sub_tmp\n        gc.collect()\nTHRESH_SEG, MIN_INSTANCE_AREA = best_thr_seg, best_min_area\nprint(f\"[VAL seg-tune] Best official={best_official:.6f} at THRESH_SEG={THRESH_SEG} | MIN_INSTANCE_AREA={MIN_INSTANCE_AREA}\")\n\n# Final validation score with the best params\ngt_df_val, sub_df_val = build_val_gt_and_pred_for_forged(thr_seg=THRESH_SEG, min_area=MIN_INSTANCE_AREA)\noverall_val = score(gt_df_val.copy(), sub_df_val.copy(), row_id_column_name=\"row_id\")\nprint(\"\\n========== Validation (OFFICIAL METRIC ONLY) ==========\")\nprint(f\"Overall (official score): {overall_val:.6f}\")\nprint(\"=======================================================\\n\")\n\n# -------------------------\n# TEST Inference & submission.csv (exact required format)\n#   - Must have a row for each image\n#   - For negatives: 'authentic'\n#   - For positives: JSON-RLE via rle_encode\n#   - Header: case_id,annotation\n# -------------------------\n@torch.no_grad()\ndef predict_one_test(image_path: str) -> str:\n    img = Image.open(image_path).convert(\"RGB\")\n    x = val_tf(img).unsqueeze(0).to(device)\n\n    # Use retrieval to build diff (or fallback to zeros)\n    if USE_RETR_IN_TEST:\n        xa = build_diff_from_retrieved(image_path)\n        diff = torch.abs(x - xa) if xa is not None else torch.zeros_like(x)\n    else:\n        diff = torch.zeros_like(x)\n\n    log_seg, log_cls = seg_with_tta(x, diff)  # TTA\n    prob_forg = torch.softmax(log_cls,1)[0,1].item()\n    if prob_forg < BEST_CLS_THR:\n        return \"authentic\"\n\n    rle = logits_to_instances_rle(log_seg, thr=THRESH_SEG, min_area=MIN_INSTANCE_AREA, use_morph=True)\n    return \"authentic\" if rle==\"[]\" else rle\n\ndef build_submission(test_dir: str, sample_csv_path: str, out_csv_path: str = \"submission.csv\"):\n    # Output must be: case_id,annotation\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\"):\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(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(p) for p in tqdm(test_files, desc=\"Predicting test\")]\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(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-06T15:37:54.254154Z","iopub.execute_input":"2025-11-06T15:37:54.254877Z","iopub.status.idle":"2025-11-06T18:27:28.185471Z","shell.execute_reply.started":"2025-11-06T15:37:54.254846Z","shell.execute_reply":"2025-11-06T18:27:28.184663Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport numpy as np\nimport cv2\nfrom PIL import Image\n\ndef _to_disp(img_tensor):\n    x = img_tensor.detach().cpu().numpy()\n    x = np.transpose(x, (1,2,0))\n    x = np.clip(x*255, 0, 255).astype(np.uint8)\n    return x\n\ndef _union_mask_from_instances(prob, thr, min_area):\n    bm = (prob > thr).astype(np.uint8)\n    bm = cv2.morphologyEx(bm, cv2.MORPH_CLOSE, np.ones((3,3),np.uint8))\n    bm = cv2.morphologyEx(bm, cv2.MORPH_OPEN,  np.ones((3,3),np.uint8))\n    num, labels = cv2.connectedComponents(bm, connectivity=8)\n    out = np.zeros_like(bm)\n    for k in range(1, num):\n        inst = (labels==k).astype(np.uint8)\n        if inst.sum() >= min_area:\n            out[inst==1] = 1\n    return out\n\n@torch.no_grad()\ndef show_auth_forged_seg(n_samples=10, df=None, figsize=(12, 3)):\n    model.eval()\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=42).reset_index(drop=True)\n\n    rows = len(samp)\n    plt.figure(figsize=(figsize[0], figsize[1]*rows))\n\n    idx_plot = 1\n    for i, row in samp.iterrows():\n        ap, fp = row['auth_path'], row['forg_path']\n\n        A = Image.open(ap).convert(\"RGB\")\n        F = Image.open(fp).convert(\"RGB\")\n        xa = val_tf(A).unsqueeze(0).to(device)\n        xf = val_tf(F).unsqueeze(0).to(device)\n\n        diff = torch.abs(xf - xa)\n        log_seg, log_cls = seg_with_tta(xf, diff)\n        prob_forg = torch.softmax(log_cls, 1)[0,1].item()\n\n        a_disp = _to_disp(xa[0])\n        f_disp = _to_disp(xf[0])\n\n        if prob_forg >= BEST_CLS_THR:\n            prob = torch.sigmoid(log_seg)[0,0].cpu().numpy()\n            mask_bin = _union_mask_from_instances(prob, THRESH_SEG, MIN_INSTANCE_AREA)\n        else:\n            mask_bin = np.zeros((IMG_SIZE, IMG_SIZE), dtype=np.uint8)\n\n        # --- Overlay امن (بدون اندیس‌گذاری ۱بعدی) ---\n        overlay = f_disp.copy()\n        if mask_bin.sum() > 0:\n            alpha = 0.45\n            red_img = np.zeros_like(overlay, dtype=np.uint8)\n            red_img[..., 0] = 255  # R=255, G=B=0\n            m3 = mask_bin[..., None].astype(bool)  # HxWx1 -> broadcast به HxWx3\n            blended = (alpha * red_img + (1 - alpha) * overlay).astype(np.uint8)\n            overlay = np.where(m3, blended, overlay)\n\n            # کانتور سفید\n            cnts, _ = cv2.findContours(mask_bin, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n            cv2.drawContours(overlay, cnts, -1, (255, 255, 255), 1)\n\n        # Authentic\n        ax1 = plt.subplot(rows, 3, idx_plot); idx_plot += 1\n        ax1.imshow(a_disp); ax1.set_title(\"Authentic\", fontsize=11); ax1.axis('off')\n\n        # Forged\n        ax2 = plt.subplot(rows, 3, idx_plot); idx_plot += 1\n        ax2.imshow(f_disp); ax2.set_title(\"Forged\", fontsize=11); ax2.axis('off')\n\n        # Predicted Seg\n        ax3 = plt.subplot(rows, 3, idx_plot); idx_plot += 1\n        ax3.imshow(overlay)\n        ax3.set_title(f\"Predicted Seg (p_forg={prob_forg:.2f})\", fontsize=11)\n        ax3.axis('off')\n\n    plt.tight_layout()\n    plt.show()\n\n# اجرا\nshow_auth_forged_seg(n_samples=10, df=val_df)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-06T18:39:16.959020Z","iopub.execute_input":"2025-11-06T18:39:16.959302Z","iopub.status.idle":"2025-11-06T18:39:21.721211Z","shell.execute_reply.started":"2025-11-06T18:39:16.959282Z","shell.execute_reply":"2025-11-06T18:39:21.720106Z"}},"outputs":[],"execution_count":null}]}