{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":19991,"databundleVersionId":1117522,"sourceType":"competition"}],"dockerImageVersionId":31011,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import numpy as np \nimport pandas as pd\nimport torch\nimport os\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport matplotlib.pyplot as plt\nfrom tqdm import tqdm\nfrom PIL import Image\nfrom torchvision.models.efficientnet import efficientnet_b2\nfrom torch.utils.data import Dataset, DataLoader\nfrom sklearn.metrics import accuracy_score, roc_curve","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:33.869077Z","iopub.execute_input":"2025-04-14T09:48:33.869277Z","iopub.status.idle":"2025-04-14T09:48:41.809439Z","shell.execute_reply.started":"2025-04-14T09:48:33.869240Z","shell.execute_reply":"2025-04-14T09:48:41.808821Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"base_path = '../input/alaska2-image-steganalysis'","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:41.810147Z","iopub.execute_input":"2025-04-14T09:48:41.810547Z","iopub.status.idle":"2025-04-14T09:48:41.814404Z","shell.execute_reply.started":"2025-04-14T09:48:41.810522Z","shell.execute_reply":"2025-04-14T09:48:41.813244Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def read_images_path(dir_name, label):\n    folder_path = os.path.join(base_path, dir_name)\n    return [[os.path.join(folder_path, filename), label] for filename in os.listdir(folder_path)]","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:41.815922Z","iopub.execute_input":"2025-04-14T09:48:41.816188Z","iopub.status.idle":"2025-04-14T09:48:41.981109Z","shell.execute_reply.started":"2025-04-14T09:48:41.816168Z","shell.execute_reply":"2025-04-14T09:48:41.980308Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def preprocess_image(image_path, size=(512, 512)):\n    # Load image\n    img = Image.open(image_path)\n    \n    # Convert to YCbCr color space (better for detecting steganography)\n    img = img.convert(\"YCbCr\")\n    \n    # Convert to numpy array with high precision (float64 to preserve subtle details)\n    img_array = np.array(img, dtype=np.float64)\n    \n    # Extract DCT residuals to better detect steganography artifacts\n    # This helps highlight the noise patterns that steganography introduces\n    y_channel = img_array[:, :, 0]\n    high_freq = y_channel - cv2.GaussianBlur(y_channel, (3, 3), 0)\n    img_array[:, :, 0] = high_freq\n    \n    # Transpose channels from (H, W, C) to (C, H, W) for PyTorch\n    img_array = np.transpose(img_array, (2, 0, 1))\n    \n    # Normalize to [0, 1] range but avoid standard ImageNet normalization\n    # which could destroy subtle steganographic patterns\n    img_array = img_array / 255.0\n    \n    # Convert to PyTorch tensor\n    img_tensor = torch.tensor(img_array, dtype=torch.float32)\n    \n    return img_tensor","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:41.981821Z","iopub.execute_input":"2025-04-14T09:48:41.982093Z","iopub.status.idle":"2025-04-14T09:48:41.994125Z","shell.execute_reply.started":"2025-04-14T09:48:41.982066Z","shell.execute_reply":"2025-04-14T09:48:41.993425Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class SteganalysisDataset(Dataset):\n    def __init__(self, image_paths, labels, augment=False):\n        self.image_paths = image_paths\n        self.labels = labels\n        self.augment = augment\n        \n    def __len__(self):\n        return len(self.image_paths)\n    \n    def __getitem__(self, index):\n        # Get image path and label\n        img_path = self.image_paths[index]\n        label = self.labels[index]\n        \n        # Process image\n        image = preprocess_image(img_path)\n        \n        # Apply data augmentation (only for training)\n        if self.augment:\n            # Flip horizontally with 50% probability (safe for steganalysis)\n            if np.random.random() > 0.5:\n                image = torch.flip(image, [2])\n                \n            # Random crop and resize back (safe for steganalysis as it preserves patterns)\n            if np.random.random() > 0.5:\n                i, j = torch.randint(0, 32, (2,))\n                image = image[:, i:512-32+i, j:512-32+j]\n                image = F.interpolate(image.unsqueeze(0), size=(512, 512), \n                                     mode='bilinear', align_corners=False).squeeze(0)\n                \n        return image, label","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:41.994850Z","iopub.execute_input":"2025-04-14T09:48:41.995160Z","iopub.status.idle":"2025-04-14T09:48:42.007107Z","shell.execute_reply.started":"2025-04-14T09:48:41.995132Z","shell.execute_reply":"2025-04-14T09:48:42.006516Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class SRMConv2D(nn.Module):\n    def __init__(self):\n        super(SRMConv2D, self).__init__()\n        \n        # Define SRM kernels (based on steganalysis research)\n        kernel1 = torch.tensor([[-1, 2, -1], [2, -4, 2], [-1, 2, -1]], dtype=torch.float32)\n        kernel2 = torch.tensor([[-1, 2, -1], [2, -4, 2], [0, 0, 0]], dtype=torch.float32)\n        kernel3 = torch.tensor([[0, 0, 0], [0, 1, 0], [0, 0, 0]], dtype=torch.float32)\n        \n        # Combine kernels into filter bank\n        self.srm_kernels = nn.Parameter(torch.stack([kernel1, kernel2, kernel3]).unsqueeze(1), \n                                       requires_grad=False)\n        \n        # Define the convolution operation\n        self.conv = nn.Conv2d(1, 3, kernel_size=3, padding=1, bias=False)\n        self.conv.weight = nn.Parameter(self.srm_kernels, requires_grad=False)\n        \n    def forward(self, x):\n        # Apply SRM filters independently to each channel\n        y_channel = self.conv(x[:, 0:1, :, :])\n        cb_channel = self.conv(x[:, 1:2, :, :])\n        cr_channel = self.conv(x[:, 2:3, :, :])\n        \n        # Combine results\n        return torch.cat([y_channel, cb_channel, cr_channel], dim=1)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.007759Z","iopub.execute_input":"2025-04-14T09:48:42.008007Z","iopub.status.idle":"2025-04-14T09:48:42.023534Z","shell.execute_reply.started":"2025-04-14T09:48:42.007980Z","shell.execute_reply":"2025-04-14T09:48:42.022789Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class SteganalysisModel(nn.Module):\n    def __init__(self, num_classes=2):\n        super(SteganalysisModel, self).__init__()\n        \n        # SRM preprocessing layer\n        self.srm = SRMConv2D()\n        \n        # Load pretrained EfficientNet\n        self.backbone = efficientnet_b2(pretrained=True)\n        \n        # Modify first layer to accept 9 channels (3 channels x 3 SRM filters)\n        self.backbone.features[0][0] = nn.Conv2d(9, 32, kernel_size=3, stride=2, padding=1, bias=False)\n        \n        # Modified classifier with attention\n        self.attention = nn.Sequential(\n            nn.Conv2d(1408, 512, kernel_size=1),\n            nn.ReLU(),\n            nn.Conv2d(512, 1, kernel_size=1),\n            nn.Sigmoid()\n        )\n        \n        # Final classifier\n        self.classifier = nn.Sequential(\n            nn.Linear(1408, 512),\n            nn.BatchNorm1d(512),\n            nn.ReLU(),\n            nn.Dropout(0.3),\n            nn.Linear(512, num_classes)\n        )\n        \n    def forward(self, x):\n        # Apply SRM filters\n        x = self.srm(x)\n        \n        # Extract features\n        features = self.backbone.features(x)\n        \n        # Apply attention\n        attention_weights = self.attention(features)\n        features = features * attention_weights\n        \n        # Global average pooling\n        features = F.adaptive_avg_pool2d(features, (1, 1))\n        features = torch.flatten(features, 1)\n        \n        # Classification\n        output = self.classifier(features)\n        \n        return output","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.024244Z","iopub.execute_input":"2025-04-14T09:48:42.024441Z","iopub.status.idle":"2025-04-14T09:48:42.034940Z","shell.execute_reply.started":"2025-04-14T09:48:42.024426Z","shell.execute_reply":"2025-04-14T09:48:42.034190Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def prepare_balanced_dataset(sample_size=15000):\n    # Load cover images\n    cover_img = read_images_path('Cover', 0)[:sample_size]\n    \n    # Load stego images (balanced among different techniques)\n    stego_per_type = sample_size\n    jmipod_img = read_images_path('JMiPOD', 1)[:stego_per_type]\n    juniward_img = read_images_path('JUNIWARD', 1)[:stego_per_type]\n    uerd_img = read_images_path('UERD', 1)[:stego_per_type]\n    \n    # Combine all images\n    data = cover_img + jmipod_img + juniward_img + uerd_img\n    df = pd.DataFrame(data=data, columns=['path', 'label'])\n    \n    # Shuffle data\n    df = df.sample(frac=1, random_state=42).reset_index(drop=True)\n    \n    # Split into train and validation\n    train_size = int(len(df) * 0.8)\n    train = df.iloc[:train_size]\n    val = df.iloc[train_size:]\n    \n    return train, val","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.035638Z","iopub.execute_input":"2025-04-14T09:48:42.036167Z","iopub.status.idle":"2025-04-14T09:48:42.050102Z","shell.execute_reply.started":"2025-04-14T09:48:42.036113Z","shell.execute_reply":"2025-04-14T09:48:42.049441Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def weighted_auc(y_true, y_score, tpr_thresholds=[0.0, 0.4, 1.0], weights=[2, 1]):\n    fpr, tpr, _ = roc_curve(y_true, y_score)\n    \n    # Calculate AUC for each region\n    auc_scores = []\n    for i in range(len(tpr_thresholds) - 1):\n        start = tpr_thresholds[i]\n        end = tpr_thresholds[i+1]\n        \n        # Filter points in current region\n        mask = (tpr >= start) & (tpr < end)\n        auc_scores.append(np.trapz(tpr[mask], fpr[mask]))\n    \n    # Calculate weighted AUC\n    weighted_auc = np.sum(np.multiply(auc_scores, weights)) / np.sum(weights)\n    \n    return weighted_auc","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.052494Z","iopub.execute_input":"2025-04-14T09:48:42.053172Z","iopub.status.idle":"2025-04-14T09:48:42.065494Z","shell.execute_reply.started":"2025-04-14T09:48:42.053147Z","shell.execute_reply":"2025-04-14T09:48:42.064724Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def calculate_metrics(model, dataloader, device='cpu'):\n    model.to(device).eval()\n    labels, probs, preds = [], [], []\n\n    with torch.no_grad():\n        for images, batch_labels in dataloader:\n            outputs = model(images.to(device))\n            batch_probs = F.softmax(outputs, dim=1)[:, 1]\n            batch_preds = torch.argmax(outputs, dim=1)\n\n            labels.extend(batch_labels.cpu().numpy())\n            probs.extend(batch_probs.cpu().numpy())\n            preds.extend(batch_preds.cpu().numpy())\n\n    labels = np.array(labels)\n    probs = np.array(probs)\n    preds = np.array(preds)\n\n    return {\n        'Weighted AUC': weighted_auc(labels, probs),\n        'Accuracy': accuracy_score(labels, preds)\n    }","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.066192Z","iopub.execute_input":"2025-04-14T09:48:42.066935Z","iopub.status.idle":"2025-04-14T09:48:42.076828Z","shell.execute_reply.started":"2025-04-14T09:48:42.066918Z","shell.execute_reply":"2025-04-14T09:48:42.076117Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(f\"Using device: {device}\")\n\n# Import OpenCV for noise extraction\nimport cv2\n\n# Set hyperparameters\nBATCH_SIZE = 16  # Smaller batch size for better gradient updates\nEPOCHS = 10  # More epochs for better training\nLR = 5e-5  # Learning rate\nWEIGHT_DECAY = 1e-5  # Add regularization","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.077476Z","iopub.execute_input":"2025-04-14T09:48:42.077675Z","iopub.status.idle":"2025-04-14T09:48:42.510515Z","shell.execute_reply.started":"2025-04-14T09:48:42.077661Z","shell.execute_reply":"2025-04-14T09:48:42.509978Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Prepare datasets\nprint(\"Preparing datasets...\")\ntrain, val = prepare_balanced_dataset(sample_size=20000)\n\n# Create datasets\ntrain_dataset = SteganalysisDataset(train['path'].tolist(), train['label'].tolist(), augment=True)\nval_dataset = SteganalysisDataset(val['path'].tolist(), val['label'].tolist(), augment=False)\n\n# Create dataloaders\ntrain_dataloader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True, \n                              num_workers=4, pin_memory=True, prefetch_factor=2, persistent_workers=True)\nval_dataloader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False, \n                           num_workers=4, pin_memory=True)\n\n# Create model\nmodel = SteganalysisModel(num_classes=2).to(device)\n\n# Set up class weights to handle imbalance\n# Higher weight for stego images (class 1)\nWEIGHTS = torch.tensor([3, 1], dtype=torch.float32).to(device)\n\n# Create optimizer with weight decay\noptimizer = torch.optim.AdamW(model.parameters(), lr=LR, weight_decay=WEIGHT_DECAY)\n\n# Learning rate scheduler\nscheduler = torch.optim.lr_scheduler.OneCycleLR(\n    optimizer,\n    max_lr=1e-4,\n    epochs=EPOCHS,\n    steps_per_epoch=len(train_dataloader),\n    pct_start=0.3\n)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:42.511188Z","iopub.execute_input":"2025-04-14T09:48:42.511448Z","iopub.status.idle":"2025-04-14T09:48:46.000478Z","shell.execute_reply.started":"2025-04-14T09:48:42.511423Z","shell.execute_reply":"2025-04-14T09:48:45.999653Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Training loop\ntrain_losses = []\naccuracies = []\nweighted_aucs = []\nbest_auc = 0\n\nfor epoch in range(EPOCHS):\n    print(f'EPOCH: {epoch+1}/{EPOCHS}')\n    if torch.cuda.is_available():\n        print(f\"GPU: {torch.cuda.get_device_name(0)}\")\n        print(f\"Memory allocated: {torch.cuda.memory_allocated(0)/1e9:.2f} GB\")\n        print(f\"Memory reserved: {torch.cuda.memory_reserved(0)/1e9:.2f} GB\")\n    \n    # Training phase\n    model.train()\n    train_loss = 0\n    \n    for images, labels in tqdm(train_dataloader, desc=\"Training\"):\n        images, labels = images.to(device), labels.to(device)\n        \n        # Zero gradients\n        optimizer.zero_grad()\n        \n        # Forward pass\n        outputs = model(images)\n        \n        # Calculate loss with class weights\n        loss = F.cross_entropy(outputs, labels, weight=WEIGHTS)\n        \n        # Backward pass\n        loss.backward()\n        \n        # Gradient clipping to prevent exploding gradients\n        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)\n        \n        # Update parameters\n        optimizer.step()\n\n        scheduler.step()\n        \n        # Track loss\n        train_loss += loss.item()\n        \n    # Calculate average training loss\n    avg_train_loss = train_loss / len(train_dataloader)\n    train_losses.append(avg_train_loss)\n    print(f'Average Training Loss: {avg_train_loss:.4f}')\n    \n    # Validation phase\n    metrics = calculate_metrics(model, val_dataloader, device)\n    acc = metrics['Accuracy']\n    current_auc = metrics['Weighted AUC']\n    \n    # Track metrics\n    accuracies.append(acc)\n    weighted_aucs.append(current_auc)\n    \n    print(f'Validation Accuracy: {acc:.4f}')\n    print(f'Validation Weighted AUC: {current_auc:.4f}')\n    \n    # Update learning rate based on validation AUC\n    scheduler.step(current_auc)\n    \n    # Save best model\n    if current_auc > best_auc:\n        best_auc = current_auc\n        torch.save(model.state_dict(), 'best_stego_model.pth')\n        print(f\"New best model saved with Weighted AUC: {best_auc:.4f}\")\n    \n    print()\n\n# Load best model\nmodel.load_state_dict(torch.load('best_stego_model.pth'))\nprint(\"Best model loaded!\")\n\n# Save complete model\ntorch.save(model, 'complete_steganalysis_model.pth')\nprint(\"Complete model saved to 'complete_steganalysis_model.pth'\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T09:48:46.001354Z","iopub.execute_input":"2025-04-14T09:48:46.001622Z","iopub.status.idle":"2025-04-14T15:25:00.843367Z","shell.execute_reply.started":"2025-04-14T09:48:46.001600Z","shell.execute_reply":"2025-04-14T15:25:00.842245Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":" # Plot training metrics\nplt.figure(figsize=(15, 5))\n\n# Plot Loss\nplt.subplot(1, 3, 1)\nplt.plot(range(1, EPOCHS+1), train_losses, marker='o', label='Training Loss')\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.title('Loss per Epoch')\nplt.legend()\n\n# Plot Accuracy\nplt.subplot(1, 3, 2)\nplt.plot(range(1, EPOCHS+1), accuracies, marker='o', label='Accuracy', color='green')\nplt.xlabel('Epoch')\nplt.ylabel('Accuracy')\nplt.title('Accuracy per Epoch')\nplt.legend()\n\n# Plot Weighted AUC\nplt.subplot(1, 3, 3)\nplt.plot(range(1, EPOCHS+1), weighted_aucs, marker='o', label='Weighted AUC', color='red')\nplt.xlabel('Epoch')\nplt.ylabel('AUC')\nplt.title('Weighted AUC per Epoch')\nplt.legend()\n\nplt.tight_layout()\nplt.savefig('training_metrics.png')\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T15:25:00.845146Z","iopub.execute_input":"2025-04-14T15:25:00.845461Z","iopub.status.idle":"2025-04-14T15:25:02.021715Z","shell.execute_reply.started":"2025-04-14T15:25:00.845422Z","shell.execute_reply":"2025-04-14T15:25:02.020920Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":" # Prepare test predictions\ntest_img = read_images_path('Test', 0)\ntest = pd.DataFrame(data=test_img, columns=['path', 'label'])\n\n# Create test dataset and dataloader\ntest_dataset = SteganalysisDataset(test['path'].tolist(), test['label'].tolist(), augment=False)\ntest_dataloader = DataLoader(test_dataset, batch_size=BATCH_SIZE, shuffle=False, \n                            num_workers=4, pin_memory=True)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T15:25:02.022524Z","iopub.execute_input":"2025-04-14T15:25:02.022730Z","iopub.status.idle":"2025-04-14T15:25:02.138048Z","shell.execute_reply.started":"2025-04-14T15:25:02.022713Z","shell.execute_reply":"2025-04-14T15:25:02.137506Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def predict(model, dataloader, device):\n    model.eval()\n    predictions = []\n    \n    with torch.no_grad():\n        for images, _ in tqdm(dataloader, desc=\"Predicting\"):\n            images = images.to(device)\n            outputs = model(images)\n            scores = F.softmax(outputs, dim=1)[:, 1]  # Probability of being stego\n            predictions.extend(scores.cpu().numpy())\n            \n    return predictions\n\n# Generate predictions\nprint(\"Generating predictions...\")\ntest_predictions = predict(model, test_dataloader, device)\nprint(\"Predictions complete.\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T15:25:02.138729Z","iopub.execute_input":"2025-04-14T15:25:02.138929Z","iopub.status.idle":"2025-04-14T15:25:55.038095Z","shell.execute_reply.started":"2025-04-14T15:25:02.138914Z","shell.execute_reply":"2025-04-14T15:25:55.037155Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Create submission file\ntest_image_ids = [os.path.basename(path) for path in test_dataset.image_paths]\nsubmission_df = pd.DataFrame({\n    'Id': test_image_ids,\n    'Label': test_predictions\n})\n\n# Format IDs as required by competition\nsubmission_df['Id'] = submission_df['Id'].str.replace('.jpg', '').astype(int)\nsubmission_df = submission_df.sort_values(by='Id')\nsubmission_df['Id'] = submission_df['Id'].astype(str).str.zfill(4) + '.jpg'\n\n# Save submission file\nsubmission_df.to_csv('submission.csv', index=False)\nprint(\"Submission file created: submission.csv\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-14T15:25:55.039271Z","iopub.execute_input":"2025-04-14T15:25:55.039873Z","iopub.status.idle":"2025-04-14T15:25:55.086424Z","shell.execute_reply.started":"2025-04-14T15:25:55.039847Z","shell.execute_reply":"2025-04-14T15:25:55.085792Z"}},"outputs":[],"execution_count":null}]}