{"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":14456136,"sourceType":"competition"},{"sourceId":4534,"sourceType":"modelInstanceVersion","modelInstanceId":3326,"modelId":986}],"dockerImageVersionId":31154,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"\n
\n\n \n
\n

\n 🧩 Unmasking the Fake with a CNN–DINOv2 Hybrid\n

\n
\n\n \n

\n A segmentation-driven pipeline for scientific image forgery detection — combining deep embeddings and adaptive CNN decoding.\n

\n\n \n
\n

🧠 Overview

\n

\n This solution implements a segmentation-based hybrid model that merges a self-supervised visual encoder \n (from DINOv2) with a lightweight CNN decoder. \n Its goal is to detect and localize image manipulations at the pixel level with high precision.\n

\n
\n\n \n
\n

⚙️ Architecture

\n
    \n
  1. Visual Encoder — extracts high-level semantic features using DINOv2 embeddings.
    2. \n
  2. CNN Decoder — converts those features into a binary segmentation mask (768→256→64→1).
    3. \n
  3. Resizing — all images and masks are resized to 256×256 for uniform input.
    4. \n
\n
\n\n \n
\n

🏋️ Training

\n \n
\n\n \n
\n

🔍 Inference & Post-Processing

\n \n
\n\n \n
\n

📊 Evaluation

\n

\n The model is evaluated on a validation subset of forged images. \n Each predicted mask is compared with its ground truth using the F1-score. \n The average score provides a reliable measure of segmentation performance.\n

\n
\n\n \n
\n

✅ Key Highlights

