{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":113558,"databundleVersionId":14174843,"sourceType":"competition"}],"dockerImageVersionId":31154,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"#

Imports Libraries

\n","metadata":{}},{"cell_type":"code","source":"import os\nimport numpy as np\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision import transforms\nfrom PIL import Image\nfrom sklearn.model_selection import train_test_split\nfrom tqdm import tqdm\n\n# =======================================\n# Configuration Variables\n# =======================================\nIMG_SIZE = 256\nBATCH_SIZE = 16\nDATA_DIR = \"/kaggle/input/recodai-luc-scientific-image-forgery-detection\" # Kaggle data path\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n# =======================================\n# Helper Functions (RLE)\n# =======================================\n\ndef rle_encode(img):\n '''\n img: numpy array of shape (height, width), 1s for mask, 0s for background\n Returns run length encoding string or \"authentic\"\n '''\n pixels = img.flatten()\n pixels = np.concatenate([[0], pixels, [0]])\n runs = np.where(pixels[1:] != pixels[:-1])[0] + 1\n runs[1::2] -= runs[::2]\n \n if np.sum(img) == 0:\n return \"authentic\"\n else:\n return ' '.join(str(x) for x in runs)\n\ndef rle_decode(mask_rle, shape):\n '''\n mask_rle: run-length as string formated (start length)\n shape: (height, width) of array to return \n Returns numpy array, 1s for mask, 0s for background\n '''\n if mask_rle == \"authentic\":\n return np.zeros(shape, dtype=np.uint8)\n s = mask_rle.split()\n starts, lengths = [np.asarray(x, dtype=int) for x in (s[0::2], s[1::2])]\n starts -= 1\n ends = starts + lengths\n img = np.zeros(shape[0]*shape[1], dtype=np.uint8)\n for lo, hi in zip(starts, ends):\n img[lo:hi] = 1\n return img.reshape(shape)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:51:48.874410Z","iopub.execute_input":"2025-11-08T05:51:48.874975Z","iopub.status.idle":"2025-11-08T05:51:48.946072Z","shell.execute_reply.started":"2025-11-08T05:51:48.874954Z","shell.execute_reply":"2025-11-08T05:51:48.945273Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Assuming you have a train_df that links image names to forgery status/masks if needed.\n# For simplicity, let's list all image files and create a dummy dataframe if needed.\n\ntrain_images = os.listdir(os.path.join(DATA_DIR, \"train_images\"))\ntrain_df = pd.DataFrame({\"image_name\": train_images})\n# You might need more sophisticated data handling if some are authentic and not in train_masks\n\n# Data Augmentations and Normalization\ntrain_transform = transforms.Compose([\n transforms.Resize((IMG_SIZE, IMG_SIZE)),\n transforms.RandomHorizontalFlip(),\n transforms.RandomVerticalFlip(),\n transforms.RandomRotation(15),\n transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2),\n transforms.ToTensor(),\n transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n])\n\nval_transform = transforms.Compose([\n transforms.Resize((IMG_SIZE, IMG_SIZE)),\n transforms.ToTensor(),\n transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n])\n\nmask_transform = transforms.Compose([\n transforms.Resize((IMG_SIZE, IMG_SIZE)),\n transforms.ToTensor()\n])\n\n# Dataset class\nclass ForgeryDataset(Dataset):\n def __init__(self, df, img_dir, mask_dir, transform=None, mask_transform=None):\n self.df = df\n self.img_dir = img_dir\n self.mask_dir = mask_dir\n self.transform = transform\n self.mask_transform = mask_transform\n\n def __len__(self):\n return len(self.df)\n\n def __getitem__(self, idx):\n img_name = self.df.iloc[idx][\"image_name\"]\n img_path = os.path.join(self.img_dir, img_name)\n mask_path = os.path.join(self.mask_dir, img_name.replace(\".jpg\", \".png\"))\n\n image = Image.open(img_path).convert(\"RGB\")\n \n # Check if mask exists, otherwise assume authentic (zero mask)\n if os.path.exists(mask_path):\n mask = Image.open(mask_path).convert(\"L\")\n else:\n mask = Image.fromarray(np.zeros((image.height, image.width), dtype=np.uint8))\n\n if self.transform:\n image = self.transform(image)\n if self.mask_transform:\n mask = self.mask_transform(mask)\n\n return image, mask\n\n# Train-validation split\ntrain_split, val_split = train_test_split(train_df, test_size=0.2, random_state=42)\n\ntrain_dataset = ForgeryDataset(train_split, os.path.join(DATA_DIR, \"train_images\"),\n os.path.join(DATA_DIR, \"train_masks\"), train_transform, mask_transform)\nval_dataset = ForgeryDataset(val_split, os.path.join(DATA_DIR, \"train_images\"),\n os.path.join(DATA_DIR, \"train_masks\"), val_transform, mask_transform)\n\ntrain_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, num_workers=2)\nval_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False, num_workers=2)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:53:48.314480Z","iopub.execute_input":"2025-11-08T05:53:48.315480Z","iopub.status.idle":"2025-11-08T05:53:48.329853Z","shell.execute_reply.started":"2025-11-08T05:53:48.315454Z","shell.execute_reply":"2025-11-08T05:53:48.328981Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.optim as optim\n\n# Example: Small U-Net\nclass SmallUNet(nn.Module):\n def __init__(self):\n super().__init__()\n def conv_block(in_ch, out_ch):\n return nn.Sequential(\n nn.Conv2d(in_ch, out_ch, 3, padding=1),\n nn.ReLU(inplace=True),\n nn.Conv2d(out_ch, out_ch, 3, padding=1),\n nn.ReLU(inplace=True)\n )\n self.enc1 = conv_block(3, 32)\n self.pool1 = nn.MaxPool2d(2)\n self.enc2 = conv_block(32, 64)\n self.pool2 = nn.MaxPool2d(2)\n self.enc3 = conv_block(64, 128)\n\n self.up2 = nn.ConvTranspose2d(128, 64, 2, stride=2)\n self.dec2 = conv_block(128, 64)\n self.up1 = nn.ConvTranspose2d(64, 32, 2, stride=2)\n self.dec1 = conv_block(64, 32)\n\n self.out = nn.Conv2d(32, 1, 1)\n\n def forward(self, x):\n e1 = self.enc1(x)\n p1 = self.pool1(e1)\n e2 = self.enc2(p1)\n p2 = self.pool2(e2)\n e3 = self.enc3(p2)\n u2 = self.up2(e3)\n d2 = self.dec2(torch.cat([u2, e2], dim=1))\n u1 = self.up1(d2)\n d1 = self.dec1(torch.cat([u1, e1], dim=1))\n return self.out(d1)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:55:39.178244Z","iopub.execute_input":"2025-11-08T05:55:39.179125Z","iopub.status.idle":"2025-11-08T05:55:39.188101Z","shell.execute_reply.started":"2025-11-08T05:55:39.179083Z","shell.execute_reply":"2025-11-08T05:55:39.187404Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"DEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nmodel = SmallUNet().to(DEVICE)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:55:49.993446Z","iopub.execute_input":"2025-11-08T05:55:49.993721Z","iopub.status.idle":"2025-11-08T05:55:50.161087Z","shell.execute_reply.started":"2025-11-08T05:55:49.993700Z","shell.execute_reply":"2025-11-08T05:55:50.160532Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"criterion = nn.BCEWithLogitsLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.001)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:55:59.841507Z","iopub.execute_input":"2025-11-08T05:55:59.842228Z","iopub.status.idle":"2025-11-08T05:55:59.846115Z","shell.execute_reply.started":"2025-11-08T05:55:59.842185Z","shell.execute_reply":"2025-11-08T05:55:59.845338Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def train_model(model, train_loader, val_loader, criterion, optimizer, num_epochs=10):\n for epoch in range(num_epochs):\n model.train()\n train_loss = 0.0\n for imgs, masks in train_loader:\n imgs, masks = imgs.to(DEVICE), masks.to(DEVICE)\n optimizer.zero_grad()\n outputs = model(imgs)\n loss = criterion(outputs, masks)\n loss.backward()\n optimizer.step()\n train_loss += loss.item()\n print(f\"Epoch {epoch+1}, Train Loss: {train_loss/len(train_loader):.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:56:07.973372Z","iopub.execute_input":"2025-11-08T05:56:07.973851Z","iopub.status.idle":"2025-11-08T05:56:07.978713Z","shell.execute_reply.started":"2025-11-08T05:56:07.973827Z","shell.execute_reply":"2025-11-08T05:56:07.977855Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =======================================\n# Prediction and Submission\n# =======================================\n\ndef create_submission(model, test_dir, sample_sub_path, submission_path):\n model.eval()\n test_images_names = os.listdir(test_dir)\n submission_data = []\n\n # Use the same validation transform for test images (without augmentation)\n test_transform = transforms.Compose([\n transforms.Resize((IMG_SIZE, IMG_SIZE)),\n transforms.ToTensor(),\n transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n ])\n\n with torch.no_grad():\n for img_name in tqdm(test_images_names):\n img_path = os.path.join(test_dir, img_name)\n image = Image.open(img_path).convert(\"RGB\")\n original_shape = image.size # (width, height)\n \n input_image = test_transform(image).unsqueeze(0).to(DEVICE)\n output = model(input_image)\n \n # Post-process the model output\n # Assuming U-Net output (logits), apply sigmoid and threshold\n mask_pred = torch.sigmoid(output).cpu().numpy().squeeze()\n # Resize the mask back to original image size\n mask_pred_resized = Image.fromarray((mask_pred * 255).astype(np.uint8)).resize(original_shape, Image.NEAREST)\n mask_pred_resized_np = np.array(mask_pred_resized) > 127 # Binary mask\n \n rle_mask = rle_encode(mask_pred_resized_np)\n \n submission_data.append({\"case_id\": img_name, \"annotation\": rle_mask})\n\n submission_df = pd.DataFrame(submission_data)\n # The competition expects \"case_id\" without extension for submission, check sample_submission.csv format\n submission_df['case_id'] = submission_df['case_id'].str.replace('.jpg', '', regex=False) \n submission_df.to_csv(submission_path, index=False)\n print(f\"Submission file saved to {submission_path}\")\n\n# Example usage (needs a working 'model' variable):\ncreate_submission(model, os.path.join(DATA_DIR, \"test_images\"), \n os.path.join(DATA_DIR, \"sample_submission.csv\"), \n \"submission.csv\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-08T05:56:41.954996Z","iopub.execute_input":"2025-11-08T05:56:41.955294Z","iopub.status.idle":"2025-11-08T05:56:42.615760Z","shell.execute_reply.started":"2025-11-08T05:56:41.955272Z","shell.execute_reply":"2025-11-08T05:56:42.615180Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}