{"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":"gpu","dataSources":[{"sourceId":113558,"databundleVersionId":14878066,"isSourceIdPinned":false,"sourceType":"competition"},{"sourceId":14016215,"sourceType":"datasetVersion","datasetId":8929132},{"sourceId":14284759,"sourceType":"datasetVersion","datasetId":9117617},{"sourceId":14387308,"sourceType":"datasetVersion","datasetId":9188253},{"sourceId":14403692,"sourceType":"datasetVersion","datasetId":9199265},{"sourceId":14406809,"sourceType":"datasetVersion","datasetId":9201167},{"sourceId":14407486,"sourceType":"datasetVersion","datasetId":9201552},{"sourceId":14417174,"sourceType":"datasetVersion","datasetId":9208044},{"sourceId":14430476,"sourceType":"datasetVersion","datasetId":9217032},{"sourceId":14459650,"sourceType":"datasetVersion","datasetId":9235629},{"sourceId":14467382,"sourceType":"datasetVersion","datasetId":9240800},{"sourceId":270092713,"sourceType":"kernelVersion"},{"sourceId":4534,"sourceType":"modelInstanceVersion","isSourceIdPinned":false,"modelInstanceId":3326,"modelId":986},{"sourceId":648498,"sourceType":"modelInstanceVersion","isSourceIdPinned":false,"modelInstanceId":489174,"modelId":504592},{"sourceId":686586,"sourceType":"modelInstanceVersion","isSourceIdPinned":false,"modelInstanceId":520737,"modelId":534998}],"dockerImageVersionId":31192,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Recod.AI Image Forgeery Detection\n### Author: Nick Pellegrin","metadata":{}},{"cell_type":"markdown","source":"## Imports and General Utility\n---","metadata":{}},{"cell_type":"code","source":"!pip uninstall tensorflow -y","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-11T17:14:26.538546Z","iopub.execute_input":"2026-01-11T17:14:26.538728Z","iopub.status.idle":"2026-01-11T17:14:48.621069Z","shell.execute_reply.started":"2026-01-11T17:14:26.538710Z","shell.execute_reply":"2026-01-11T17:14:48.620329Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# General Imports\nimport os\nimport gc\nimport math\nimport json\nimport random\nimport itertools\nimport dataclasses\nfrom pathlib import Path\nfrom typing import Optional, List, Tuple\nfrom functools import lru_cache\nfrom tqdm.notebook import tqdm\nfrom copy import deepcopy\n\n# Data Imports\nimport numpy as np\nimport pandas as pd\npd.set_option('display.max_colwidth', None)\npd.set_option('display.max_rows', 1000000)\n\n# Image Imports\nimport cv2\nfrom PIL import Image\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\n\n# Model Imports\nimport timm\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nimport torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader, RandomSampler, SequentialSampler\nfrom transformers import AutoImageProcessor, AutoModel\nfrom sklearn.model_selection import StratifiedGroupKFold, train_test_split\nfrom sklearn.metrics import roc_auc_score, f1_score\nfrom scipy.optimize import linear_sum_assignment\nfrom scipy.ndimage import label\nfrom safetensors.torch import save_file, load_file\n\n# Scoring\n# import sys\n# sys.path.append(\"/kaggle/input/recodai-f1\")\n# from metric import score, rle_encode\n\n\n\n\n\n    \n# Data Directories\nOUTPUT_DIR: str = '/kaggle/working/'\nDATA_DIR: str = '/kaggle/input/recodai-luc-scientific-image-forgery-detection/'\ntrain_images_authentic_dir: Path = Path(DATA_DIR) / 'train_images/authentic/'\ntrain_images_forged_dir:    Path = Path(DATA_DIR) / 'train_images/forged/'\ntrain_masks_forged_dir:     Path = Path(DATA_DIR) / 'train_masks/'\ntrain_suppl_images_dir:     Path = Path(DATA_DIR) / 'supplemental_images/'\ntrain_suppl_masks_dir:      Path = Path(DATA_DIR) / 'supplemental_masks/'\ntest_images_dir:            Path = Path(DATA_DIR) / 'test_images/'\nsample_submission:          Path = Path(DATA_DIR) / 'sample_submission.csv'\nimg_size = 518\n\ndinov2_base: str = '/kaggle/input/dinov2/pytorch/base/1'\npretrained_classifiers: Path = Path('/kaggle/input/effnetv2-classifiers/')\npretrained_segmenters: Path = Path('/kaggle/input/dinov2-segmenters/')\n\n\n\n# Seed Everything\nseed=42\nrandom.seed(seed)\nos.environ[\"PYTHONHASHSEED\"] = str(seed)\nnp.random.seed(seed)\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(f\"device: {device}\")\ntorch.manual_seed(seed)\ntorch.cuda.manual_seed_all(seed)\ntorch.backends.cudnn.deterministic = True\ntorch.backends.cudnn.benchmark = False\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-11T17:15:28.920845Z","iopub.execute_input":"2026-01-11T17:15:28.921188Z","iopub.status.idle":"2026-01-11T17:16:19.331987Z","shell.execute_reply.started":"2026-01-11T17:15:28.921156Z","shell.execute_reply":"2026-01-11T17:16:19.331226Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Training\n---","metadata":{}},{"cell_type":"markdown","source":"def build_dataframe():\n    authentic_imgs = sorted(list(train_images_authentic_dir.glob(\"*.png\")))\n    forged_imgs = sorted(list(train_images_forged_dir.glob(\"*.png\")))\n    forged_masks = sorted(list(train_masks_forged_dir.glob(\"*.npy\")))\n\n    rows = []\n\n    for img in authentic_imgs:\n        rows.append({\n            \"unique_id\": f\"{img.stem}_real\",\n            \"id\": str(img.stem),\n            \"filepath\": str(img),\n            \"label\": 0,\n            \"mask_path\": None\n        })\n\n    mask_map = {m.stem: m for m in forged_masks}\n    for img in forged_imgs:\n        stem = img.stem \n        mask_path = mask_map.get(stem, None)\n        rows.append({\n            \"unique_id\": f\"{stem}_fake\",\n            \"id\": str(img.stem),\n            \"filepath\": str(img),\n            \"label\": 1,\n            \"mask_path\": str(mask_path) if mask_path is not None else None\n        })\n\n    df = pd.DataFrame(rows)\n    df = df.sample(frac=1, random_state=seed).reset_index(drop=True)\n    return df\n\n\n'''\ndef get_train_transforms(image_size=512):\n    return A.Compose([\n        # A.HorizontalFlip(p=0.5),\n        # A.VerticalFlip(p=0.1),\n        # A.Affine(\n        #     scale=(0.95, 1.05),\n        #     rotate=(-5, 5),\n        #     translate_percent=0.05,\n        #     interpolation=cv2.INTER_LINEAR,\n        #     mask_interpolation=cv2.INTER_NEAREST,\n        #     p=0.3\n        # ),\n        A.RandomBrightnessContrast(0.08, 0.08, p=0.3),\n        # A.HueSaturationValue(8, 10, 8, p=0.3),\n        A.OneOf([\n            A.ImageCompression(quality_range=(70, 95)),\n            A.GaussianBlur(blur_limit=(3, 5)),\n        ], p=0.2),\n        A.ColorJitter(\n            brightness=0.08,\n            contrast=0.08,\n            saturation=0.05,\n            hue=0.01,\n            p=0.3\n        ),\n    ])\n'''\ndef get_train_transforms(image_size=512):\n    return A.Compose([])\n\ndef get_val_transforms(image_size=512):\n    return A.Compose([])\n\n\n\nclass ForgeryDataset(Dataset):\n    def __init__(self, df, num_masks=1, image_size=512, train=True):\n        self.df = df\n        self.train = train\n        self.num_masks = num_masks\n        self.train_transforms = get_train_transforms(image_size=image_size)\n        self.val_transforms   = get_val_transforms(image_size=image_size)\n\n    def __len__(self):\n        return len(self.df)\n\n\n    def __getitem__(self, idx):\n        row = self.df.iloc[idx]\n        img = Image.open(row[\"filepath\"]).convert(\"RGB\")\n        w, h = img.size\n        if row[\"mask_path\"] is not None:\n            m = np.load(row[\"mask_path\"])\n            if m.ndim == 3: \n                m = np.max(m, axis=0)\n            m = (m > 0).astype(np.uint8)\n        else:\n            m = np.zeros((h, w), np.uint8)\n        img_r = np.array(img.resize((img_size, img_size)))\n        m_r = cv2.resize(m, (img_size, img_size), interpolation=cv2.INTER_NEAREST)\n        if self.train:\n            transformed = self.train_transforms(image=img_r, mask=m_r)\n        else:\n            transformed = self.val_transforms(image=img_r, mask=m_r)\n        img_t  = transformed[\"image\"]\n        img_t = torch.from_numpy(np.array(img_t, np.float32)/255.).permute(2,0,1)\n        m_t = transformed[\"mask\"]\n        m_t = torch.from_numpy(m_t[None, ...].astype(np.float32))\n\n        return img_t, m_t\n\n\n\n\n\n\n\nclass TinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True))\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n\n\n\nclass Segmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder = encoder\n        self.processor = processor\n        self.seg_head = TinyDecoder(768,1)\n        for p in self.encoder.parameters(): \n            p.requires_grad = False\n        # self.seg_head.apply(self._init_weights)\n\n    def _init_layer(self, layer):\n        if hasattr(layer, \"weight\"):\n            nn.init.xavier_uniform_(layer.weight)\n        if hasattr(layer, 'bias') and layer.bias is not None:\n            layer.bias.data.fill_(0.)      \n    \n    def _init_bn(self, bn):\n        if hasattr(bn, \"weight\"):\n            bn.weight.data.fill_(1.)\n        if hasattr(bn, \"bias\"):\n            bn.bias.data.fill_(0.)\n    \n    def _init_weights(self, m):\n        if isinstance(m, (nn.Linear, nn.Conv2d)):\n            self._init_layer(m)\n        elif isinstance(m, nn.BatchNorm1d):\n            self._init_bn(m)\n    \n    def forward(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).last_hidden_state\n        b, n, c = feats.shape\n        fmap = feats[:,1:,:].permute(0,2,1)\n        s = int(math.sqrt(n - 1))\n        fmap = fmap.reshape(b, c, s, s)\n        return self.seg_head(fmap, (img_size, img_size))\n\n\n\n\n\ndef stratified_group_train_val_split(df, val_size=0.2, seed=42):\n    group_labels = df.groupby(\"id\")[\"label\"].max()\n    groups = group_labels.index.values\n    y = group_labels.values\n    train_groups, val_groups = train_test_split(\n        groups,\n        test_size=val_size,\n        stratify=y,\n        random_state=seed,\n    )\n\n    train_df = df[df[\"id\"].isin(train_groups)].reset_index(drop=True)\n    val_df   = df[df[\"id\"].isin(val_groups)].reset_index(drop=True)\n\n    return train_df, val_df\n\n\n\n\n\n\n''' Training Loop '''\n\nnum_epochs = 16\nbatch_size = 2\nacc_steps = 8 # eff batch size of 2 * 8 = 16\nlr_max = 1e-5 # or 1.5e-4 with 5 epochs\nlr_min = 1e-6\nweight_decay = 1e-4 # 2e-5\nmax_grad_norm = 5.0\nearly_stopping = 3\n\ndf = build_dataframe()\ntrain_df, val_df = stratified_group_train_val_split(df, val_size=0.2, seed=seed)\n\n# --- Training DataLoader ---\ntrain_dataset = ForgeryDataset(train_df, num_masks=1, image_size=img_size, train=True)\ntrain_loader = DataLoader(\n    train_dataset, \n    batch_size=batch_size, \n    shuffle=True, \n    num_workers=4, \n    drop_last=True, \n    pin_memory=True,\n)\n\n# --- Validation DataLoader ---\nval_dataset = ForgeryDataset(val_df, num_masks=1, image_size=img_size, train=False)\nval_loader = DataLoader(\n    val_dataset, \n    batch_size=batch_size,\n    shuffle=False, \n    num_workers=4, \n    pin_memory=True,\n)\n\n\n# --- Create new model ---\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\n# ema_model = SegmenterEMA(model, device=\"cpu\")\n# ema_model = nn.DataParallel(ema_model) if torch.cuda.device_count() > 1 else ema_model\n# for param in ema_model.parameters():\n#     param.requires_grad = False\n\noptimizer = optim.AdamW(model.seg_head.parameters(), weight_decay=weight_decay, lr=lr_max)\nlr_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, num_epochs, eta_min=lr_min)\ncriterion = nn.BCEWithLogitsLoss()\nscaler = torch.amp.GradScaler(\"cuda\")\npatience = early_stopping\nn_patience = 0\nbest_model_score = float('inf')\nbest_model_state = None\n\n\n\n\"\"\" Training Stage 1 \"\"\"\nfor epoch in range(num_epochs):  \n    # --- Train ---\n    model.train()\n    train_loss = 0.0\n    pbar_train = tqdm(train_loader, desc=f\"Epoch {epoch+1}/{num_epochs} | Training\", leave=False)\n    for step, batch in enumerate(pbar_train):\n        if batch is None:\n            continue\n        inputs, labels = batch\n        inputs = inputs.to(device)\n        labels = labels.to(device)\n        with torch.amp.autocast(device_type=\"cuda\"):\n            mask_logits = model(inputs)\n            loss = criterion(mask_logits, labels)\n        loss = loss / acc_steps\n        scaler.scale(loss).backward()\n        if (step + 1) % acc_steps == 0 or (step + 1) == len(train_loader):\n            scaler.unscale_(optimizer)\n            torch.nn.utils.clip_grad_norm_(\n                filter(lambda p: p.requires_grad, model.parameters()),\n                max_grad_norm\n            )\n            scaler.step(optimizer)\n            scaler.update()\n            optimizer.zero_grad()\n        batch_loss = loss.item() * acc_steps\n        train_loss += batch_loss\n        pbar_train.set_postfix(loss=f\"{batch_loss:.4f}\", lr=optimizer.param_groups[0][\"lr\"])\n    lr_scheduler.step()\n    train_loss /= len(train_loader)\n\n    \n    # --- Validate ---\n    model.eval()\n    val_loss, val_score = 0.0, 0.0\n    val_preds, val_labels = [], []\n    with torch.no_grad():\n        pbar_val = tqdm(val_loader, desc=f\"Epoch {epoch+1}/{num_epochs} | Validating\", leave=False)\n        for batch in pbar_val:\n            if batch is None:\n                continue\n            inputs, labels = batch\n            inputs = inputs.to(device)\n            labels = labels.to(device)\n            with torch.amp.autocast(device_type=\"cuda\"):\n                mask_logits = model(inputs)\n                loss = criterion(mask_logits, labels)\n            val_loss += loss.item()\n            pbar_val.set_postfix(loss=f\"{val_loss:.4f}\")\n    val_loss /= len(val_loader)\n    print(f\"Epoch {epoch+1} | Loss: {val_loss:.5f}\")\n\n    # --- Evaluate ---\n    if val_loss < best_model_score:\n        best_model_score = val_loss\n        best_model_state = {k: v.clone() for k, v in model.state_dict().items()}\n        n_patience = 0\n    else:\n        n_patience += 1\n        if n_patience >= patience:\n            print(f\"Early stopping at epoch {epoch+1}\")\n            break\n\nsave_file(best_model_state, \"dinov2_best_v9_stage1.safetensors\")\n\n# --- Cleanup ---\ndel model\ndel processor\ndel encoder\ntorch.cuda.empty_cache()\ngc.collect()\n\n\n\n\n\n\nlr_seg_head = 1e-5\nlr_encoder = 5e-7\nlr_min = 5e-7\nunfreeze_layers = 12\nweight_decay = 1e-4 # 2e-5\nmax_grad_norm = 5.0\nearly_stopping = 5\n\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(load_file(\"dinov2_best_v9_stage1.safetensors\"))\n\ntry:\n    if hasattr(model.encoder, 'encoder') and hasattr(model.encoder.encoder, 'layer'):\n         layers_to_unfreeze = model.encoder.encoder.layer[-unfreeze_layers:]\n    elif hasattr(model_seg.encoder, 'layers'):\n         layers_to_unfreeze = model.encoder.layers[-unfreeze_layers:]\n    else:\n         layers_to_unfreeze = model.encoder.base_model.encoder.layer[-unfreeze_layers:]\nexcept Exception as e:\n    layers_to_unfreeze = model.encoder.encoder.layer[-unfreeze_layers:]\n\nfor param in layers_to_unfreeze.parameters():\n    param.requires_grad = True\n\noptimizer = optim.AdamW([\n    {'params': model.seg_head.parameters(), 'lr': lr_seg_head},\n    {'params': model.encoder.parameters(), 'lr': lr_encoder}\n], weight_decay=weight_decay)\n\nlr_scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, num_epochs, eta_min=lr_min)\ncriterion = nn.BCEWithLogitsLoss()\nscaler = torch.amp.GradScaler(\"cuda\")\npatience = early_stopping\nn_patience = 0\nbest_model_score = float('inf')\nbest_model_state = None\n\n\n\n\"\"\" Training Stage 2 \"\"\"\nfor epoch in range(num_epochs):  \n    # --- Train ---\n    model.train()\n    train_loss = 0.0\n    pbar_train = tqdm(train_loader, desc=f\"Epoch {epoch+1}/{num_epochs} | Training\", leave=False)\n    for step, batch in enumerate(pbar_train):\n        if batch is None:\n            continue\n        inputs, labels = batch\n        inputs = inputs.to(device)\n        labels = labels.to(device)\n        with torch.amp.autocast(device_type=\"cuda\"):\n            mask_logits = model(inputs)\n            loss = criterion(mask_logits, labels)\n        loss = loss / acc_steps\n        scaler.scale(loss).backward()\n        if (step + 1) % acc_steps == 0 or (step + 1) == len(train_loader):\n            scaler.unscale_(optimizer)\n            torch.nn.utils.clip_grad_norm_(\n                filter(lambda p: p.requires_grad, model.parameters()),\n                max_grad_norm\n            )\n            scaler.step(optimizer)\n            scaler.update()\n            optimizer.zero_grad()\n        batch_loss = loss.item() * acc_steps\n        train_loss += batch_loss\n        pbar_train.set_postfix(loss=f\"{batch_loss:.4f}\", lr=optimizer.param_groups[0][\"lr\"])\n    lr_scheduler.step()\n    train_loss /= len(train_loader)\n\n    \n    # --- Validate ---\n    model.eval()\n    val_loss, val_score = 0.0, 0.0\n    val_preds, val_labels = [], []\n    with torch.no_grad():\n        pbar_val = tqdm(val_loader, desc=f\"Epoch {epoch+1}/{num_epochs} | Validating\", leave=False)\n        for batch in pbar_val:\n            if batch is None:\n                continue\n            inputs, labels = batch\n            inputs = inputs.to(device)\n            labels = labels.to(device)\n            with torch.amp.autocast(device_type=\"cuda\"):\n                mask_logits = model(inputs)\n                loss = criterion(mask_logits, labels)\n            val_loss += loss.item()\n            pbar_val.set_postfix(loss=f\"{val_loss:.4f}\")\n    val_loss /= len(val_loader)\n    print(f\"Epoch {epoch+1} | Loss: {val_loss:.5f}\")\n\n    # --- Evaluate ---\n    if val_loss < best_model_score:\n        best_model_score = val_loss\n        best_model_state = {k: v.clone() for k, v in model.state_dict().items()}\n        n_patience = 0\n    else:\n        n_patience += 1\n        if n_patience >= patience:\n            print(f\"Early stopping at epoch {epoch+1}\")\n            break\n\nsave_file(best_model_state, \"dinov2_best_v9_stage2.safetensors\")\n\n# --- Cleanup ---\ndel model\ntorch.cuda.empty_cache()\ngc.collect()","metadata":{"execution":{"iopub.status.busy":"2026-01-11T17:16:34.452003Z","iopub.execute_input":"2026-01-11T17:16:34.453247Z","iopub.status.idle":"2026-01-11T17:17:07.436808Z","shell.execute_reply.started":"2026-01-11T17:16:34.453222Z","shell.execute_reply":"2026-01-11T17:17:07.435604Z"},"jupyter":{"source_hidden":true}}},{"cell_type":"markdown","source":"## Inference Pipeline\n---","metadata":{}},{"cell_type":"markdown","source":"def build_dataframe():\n    authentic_imgs = sorted(list(train_images_authentic_dir.glob(\"*.png\")))\n    forged_imgs = sorted(list(train_images_forged_dir.glob(\"*.png\")))\n    forged_masks = sorted(list(train_masks_forged_dir.glob(\"*.npy\")))\n\n    rows = []\n\n    for img in authentic_imgs:\n        rows.append({\n            \"unique_id\": f\"{img.stem}_real\",\n            \"id\": str(img.stem),\n            \"filepath\": str(img),\n            \"label\": 0,\n            \"mask_path\": None\n        })\n\n    mask_map = {m.stem: m for m in forged_masks}\n    for img in forged_imgs:\n        stem = img.stem \n        mask_path = mask_map.get(stem, None)\n        rows.append({\n            \"unique_id\": f\"{stem}_fake\",\n            \"id\": str(img.stem),\n            \"filepath\": str(img),\n            \"label\": 1,\n            \"mask_path\": str(mask_path) if mask_path is not None else None\n        })\n\n    df = pd.DataFrame(rows)\n    df = df.sample(frac=1, random_state=seed).reset_index(drop=True)\n    return df\n\n\ndef stratified_group_train_val_split(df, val_size=0.2, seed=42):\n    group_labels = df.groupby(\"id\")[\"label\"].max()\n    groups = group_labels.index.values\n    y = group_labels.values\n    train_groups, val_groups = train_test_split(\n        groups,\n        test_size=val_size,\n        stratify=y,\n        random_state=seed,\n    )\n\n    train_df = df[df[\"id\"].isin(train_groups)].reset_index(drop=True)\n    val_df   = df[df[\"id\"].isin(val_groups)].reset_index(drop=True)\n\n    return train_df, val_df\n\n\n\n\nclass TinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True))\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n\n\n\nclass Segmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder = encoder\n        self.processor = processor\n        self.seg_head = TinyDecoder(768,1)\n        for p in self.encoder.parameters(): \n            p.requires_grad = False\n        self.seg_head.apply(self._init_weights)\n\n    def _init_layer(self, layer):\n        if hasattr(layer, \"weight\"):\n            nn.init.xavier_uniform_(layer.weight)\n        if hasattr(layer, 'bias') and layer.bias is not None:\n            layer.bias.data.fill_(0.)      \n    \n    def _init_bn(self, bn):\n        if hasattr(bn, \"weight\"):\n            bn.weight.data.fill_(1.)\n        if hasattr(bn, \"bias\"):\n            bn.bias.data.fill_(0.)\n    \n    def _init_weights(self, m):\n        if isinstance(m, (nn.Linear, nn.Conv2d)):\n            self._init_layer(m)\n        elif isinstance(m, nn.BatchNorm1d):\n            self._init_bn(m)\n    \n    def forward(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).last_hidden_state\n        b, n, c = feats.shape\n        fmap = feats[:,1:,:].permute(0,2,1)\n        s = int(math.sqrt(n - 1))\n        fmap = fmap.reshape(b, c, s, s)\n        return self.seg_head(fmap, (img_size, img_size))\n\n\n\n\ndef hysteresis_threshold(preds, t_low, t_high):\n    hi = (preds > t_high).astype(np.uint8)\n    lo = (preds > t_low).astype(np.uint8)\n    num_labels, labels = cv2.connectedComponents(lo)\n    out = np.zeros_like(lo)\n    for lab in range(1, num_labels):\n        component = (labels == lab)\n        if (hi & component).any():\n            out[component] = 1\n\n    return out\n        \n\n\ndef postprocess(preds, original_size, alpha_grad=0.35):\n    gx = cv2.Sobel(preds, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(preds, cv2.CV_32F, 0, 1, ksize=3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_mag / (grad_mag.max() + 1e-6)\n    enhanced = (1 - alpha_grad) * preds + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (3, 3), 0)\n    \n    # thr = np.mean(enhanced) + 0.3 * np.std(enhanced)\n    # mask = (enhanced > thr).astype(np.uint8)\n    # mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))\n    # mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((3, 3), np.uint8))\n    \n    t_high = enhanced.mean() + 0.6 * enhanced.std()\n    t_low  = enhanced.mean() + 0.2 * enhanced.std()\n    mask = hysteresis_threshold(enhanced, t_low, t_high)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5,5)))\n    \n    mask = cv2.resize(mask, original_size, interpolation=cv2.INTER_NEAREST)\n    \n    return mask\n'''\ndef postprocess(\n    preds,\n    original_size,\n    alpha_grad=0.35,\n    beta_high=0.55,\n    beta_low=0.55,\n    min_mean_inside=0.15,\n    min_effective_area=600\n):\n    gx = cv2.Sobel(preds, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(preds, cv2.CV_32F, 0, 1, ksize=3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_mag / (grad_mag.max() + 1e-6)\n    enhanced = (1 - alpha_grad) * preds + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (3, 3), 0)\n    pmax = enhanced.max()\n    if pmax < 0.15:\n        return np.zeros(original_size[::-1], dtype=np.uint8)\n    t_high = beta_high * pmax\n    t_low  = beta_low  * pmax\n    mask = hysteresis_threshold(enhanced, t_low, t_high)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))\n    \n    mask = cv2.resize(mask, original_size, interpolation=cv2.INTER_NEAREST)\n    return mask\n'''\n\n\ndef infer_image_results(model, image):\n    image_array = np.array(image.resize((img_size, img_size)))\n    image_tensor = torch.from_numpy(np.array(image_array, np.float32)/255.).permute(2,0,1)\n    image_tensor = image_tensor.unsqueeze(0).to(device)\n\n    with torch.no_grad():\n        logits = model(image_tensor)\n        probs = torch.sigmoid(logits)[0, 0].cpu().numpy()\n    mask = postprocess(probs, image.size)\n\n    \n    area = int(mask.sum())\n    if area > 0:\n        mean_inside = float(\n            probs[cv2.resize(mask, (img_size, img_size), interpolation=cv2.INTER_NEAREST) == 1].mean()\n        )\n    else:\n        mean_inside = 0.0\n\n    # Not needed for this analysis\n    # if area < 200 or mean_inside < 0.22:\n    #     return \"authentic\"\n    probs = cv2.resize(probs, image.size, interpolation=cv2.INTER_NEAREST)\n    return logits, probs, mask, area, mean_inside\n\n\ndef analyze_case(gt, mask, probs):\n    gt_flat   = gt.flatten()\n    mask_flat = mask.flatten()\n\n    tp = np.sum((gt_flat == 1) & (mask_flat == 1))\n    fp = np.sum((gt_flat == 0) & (mask_flat == 1))\n    fn = np.sum((gt_flat == 1) & (mask_flat == 0))\n\n    f1 = 0.0 if tp == 0 else (2 * tp) / (2 * tp + fp + fn)\n\n    gt_area   = gt.sum()\n    pred_area = mask.sum()\n\n    coverage = tp / (gt_area + 1e-6)          # recall proxy\n    purity   = tp / (pred_area + 1e-6)        # precision proxy\n\n    # Boundary sharpness\n    gx = cv2.Sobel(probs, cv2.CV_32F, 1, 0, 3)\n    gy = cv2.Sobel(probs, cv2.CV_32F, 0, 1, 3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    boundary_strength = grad_mag[gt == 1].mean() if gt_area > 0 else 0\n\n    return {\n        \"f1\": f1,\n        \"gt_area\": int(gt_area),\n        \"pred_area\": int(pred_area),\n        \"coverage\": coverage,\n        \"purity\": purity,\n        \"boundary_strength\": boundary_strength,\n    }\n\n\n\n\n\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(load_file('/kaggle/input/dinov2-stage2/dinov2_best_v8_stage2.safetensors'))\nmodel.eval()\n\n\n\ndf = build_dataframe()\ntrain_df, val_df = stratified_group_train_val_split(df, val_size=0.15, seed=seed)\n\ncount = 0\nall_f1 = []\nfor index, row in val_df.iterrows():\n    if row['mask_path'] == None:\n        continue # Only evaluate forged images\n    \n    img = Image.open(row['filepath']).convert(\"RGB\")\n    gt = np.load(row[\"mask_path\"])\n    if gt.ndim == 3: \n        gt = np.max(gt, axis=0)\n    gt = (gt > 0).astype(np.uint8)\n    \n\n    logits, probs, mask, area, mean_inside = infer_image_results(model=model, image=img)\n\n    stats = analyze_case(gt, mask, probs)\n    all_f1.append(stats['f1'])\n\n    print(f\"\\nImage: {row['filepath']}\")\n    print(\n        f\"F1={stats['f1']:.3f} | \"\n        f\"GT area={stats['gt_area']} | \"\n        f\"Pred area={stats['pred_area']} | \"\n        f\"Coverage={stats['coverage']:.3f} | \"\n        f\"Purity={stats['purity']:.3f} | \"\n        f\"Mean-in={mean_inside:.3f}\"\n    )\n    \n    import matplotlib.pyplot as plt\n    fig, axs = plt.subplots(1, 4, figsize=(16, 4))\n    axs[0].imshow(img)\n    axs[0].set_title(\"Image\")\n    \n    axs[1].imshow(gt, cmap=\"gray\")\n    axs[1].set_title(\"GT mask\")\n    \n    axs[2].imshow(probs, cmap=\"jet\")\n    axs[2].set_title(\"Pred prob\")\n    \n    axs[3].imshow(mask, cmap=\"gray\")\n    axs[3].set_title(\"Postprocessed mask\")\n    \n    for ax in axs:\n        ax.axis(\"off\")\n    \n    plt.show()\n\n\n    count += 1\n    if count >= 15:\n        break\n\nprint(f\"Average F1: {np.mean(all_f1)}\")\n\n","metadata":{"execution":{"iopub.status.busy":"2026-01-11T01:54:54.071059Z","iopub.execute_input":"2026-01-11T01:54:54.071830Z","iopub.status.idle":"2026-01-11T01:55:20.145031Z","shell.execute_reply.started":"2026-01-11T01:54:54.071798Z","shell.execute_reply":"2026-01-11T01:55:20.143896Z"},"jupyter":{"source_hidden":true}}},{"cell_type":"code","source":"\"\"\" Segmentation Model \"\"\"\n'''\ndef init_layer(layer):\n    if hasattr(layer, \"weight\"):\n        nn.init.xavier_uniform_(layer.weight)\n    if hasattr(layer, 'bias') and layer.bias is not None:\n        layer.bias.data.fill_(0.)      \n    \ndef init_bn(bn):\n    if hasattr(bn, \"weight\"):\n        bn.weight.data.fill_(1.)\n    if hasattr(bn, \"bias\"):\n        bn.bias.data.fill_(0.)\n\nclass TinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.net = nn.Sequential(\n            nn.Conv2d(in_ch, 192, 3, padding=1), nn.ReLU(), # 256\n            nn.Conv2d(192, 64, 3, padding=1), nn.ReLU(),\n            nn.Conv2d(64, out_ch, 1)\n        )\n\n    def forward(self, f, size):\n        return self.net(F.interpolate(f, size=size, mode=\"bilinear\", align_corners=False))\n\n\nclass Segmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder = encoder\n        self.processor = processor\n        self.seg_head = TinyDecoder(768, 1)\n        for p in self.encoder.parameters():\n            p.requires_grad = False\n        self.seg_head.apply(self._init_weights)\n\n    def _init_weights(self, m):\n        if isinstance(m, (nn.Linear, nn.Conv2d)):\n            init_layer(m)\n        elif isinstance(m, nn.BatchNorm1d):\n            init_bn(m)\n\n    def forward(self, x):\n        imgs = (x*255).clamp(0, 255).byte().permute(0, 2, 3, 1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        with torch.no_grad():\n            feats = self.encoder(**inputs).last_hidden_state\n        B, N, C = feats.shape\n        fmap = feats[:, 1:, :].permute(0, 2, 1)\n        s = int(math.sqrt(N-1))\n        fmap = fmap.reshape(B, C, s, s)\n        return self.seg_head(fmap, (img_size, img_size))\n        return masks_logits, conf_logits\n'''\n\nclass TinyDecoder(nn.Module):\n    def __init__(self, in_ch=768, out_ch=1):\n        super().__init__()\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, 384, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block2 = nn.Sequential(\n            nn.Conv2d(384, 192, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(0.1))\n        self.block3 = nn.Sequential(\n            nn.Conv2d(192, 96, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True))\n        self.conv_out = nn.Conv2d(96, out_ch, kernel_size=1)\n    \n    def forward(self, f, target_size):\n        x = F.interpolate(self.block1(f), size=(74, 74), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block2(x), size=(148, 148), mode='bilinear', align_corners=False)\n        x = F.interpolate(self.block3(x), size=(296, 296), mode='bilinear', align_corners=False)\n        x = self.conv_out(x)\n        x = F.interpolate(x, size=target_size, mode='bilinear', align_corners=False)\n        \n        return x\n\n\n\nclass Segmenter(nn.Module):\n    def __init__(self, encoder, processor):\n        super().__init__()\n        self.encoder = encoder\n        self.processor = processor\n        self.seg_head = TinyDecoder(768,1)\n        for p in self.encoder.parameters(): \n            p.requires_grad = False\n        self.seg_head.apply(self._init_weights)\n\n    def _init_layer(self, layer):\n        if hasattr(layer, \"weight\"):\n            nn.init.xavier_uniform_(layer.weight)\n        if hasattr(layer, 'bias') and layer.bias is not None:\n            layer.bias.data.fill_(0.)      \n    \n    def _init_bn(self, bn):\n        if hasattr(bn, \"weight\"):\n            bn.weight.data.fill_(1.)\n        if hasattr(bn, \"bias\"):\n            bn.bias.data.fill_(0.)\n    \n    def _init_weights(self, m):\n        if isinstance(m, (nn.Linear, nn.Conv2d)):\n            self._init_layer(m)\n        elif isinstance(m, nn.BatchNorm1d):\n            self._init_bn(m)\n    \n    def forward(self,x):\n        imgs = (x*255).clamp(0,255).byte().permute(0,2,3,1).cpu().numpy()\n        inputs = self.processor(images=list(imgs), return_tensors=\"pt\").to(x.device)\n        # with torch.no_grad(): \n        #     feats = self.encoder(**inputs).last_hidden_state\n        feats = self.encoder(**inputs).last_hidden_state\n        b, n, c = feats.shape\n        fmap = feats[:,1:,:].permute(0,2,1)\n        s = int(math.sqrt(n - 1))\n        fmap = fmap.reshape(b, c, s, s)\n        return self.seg_head(fmap, (img_size, img_size))\n\n\n\n\n\n\ndef my_rle_encode(mask):\n    pixels = mask.T.flatten()\n    dots = np.where(pixels == 1)[0]\n    if len(dots) == 0:\n        return \"authentic\"\n    \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    \n    return json.dumps([int(x) for x in run_lengths])\n\n\n\ndef hysteresis_threshold(preds, t_low, t_high):\n    hi = (preds > t_high).astype(np.uint8)\n    lo = (preds > t_low).astype(np.uint8)\n    num_labels, labels = cv2.connectedComponents(lo)\n    out = np.zeros_like(lo)\n    for lab in range(1, num_labels):\n        component = (labels == lab)\n        if (hi & component).any():\n            out[component] = 1\n\n    return out\n        \n\n \ndef postprocess(preds, original_size, alpha_grad=0.35):\n    gx = cv2.Sobel(preds, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(preds, cv2.CV_32F, 0, 1, ksize=3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_mag / (grad_mag.max() + 1e-6)\n    enhanced = (1 - alpha_grad) * preds + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (3, 3), 0)\n    \n    # thr = np.mean(enhanced) + 0.3 * np.std(enhanced)\n    # mask = (enhanced > thr).astype(np.uint8)\n    # mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8))\n    # mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((3, 3), np.uint8))\n    \n    t_high = enhanced.mean() + 0.6 * enhanced.std()\n    t_low  = enhanced.mean() + 0.2 * enhanced.std()\n    mask = hysteresis_threshold(enhanced, t_low, t_high)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5,5)))\n    \n    mask = cv2.resize(mask, original_size, interpolation=cv2.INTER_NEAREST)\n    \n    return mask\n\n\n\n\n@torch.no_grad()\ndef predict_with_tta(model, image):\n    predictions = []\n    pred = torch.sigmoid(model(image))\n    predictions.append(pred)\n    # pred = torch.sigmoid(model(torch.flip(image, dims=[3])))\n    # predictions.append(torch.flip(pred, dims=[3]))\n    # pred = torch.sigmoid(model(torch.flip(image, dims=[2])))\n    # predictions.append(torch.flip(pred, dims=[2]))\n    # pred = torch.sigmoid(model(torch.rot90(image, 1, [2, 3])))\n    # predictions.append(torch.rot90(pred, -1, [2, 3]))\n    return torch.stack(predictions).mean(0)[0, 0].cpu().numpy()\n\n\n\n@torch.no_grad()\ndef predict_with_tta_ensemble(models, image):\n    predictions = []\n    for model in models:\n        pred = predict_with_tta(model, image)\n        predictions.append(pred)\n    return np.mean(predictions, axis=0)\n\n\n\ndef infer_image_annotation(classifier_models, segmentation_models, image):\n    image_array = np.array(image.resize((img_size, img_size)))\n    image_tensor = torch.from_numpy(np.array(image_array, np.float32)/255.).permute(2,0,1)\n    image_tensor = image_tensor.unsqueeze(0).to(device)\n    \n    # prob = ensemble_classifier(classifier_models, image_tensor)\n    # if prob < 0.2:\n    #     return \"authentic\"\n    \n    pred = predict_with_tta_ensemble(segmentation_models, image_tensor)\n    mask = postprocess(pred, image.size)\n    w, h = image.size\n    \n    area = int(mask.sum())\n    if area > 0:\n        mean_inside = float(\n            pred[cv2.resize(mask, (img_size, img_size), interpolation=cv2.INTER_NEAREST) == 1].mean()\n        )\n    else:\n        mean_inside = 0.0\n\n    if area < 400 or mean_inside < 0.3:\n        return \"authentic\"  \n    \n    return my_rle_encode((mask > 0).astype(np.uint8))\n        \n\"\"\" ------------------------------------------------------------------------------------------------------- \"\"\"\n\n@torch.no_grad()\ndef segment_prob_map(pil):\n    x = torch.from_numpy(np.array(pil.resize((img_size, img_size)), np.float32)/255.).permute(2,0,1)[None].to(device)\n    prob = torch.sigmoid(model(x))[0,0].cpu().numpy()\n    return prob\n\ndef enhanced_adaptive_mask(prob, alpha_grad=0.35):\n    gx = cv2.Sobel(prob, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(prob, cv2.CV_32F, 0, 1, ksize=3)\n    grad_mag = np.sqrt(gx**2 + gy**2)\n    grad_norm = grad_mag / (grad_mag.max() + 1e-6)\n    enhanced = (1 - alpha_grad) * prob + alpha_grad * grad_norm\n    enhanced = cv2.GaussianBlur(enhanced, (3,3), 0)\n    thr = np.mean(enhanced) + 0.3 * np.std(enhanced)\n    mask = (enhanced > thr).astype(np.uint8)\n    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, np.ones((5,5), np.uint8))\n    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, np.ones((3,3), np.uint8))\n    return mask, thr\n\ndef finalize_mask(prob, orig_size):\n    mask, thr = enhanced_adaptive_mask(prob)\n    mask = cv2.resize(mask, orig_size, interpolation=cv2.INTER_NEAREST)\n    return mask, thr\n\ndef pipeline_final(pil):\n    prob = segment_prob_map(pil)\n    mask, thr = finalize_mask(prob, pil.size)\n    area = int(mask.sum())\n    mean_inside = float(prob[cv2.resize(mask,(img_size,img_size),interpolation=cv2.INTER_NEAREST)==1].mean()) if area>0 else 0.0\n    if area < 400 or mean_inside < 0.3:\n        return \"authentic\"\n    return my_rle_encode((mask > 0).astype(np.uint8))\n\n\n\n\n\nclassifiers = []\n'''\nfor model_file in sorted(list(pretrained_classifiers.glob(\"*.safetensors\"))):\n    model = ForgeryClassifier(\"tf_efficientnetv2_s\", pretrained=False, anti_alias=False).to(device)\n    model.load_state_dict(load_file(model_file))\n    model.eval()\n    classifiers.append(model)\n'''\n'''\nsegmenters = []\nfor model_file in sorted(list(pretrained_segmenters.glob(\"*.safetensors\"))):\n    processor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\n    encoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\n    model = Segmenter(encoder=encoder, processor=processor).to(device)\n    model.load_state_dict(load_file(model_file))\n    model.eval()\n    segmenters.append(model)\n'''\n'''\nsegmenter_special = []\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(torch.load('/kaggle/input/m/ravaghi/dinov2/pytorch/base/1/model.pt'))\nmodel.eval()\nsegmenter_special.append(model)\n'''\n'''\nsegmenter_special = []\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(torch.load('/kaggle/input/dinobestmodel/pytorch/default/1/dino197.pth')['model_state_dict'])\nmodel.eval()\nsegmenter_special.append(model)\n'''  \n\n\nsegmenter_special = []\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(load_file('/kaggle/input/dinov2-v9-stage2/dinov2_best_v9_stage2.safetensors'))\nmodel.eval()\nsegmenter_special.append(model)\n\n'''\nif area < 200 or mean_inside < 0.22:\nsegmenter_special = []\nprocessor = AutoImageProcessor.from_pretrained(dinov2_base, use_fast=True, local_files_only=True)\nencoder = AutoModel.from_pretrained(dinov2_base, local_files_only=True)\nmodel = Segmenter(encoder=encoder, processor=processor).to(device)\nmodel.load_state_dict(torch.load('/kaggle/input/cnndinov2-u52/model_seg_final.pt'))\nmodel.eval()\nsegmenter_special.append(model)\n'''\n\n\n\ntest_images = sorted(list(test_images_dir.glob(\"*.png\")))\npredictions = []\nfor img_path in tqdm(test_images):\n    img = Image.open(img_path).convert(\"RGB\")\n    w, h = img.size\n    case_id = img_path.stem\n    # annotation = infer_image_annotation(\n    #     classifier_models=classifiers, \n    #     segmentation_models=segmenter_special,\n    #     # segmentation_models=segmenters,\n    #     image=img\n    # )\n    annotation = pipeline_final(img)\n\n    predictions.append({\n        \"case_id\": case_id,\n        \"annotation\": annotation\n    })\n\n\npredictions = pd.DataFrame(predictions)\npredictions[\"case_id\"] = predictions[\"case_id\"].astype(str)\n\nsubmission = pd.read_csv(sample_submission)\nsubmission[\"case_id\"] = submission[\"case_id\"].astype(str)\n\nsubmission = submission[[\"case_id\"]].merge(predictions, on=\"case_id\", how=\"left\")\nsubmission[\"annotation\"] = submission[\"annotation\"].fillna(\"authentic\")\nsubmission[[\"case_id\", \"annotation\"]].to_csv(\"submission.csv\", index=False)\nsubmission.head()\n        \n        \n    \n\n","metadata":{"execution":{"iopub.status.busy":"2025-12-27T03:37:02.568020Z","iopub.execute_input":"2025-12-27T03:37:02.568308Z","iopub.status.idle":"2025-12-27T03:37:25.801960Z","shell.execute_reply.started":"2025-12-27T03:37:02.568286Z","shell.execute_reply":"2025-12-27T03:37:25.801374Z"},"trusted":true},"outputs":[],"execution_count":null}]}