\n \n
\n\n ","metadata":{}},{"cell_type":"code","source":"yaml_text = \"\"\"\n# Dataset paths\ndata:\n train_dir: \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train\"\n test_dir: \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test\"\n train_csv: \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train.csv\"\n test_csv: \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test.csv\"\n dinov2_path: \"/kaggle/input/dinov2/pytorch/base/1\"\n\ntraining:\n n_folds: 5\n seed: 42\n epochs: 30\n batch_size: 16\n num_workers: 4\n image_size: 384\n optimizer: \"AdamW\"\n learning_rate: 1e-4\n weight_decay: 1e-2\n scheduler: \"CosineAnnealingWarmRestarts\"\n warmup_epochs: 3\n min_lr: 1e-6\n loss_fn: \"focal\"\n focal_gamma: 2.0\n label_smoothing: 0.1\n patience: 7\n\nmodels:\n - name: \"swin_base_patch4_window12_384\"\n image_size: 384\n pretrained: true\n - name: \"convnext_base\"\n image_size: 512\n pretrained: true\n - name: \"vit_base_patch16_384\"\n image_size: 384\n pretrained: true\n - name: \"tf_efficientnetv2_l\"\n image_size: 480\n pretrained: true\n\naugmentation:\n train:\n horizontal_flip: 0.5\n vertical_flip: 0.3\n rotate_limit: 15\n brightness_limit: 0.2\n contrast_limit: 0.2\n blur_limit: 3\n jpeg_quality: [70, 100]\n noise_var: [10, 50]\n mixup_alpha: 0.2\n cutmix_prob: 0.5\n tta:\n enabled: true\n n_augments: 5\n\ninference:\n tta: true\n threshold: 0.5\n calibration: true\n\"\"\"\n\nwith open(\"config.yaml\", \"w\") as f:\n f.write(yaml_text)\n\nprint(\"config.yaml created!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:19:14.774608Z","iopub.execute_input":"2025-11-17T07:19:14.774881Z","iopub.status.idle":"2025-11-17T07:19:14.780354Z","shell.execute_reply.started":"2025-11-17T07:19:14.774862Z","shell.execute_reply":"2025-11-17T07:19:14.779603Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport cv2\nimport numpy as np\nimport pandas as pd\nimport torch\nfrom torch.utils.data import Dataset, DataLoader\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\nfrom sklearn.model_selection import StratifiedKFold, GroupKFold\nimport yaml\n\nclass ForgeryDataset(Dataset):\n def __init__(self, df, image_dir, transform=None, is_test=False):\n self.df = df.reset_index(drop=True)\n self.image_dir = image_dir\n self.transform = transform\n self.is_test = is_test\n \n def __len__(self):\n return len(self.df)\n \n def __getitem__(self, idx):\n row = self.df.iloc[idx]\n \n # Load image\n img_path = os.path.join(self.image_dir, row['filename'])\n image = cv2.imread(img_path)\n image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n \n # Apply transforms\n if self.transform:\n augmented = self.transform(image=image)\n image = augmented['image']\n \n if self.is_test:\n return {\n 'image': image,\n 'filename': row['filename']\n }\n else:\n label = torch.tensor(row['label'], dtype=torch.float32)\n return {\n 'image': image,\n 'label': label,\n 'filename': row['filename']\n }\n\ndef get_train_transforms(config):\n \"\"\"Strong augmentations for training\"\"\"\n img_size = config['training']['image_size']\n aug_config = config['augmentation']['train']\n \n return A.Compose([\n A.Resize(img_size, img_size),\n A.HorizontalFlip(p=aug_config['horizontal_flip']),\n A.VerticalFlip(p=aug_config['vertical_flip']),\n A.Rotate(limit=aug_config['rotate_limit'], p=0.5),\n A.RandomBrightnessContrast(\n brightness_limit=aug_config['brightness_limit'],\n contrast_limit=aug_config['contrast_limit'],\n p=0.5\n ),\n A.OneOf([\n A.GaussianBlur(blur_limit=aug_config['blur_limit']),\n A.GaussNoise(var_limit=aug_config['noise_var']),\n A.ImageCompression(\n quality_lower=aug_config['jpeg_quality'][0],\n quality_upper=aug_config['jpeg_quality'][1]\n ),\n ], p=0.5),\n A.ShiftScaleRotate(shift_limit=0.1, scale_limit=0.1, rotate_limit=15, p=0.5),\n A.CoarseDropout(max_holes=8, max_height=32, max_width=32, p=0.3),\n A.Normalize(\n mean=[0.485, 0.456, 0.406],\n std=[0.229, 0.224, 0.225]\n ),\n ToTensorV2()\n ])\n\ndef get_valid_transforms(config):\n \"\"\"Minimal transforms for validation\"\"\"\n img_size = config['training']['image_size']\n \n return A.Compose([\n A.Resize(img_size, img_size),\n A.Normalize(\n mean=[0.485, 0.456, 0.406],\n std=[0.229, 0.224, 0.225]\n ),\n ToTensorV2()\n ])\n\ndef get_tta_transforms(config):\n \"\"\"Test-time augmentation transforms\"\"\"\n img_size = config['training']['image_size']\n \n transforms = [\n # Original\n A.Compose([\n A.Resize(img_size, img_size),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2()\n ]),\n # Horizontal flip\n A.Compose([\n A.Resize(img_size, img_size),\n A.HorizontalFlip(p=1.0),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2()\n ]),\n # Vertical flip\n A.Compose([\n A.Resize(img_size, img_size),\n A.VerticalFlip(p=1.0),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2()\n ]),\n # Slight rotation\n A.Compose([\n A.Resize(img_size, img_size),\n A.Rotate(limit=5, p=1.0),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2()\n ]),\n # Scale\n A.Compose([\n A.Resize(int(img_size * 1.1), int(img_size * 1.1)),\n A.CenterCrop(img_size, img_size),\n A.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n ToTensorV2()\n ]),\n ]\n \n return transforms\n\ndef create_folds(df, config):\n \"\"\"Create stratified K-fold splits\"\"\"\n n_folds = config['training']['n_folds']\n seed = config['training']['seed']\n \n df['fold'] = -1\n \n # Check if there are groups (e.g., multiple images from same source)\n if 'group' in df.columns:\n print(\"Using GroupKFold\")\n gkf = GroupKFold(n_splits=n_folds)\n for fold, (train_idx, val_idx) in enumerate(gkf.split(df, df['label'], df['group'])):\n df.loc[val_idx, 'fold'] = fold\n else:\n print(\"Using StratifiedKFold\")\n skf = StratifiedKFold(n_splits=n_folds, shuffle=True, random_state=seed)\n for fold, (train_idx, val_idx) in enumerate(skf.split(df, df['label'])):\n df.loc[val_idx, 'fold'] = fold\n \n return df\n\ndef get_dataloaders(train_df, fold, config):\n \"\"\"Create train and validation dataloaders\"\"\"\n train_fold = train_df[train_df['fold'] != fold].reset_index(drop=True)\n valid_fold = train_df[train_df['fold'] == fold].reset_index(drop=True)\n \n train_dataset = ForgeryDataset(\n train_fold,\n config['data']['train_dir'],\n transform=get_train_transforms(config)\n )\n \n valid_dataset = ForgeryDataset(\n valid_fold,\n config['data']['train_dir'],\n transform=get_valid_transforms(config)\n )\n \n train_loader = DataLoader(\n train_dataset,\n batch_size=config['training']['batch_size'],\n shuffle=True,\n num_workers=config['training']['num_workers'],\n pin_memory=True,\n drop_last=True\n )\n \n valid_loader = DataLoader(\n valid_dataset,\n batch_size=config['training']['batch_size'] * 2,\n shuffle=False,\n num_workers=config['training']['num_workers'],\n pin_memory=True\n )\n \n return train_loader, valid_loader\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:19:26.167471Z","iopub.execute_input":"2025-11-17T07:19:26.168205Z","iopub.status.idle":"2025-11-17T07:19:59.053816Z","shell.execute_reply.started":"2025-11-17T07:19:26.168173Z","shell.execute_reply":"2025-11-17T07:19:59.053105Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport timm\nfrom torch.cuda.amp import autocast\n\nclass ForgeryClassifier(nn.Module):\n def __init__(self, model_name, pretrained=True, num_classes=1):\n super().__init__()\n self.model_name = model_name\n \n # Load backbone\n self.backbone = timm.create_model(\n model_name,\n pretrained=pretrained,\n num_classes=0, # Remove classifier\n global_pool=''\n )\n \n # Get feature dimension\n with torch.no_grad():\n dummy_input = torch.randn(1, 3, 384, 384)\n features = self.backbone(dummy_input)\n if len(features.shape) == 4: # Conv features\n self.feature_dim = features.shape[1]\n else:\n self.feature_dim = features.shape\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:21:37.008835Z","iopub.execute_input":"2025-11-17T07:21:37.009548Z","iopub.status.idle":"2025-11-17T07:21:40.829897Z","shell.execute_reply.started":"2025-11-17T07:21:37.009524Z","shell.execute_reply":"2025-11-17T07:21:40.829140Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport timm\nfrom torch.cuda.amp import autocast\nimport math\n\nclass ForgeryClassifier(nn.Module):\n \"\"\"\n Advanced Forgery Detection Model\n Supports: Swin, ConvNeXt, ViT, EfficientNet, and more\n \"\"\"\n def __init__(\n self,\n model_name,\n pretrained=True,\n num_classes=1,\n dropout=0.3,\n use_gem_pooling=False,\n use_attention=False\n ):\n super().__init__()\n self.model_name = model_name\n self.use_gem_pooling = use_gem_pooling\n self.use_attention = use_attention\n \n print(f\"🔨 Building model: {model_name}\")\n \n # Load backbone from timm\n self.backbone = timm.create_model(\n model_name,\n pretrained=pretrained,\n num_classes=0, # Remove classifier head\n global_pool='' # Remove pooling\n )\n \n # Get feature dimension\n self.feature_dim = self._get_feature_dim()\n print(f\"✅ Feature dimension: {self.feature_dim}\")\n \n # Pooling layer\n if use_gem_pooling:\n self.global_pool = GeM()\n else:\n self.global_pool = nn.AdaptiveAvgPool2d(1)\n \n # Attention mechanism (optional)\n if use_attention:\n self.attention = SpatialAttention(self.feature_dim)\n \n # Classifier head\n self.head = nn.Sequential(\n nn.Linear(self.feature_dim, 512),\n nn.BatchNorm1d(512),\n nn.ReLU(inplace=True),\n nn.Dropout(dropout),\n nn.Linear(512, 256),\n nn.BatchNorm1d(256),\n nn.ReLU(inplace=True),\n nn.Dropout(dropout / 2),\n nn.Linear(256, num_classes)\n )\n \n # Initialize weights\n self._init_weights()\n \n def _get_feature_dim(self):\n \"\"\"Automatically detect feature dimension\"\"\"\n with torch.no_grad():\n dummy_input = torch.randn(1, 3, 224, 224)\n features = self.backbone(dummy_input)\n \n if isinstance(features, (list, tuple)):\n features = features[-1]\n \n if len(features.shape) == 4: # Conv features (B, C, H, W)\n return features.shape[1]\n elif len(features.shape) == 3: # Transformer features (B, N, C)\n return features.shape[2]\n else:\n return features.shape[1]\n \n def _init_weights(self):\n \"\"\"Initialize classifier head weights\"\"\"\n for m in self.head.modules():\n if isinstance(m, nn.Linear):\n nn.init.xavier_uniform_(m.weight)\n if m.bias is not None:\n nn.init.constant_(m.bias, 0)\n elif isinstance(m, nn.BatchNorm1d):\n nn.init.constant_(m.weight, 1)\n nn.init.constant_(m.bias, 0)\n \n @autocast()\n def forward(self, x):\n # Extract features from backbone\n features = self.backbone(x)\n \n # Handle different output formats\n if isinstance(features, (list, tuple)):\n features = features[-1]\n \n # Handle transformer vs CNN features\n if len(features.shape) == 3: # Transformer (B, N, C)\n # Use CLS token or mean pooling\n if hasattr(self.backbone, 'num_prefix_tokens') and self.backbone.num_prefix_tokens > 0:\n features = features[:, 0] # CLS token\n else:\n features = features.mean(dim=1) # Mean pooling\n \n elif len(features.shape) == 4: # CNN (B, C, H, W)\n # Apply attention if enabled\n if self.use_attention:\n features = self.attention(features)\n \n # Global pooling\n if self.use_gem_pooling:\n features = self.global_pool(features)\n else:\n features = self.global_pool(features)\n \n features = features.flatten(1)\n \n # Classification head\n output = self.head(features)\n \n return output\n \n def freeze_backbone(self):\n \"\"\"Freeze backbone for initial training\"\"\"\n print(\"🔒 Freezing backbone\")\n for param in self.backbone.parameters():\n param.requires_grad = False\n \n def unfreeze_backbone(self):\n \"\"\"Unfreeze backbone for fine-tuning\"\"\"\n print(\"🔓 Unfreezing backbone\")\n for param in self.backbone.parameters():\n param.requires_grad = True\n \n def get_optimizer_params(self, lr, weight_decay):\n \"\"\"Get parameter groups with differential learning rates\"\"\"\n backbone_params = []\n head_params = []\n \n for name, param in self.named_parameters():\n if not param.requires_grad:\n continue\n \n if 'backbone' in name:\n backbone_params.append(param)\n else:\n head_params.append(param)\n \n return [\n {'params': backbone_params, 'lr': lr, 'weight_decay': weight_decay},\n {'params': head_params, 'lr': lr * 10, 'weight_decay': weight_decay}\n ]\n\n\nclass GeM(nn.Module):\n \"\"\"Generalized Mean Pooling\"\"\"\n def __init__(self, p=3, eps=1e-6):\n super().__init__()\n self.p = nn.Parameter(torch.ones(1) * p)\n self.eps = eps\n \n def forward(self, x):\n return self.gem(x, p=self.p, eps=self.eps)\n \n def gem(self, x, p=3, eps=1e-6):\n return F.avg_pool2d(x.clamp(min=eps).pow(p), (x.size(-2), x.size(-1))).pow(1./p)\n\n\nclass SpatialAttention(nn.Module):\n \"\"\"Spatial Attention Module\"\"\"\n def __init__(self, in_channels):\n super().__init__()\n self.conv = nn.Conv2d(in_channels, 1, kernel_size=1)\n self.sigmoid = nn.Sigmoid()\n \n def forward(self, x):\n attention = self.conv(x)\n attention = self.sigmoid(attention)\n return x * attention\n\n\nclass EnsembleModel(nn.Module):\n \"\"\"Ensemble multiple models\"\"\"\n def __init__(self, models, weights=None):\n super().__init__()\n self.models = nn.ModuleList(models)\n \n if weights is None:\n self.weights = [1.0 / len(models)] * len(models)\n else:\n self.weights = weights\n \n def forward(self, x):\n outputs = []\n for model, weight in zip(self.models, self.weights):\n output = model(x)\n outputs.append(output * weight)\n \n return torch.stack(outputs).sum(dim=0)\n\n\ndef build_model(config, model_config=None):\n \"\"\"\n Build model from config\n \n Args:\n config: Main config dict\n model_config: Specific model config (if None, uses first model in config)\n \"\"\"\n if model_config is None:\n model_config = config['models'][0]\n \n model_name = model_config['name']\n pretrained = model_config.get('pretrained', True)\n \n # Model-specific settings\n dropout = model_config.get('dropout', 0.3)\n use_gem = model_config.get('use_gem_pooling', False)\n use_attention = model_config.get('use_attention', False)\n \n model = ForgeryClassifier(\n model_name=model_name,\n pretrained=pretrained,\n num_classes=1,\n dropout=dropout,\n use_gem_pooling=use_gem,\n use_attention=use_attention\n )\n \n return model\n\n\ndef load_checkpoint(model, checkpoint_path, device='cuda'):\n \"\"\"Load model checkpoint\"\"\"\n print(f\"📂 Loading checkpoint: {checkpoint_path}\")\n checkpoint = torch.load(checkpoint_path, map_location=device)\n \n if 'model_state_dict' in checkpoint:\n model.load_state_dict(checkpoint['model_state_dict'])\n else:\n model.load_state_dict(checkpoint)\n \n print(\"✅ Checkpoint loaded successfully\")\n return model\n\n\n# ============================================\n# Model Factory - Easy model selection\n# ============================================\n\nMODEL_REGISTRY = {\n 'swin_small': 'swin_small_patch4_window7_224',\n 'swin_base_384': 'swin_base_patch4_window12_384',\n 'swin_base': 'swin_base_patch4_window7_224',\n 'swin_large_384': 'swin_large_patch4_window12_384',\n \n 'convnext_tiny': 'convnext_tiny',\n 'convnext_small': 'convnext_small',\n 'convnext_base': 'convnext_base',\n 'convnext_large': 'convnext_large',\n \n 'vit_base': 'vit_base_patch16_224',\n 'vit_base_384': 'vit_base_patch16_384',\n 'vit_large': 'vit_large_patch16_224',\n 'vit_large_384': 'vit_large_patch16_384',\n \n 'efficientnet_b5': 'tf_efficientnet_b5',\n 'efficientnet_b7': 'tf_efficientnet_b7',\n 'efficientnetv2_m': 'tf_efficientnetv2_m',\n 'efficientnetv2_l': 'tf_efficientnetv2_l',\n 'efficientnetv2_xl': 'tf_efficientnetv2_xl',\n \n 'resnet50': 'resnet50',\n 'resnet101': 'resnet101',\n 'resnet152': 'resnet152',\n \n 'densenet121': 'densenet121',\n 'densenet169': 'densenet169',\n \n 'maxvit_base': 'maxvit_base_tf_384',\n 'coatnet_2': 'coatnet_2_rw_224',\n}\n\n\ndef create_model(\n model_name,\n pretrained=True,\n num_classes=1,\n dropout=0.3,\n use_gem_pooling=False,\n use_attention=False\n):\n \"\"\"\n Quick model creation\n \n Usage:\n model = create_model('swin_base_384', pretrained=True)\n \"\"\"\n # Resolve model name\n if model_name in MODEL_REGISTRY:\n model_name = MODEL_REGISTRY[model_name]\n \n model = ForgeryClassifier(\n model_name=model_name,\n pretrained=pretrained,\n num_classes=num_classes,\n dropout=dropout,\n use_gem_pooling=use_gem_pooling,\n use_attention=use_attention\n )\n \n return model\n\n\ndef count_parameters(model):\n \"\"\"Count trainable parameters\"\"\"\n total = sum(p.numel() for p in model.parameters())\n trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)\n \n print(f\"📊 Total parameters: {total:,}\")\n print(f\"📊 Trainable parameters: {trainable:,}\")\n \n return total, trainable\n\n\n# ============================================\n# Test function\n# ============================================\n\ndef test_model():\n \"\"\"Test model creation\"\"\"\n print(\"=\" * 50)\n print(\"Testing Model Creation\")\n print(\"=\" * 50)\n \n models_to_test = [\n 'swin_base_patch4_window12_384',\n 'convnext_base',\n 'vit_base_patch16_384',\n 'tf_efficientnetv2_l'\n ]\n \n for model_name in models_to_test:\n print(f\"\\n🔨 Creating {model_name}\")\n try:\n model = create_model(model_name, pretrained=False)\n count_parameters(model)\n \n # Test forward pass\n dummy_input = torch.randn(2, 3, 384, 384)\n with torch.no_grad():\n output = model(dummy_input)\n print(f\"✅ Output shape: {output.shape}\")\n \n except Exception as e:\n print(f\"❌ Error: {e}\")\n \n print(\"\\n\" + \"=\" * 50)\n print(\"✅ All tests completed!\")\n\n\nif __name__ == '__main__':\n test_model()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:22:44.938083Z","iopub.execute_input":"2025-11-17T07:22:44.938686Z","iopub.status.idle":"2025-11-17T07:22:53.797815Z","shell.execute_reply.started":"2025-11-17T07:22:44.938664Z","shell.execute_reply":"2025-11-17T07:22:53.797030Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport timm\n\nclass FastForgeryDetector(nn.Module):\n def __init__(self, backbone=\"convnext_base\", pretrained=True):\n super().__init__()\n\n self.backbone = timm.create_model(\n backbone,\n pretrained=pretrained,\n num_classes=0,\n global_pool=\"\"\n )\n\n # find feature size\n self.out_dim = self._get_dim()\n\n self.pool = nn.AdaptiveAvgPool2d(1)\n\n self.fc = nn.Sequential(\n nn.Linear(self.out_dim, 256),\n nn.ReLU(True),\n nn.Dropout(0.3),\n nn.Linear(256, 1)\n )\n\n def _get_dim(self):\n with torch.no_grad():\n x = torch.randn(1, 3, 224, 224)\n y = self.backbone(x)\n if y.dim() == 4:\n return y.shape[1]\n else:\n return y.shape[-1]\n\n def forward(self, x):\n y = self.backbone(x)\n if y.dim() == 3:\n y = y.mean(1)\n else:\n y = self.pool(y).flatten(1)\n return self.fc(y)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:24:45.551003Z","iopub.execute_input":"2025-11-17T07:24:45.551531Z","iopub.status.idle":"2025-11-17T07:24:45.557545Z","shell.execute_reply.started":"2025-11-17T07:24:45.551513Z","shell.execute_reply":"2025-11-17T07:24:45.556909Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\nclass FocalLoss(nn.Module):\n \"\"\"\n Focal Loss for imbalanced classification\n \"\"\"\n def __init__(self, alpha=0.25, gamma=2.0, reduction='mean'):\n super().__init__()\n self.alpha = alpha\n self.gamma = gamma\n self.reduction = reduction\n \n def forward(self, inputs, targets):\n BCE_loss = F.binary_cross_entropy_with_logits(\n inputs, targets, reduction='none'\n )\n pt = torch.exp(-BCE_loss)\n F_loss = self.alpha * (1 - pt) ** self.gamma * BCE_loss\n \n if self.reduction == 'mean':\n return F_loss.mean()\n elif self.reduction == 'sum':\n return F_loss.sum()\n else:\n return F_loss\n\n\nclass LabelSmoothingBCELoss(nn.Module):\n \"\"\"\n Binary Cross Entropy with Label Smoothing\n \"\"\"\n def __init__(self, smoothing=0.1):\n super().__init__()\n self.smoothing = smoothing\n \n def forward(self, inputs, targets):\n # Apply label smoothing\n targets = targets * (1 - self.smoothing) + 0.5 * self.smoothing\n return F.binary_cross_entropy_with_logits(inputs, targets)\n\n\nclass ComboLoss(nn.Module):\n \"\"\"\n Combination of BCE and Focal Loss\n \"\"\"\n def __init__(self, alpha=0.5, focal_gamma=2.0, label_smoothing=0.0):\n super().__init__()\n self.alpha = alpha\n self.focal = FocalLoss(gamma=focal_gamma)\n self.bce = LabelSmoothingBCELoss(smoothing=label_smoothing)\n \n def forward(self, inputs, targets):\n focal_loss = self.focal(inputs, targets)\n bce_loss = self.bce(inputs, targets)\n return self.alpha * focal_loss + (1 - self.alpha) * bce_loss\n\n\ndef get_loss_function(config):\n \"\"\"\n Get loss function from config\n \"\"\"\n loss_name = config['training'].get('loss_fn', 'bce')\n \n if loss_name == 'bce':\n return nn.BCEWithLogitsLoss()\n \n elif loss_name == 'focal':\n gamma = config['training'].get('focal_gamma', 2.0)\n return FocalLoss(gamma=gamma)\n \n elif loss_name == 'label_smoothing':\n smoothing = config['training'].get('label_smoothing', 0.1)\n return LabelSmoothingBCELoss(smoothing=smoothing)\n \n elif loss_name == 'combo':\n gamma = config['training'].get('focal_gamma', 2.0)\n smoothing = config['training'].get('label_smoothing', 0.1)\n return ComboLoss(focal_gamma=gamma, label_smoothing=smoothing)\n \n else:\n raise ValueError(f\"Unknown loss function: {loss_name}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:29:55.640690Z","iopub.execute_input":"2025-11-17T07:29:55.641008Z","iopub.status.idle":"2025-11-17T07:29:55.651210Z","shell.execute_reply.started":"2025-11-17T07:29:55.640984Z","shell.execute_reply":"2025-11-17T07:29:55.650569Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\nclass FocalLoss(nn.Module):\n \"\"\"\n Focal Loss for imbalanced classification\n \"\"\"\n def __init__(self, alpha=0.25, gamma=2.0, reduction='mean'):\n super().__init__()\n self.alpha = alpha\n self.gamma = gamma\n self.reduction = reduction\n \n def forward(self, inputs, targets):\n BCE_loss = F.binary_cross_entropy_with_logits(\n inputs, targets, reduction='none'\n )\n pt = torch.exp(-BCE_loss)\n F_loss = self.alpha * (1 - pt) ** self.gamma * BCE_loss\n \n if self.reduction == 'mean':\n return F_loss.mean()\n elif self.reduction == 'sum':\n return F_loss.sum()\n else:\n return F_loss\n\n\nclass LabelSmoothingBCELoss(nn.Module):\n \"\"\"\n Binary Cross Entropy with Label Smoothing\n \"\"\"\n def __init__(self, smoothing=0.1):\n super().__init__()\n self.smoothing = smoothing\n \n def forward(self, inputs, targets):\n # Apply label smoothing\n targets = targets * (1 - self.smoothing) + 0.5 * self.smoothing\n return F.binary_cross_entropy_with_logits(inputs, targets)\n\n\nclass ComboLoss(nn.Module):\n \"\"\"\n Combination of BCE and Focal Loss\n \"\"\"\n def __init__(self, alpha=0.5, focal_gamma=2.0, label_smoothing=0.0):\n super().__init__()\n self.alpha = alpha\n self.focal = FocalLoss(gamma=focal_gamma)\n self.bce = LabelSmoothingBCELoss(smoothing=label_smoothing)\n \n def forward(self, inputs, targets):\n focal_loss = self.focal(inputs, targets)\n bce_loss = self.bce(inputs, targets)\n return self.alpha * focal_loss + (1 - self.alpha) * bce_loss\n\n\ndef get_loss_function(config):\n \"\"\"\n Get loss function from config\n \"\"\"\n loss_name = config['training'].get('loss_fn', 'bce')\n \n if loss_name == 'bce':\n return nn.BCEWithLogitsLoss()\n \n elif loss_name == 'focal':\n gamma = config['training'].get('focal_gamma', 2.0)\n return FocalLoss(gamma=gamma)\n \n elif loss_name == 'label_smoothing':\n smoothing = config['training'].get('label_smoothing', 0.1)\n return LabelSmoothingBCELoss(smoothing=smoothing)\n \n elif loss_name == 'combo':\n gamma = config['training'].get('focal_gamma', 2.0)\n smoothing = config['training'].get('label_smoothing', 0.1)\n return ComboLoss(focal_gamma=gamma, label_smoothing=smoothing)\n \n else:\n raise ValueError(f\"Unknown loss function: {loss_name}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:30:11.884608Z","iopub.execute_input":"2025-11-17T07:30:11.885231Z","iopub.status.idle":"2025-11-17T07:30:11.894540Z","shell.execute_reply.started":"2025-11-17T07:30:11.885207Z","shell.execute_reply":"2025-11-17T07:30:11.893872Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import numpy as np\nfrom sklearn.metrics import (\n roc_auc_score, \n accuracy_score, \n f1_score, \n precision_score, \n recall_score,\n confusion_matrix,\n log_loss\n)\nimport torch\n\nclass AverageMeter:\n \"\"\"Computes and stores the average and current value\"\"\"\n def __init__(self):\n self.reset()\n \n def reset(self):\n self.val = 0\n self.avg = 0\n self.sum = 0\n self.count = 0\n \n def update(self, val, n=1):\n self.val = val\n self.sum += val * n\n self.count += n\n self.avg = self.sum / self.count\n\n\nclass MetricTracker:\n \"\"\"Track and compute metrics\"\"\"\n def __init__(self):\n self.reset()\n \n def reset(self):\n self.predictions = []\n self.targets = []\n self.losses = []\n \n def update(self, preds, targets, loss=None):\n \"\"\"\n Args:\n preds: torch.Tensor or numpy array (logits or probabilities)\n targets: torch.Tensor or numpy array\n loss: float\n \"\"\"\n # Convert to numpy\n if torch.is_tensor(preds):\n preds = preds.detach().cpu().numpy()\n if torch.is_tensor(targets):\n targets = targets.detach().cpu().numpy()\n \n self.predictions.extend(preds.flatten())\n self.targets.extend(targets.flatten())\n \n if loss is not None:\n self.losses.append(loss)\n \n def compute_metrics(self, threshold=0.5):\n \"\"\"Compute all metrics\"\"\"\n predictions = np.array(self.predictions)\n targets = np.array(self.targets)\n \n # Convert logits to probabilities\n probs = torch.sigmoid(torch.tensor(predictions)).numpy()\n \n # Binary predictions\n binary_preds = (probs >= threshold).astype(int)\n \n metrics = {}\n \n # Loss\n if len(self.losses) > 0:\n metrics['loss'] = np.mean(self.losses)\n \n # AUC\n try:\n metrics['auc'] = roc_auc_score(targets, probs)\n except:\n metrics['auc'] = 0.0\n \n # Log Loss\n try:\n metrics['log_loss'] = log_loss(targets, probs)\n except:\n metrics['log_loss'] = 0.0\n \n # Accuracy\n metrics['accuracy'] = accuracy_score(targets, binary_preds)\n \n # F1 Score\n metrics['f1'] = f1_score(targets, binary_preds, zero_division=0)\n \n # Precision\n metrics['precision'] = precision_score(targets, binary_preds, zero_division=0)\n \n # Recall\n metrics['recall'] = recall_score(targets, binary_preds, zero_division=0)\n \n # Confusion Matrix\n tn, fp, fn, tp = confusion_matrix(targets, binary_preds).ravel()\n metrics['true_positives'] = tp\n metrics['false_positives'] = fp\n metrics['true_negatives'] = tn\n metrics['false_negatives'] = fn\n \n # Specificity\n metrics['specificity'] = tn / (tn + fp) if (tn + fp) > 0 else 0.0\n \n return metrics\n \n def get_predictions(self):\n \"\"\"Get raw predictions and targets\"\"\"\n predictions = np.array(self.predictions)\n targets = np.array(self.targets)\n probs = torch.sigmoid(torch.tensor(predictions)).numpy()\n return probs, targets\n\n\ndef print_metrics(metrics, prefix=''):\n \"\"\"Pretty print metrics\"\"\"\n print(f\"\\n{prefix} Metrics:\")\n print(\"=\" * 60)\n for key, value in metrics.items():\n if isinstance(value, float):\n print(f\"{key:20s}: {value:.4f}\")\n else:\n print(f\"{key:20s}: {value}\")\n print(\"=\" * 60)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:30:36.588775Z","iopub.execute_input":"2025-11-17T07:30:36.589363Z","iopub.status.idle":"2025-11-17T07:30:36.601305Z","shell.execute_reply.started":"2025-11-17T07:30:36.589342Z","shell.execute_reply":"2025-11-17T07:30:36.600501Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile dataset.py\nimport os\nimport cv2\nimport torch\nimport numpy as np\nimport pandas as pd\nfrom sklearn.model_selection import StratifiedKFold\nfrom torch.utils.data import Dataset, DataLoader\nimport albumentations as A\nfrom albumentations.pytorch import ToTensorV2\n\ndef create_folds(df, n_folds=5, seed=42):\n df[\"fold\"] = -1\n skf = StratifiedKFold(n_splits=n_folds, shuffle=True, random_state=seed)\n for fold, (_, val_idx) in enumerate(skf.split(df, df.label)):\n df.loc[val_idx, \"fold\"] = fold\n return df\n\ndef get_transforms(image_size):\n return A.Compose([\n A.Resize(image_size, image_size),\n A.HorizontalFlip(p=0.5),\n A.VerticalFlip(p=0.3),\n A.Rotate(limit=15, p=0.5),\n A.RandomBrightnessContrast(p=0.5),\n A.GaussianBlur(p=0.3),\n A.ImageCompression(quality_lower=70, quality_upper=100, p=0.5),\n ToTensorV2()\n ])\n\nclass ForgeryDataset(Dataset):\n def __init__(self, df, img_dir, transforms=None):\n self.df = df\n self.img_dir = img_dir\n self.transforms = transforms\n\n def __len__(self):\n return len(self.df)\n\n def __getitem__(self, idx):\n row = self.df.iloc[idx]\n img_path = os.path.join(self.img_dir, row.image_name)\n image = cv2.imread(img_path)\n image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n\n if self.transforms:\n image = self.transforms(image=image)[\"image\"]\n\n label = torch.tensor(row.label, dtype=torch.float32)\n return image, label\n\ndef get_dataloaders(df, img_dir, fold, batch_size, image_size, num_workers=4):\n train_df = df[df.fold != fold].reset_index(drop=True)\n val_df = df[df.fold == fold].reset_index(drop=True)\n\n train_dataset = ForgeryDataset(train_df, img_dir, transforms=get_transforms(image_size))\n val_dataset = ForgeryDataset(val_df, img_dir, transforms=get_transforms(image_size))\n\n train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=num_workers)\n val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers)\n\n return train_loader, val_loader\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:32:11.964236Z","iopub.execute_input":"2025-11-17T07:32:11.964810Z","iopub.status.idle":"2025-11-17T07:32:11.970442Z","shell.execute_reply.started":"2025-11-17T07:32:11.964791Z","shell.execute_reply":"2025-11-17T07:32:11.969667Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile model.py\nimport timm\nimport torch.nn as nn\n\ndef build_model(model_name, pretrained=True):\n model = timm.create_model(model_name, pretrained=pretrained)\n in_features = model.get_classifier().in_features\n model.fc = nn.Linear(in_features, 1)\n return model\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:32:21.647502Z","iopub.execute_input":"2025-11-17T07:32:21.648104Z","iopub.status.idle":"2025-11-17T07:32:21.652902Z","shell.execute_reply.started":"2025-11-17T07:32:21.648082Z","shell.execute_reply":"2025-11-17T07:32:21.651985Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile loss.py\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\nclass FocalLoss(nn.Module):\n def __init__(self, gamma=2.0):\n super().__init__()\n self.gamma = gamma\n\n def forward(self, inputs, targets):\n bce = F.binary_cross_entropy_with_logits(inputs, targets, reduction=\"none\")\n pt = torch.exp(-bce)\n focal = (1 - pt)**self.gamma * bce\n return focal.mean()\n\ndef get_loss_function(name=\"focal\", gamma=2.0, smoothing=0.1):\n if name == \"focal\":\n return FocalLoss(gamma)\n elif name == \"bce\":\n return nn.BCEWithLogitsLoss()\n elif name == \"label_smoothing\":\n return nn.BCEWithLogitsLoss(label_smoothing=smoothing)\n else:\n raise ValueError(\"Unknown loss function\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:32:31.599074Z","iopub.execute_input":"2025-11-17T07:32:31.599342Z","iopub.status.idle":"2025-11-17T07:32:31.604282Z","shell.execute_reply.started":"2025-11-17T07:32:31.599323Z","shell.execute_reply":"2025-11-17T07:32:31.603590Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile train_utils.py\nimport torch\nimport numpy as np\nfrom tqdm import tqdm\n\ndef train_one_epoch(model, loader, criterion, optimizer, device):\n model.train()\n total_loss = 0\n for images, labels in tqdm(loader):\n images, labels = images.to(device), labels.to(device)\n\n optimizer.zero_grad()\n preds = model(images).view(-1)\n loss = criterion(preds, labels)\n loss.backward()\n optimizer.step()\n\n total_loss += loss.item()\n return total_loss / len(loader)\n\ndef validate(model, loader, criterion, device):\n model.eval()\n total_loss = 0\n preds_all = []\n labels_all = []\n\n with torch.no_grad():\n for images, labels in loader:\n images, labels = images.to(device), labels.to(device)\n preds = model(images).view(-1)\n\n total_loss += criterion(preds, labels).item()\n preds_all.extend(preds.cpu().numpy())\n labels_all.extend(labels.cpu().numpy())\n\n return total_loss / len(loader), np.array(preds_all), np.array(labels_all)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:32:39.275711Z","iopub.execute_input":"2025-11-17T07:32:39.275996Z","iopub.status.idle":"2025-11-17T07:32:39.281056Z","shell.execute_reply.started":"2025-11-17T07:32:39.275946Z","shell.execute_reply":"2025-11-17T07:32:39.280251Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from dataset import create_folds, get_dataloaders\nfrom model import build_model\nfrom loss import get_loss_function\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:32:47.137499Z","iopub.execute_input":"2025-11-17T07:32:47.138008Z","iopub.status.idle":"2025-11-17T07:32:47.143966Z","shell.execute_reply.started":"2025-11-17T07:32:47.137982Z","shell.execute_reply":"2025-11-17T07:32:47.143262Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile engine.py\nimport torch\nimport os\nimport numpy as np\nfrom sklearn.metrics import roc_auc_score\n\nclass Trainer:\n def __init__(self, model, criterion, optimizer, scheduler=None, device='cuda', use_amp=True, grad_clip=None, accumulation_steps=1, mixup_alpha=0.0, cutmix_alpha=0.0):\n self.model = model.to(device)\n self.criterion = criterion\n self.optimizer = optimizer\n self.scheduler = scheduler\n self.device = device\n self.use_amp = use_amp\n self.grad_clip = grad_clip\n self.accumulation_steps = accumulation_steps\n self.scaler = torch.cuda.amp.GradScaler(enabled=use_amp)\n\n def train_epoch(self, loader, epoch):\n self.model.train()\n total_loss = 0\n preds_all, labels_all = [], []\n for step, (images, labels) in enumerate(loader):\n images, labels = images.to(self.device), labels.to(self.device)\n with torch.cuda.amp.autocast(enabled=self.use_amp):\n outputs = self.model(images).view(-1)\n loss = self.criterion(outputs, labels)\n self.scaler.scale(loss).backward()\n if self.grad_clip:\n torch.nn.utils.clip_grad_norm_(self.model.parameters(), self.grad_clip)\n self.scaler.step(self.optimizer)\n self.scaler.update()\n self.optimizer.zero_grad()\n total_loss += loss.item()\n preds_all.extend(outputs.detach().cpu().numpy())\n labels_all.extend(labels.detach().cpu().numpy())\n\n auc = roc_auc_score(labels_all, preds_all)\n return {'loss': total_loss / len(loader), 'auc': auc}\n\n def validate_epoch(self, loader, epoch):\n self.model.eval()\n total_loss = 0\n preds_all, labels_all = [], []\n with torch.no_grad():\n for images, labels in loader:\n images, labels = images.to(self.device), labels.to(self.device)\n outputs = self.model(images).view(-1)\n loss = self.criterion(outputs, labels)\n total_loss += loss.item()\n preds_all.extend(outputs.cpu().numpy())\n labels_all.extend(labels.cpu().numpy())\n auc = roc_auc_score(labels_all, preds_all)\n return {'loss': total_loss / len(loader), 'auc': auc}\n\nclass EarlyStopping:\n def __init__(self, patience=5, mode='max', verbose=False):\n self.patience = patience\n self.mode = mode\n self.verbose = verbose\n self.best_score = None\n self.counter = 0\n self.early_stop = False\n\n def __call__(self, score):\n if self.best_score is None:\n self.best_score = score\n return False\n if (self.mode == 'max' and score > self.best_score) or (self.mode == 'min' and score < self.best_score):\n self.best_score = score\n self.counter = 0\n return False\n else:\n self.counter += 1\n if self.verbose:\n print(f\"EarlyStopping counter: {self.counter} / {self.patience}\")\n if self.counter >= self.patience:\n self.early_stop = True\n return self.early_stop\n\ndef save_checkpoint(model, optimizer, scheduler, epoch, metrics, filename):\n state = {\n 'model_state_dict': model.state_dict(),\n 'optimizer_state_dict': optimizer.state_dict(),\n 'scheduler_state_dict': scheduler.state_dict() if scheduler else None,\n 'epoch': epoch,\n 'metrics': metrics\n }\n torch.save(state, filename)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:33:44.286879Z","iopub.execute_input":"2025-11-17T07:33:44.287702Z","iopub.status.idle":"2025-11-17T07:33:44.293697Z","shell.execute_reply.started":"2025-11-17T07:33:44.287671Z","shell.execute_reply":"2025-11-17T07:33:44.292942Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%writefile metrics.py\ndef print_metrics(metrics, prefix=''):\n print(f\"{prefix} Loss: {metrics['loss']:.4f} | {prefix} AUC: {metrics['auc']:.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:33:53.040488Z","iopub.execute_input":"2025-11-17T07:33:53.040752Z","iopub.status.idle":"2025-11-17T07:33:53.045566Z","shell.execute_reply.started":"2025-11-17T07:33:53.040733Z","shell.execute_reply":"2025-11-17T07:33:53.045001Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from engine import Trainer, EarlyStopping, save_checkpoint\nfrom metrics import print_metrics\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:34:01.811265Z","iopub.execute_input":"2025-11-17T07:34:01.811490Z","iopub.status.idle":"2025-11-17T07:34:01.817521Z","shell.execute_reply.started":"2025-11-17T07:34:01.811475Z","shell.execute_reply":"2025-11-17T07:34:01.816969Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import argparse\n\nparser = argparse.ArgumentParser()\nparser.add_argument('--config', type=str, default='config.yaml')\nparser.add_argument('--folds', type=str, default='all', help='all or 0,1,2')\nparser.add_argument('--output_dir', type=str, default='./checkpoints')\n\n# In notebooks, ignore extra args from Jupyter\nargs = parser.parse_args([])\n\nprint(args.config, args.folds, args.output_dir)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:35:01.656205Z","iopub.execute_input":"2025-11-17T07:35:01.656710Z","iopub.status.idle":"2025-11-17T07:35:01.662464Z","shell.execute_reply.started":"2025-11-17T07:35:01.656690Z","shell.execute_reply":"2025-11-17T07:35:01.661716Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Args:\n config = 'config.yaml'\n folds = 'all'\n output_dir = './checkpoints'\n\nargs = Args()\nprint(args.config, args.folds, args.output_dir)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:35:10.427366Z","iopub.execute_input":"2025-11-17T07:35:10.427937Z","iopub.status.idle":"2025-11-17T07:35:10.431836Z","shell.execute_reply.started":"2025-11-17T07:35:10.427916Z","shell.execute_reply":"2025-11-17T07:35:10.431174Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import argparse\n\nparser = argparse.ArgumentParser()\nparser.add_argument('--config', type=str, default='config.yaml')\nparser.add_argument('--folds', type=str, default='all')\nparser.add_argument('--output_dir', type=str, default='./checkpoints')\n\n# Ignore unknown args from notebook\nargs = parser.parse_args(args=[])\nprint(args.config, args.folds, args.output_dir)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:36:15.094068Z","iopub.execute_input":"2025-11-17T07:36:15.094306Z","iopub.status.idle":"2025-11-17T07:36:15.099910Z","shell.execute_reply.started":"2025-11-17T07:36:15.094289Z","shell.execute_reply":"2025-11-17T07:36:15.099339Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Args:\n config = 'config.yaml'\n folds = 'all'\n output_dir = './checkpoints'\n\nargs = Args()\nprint(args.config, args.folds, args.output_dir)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:36:26.205502Z","iopub.execute_input":"2025-11-17T07:36:26.206014Z","iopub.status.idle":"2025-11-17T07:36:26.210772Z","shell.execute_reply.started":"2025-11-17T07:36:26.205993Z","shell.execute_reply":"2025-11-17T07:36:26.209988Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import yaml\n\n# Load configuration\nwith open('config.yaml', 'r') as f:\n config = yaml.safe_load(f)\n\nprint(config.keys()) # check that config loaded correctly\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:36:48.928246Z","iopub.execute_input":"2025-11-17T07:36:48.928657Z","iopub.status.idle":"2025-11-17T07:36:48.936799Z","shell.execute_reply.started":"2025-11-17T07:36:48.928637Z","shell.execute_reply":"2025-11-17T07:36:48.936154Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\n# Define the directory path that is failing\ndata_dir = '/kaggle/input/recodai-luc-scientific-image-forgery-detection/'\n\n# Check if the directory exists\nif os.path.isdir(data_dir):\n print(f\"Directory found. Listing contents of {data_dir}:\")\n # List all files and folders inside that directory\n for filename in os.listdir(data_dir):\n print(f\"- {filename}\")\nelse:\n print(f\"Directory not found: {data_dir}\")\n print(\"Please ensure the dataset is added to your notebook correctly.\")\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:45:55.561152Z","iopub.execute_input":"2025-11-17T07:45:55.561897Z","iopub.status.idle":"2025-11-17T07:45:55.569521Z","shell.execute_reply.started":"2025-11-17T07:45:55.561874Z","shell.execute_reply":"2025-11-17T07:45:55.568857Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def main():\n # ... (previous code) ...\n\n # FIX 1: Use parse_known_args()\n args, unknown = parser.parse_known_args()\n\n # Load config\n with open(args.config, 'r') as f:\n config = yaml.safe_load(f)\n\n # ... (rest of main function) ...\n\n # Use the absolute path directly that you verified in Step 1\n train_csv_path = '/kaggle/input/recodai-luc-scientific-image-forgery-detection/train.csv'\n \n # Load and create folds\n train_df = pd.read_csv(train_csv_path) # Use the verified path\n train_df = create_folds(train_df, config)\n\n # ... (rest of main function) ...\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:46:10.597667Z","iopub.execute_input":"2025-11-17T07:46:10.598251Z","iopub.status.idle":"2025-11-17T07:46:10.602401Z","shell.execute_reply.started":"2025-11-17T07:46:10.598229Z","shell.execute_reply":"2025-11-17T07:46:10.601766Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\npotential_path = '/kaggle/input/recodai-luc-scientific-image-forgery-detection/'\n\nif os.path.exists(potential_path):\n print(f\"Directory found. Files inside {potential_path}:\")\n for fname in os.listdir(potential_path):\n print(f\"- {fname}\")\nelse:\n print(f\"Directory not found: {potential_path}\")\n print(\"Check if you added the dataset to your Kaggle notebook correctly.\")\n # You might also want to list the current working directory to find a relative path:\n print(f\"Current working directory: {os.getcwd()}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:48:45.994506Z","iopub.execute_input":"2025-11-17T07:48:45.994766Z","iopub.status.idle":"2025-11-17T07:48:46.001276Z","shell.execute_reply.started":"2025-11-17T07:48:45.994748Z","shell.execute_reply":"2025-11-17T07:48:46.000629Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import yaml\n\nwith open(\"config.yaml\", \"r\") as f:\n config = yaml.safe_load(f)\n\nprint(config.keys()) # dict_keys(['data', 'training', 'models', 'augmentation', 'inference'])\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:49:32.624807Z","iopub.execute_input":"2025-11-17T07:49:32.625525Z","iopub.status.idle":"2025-11-17T07:49:32.634013Z","shell.execute_reply.started":"2025-11-17T07:49:32.625499Z","shell.execute_reply":"2025-11-17T07:49:32.633291Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"train_dir = config['data']['train_dir']\nprint(train_dir)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:49:45.957200Z","iopub.execute_input":"2025-11-17T07:49:45.957853Z","iopub.status.idle":"2025-11-17T07:49:45.961681Z","shell.execute_reply.started":"2025-11-17T07:49:45.957831Z","shell.execute_reply":"2025-11-17T07:49:45.960932Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"config = {\n 'data': {\n 'train_dir': \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train\",\n 'test_dir': \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test\",\n 'train_csv': \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/train.csv\",\n 'test_csv': \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test.csv\",\n 'dinov2_path': \"/kaggle/input/dinov2/pytorch/base/1\"\n },\n 'training': {\n 'n_folds': 5,\n 'seed': 42,\n 'epochs': 30,\n 'batch_size': 16,\n 'num_workers': 4,\n 'image_size': 384,\n 'optimizer': \"AdamW\",\n 'learning_rate': 1e-4,\n 'weight_decay': 1e-2,\n 'scheduler': \"CosineAnnealingWarmRestarts\",\n 'warmup_epochs': 3,\n 'min_lr': 1e-6,\n 'loss_fn': \"focal\",\n 'focal_gamma': 2.0,\n 'label_smoothing': 0.1,\n 'patience': 7\n },\n 'models': [\n {'name': \"swin_base_patch4_window12_384\", 'image_size': 384, 'pretrained': True},\n {'name': \"convnext_base\", 'image_size': 512, 'pretrained': True},\n {'name': \"vit_base_patch16_384\", 'image_size': 384, 'pretrained': True},\n {'name': \"tf_efficientnetv2_l\", 'image_size': 480, 'pretrained': True}\n ],\n 'augmentation': {\n 'train': {\n 'horizontal_flip': 0.5,\n 'vertical_flip': 0.3,\n 'rotate_limit': 15,\n 'brightness_limit': 0.2,\n 'contrast_limit': 0.2,\n 'blur_limit': 3,\n 'jpeg_quality': [70, 100],\n 'noise_var': [10, 50],\n 'mixup_alpha': 0.2,\n 'cutmix_prob': 0.5\n },\n 'tta': {\n 'enabled': True,\n 'n_augments': 5\n }\n },\n 'inference': {\n 'tta': True,\n 'threshold': 0.5,\n 'calibration': True\n }\n}\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:49:57.325797Z","iopub.execute_input":"2025-11-17T07:49:57.326609Z","iopub.status.idle":"2025-11-17T07:49:57.332542Z","shell.execute_reply.started":"2025-11-17T07:49:57.326578Z","shell.execute_reply":"2025-11-17T07:49:57.332005Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\n\n!ls /kaggle/input/\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:50:50.694766Z","iopub.execute_input":"2025-11-17T07:50:50.695118Z","iopub.status.idle":"2025-11-17T07:50:50.884360Z","shell.execute_reply.started":"2025-11-17T07:50:50.695096Z","shell.execute_reply":"2025-11-17T07:50:50.883608Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport torch\nfrom model import build_model\nfrom dataset import get_dataloaders\nfrom loss import get_loss_function\nfrom engine import Trainer, EarlyStopping, save_checkpoint\nfrom metrics import print_metrics\n\n# -------------------------------\n# Optimizer\n# -------------------------------\ndef get_optimizer(model, config):\n lr = config['training']['learning_rate']\n weight_decay = config['training']['weight_decay']\n optimizer_name = config['training']['optimizer']\n \n param_groups = model.get_optimizer_params(lr, weight_decay)\n \n if optimizer_name == 'AdamW':\n optimizer = torch.optim.AdamW(param_groups)\n elif optimizer_name == 'Adam':\n optimizer = torch.optim.Adam(param_groups)\n elif optimizer_name == 'SGD':\n optimizer = torch.optim.SGD(param_groups, momentum=0.9)\n else:\n raise ValueError(f\"Unknown optimizer: {optimizer_name}\")\n \n return optimizer\n\n# -------------------------------\n# Scheduler\n# -------------------------------\ndef get_scheduler(optimizer, config, steps_per_epoch):\n scheduler_name = config['training']['scheduler']\n epochs = config['training']['epochs']\n warmup_epochs = config['training'].get('warmup_epochs', 0)\n min_lr = config['training'].get('min_lr', 1e-6)\n\n if scheduler_name == 'CosineAnnealingWarmRestarts':\n scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(\n optimizer, T_0=epochs, T_mult=1, eta_min=min_lr\n )\n elif scheduler_name == 'CosineAnnealingLR':\n scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(\n optimizer, T_max=epochs, eta_min=min_lr\n )\n elif scheduler_name == 'ReduceLROnPlateau':\n scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(\n optimizer, mode='min', factor=0.5, patience=3, min_lr=min_lr, verbose=True\n )\n elif scheduler_name == 'OneCycleLR':\n scheduler = torch.optim.lr_scheduler.OneCycleLR(\n optimizer,\n max_lr=config['training']['learning_rate'] * 10,\n epochs=epochs,\n steps_per_epoch=steps_per_epoch,\n pct_start=warmup_epochs / epochs\n )\n else:\n raise ValueError(f\"Unknown scheduler: {scheduler_name}\")\n\n return scheduler\n\n# -------------------------------\n# Fold training function\n# -------------------------------\ndef run_fold_training(fold, config, output_dir):\n print(f\"🎯 Starting training for fold {fold}\")\n \n # Build model\n model = build_model(config)\n device = 'cuda' if torch.cuda.is_available() else 'cpu'\n model.to(device)\n \n # Get dataloaders\n train_loader, valid_loader, steps_per_epoch = get_dataloaders(config, fold)\n \n # Setup training components\n optimizer = get_optimizer(model, config)\n scheduler = get_scheduler(optimizer, config, steps_per_epoch)\n criterion = get_loss_function(config)\n early_stopping = EarlyStopping(patience=config['training']['early_stopping_patience'])\n trainer = Trainer(model, criterion, optimizer, scheduler, config)\n \n best_auc = 0.0\n best_epoch = 0\n \n # Training loop\n for epoch in range(1, config['training']['epochs'] + 1):\n print(f\"\\n📅 Epoch {epoch}/{config['training']['epochs']}\")\n \n # Train\n trainer.train_one_epoch(train_loader)\n \n # Validate\n valid_metrics = trainer.validate_one_epoch(valid_loader)\n print_metrics(valid_metrics, epoch)\n \n # Save best model\n if valid_metrics['auc'] > best_auc:\n best_auc = valid_metrics['auc']\n best_epoch = epoch\n checkpoint_path = os.path.join(output_dir, f'fold{fold}_best.pth')\n save_checkpoint(model, optimizer, scheduler, epoch, valid_metrics, checkpoint_path)\n \n # Early stopping\n if early_stopping(valid_metrics['auc']):\n print(\"⏹ Early stopping triggered\")\n break\n\n print(f\"\\n✅ Fold {fold} best AUC: {best_auc:.4f} at epoch {best_epoch}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:52:26.192482Z","iopub.execute_input":"2025-11-17T07:52:26.193043Z","iopub.status.idle":"2025-11-17T07:52:26.206469Z","shell.execute_reply.started":"2025-11-17T07:52:26.193016Z","shell.execute_reply":"2025-11-17T07:52:26.205584Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import yaml\n\nwith open(\"config.yaml\", \"r\") as f:\n config = yaml.safe_load(f)\n\nprint(config.keys())\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:53:31.998493Z","iopub.execute_input":"2025-11-17T07:53:31.999221Z","iopub.status.idle":"2025-11-17T07:53:32.007732Z","shell.execute_reply.started":"2025-11-17T07:53:31.999200Z","shell.execute_reply":"2025-11-17T07:53:32.006997Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def get_valid_transforms(config):\n \"\"\"Minimal transforms for validation / testing\"\"\"\n img_size = config['training']['image_size']\n \n return A.Compose([\n A.Resize(img_size, img_size),\n A.Normalize(\n mean=[0.485, 0.456, 0.406],\n std=[0.229, 0.224, 0.225]\n ),\n ToTensorV2()\n ])\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:54:50.277449Z","iopub.execute_input":"2025-11-17T07:54:50.277896Z","iopub.status.idle":"2025-11-17T07:54:50.281997Z","shell.execute_reply.started":"2025-11-17T07:54:50.277872Z","shell.execute_reply":"2025-11-17T07:54:50.281437Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import albumentations as A\nfrom albumentations.pytorch import ToTensorV2\n\ndef get_valid_transforms(config):\n img_size = config['training']['image_size']\n return A.Compose([\n A.Resize(img_size, img_size),\n A.Normalize(\n mean=[0.485, 0.456, 0.406],\n std=[0.229, 0.224, 0.225]\n ),\n ToTensorV2()\n ])\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:55:27.222155Z","iopub.execute_input":"2025-11-17T07:55:27.222435Z","iopub.status.idle":"2025-11-17T07:55:27.226848Z","shell.execute_reply.started":"2025-11-17T07:55:27.222418Z","shell.execute_reply":"2025-11-17T07:55:27.226295Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os, cv2, json, math, random, torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom sklearn.model_selection import train_test_split\nfrom torch.utils.data import Dataset, DataLoader\nimport torch.nn as nn, torch.nn.functional as F, torch.optim as optim\nfrom transformers import AutoImageProcessor, AutoModel\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nBASE_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\"\nAUTH_DIR = f\"{BASE_DIR}/train_images/authentic\"\nFORG_DIR = f\"{BASE_DIR}/train_images/forged\"\nMASK_DIR = f\"{BASE_DIR}/train_masks\"\nTEST_DIR = f\"{BASE_DIR}/test_images\"\nDINO_PATH = \"/kaggle/input/dinov2/pytorch/base/1\"\n\nIMG_SIZE = 512\nBATCH_SIZE = 1\nEPOCHS_SEG = 4\nLR_SEG = 1e-5\nWEIGHT_DECAY = 2e-5\n\n\nclass ForgerySegDataset(Dataset):\n def __init__(self, auth_paths, forg_paths, mask_dir, img_size=256):\n self.samples = []\n for p in forg_paths:\n m = os.path.join(mask_dir, Path(p).stem + \".npy\")\n if os.path.exists(m):\n self.samples.append((p, m))\n for p in auth_paths:\n self.samples.append((p, None))\n self.img_size = img_size\n def __len__(self): return len(self.samples)\n def __getitem__(self, idx):\n img_path, mask_path = self.samples[idx]\n img = Image.open(img_path).convert(\"RGB\")\n w, h = img.size\n if mask_path is None:\n mask = np.zeros((h, w), np.uint8)\n else:\n m = np.load(mask_path)\n if m.ndim == 3: m = np.max(m, axis=0)\n mask = (m > 0).astype(np.uint8)\n img_r = img.resize((IMG_SIZE, IMG_SIZE))\n mask_r = cv2.resize(mask, (IMG_SIZE, IMG_SIZE), interpolation=cv2.INTER_NEAREST)\n img_t = torch.from_numpy(np.array(img_r, np.float32)/255.).permute(2,0,1)\n mask_t = torch.from_numpy(mask_r[None, ...].astype(np.float32))\n return img_t, mask_t\n\n\n# MODEL (DINOv2 + Decoder)\n\nfrom transformers import AutoImageProcessor, AutoModel\nprocessor = AutoImageProcessor.from_pretrained(DINO_PATH, local_files_only=True)\nencoder = AutoModel.from_pretrained(DINO_PATH, local_files_only=True).eval().to(device)\n\nclass DinoTinyDecoder(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,256,3,padding=1), nn.ReLU(),\n nn.Conv2d(256,64,3,padding=1), nn.ReLU(),\n nn.Conv2d(64,out_ch,1)\n )\n def forward(self, f, size):\n return self.net(F.interpolate(f, size=size, mode=\"bilinear\", align_corners=False))\n\nclass DinoSegmenter(nn.Module):\n def __init__(self, encoder, processor):\n super().__init__()\n self.encoder, self.processor = encoder, processor\n for p in self.encoder.parameters(): p.requires_grad = False\n self.seg_head = DinoTinyDecoder(768,1)\n def forward_features(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(): 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 fmap\n def forward_seg(self,x):\n fmap = self.forward_features(x)\n return self.seg_head(fmap,(IMG_SIZE,IMG_SIZE))\n\n# TRAINING\n\nauth_imgs = sorted([str(Path(AUTH_DIR)/f) for f in os.listdir(AUTH_DIR)])\nforg_imgs = sorted([str(Path(FORG_DIR)/f) for f in os.listdir(FORG_DIR)])\ntrain_auth, val_auth = train_test_split(auth_imgs, test_size=0.2, random_state=42)\ntrain_forg, val_forg = train_test_split(forg_imgs, test_size=0.2, random_state=42)\n\ntrain_loader = DataLoader(ForgerySegDataset(train_auth, train_forg, MASK_DIR),\n batch_size=BATCH_SIZE, shuffle=True, num_workers=2)\n\nmodel_seg = DinoSegmenter(encoder, processor).to(device)\nopt_seg = optim.AdamW(model_seg.seg_head.parameters(), lr=LR_SEG, weight_decay=WEIGHT_DECAY)\ncrit_seg = nn.BCEWithLogitsLoss()\n\nfor e in range(EPOCHS_SEG):\n model_seg.train()\n total_loss = 0\n for x,m in tqdm(train_loader, desc=f\"[Segmentation] Epoch {e+1}/{EPOCHS_SEG}\"):\n x,m = x.to(device),m.to(device)\n loss = crit_seg(model_seg.forward_seg(x),m)\n opt_seg.zero_grad(); loss.backward(); opt_seg.step()\n total_loss += loss.item()\n print(f\" → avg_loss={total_loss/len(train_loader):.4f}\")\ntorch.save(model_seg.state_dict(),\"model_seg_final.pt\")\n\n\n# INFERENCE UTILS\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_seg.forward_seg(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 # condition de filtrage\n if area < 400 or mean_inside < 0.35:\n return \"authentic\", None, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n return \"forged\", mask, {\"area\": area, \"mean_inside\": mean_inside, \"thr\": thr}\n\n\nfrom sklearn.metrics import f1_score\nval_items = [(p, 1) for p in val_forg[:10]]\nresults = []\nfor p,_ in tqdm(val_items, desc=\"Validation forged-only\"):\n pil = Image.open(p).convert(\"RGB\")\n label, m_pred, dbg = pipeline_final(pil)\n m_gt = np.load(Path(MASK_DIR)/f\"{Path(p).stem}.npy\")\n if m_gt.ndim==3: m_gt=np.max(m_gt,axis=0)\n m_gt=(m_gt>0).astype(np.uint8)\n m_pred=(m_pred>0).astype(np.uint8) if m_pred is not None else np.zeros_like(m_gt)\n f1 = f1_score(m_gt.flatten(), m_pred.flatten(), zero_division=0)\n results.append((Path(p).stem, f1, dbg))\nprint(\"\\n F1-score par image falsifiée:\\n\")\nfor cid,f1,dbg in results:\n print(f\"{cid} — F1={f1:.4f} | area={dbg['area']} mean={dbg['mean_inside']:.3f} thr={dbg['thr']:.3f}\")\nprint(f\"\\n Moyenne F1 (falsifiées) = {np.mean([r[1] for r in results]):.4f}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-17T07:59:08.260622Z","iopub.execute_input":"2025-11-17T07:59:08.260878Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os, json, cv2\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\n\n# --- RLE Encoder for Kaggle Submission ---\ndef rle_encode(mask: np.ndarray, fg_val: int = 1) -> str:\n pixels = mask.T.flatten()\n dots = np.where(pixels == fg_val)[0]\n if len(dots) == 0:\n return \"authentic\"\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 json.dumps([int(x) for x in run_lengths])\n\n# --- Paths ---\nTEST_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/test_images\"\nSAMPLE_SUB = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection/sample_submission.csv\"\nOUT_PATH = \"submission.csv\"\n\nrows = []\nfor f in tqdm(sorted(os.listdir(TEST_DIR)), desc=\"Inference on Test Set\"):\n pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n label, mask, dbg = pipeline_final(pil) # utilise la version améliorée\n\n # Sécurisation masque\n if mask is None:\n mask = np.zeros(pil.size[::-1], np.uint8)\n else:\n mask = np.array(mask, dtype=np.uint8)\n\n # Annotation finale\n if label == \"authentic\":\n annot = \"authentic\"\n else:\n annot = rle_encode((mask > 0).astype(np.uint8))\n\n rows.append({\n \"case_id\": Path(f).stem,\n \"annotation\": annot,\n \"area\": int(dbg.get(\"area\", mask.sum())),\n \"mean\": float(dbg.get(\"mean_inside\", 0.0)),\n \"thr\": float(dbg.get(\"thr\", 0.0))\n })\n\n\nsub = pd.DataFrame(rows)\nss = pd.read_csv(SAMPLE_SUB)\nss[\"case_id\"] = ss[\"case_id\"].astype(str)\nsub[\"case_id\"] = sub[\"case_id\"].astype(str)\nfinal = ss[[\"case_id\"]].merge(sub, on=\"case_id\", how=\"left\")\nfinal[\"annotation\"] = final[\"annotation\"].fillna(\"authentic\")\nfinal[[\"case_id\", \"annotation\"]].to_csv(OUT_PATH, index=False)\n\nprint(f\"\\n✅ Saved submission file: {OUT_PATH}\")\nprint(final.head(10))\n\n\nsample_files = sorted(os.listdir(TEST_DIR))[:5]\nfor f in sample_files:\n pil = Image.open(Path(TEST_DIR)/f).convert(\"RGB\")\n label, mask, dbg = pipeline_final(pil)\n mask = np.array(mask, dtype=np.uint8) if mask is not None else np.zeros(pil.size[::-1], np.uint8)\n\n print(f\"{'🔴' if label=='forged' else '🟢'} {f}: {label} | area={mask.sum()} mean={dbg.get('mean_inside', 0):.3f}\")\n\n if label == \"authentic\":\n plt.figure(figsize=(5,5))\n plt.imshow(pil)\n plt.title(f\"{f} — Authentic\")\n plt.axis(\"off\")\n plt.show()\n else:\n plt.figure(figsize=(10,5))\n plt.subplot(1,2,1)\n plt.imshow(pil)\n plt.title(\"Original Image\")\n plt.axis(\"off\")\n plt.subplot(1,2,2)\n plt.imshow(pil)\n plt.imshow(mask, alpha=0.45, cmap=\"Reds\")\n plt.title(f\"Predicted Forged Mask\\nArea={mask.sum()} | Mean={dbg.get('mean_inside', 0):.3f}\")\n plt.axis(\"off\")\n plt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch, cv2, math, numpy as np, matplotlib.pyplot as plt\nfrom pathlib import Path\nfrom PIL import Image\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nIMG_SIZE = 256\n\n# 1️ Predict probability map (from model)\n@torch.no_grad()\ndef predict_prob_map(pil):\n \"\"\"Return DINOv2 segmentation probability map [0,1].\"\"\"\n img = pil.resize((IMG_SIZE, IMG_SIZE))\n x = torch.from_numpy(np.array(img, np.float32) / 255.).permute(2, 0, 1)[None].to(device)\n logits = model_seg.forward_seg(x)\n prob = torch.sigmoid(logits)[0, 0].cpu().numpy()\n return prob\n\n\n# 2️ Post-processing consistent with pipeline_final\ndef adaptive_mask(prob, alpha_grad=0.35):\n \"\"\"Adaptive enhancement + morphological refinement.\"\"\"\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\n enhanced = (1 - alpha_grad) * prob + 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 return mask, float(thr)\n\n\n# 3️ Unified visualization pipeline (uses same filtering logic)\ndef pipeline_visual(pil):\n prob = predict_prob_map(pil)\n mask, thr = adaptive_mask(prob)\n area = int(mask.sum())\n mean_inside = float(prob[mask == 1].mean()) if area > 0 else 0.0\n\n # same decision rule as pipeline_final\n AREA_THR, MEAN_THR = 800, 0.45\n label = \"forged\" if (area >= AREA_THR and mean_inside >= MEAN_THR) else \"authentic\"\n return label, mask, thr, area, mean_inside\n\n\n# 4️ Visualization (for validation forged samples)\nsample_forged = val_forg[:5]\nn = len(sample_forged)\nfig, axes = plt.subplots(n, 3, figsize=(12, n * 3))\nif n == 1:\n axes = np.expand_dims(axes, axis=0)\n\nfor i, p in enumerate(sample_forged):\n pil = Image.open(p).convert(\"RGB\")\n label, m_pred, thr, area, mean = pipeline_visual(pil)\n\n # Ground Truth mask\n m_gt = np.load(Path(MASK_DIR)/f\"{Path(p).stem}.npy\")\n if m_gt.ndim == 3: m_gt = np.max(m_gt, axis=0)\n m_gt = (m_gt > 0).astype(np.uint8)\n\n # Resize all for consistency\n img_disp = cv2.resize(np.array(pil), (256, 256))\n gt_disp = cv2.resize(m_gt, (256, 256))\n pr_disp = cv2.resize(m_pred, (256, 256))\n\n # === Column 1: Original ===\n axes[i, 0].imshow(img_disp)\n axes[i, 0].set_title(\"🖼️ Original Image\", fontsize=11, weight=\"bold\")\n axes[i, 0].axis(\"off\")\n\n # === Column 2: Ground Truth ===\n axes[i, 1].imshow(gt_disp, cmap=\"gray\")\n axes[i, 1].set_title(\"✅ Ground Truth\", fontsize=11, weight=\"bold\")\n axes[i, 1].axis(\"off\")\n\n # === Column 3: Predicted Mask ===\n axes[i, 2].imshow(img_disp)\n axes[i, 2].imshow(pr_disp, cmap=\"coolwarm\", alpha=0.45)\n axes[i, 2].set_title(f\"🔮 Predicted ({label})\\nThr={thr:.3f} | Area={area} | Mean={mean:.3f}\",\n fontsize=10)\n axes[i, 2].axis(\"off\")\n\nplt.subplots_adjust(top=0.92, hspace=0.35)\nfig.suptitle(\"🔍 Segmentation of Forged Samples — CNN–DINOv2 Hybrid\", \n fontsize=16, fontweight=\"bold\", color=\"#b30000\")\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport cv2, numpy as np\nfrom pathlib import Path\nfrom PIL import Image\n\n# Select a few authentic examples\nsample_auth = val_auth[:5]\nn = len(sample_auth)\n\nfig, axes = plt.subplots(n, 2, figsize=(9, n * 3))\nif n == 1:\n axes = np.expand_dims(axes, axis=0)\n\nfor i, p in enumerate(sample_auth):\n pil = Image.open(p).convert(\"RGB\")\n label, m_pred, thr, area, mean = pipeline_visual(pil) # <-- version alignée avec ta nouvelle pipeline\n\n # Predicted mask (should be empty for authentic images)\n m_pred = (m_pred > 0).astype(np.uint8) if m_pred is not None else np.zeros((IMG_SIZE, IMG_SIZE))\n\n # Resize for consistent display\n img_disp = cv2.resize(np.array(pil), (256, 256))\n pr_disp = cv2.resize(m_pred, (256, 256))\n\n # === Column 1: Original Image ===\n axes[i, 0].imshow(img_disp)\n axes[i, 0].set_title(\"🖼️ Original Image\", fontsize=11, weight=\"bold\")\n axes[i, 0].axis(\"off\")\n\n # === Column 2: Predicted Mask ===\n axes[i, 1].imshow(img_disp)\n axes[i, 1].imshow(pr_disp, cmap=\"coolwarm\", alpha=0.45)\n axes[i, 1].set_title(\n f\"🟢 Predicted: {label.upper()}\\nArea={area} | Mean={mean:.3f} | Thr={thr:.3f}\",\n fontsize=10\n )\n axes[i, 1].axis(\"off\")\n\n for j in range(2):\n axes[i, j].set_aspect(\"equal\")\n\nplt.subplots_adjust(top=0.90, hspace=0.35)\nfig.suptitle(\"🟢 Segmentation of Authentic Images — CNN–DINOv2 Hybrid\",\n fontsize=16, fontweight=\"bold\", color=\"#009933\")\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}