{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":113558,"databundleVersionId":14878066,"sourceType":"competition"},{"sourceId":14488482,"sourceType":"datasetVersion","datasetId":9128438},{"sourceId":4534,"sourceType":"modelInstanceVersion","modelInstanceId":3326,"modelId":986}],"dockerImageVersionId":31234,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Recod.ai/LUC - Scientific Image Forgery Detection","metadata":{}},{"cell_type":"markdown","source":"## 1. Setup","metadata":{}},{"cell_type":"code","source":"from pathlib import Path\nfrom typing import Any, Literal\nimport yaml\nfrom tqdm import tqdm\nimport math\nimport json\nimport numpy as np\nimport numpy.typing as npt\nimport pandas as pd\nimport cv2\nimport scipy.optimize\nimport numba\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport timm\nimport lightgbm as lgb\nfrom transformers import AutoModel\nimport matplotlib.pyplot as plt\nfrom matplotlib.colors import ListedColormap\nimport matplotlib.patches as mpatches","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:53:34.419462Z","iopub.execute_input":"2026-01-13T17:53:34.420145Z","iopub.status.idle":"2026-01-13T17:53:40.157320Z","shell.execute_reply.started":"2026-01-13T17:53:34.420114Z","shell.execute_reply":"2026-01-13T17:53:40.156502Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"competition_dataset_directory = Path('/kaggle/input/recodai-luc-scientific-image-forgery-detection')\nexternal_dataset_directory = Path('/kaggle/input/recod-ailuc-sifd-dataset')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:53:40.158467Z","iopub.execute_input":"2026-01-13T17:53:40.159289Z","iopub.status.idle":"2026-01-13T17:53:40.164013Z","shell.execute_reply.started":"2026-01-13T17:53:40.159264Z","shell.execute_reply":"2026-01-13T17:53:40.163183Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 2. RLE Utilities","metadata":{}},{"cell_type":"code","source":"@numba.jit(nopython=True)\ndef _rle_encode_jit(x: npt.NDArray, fg_val: int = 1) -> list[int]:\n\n    \"\"\"\n    Perform efficient Run-Length Encoding (RLE) on a single binary mask using Numba.\n    The function flattens the array in column-major order (Fortran-style) and  encodes the positions of the foreground pixels.\n    The output uses 1-based indexing.\n\n    Parameters\n    ----------\n    x: npt.NDArray\n        A 2D binary mask array of shape (Height, Width).\n    \n    fg_val: int, default=1\n        The value inside the mask to consider as foreground (the pixels to encode).\n\n    Returns\n    -------\n    list[int]\n        A list of integers representing the RLE pairs [start, length, start, length, ...].\n    \"\"\"\n\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\n\ndef rle_encode(masks: list[npt.NDArray], fg_val: int = 1) -> str:\n\n    \"\"\"\n    Convert a collection of binary masks into a serialized RLE string format.\n    Each mask in the input list (or channel in a 3D array) is encoded separately and formatted as a JSON list.\n    Multiple masks are separated by semicolons.\n\n    Parameters\n    ----------\n    masks: list[npt.NDArray] or npt.NDArray\n        A list of 2D arrays or a single 3D array of shape (N, Height, Width), where N is the number of instances.\n    \n    fg_val: int, default=1\n        The value inside the masks to consider as foreground.\n\n    Returns\n    -------\n    str\n        A string containing the JSON-encoded RLEs separated by semicolons.\n    \"\"\"\n\n    return ';'.join([json.dumps(_rle_encode_jit(x, fg_val)) for x in masks])\n\n\n@numba.njit\ndef _rle_decode_jit(mask_rle: npt.NDArray, height: int, width: int) -> npt.NDArray:\n\n    \"\"\"\n    Reconstruct a flattened binary mask from Run-Length Encoded (RLE) arrays using Numba.\n    This low-level function processes the RLE pairs to rebuild the binary array.\n    It assumes the input RLE uses 1-based indexing and validates that segments  do not overlap.\n\n    Parameters\n    ----------\n    mask_rle: npt.NDArray\n        A 1D array of integers containing RLE pairs [start, length, start, length, ...].\n    \n    height: int\n        The height of the target image.\n    \n    width: int\n        The width of the target image.\n\n    Returns\n    -------\n    npt.NDArray\n        A flattened 1D binary array (bool) of size height * width, where 1 indicates foreground.\n    \"\"\"\n\n    if len(mask_rle) % 2 != 0:\n        raise ValueError('One or more rows has an odd number of values.')\n\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\n\ndef rle_decode(mask_rle: str, shape: tuple[int, int]) -> npt.NDArray:\n\n    \"\"\"\n    Deserialize an RLE string back into a 2D binary mask.\n    This function parses the JSON string, converts it to a NumPy array.\n    It reshapes the reconstructed flattened mask back into the original 2D image dimensions using Fortran-style (column-major) order.\n\n    Parameters\n    ----------\n    mask_rle: str\n        A JSON-formatted string representing the RLE list (e.g., \"[1, 10, 50, 5]\").\n        Empty predictions should be handled before calling this or will result in an empty mask.\n    \n    shape: tuple[int, int]\n        The target shape (Height, Width) of the decoded mask.\n\n    Returns\n    -------\n    npt.NDArray\n        The reconstructed 2D binary mask array of shape (Height, Width).\n    \"\"\"\n\n    mask_rle = json.loads(mask_rle)\n    mask_rle = np.asarray(mask_rle, dtype=np.int32)\n    return _rle_decode_jit(mask_rle, shape[0], shape[1]).reshape(shape, order='F')\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:52:52.346288Z","iopub.execute_input":"2026-01-13T17:52:52.346526Z","iopub.status.idle":"2026-01-13T17:52:52.487260Z","shell.execute_reply.started":"2026-01-13T17:52:52.346505Z","shell.execute_reply":"2026-01-13T17:52:52.486689Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 3. Visualization Utilities","metadata":{}},{"cell_type":"code","source":"def _to_3d_mask(mask: np.ndarray) -> np.ndarray:\n\n    \"\"\"\n    Ensure a mask has shape (C, H, W).\n\n    Parameters\n    ----------\n    mask: np.ndarray\n        Input mask array of shape (H, W) or (C, H, W).\n\n    Returns\n    -------\n    np.ndarray\n        Mask array with shape (C, H, W).\n    \"\"\"\n\n    if mask is None:\n        return mask\n    \n    if mask.ndim == 2:\n        return mask[np.newaxis, :, :]\n    \n    if mask.ndim == 3:\n        return mask\n\n\ndef _dim_image(image: np.ndarray, factor: float = 0.65) -> np.ndarray:\n\n    \"\"\"\n    Dim an image by scaling pixel intensities.\n\n    Parameters\n    ----------\n    image: np.ndarray\n        Input image array (H, W, C), uint8 or float.\n\n    factor: float, optional\n        Multiplicative dimming factor, by default 0.65.\n\n    Returns\n    -------\n    x: np.ndarray\n        Dimmed image array with the same shape as input.\n    \"\"\"\n\n    if image.dtype == np.uint8:\n        \n        x = image.astype(np.float32) / 255.0\n        x = np.clip(x * factor, 0.0, 1.0)\n\n        return (x * 255.0).astype(np.uint8)\n    \n    x = image.astype(np.float32)\n    x = np.clip(x * factor, 0.0, 1.0)\n    \n    return x\n\n\ndef visualize_image(\n    image: np.ndarray,\n    image_metadata: dict,\n    mask: np.ndarray | None = None,\n    mask_prediction: np.ndarray | None = None,\n    path: str | Path | None = None\n) -> None:\n    \n    \"\"\"\n    Visualize authentic and forged images, overlaying the forgery mask if present.\n\n    Parameters\n    ----------\n    image_authentic: np.ndarray, optional\n        The authentic image array.\n\n    image_metadata: np.dict\n        Dictionary of image metadata.\n\n    mask: np.ndarray, optional\n        Binary segmentation mask where non-zero values indicate forged regions.\n\n    mask_prediction: np.ndarray, optional\n        Binary prediction mask where non-zero values indicate forged regions.\n\n    path: str or pathlib.Path, optional\n        If provided, saves the plot image to the given file path.\n        If None, displays the plot interactively.\n    \"\"\"\n    \n    fig, axes = plt.subplots(1, 2, figsize=(16, 8), dpi=120, constrained_layout=True)\n\n    axes[0].imshow(image)\n    axes[0].set_title(f'Image {image.shape}')\n    axes[0].axis('off')\n\n    axes[1].imshow(_dim_image(image, factor=0.65))\n    axes[1].axis('off')\n\n    mask = _to_3d_mask(mask) if mask is not None else None\n    mask_prediction = _to_3d_mask(mask_prediction) if mask_prediction is not None else None\n\n    legend_handles = []\n    title_lines = [f'Image {image.shape}']\n\n    if mask is not None:\n        has_gt = False\n        cmap_gt = ListedColormap(['#00FF00'])\n        for c in range(mask.shape[0]):\n            m = (mask[c] > 0).astype(np.uint8)\n            if m.max() == 0:\n                continue\n            has_gt = True\n\n            overlay = np.ma.masked_where(m == 0, m)\n            axes[1].imshow(overlay, cmap=cmap_gt, alpha=0.45, interpolation='nearest')\n            axes[1].contour(m, levels=[0.5], linewidths=2.5)\n\n        if has_gt:\n            title_lines.append(f'Annotation {mask.shape} - Pixels {mask.sum()}')\n            legend_handles.append(mpatches.Patch(color='#00FF00', label='Ground Truth', alpha=0.7))\n\n    if mask_prediction is not None:\n        has_pred = False\n        cmap_pred = ListedColormap(['#FF0000'])\n        for c in range(mask_prediction.shape[0]):\n            m = (mask_prediction[c] > 0).astype(np.uint8)\n            if m.max() == 0:\n                continue\n            has_pred = True\n\n            overlay = np.ma.masked_where(m == 0, m)\n            axes[1].imshow(overlay, cmap=cmap_pred, alpha=0.45, interpolation='nearest')\n            axes[1].contour(m, levels=[0.5], linewidths=2.5)\n\n        if has_pred:\n            title_lines.append(f'Prediction {mask_prediction.shape} - Pixels {mask_prediction.sum()}')\n            legend_handles.append(mpatches.Patch(color='#FF0000', label='Prediction', alpha=0.7))\n\n    axes[1].set_title('\\n'.join(title_lines))\n\n    if legend_handles:\n        axes[1].legend(handles=legend_handles, loc='upper right')\n\n    suptitle = f'image_id {image_metadata[\"image_id\"]}'\n    if suptitle:\n        fig.suptitle(suptitle, fontsize=12)\n\n    if path:\n        plt.savefig(path, bbox_inches='tight')\n        plt.close(fig)\n    else:\n        plt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:52:52.488745Z","iopub.execute_input":"2026-01-13T17:52:52.488984Z","iopub.status.idle":"2026-01-13T17:52:52.501903Z","shell.execute_reply.started":"2026-01-13T17:52:52.488962Z","shell.execute_reply":"2026-01-13T17:52:52.501202Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 4. Models","metadata":{}},{"cell_type":"code","source":"def named_trainable_parameters(module: nn.Module) -> list[tuple[str, nn.Parameter]]:\n    return [(n, p) for n, p in module.named_parameters() if p.requires_grad]\n\n\nclass DinoDecoder(nn.Module):\n\n    def __init__(self, in_ch: int, mid_ch: int = 256, out_ch: int = 1, dropout: float = 0.1):\n\n        \"\"\"\n        Lightweight decoder that upsamples a DINO feature map into segmentation logits.\n\n        Parameters\n        ----------\n        in_ch: int\n            Number of input feature channels.\n\n        mid_ch: int\n            Decoder hidden channel width.\n\n        out_ch: int\n            Number of output channels (logits).\n\n        dropout: float\n            Dropout probability.\n        \"\"\"\n\n        super().__init__()\n\n        self.block1 = nn.Sequential(\n            nn.Conv2d(in_ch, mid_ch, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(dropout),\n        )\n        self.block2 = nn.Sequential(\n            nn.Conv2d(mid_ch, mid_ch // 2, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(dropout),\n        )\n        self.block3 = nn.Sequential(\n            nn.Conv2d(mid_ch // 2, mid_ch // 4, kernel_size=3, padding=1),\n            nn.ReLU(inplace=True),\n        )\n        self.out_conv = nn.Conv2d(mid_ch // 4, out_ch, kernel_size=1)\n\n    def forward(self, feature_map: torch.Tensor, out_size: tuple[int, int]) -> torch.Tensor:\n\n        \"\"\"\n        Produce segmentation logits from a feature map.\n\n        Parameters\n        ----------\n        feature_map: torch.Tensor\n            Feature map of shape (B, C, h, w).\n\n        out_size: tuple[int, int]\n            Target output spatial size (H, W).\n\n        Returns\n        -------\n        torch.Tensor\n            Logits of shape (B, out_ch, H, W).\n        \"\"\"\n\n        x = self.block1(feature_map)\n        x = F.interpolate(x, scale_factor=2.0, mode='bilinear', align_corners=False)\n        x = self.block2(x)\n        x = F.interpolate(x, scale_factor=2.0, mode='bilinear', align_corners=False)\n        x = self.block3(x)\n        x = self.out_conv(x)\n        x = F.interpolate(x, size=out_size, mode='bilinear', align_corners=False)\n\n        return x\n\n\nclass DinoV2Segmenter(nn.Module):\n\n    def __init__(\n        self,\n        dino_ckpt: str,\n        *,\n        unfreeze_last_n: int = 0,\n        train_encoder_norm: bool = True,\n        freeze_embeddings: bool = True,\n        decoder_mid_ch: int = 256,\n    ):\n        \n        \"\"\"\n        DINOv2-based segmentation model with a lightweight convolutional decoder.\n\n        Parameters\n        ----------\n        dino_ckpt: str\n            Hugging Face checkpoint name or path.\n\n        unfreeze_last_n: int\n            Number of last transformer blocks to unfreeze (-1 = all).\n\n        train_encoder_norm: bool\n            Whether to unfreeze encoder normalization layers.\n\n        freeze_embeddings: bool\n            Whether to keep embedding layers frozen.\n\n        decoder_mid_ch: int\n            Decoder hidden channel width.\n        \"\"\"\n\n        super().__init__()\n\n        self.encoder = AutoModel.from_pretrained(dino_ckpt)\n        hidden_dim = int(self.encoder.config.hidden_size)\n        self.decoder = DinoDecoder(in_ch=hidden_dim, mid_ch=decoder_mid_ch, out_ch=1)\n\n        self._configure_trainable_encoder_parts(\n            unfreeze_last_n=unfreeze_last_n,\n            train_encoder_norm=train_encoder_norm,\n            freeze_embeddings=freeze_embeddings,\n        )\n\n    def _get_block_list(self) -> nn.ModuleList:\n\n        \"\"\"\n        Return the list of transformer blocks from the encoder.\n\n        Returns\n        -------\n        nn.ModuleList\n            Transformer block modules in forward order.\n        \"\"\"\n\n        try:\n            return self.encoder.encoder.layer\n        except Exception as e:\n            raise AttributeError(\n                'Could not find transformer blocks at `self.encoder.encoder.layer`. '\n                'Inspect your model structure and adjust `_get_block_list`.'\n            ) from e\n\n    def _configure_trainable_encoder_parts(\n        self,\n        *,\n        unfreeze_last_n: int,\n        train_encoder_norm: bool,\n        freeze_embeddings: bool,\n    ) -> None:\n        \n        \"\"\"\n        Freeze the encoder and selectively unfreeze specified components for training.\n\n        Parameters\n        ----------\n        unfreeze_last_n: int\n            Number of last transformer blocks to unfreeze (-1 = unfreeze all).\n\n        train_encoder_norm: bool\n            Whether to unfreeze encoder normalization layers.\n\n        freeze_embeddings: bool\n            Whether to keep embedding layers frozen.\n        \"\"\"\n        \n        for p in self.encoder.parameters():\n            p.requires_grad = False\n\n        blocks = self._get_block_list()\n        num_blocks = len(blocks)\n\n        if unfreeze_last_n == -1:\n            unfreeze_last_n = num_blocks\n\n        if unfreeze_last_n < 0:\n            raise ValueError(f'unfreeze_last_n must be >= 0, got {unfreeze_last_n}')\n        if unfreeze_last_n > num_blocks:\n            raise ValueError(f'unfreeze_last_n={unfreeze_last_n} exceeds num_blocks={num_blocks}')\n\n        if not freeze_embeddings:\n            for p in self.encoder.embeddings.parameters():\n                p.requires_grad = True\n\n        if unfreeze_last_n > 0:\n            for i in range(num_blocks - unfreeze_last_n, num_blocks):\n                for p in blocks[i].parameters():\n                    p.requires_grad = True\n\n            if train_encoder_norm:\n                if hasattr(self.encoder, 'layernorm'):\n                    for p in self.encoder.layernorm.parameters():\n                        p.requires_grad = True\n                elif hasattr(self.encoder, 'encoder') and hasattr(self.encoder.encoder, 'layernorm'):\n                    for p in self.encoder.encoder.layernorm.parameters():\n                        p.requires_grad = True\n\n        for p in self.decoder.parameters():\n            p.requires_grad = True\n\n    def _tokens_to_map(self, tokens: torch.Tensor) -> torch.Tensor:\n\n        \"\"\"\n        Convert transformer tokens into a 2D spatial feature map.\n\n        Parameters\n        ----------\n        tokens: torch.Tensor\n            Token tensor of shape (B, N, C).\n\n        Returns\n        -------\n        torch.Tensor\n            Feature map of shape (B, C, s, s).\n        \"\"\"\n\n        B, N, C = tokens.shape\n        num_reg = int(getattr(self.encoder.config, 'num_register_tokens', 0) or 0)\n        patch_tokens = tokens[:, 1 + num_reg :, :]\n        n_patch = patch_tokens.shape[1]\n        s = int(math.isqrt(n_patch))\n        if s * s != n_patch:\n            raise ValueError(f'Patch token count {n_patch} is not a perfect square; cannot reshape.')\n        return patch_tokens.permute(0, 2, 1).reshape(B, C, s, s)\n    \n    def param_groups(\n        self,\n        *,\n        lr_decoder: float,\n        lr_encoder: float,\n        weight_decay: float,\n    ) -> list[dict]:\n\n        \"\"\"\n        Build optimizer parameter groups for encoder and decoder.\n\n        Parameters\n        ----------\n        lr_decoder: float\n            Learning rate for decoder parameters.\n\n        lr_encoder: float\n            Learning rate for encoder parameters.\n\n        weight_decay: float\n            Weight decay value.\n\n        Returns\n        -------\n        list[dict]\n            Optimizer parameter groups.\n        \"\"\"\n\n        decoder_params = [p for _, p in named_trainable_parameters(self.decoder)]\n        encoder_params = [p for _, p in named_trainable_parameters(self.encoder)]\n\n        groups: list[dict] = []\n        if encoder_params:\n            groups.append(\n                {'params': encoder_params, 'lr': lr_encoder, 'weight_decay': weight_decay}\n            )\n        if decoder_params:\n            groups.append(\n                {'params': decoder_params, 'lr': lr_decoder, 'weight_decay': weight_decay}\n            )\n\n        return groups\n\n    def forward(self, x: torch.Tensor) -> torch.Tensor:\n\n        \"\"\"\n        Run encoder and decoder to produce segmentation logits.\n\n        Parameters\n        ----------\n        x: torch.Tensor\n            Input image tensor of shape (B, C, H, W).\n\n        Returns\n        -------\n        torch.Tensor\n            Segmentation logits of shape (B, 1, H, W).\n        \"\"\"\n\n        H, W = x.shape[-2:]\n        out = self.encoder(pixel_values=x).last_hidden_state\n        feature_map = self._tokens_to_map(out)\n        outputs = self.decoder(feature_map, (H, W))\n\n        return outputs\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:52:52.502944Z","iopub.execute_input":"2026-01-13T17:52:52.503381Z","iopub.status.idle":"2026-01-13T17:52:52.526264Z","shell.execute_reply.started":"2026-01-13T17:52:52.503358Z","shell.execute_reply":"2026-01-13T17:52:52.525621Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def load_dino_models(\n    model_directory: str | Path,\n    device: str | torch.device = 'cpu',\n) -> tuple[dict[str, Any], dict[str, torch.nn.Module]]:\n    \n    \"\"\"\n    Loads a YAML configuration and trained torch models from the given directory.\n\n    Parameters\n    ----------\n    model_directory: str | Path\n        The path to the directory containing config.yaml and torch model files (matching the glob pattern 'model*').\n\n    device: str | torch.device, default='cpu'\n        Model location.\n\n    Returns\n    -------\n    config: dict[str, Any]\n        Dictionary parsed from config file.\n\n    models: dict[str, torch.nn.Module]\n        Models dictionaries.\n    \"\"\"\n\n    model_directory = Path(model_directory)\n    device = torch.device(device)\n\n    if not model_directory.exists():\n        raise FileNotFoundError(f'Directory not found: {model_directory}')\n\n    config_path = model_directory / 'config.yaml'\n    if not config_path.exists():\n        raise FileNotFoundError(f'Config file not found at: {config_path}')\n\n    config = yaml.load(open(config_path), Loader=yaml.FullLoader)\n    config['model']['model_args']['dino_ckpt'] = '/kaggle/input/dinov2/pytorch/base/1'\n\n    models = {}\n    model_files = sorted(model_directory.glob('model*'))\n\n    for model_path in tqdm(model_files, desc=f'Loading models from {model_directory}'):\n\n        model = eval(config['model']['model_class'])(**config['model']['model_args'])\n        model.load_state_dict(torch.load(model_path, map_location=torch.device(device)))\n        model.eval()\n        model.to(device)\n\n        model_name = model_path.stem\n        models[model_name] = model\n\n    return config, models","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:52:52.527220Z","iopub.execute_input":"2026-01-13T17:52:52.527470Z","iopub.status.idle":"2026-01-13T17:52:52.544828Z","shell.execute_reply.started":"2026-01-13T17:52:52.527449Z","shell.execute_reply":"2026-01-13T17:52:52.544238Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device('cuda')\n\nconfig, models = load_dino_models(\n    model_directory=external_dataset_directory / 'dinov2_segmenter',\n    device=device\n)\nmodels = {k: v for k, v in models.items() if 'loss' in k}\nprint(json.dumps(config, indent=2))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T17:52:52.545801Z","iopub.execute_input":"2026-01-13T17:52:52.546222Z","iopub.status.idle":"2026-01-13T17:53:32.686745Z","shell.execute_reply.started":"2026-01-13T17:52:52.546199Z","shell.execute_reply":"2026-01-13T17:53:32.686031Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def load_lgb_models(\n    model_directory: str | Path\n) -> tuple[dict[str, Any], dict[str, lgb.Booster]]:\n    \n    \"\"\"\n    Loads a YAML configuration and trained LightGBM models from the given directory.\n\n    Parameters\n    ----------\n    model_directory: str | Path\n        The path to the directory containing 'config.yaml' and LightGBM model files (matching the glob pattern 'model*').\n\n    Returns\n    -------\n    tuple[dict[str, Any], dict[str, lgb.Booster]]\n        A tuple containing:\n        1. config: The dictionary parsed from config.yaml.\n        2. models: A dictionary where keys are filenames without extension and values are the loaded lgb.Booster objects.\n    \"\"\"\n\n    model_directory = Path(model_directory)\n    \n    if not model_directory.exists():\n        raise FileNotFoundError(f'Directory not found: {model_directory}')\n\n    config_path = model_directory / 'config.yaml'\n    if not config_path.exists():\n        raise FileNotFoundError(f'Config file not found at: {config_path}')\n\n    config = yaml.load(open(config_path), Loader=yaml.FullLoader)\n\n    models = {}\n    model_files = sorted(model_directory.glob('model*'))\n    \n    for model_path in tqdm(model_files, desc=f'Loading models from {model_directory}'):\n\n        model = lgb.Booster(model_file=model_path)\n        model_name = model_path.stem \n        models[model_name] = model\n\n    return config, models\n\n\ndef lgb_predict(\n    df: pd.DataFrame,\n    model_name: str,\n    config: dict[str, Any],\n    models: dict[str, lgb.Booster],\n    aggregation: Literal['mean', 'median', 'min', 'max'] = 'mean',\n    verbose: bool = False\n) -> pd.DataFrame:\n    \n    \"\"\"\n    Generates ensemble predictions by stacking LightGBM model outputs and applying an aggregation function.\n\n    Parameters\n    ----------\n    df: pd.DataFrame\n        The input DataFrame containing the features required for prediction.\n    \n    model_name: str\n        Identifier for the model, used to name the output column (e.g., '{model_name}_prediction').\n    \n    config: dict[str, Any]\n        Configuration dictionary. Must contain ['training']['features'].\n    \n    models : dict[str, lgb.Booster]\n        A dictionary of trained LightGBM Booster objects.\n    \n    aggregation: {'mean', 'median', 'min', 'max'}, default='mean'\n        The method used to combine predictions from multiple models.\n    \n    verbose: bool, default=False\n        If True, prints statistics for each individual model.\n\n    Returns\n    -------\n    df: pd.DataFrame\n        The input DataFrame with the added ensemble prediction column.\n    \"\"\"\n\n    feature_columns = config['training']['features']\n\n    n_samples = len(df)\n    n_models = len(models)\n    predictions = np.zeros((n_samples, n_models), dtype=np.float32)\n    features = df[feature_columns].to_numpy()\n\n    model_items = list(models.items())\n    for i, (model_filename, model) in enumerate(tqdm(model_items)):\n        model_predictions = model.predict(features, num_iteration=config['fit_parameters']['boosting_rounds'])\n        predictions[:, i] = model_predictions\n        if verbose:\n            print(\n                f'LightGBM ({model_filename}) - '\n                f'mean: {np.mean(model_predictions):.4f} | std: {np.std(model_predictions):.4f} | min: {np.min(model_predictions):.4f} | max: {np.max(model_predictions):.4f}'\n            )\n    \n    if aggregation == 'mean':\n        predictions = np.mean(predictions, axis=1)\n    elif aggregation == 'median':   \n        predictions = np.median(predictions, axis=1)\n    elif aggregation == 'max':\n        predictions = np.max(predictions, axis=1)\n    elif aggregation == 'min':\n        predictions = np.min(predictions, axis=1)\n    else:\n        raise ValueError(f'Unknown aggregation method: {aggregation}')\n    \n    print(\n        f'Aggregated predictions ({aggregation})\\n'\n        f'mean: {np.mean(predictions):.4f} | std: {np.std(predictions):.4f} | min: {np.min(predictions):.4f} | max: {np.max(predictions):.4f}'\n    )\n\n    df[f'{model_name}_prediction'] = predictions\n\n    return df\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:04:24.805975Z","iopub.execute_input":"2026-01-13T18:04:24.806277Z","iopub.status.idle":"2026-01-13T18:04:24.816814Z","shell.execute_reply.started":"2026-01-13T18:04:24.806250Z","shell.execute_reply":"2026-01-13T18:04:24.816192Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"lgb_config, lgb_models = load_lgb_models(\n    model_directory=external_dataset_directory / 'lightgbm_classifier'\n)\nprint(json.dumps(lgb_config, indent=2))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:01:04.772778Z","iopub.execute_input":"2026-01-13T18:01:04.773392Z","iopub.status.idle":"2026-01-13T18:01:05.025007Z","shell.execute_reply.started":"2026-01-13T18:01:04.773363Z","shell.execute_reply":"2026-01-13T18:01:05.024320Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 5. Transforms","metadata":{}},{"cell_type":"code","source":"def apply_tta(x: torch.Tensor, k: int, hflip: bool) -> torch.Tensor:\n\n    \"\"\"\n    Apply test-time augmentation.\n\n    Parameters\n    ----------\n    x: torch.Tensor\n        Input tensor with spatial dimensions in the last two axes.\n\n    k: int\n        Number of 90 degree counter-clockwise rotations.\n\n    hflip: bool\n        Whether to apply horizontal flip.\n\n    Returns\n    -------\n    x: torch.Tensor\n        Augmented tensor.\n    \"\"\"\n\n    if hflip:\n        x = torch.flip(x, dims=[-1])\n\n    if k > 0:\n        x = torch.rot90(x, k, dims=(-2, -1))\n        \n    return x\n\n\ndef invert_tta(x: torch.Tensor, k: int, hflip: bool) -> torch.Tensor:\n\n    \"\"\"\n    Invert test-time augmentation.\n\n    Parameters\n    ----------\n    x: torch.Tensor\n        Tensor in augmented orientation.\n\n    k: int\n        Number of 90 degree counter-clockwise rotations.\n\n    hflip: bool\n        Whether horizontal flip was applied.\n\n    Returns\n    -------\n    x: torch.Tensor\n        Tensor restored to original orientation.\n    \"\"\"\n\n    if k > 0:\n        x = torch.rot90(x, (4 - k) % 4, dims=(-2, -1))\n\n    if hflip:\n        x = torch.flip(x, dims=[-1])\n\n    return x","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:01:10.898379Z","iopub.execute_input":"2026-01-13T18:01:10.899090Z","iopub.status.idle":"2026-01-13T18:01:10.904437Z","shell.execute_reply.started":"2026-01-13T18:01:10.899059Z","shell.execute_reply":"2026-01-13T18:01:10.903714Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## 6. Inference","metadata":{}},{"cell_type":"code","source":"image_directory = competition_dataset_directory / 'test_images'\n\ndf = pd.read_csv(competition_dataset_directory / 'sample_submission.csv')\nprint(f'Test Set Shape {df.shape}')\ndisplay(df)\n\nsubmission_mode = df.shape[0] != 1\nprint(f'Submission Mode {submission_mode}')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:01:12.425251Z","iopub.execute_input":"2026-01-13T18:01:12.426005Z","iopub.status.idle":"2026-01-13T18:01:12.456348Z","shell.execute_reply.started":"2026-01-13T18:01:12.425976Z","shell.execute_reply":"2026-01-13T18:01:12.455809Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device('cuda')\namp = True\n\nresize_height = 512\nresize_width = 512\nnormalization_mean = np.array((0.485, 0.456, 0.406), dtype=np.float32)\nnormalization_std = np.array((0.229, 0.224, 0.225), dtype=np.float32)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:01:13.361005Z","iopub.execute_input":"2026-01-13T18:01:13.361303Z","iopub.status.idle":"2026-01-13T18:01:13.365822Z","shell.execute_reply.started":"2026-01-13T18:01:13.361277Z","shell.execute_reply":"2026-01-13T18:01:13.365107Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def _safe_skew_kurt(x: np.ndarray) -> tuple[float, float]:\n\n    \"\"\"\n    Compute population skewness and excess kurtosis in a numerically safe way.\n\n    Parameters\n    ----------\n    x: np.ndarray\n        Input array of values.\n\n    Returns\n    -------\n    skew: float\n        Population skewness.\n\n    kurt: float\n        Population excess kurtosis..\n    \"\"\"\n\n    x = x.astype(np.float64)\n    mu = x.mean()\n    s = x.std()\n    if s < 1e-12:\n        return 0.0, 0.0\n    z = (x - mu) / s\n    skew = float(np.mean(z**3))\n    kurt = float(np.mean(z**4) - 3.0)\n\n    return skew, kurt\n\n\ndef extract_probability_stats(probabilities) -> dict:\n\n    \"\"\"\n    Extract global, spatial, and distributional statistics from a probability map.\n\n    Parameters\n    ----------\n    probabilities: np.ndarray\n        2D array of shape (H, W) with values in [0, 1].\n\n    Returns\n    -------\n    stats: dict\n        Dictionary mapping feature names to scalar float values.\n    \"\"\"\n\n    p = probabilities.astype(np.float32)\n    H, W = p.shape\n    N = H * W\n    flat = p.ravel()\n\n    stats = {\n        'p_min': float(flat.min()),\n        'p_max': float(flat.max()),\n        'p_mean': float(flat.mean()),\n        'p_std': float(flat.std()),\n        'p_q90': float(np.quantile(flat, 0.90)),\n        'p_q95': float(np.quantile(flat, 0.95)),\n        'p_q99': float(np.quantile(flat, 0.99)),\n        'p_q999': float(np.quantile(flat, 0.999)),\n    }\n\n    skew, kurt = _safe_skew_kurt(flat)\n    stats['p_skew'] = skew\n    stats['p_kurtosis'] = kurt\n\n    hist, _ = np.histogram(flat, bins=64, range=(0.0, 1.0), density=True)\n    hist = hist + 1e-12\n    hist = hist / hist.sum()\n    stats['p_entropy'] = float(-np.sum(hist * np.log(hist)))\n    stats['frac_040_060'] = float(((p >= 0.4) & (p <= 0.6)).mean())\n\n    for t in (0.2, 0.3, 0.4, 0.5, 0.55, 0.6, 0.7, 0.8, 0.9):\n        stats[f'frac_gt_{t:.2f}'] = float((p > t).mean())\n\n    for k in (50, 200, 1000):\n        kk = min(k, N)\n        stats[f'topk_mean_{k}'] = float(np.mean(np.partition(flat, -kk)[-kk:]))\n\n    stats['q99_minus_q95'] = float(stats['p_q99'] - stats['p_q95'])\n    stats['q999_minus_q99'] = float(stats['p_q999'] - stats['p_q99'])\n\n    wsum = float(flat.sum()) + 1e-12\n    ys = np.arange(H, dtype=np.float32)[:, None]\n    xs = np.arange(W, dtype=np.float32)[None, :]\n    cy = float((p * ys).sum() / wsum) / max(H - 1, 1)\n    cx = float((p * xs).sum() / wsum) / max(W - 1, 1)\n    stats['com_y'] = cy\n    stats['com_x'] = cx\n    stats['com_dist_center'] = float(np.sqrt((cy - 0.5) ** 2 + (cx - 0.5) ** 2))\n\n    k = max(1, int(0.08 * min(H, W)))\n    border = np.zeros((H, W), np.uint8)\n    border[:k, :] = 1; border[-k:, :] = 1; border[:, :k] = 1; border[:, -k:] = 1\n    stats['border_mass_frac'] = float((p * border).sum() / wsum)\n\n    gx = cv2.Sobel(p, cv2.CV_32F, 1, 0, ksize=3)\n    gy = cv2.Sobel(p, cv2.CV_32F, 0, 1, ksize=3)\n    g = np.sqrt(gx * gx + gy * gy)\n    stats['grad_mean'] = float(g.mean())\n    stats['grad_q95'] = float(np.quantile(g.ravel(), 0.95))\n    stats['lap_var'] = float(cv2.Laplacian(p, cv2.CV_32F).var())\n\n    return stats\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:01:16.500081Z","iopub.execute_input":"2026-01-13T18:01:16.500378Z","iopub.status.idle":"2026-01-13T18:01:16.513393Z","shell.execute_reply.started":"2026-01-13T18:01:16.500351Z","shell.execute_reply":"2026-01-13T18:01:16.512736Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for idx, row in df.iterrows():\n\n    image_path = image_directory / f'{row['case_id']}.png'\n    image = cv2.imread(image_path)\n    \n    if image is None:\n        # Kaggle sample submission file error fallback \n        df.loc[idx, 'annotation'] = 'authentic'\n        continue\n\n    image_raw_shape = image.shape[:2]\n    image_raw = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n    image = cv2.resize(\n        image,\n        (resize_height, resize_width),\n        interpolation=cv2.INTER_LINEAR,\n    )\n    image = image.astype(np.float32) / 255.0\n    image = (image - normalization_mean) / normalization_std\n    image = np.transpose(image, (2, 0, 1))\n\n    inputs = torch.from_numpy(image)\n    inputs = inputs.unsqueeze(0).to(device)\n    prediction = None\n\n    for model_filename, model in tqdm(models.items()):\n        with torch.no_grad():\n            for hflip in (False, True):\n                for k in (0, 1, 2, 3):\n                    inputs_tta = apply_tta(x=inputs, k=k, hflip=hflip)\n                    if amp:\n                        with torch.autocast(device_type=device.type, dtype=torch.float16):\n                            outputs = model(inputs_tta)\n                    else:\n                        outputs = model(inputs_tta)\n\n                    outputs = invert_tta(x=outputs, k=k, hflip=hflip).float()\n                    if prediction is None:\n                        prediction = outputs\n                    else:\n                        prediction += outputs\n\n    prediction /= 40\n    prediction = torch.sigmoid(prediction)\n    prediction = prediction[0, 0, :, :].cpu().numpy()\n    \n    prediction_stats = extract_probability_stats(probabilities=prediction)\n    df_prediction_stats = pd.DataFrame([prediction_stats])\n    df_prediction_stats = lgb_predict(\n        df=df_prediction_stats,\n        model_name='lgb',\n        config=lgb_config,\n        models=lgb_models,\n        aggregation='mean',\n        verbose=not submission_mode\n    )\n    forged_probability = float(df_prediction_stats.loc[0, 'lgb_prediction'])\n\n    probability_q999 = float(np.quantile(prediction, 0.999))\n    if probability_q999 >= 0.95:\n        prediction = cv2.resize(prediction, (image_raw_shape[1], image_raw_shape[0]), interpolation=cv2.INTER_CUBIC)\n        prediction = (prediction > 0.4).astype(np.uint8)\n        annotation = rle_encode([prediction])\n    else:\n        annotation = 'authentic'\n\n       \n    df.loc[idx, 'annotation'] = annotation\n\n    if not submission_mode:\n        visualize_image(\n            image=image_raw,\n            image_metadata={'image_id': row['case_id']},\n            mask=None,\n            mask_prediction=prediction,\n            path=None\n        )","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-14T06:18:05.468798Z","iopub.execute_input":"2026-01-14T06:18:05.469152Z","iopub.status.idle":"2026-01-14T06:18:05.495768Z","shell.execute_reply.started":"2026-01-14T06:18:05.469120Z","shell.execute_reply":"2026-01-14T06:18:05.494767Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df.to_csv('submission.csv', index=False)\ndisplay(df)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-13T18:10:29.257418Z","iopub.execute_input":"2026-01-13T18:10:29.257743Z","iopub.status.idle":"2026-01-13T18:10:29.269985Z","shell.execute_reply.started":"2026-01-13T18:10:29.257720Z","shell.execute_reply":"2026-01-13T18:10:29.269266Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}