{"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":19991,"databundleVersionId":1117522,"isSourceIdPinned":false,"sourceType":"competition"},{"sourceId":2078376,"sourceType":"datasetVersion","datasetId":1245748},{"sourceId":2118910,"sourceType":"datasetVersion","datasetId":1271386},{"sourceId":4043617,"sourceType":"datasetVersion","datasetId":2395063},{"sourceId":6411883,"sourceType":"datasetVersion","datasetId":3697835},{"sourceId":8013000,"sourceType":"datasetVersion","datasetId":4720723},{"sourceId":12532410,"sourceType":"datasetVersion","datasetId":7909808}],"dockerImageVersionId":31090,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"#resnet18+lsb dt+train code\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision import models, transforms\nfrom sklearn.metrics import accuracy_score, precision_recall_fscore_support, confusion_matrix\nimport numpy as np\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nimport pandas as pd\n\n# Custom Dataset for Stego and Cover Images\nclass StegoDataset(Dataset):\n    def __init__(self, cover_dir, stego_dir, transform=None):\n        self.cover_dir = cover_dir\n        self.stego_dir = stego_dir\n        self.transform = transform\n        self.cover_images = [os.path.join(cover_dir, f) for f in os.listdir(cover_dir) if f.endswith(('.png', '.jpg'))]\n        self.stego_images = [os.path.join(stego_dir, f) for f in os.listdir(stego_dir) if f.endswith(('.png', '.jpg'))]\n        self.all_images = self.cover_images + self.stego_images\n        self.labels = [0] * len(self.cover_images) + [1] * len(self.stego_images)\n\n    def __len__(self):\n        return len(self.all_images)\n\n    def __getitem__(self, idx):\n        img_path = self.all_images[idx]\n        label = self.labels[idx]\n        image = Image.open(img_path).convert('RGB')\n        if self.transform:\n            image = self.transform(image)\n        return image, label\n\n# Data Transforms\ndata_transforms = {\n    'train': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.RandomHorizontalFlip(),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n    'val': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n}\n\n# Dataset and DataLoader\ncover_dir = '/kaggle/input/lsb-stego/lsb_stego/lsb_stego/cover'\nstego_dir = '/kaggle/input/lsb-stego/lsb_stego/lsb_stego/lsb'\ndataset = StegoDataset(cover_dir, stego_dir, transform=data_transforms['train'])\n\n# Split dataset into train and validation\ntrain_size = int(0.8 * len(dataset))\nval_size = len(dataset) - train_size\ntrain_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])\nval_dataset.dataset.transform = data_transforms['val']\n\ntrain_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, num_workers=2)\nval_loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=2)\n\n# Load Pretrained ResNet18\nmodel = models.resnet18(pretrained=True)\nnum_ftrs = model.fc.in_features\nmodel.fc = nn.Linear(num_ftrs, 2)  # Binary classification (cover vs stego)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nmodel = model.to(device)\n\n# Loss Function and Optimizer\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.001)\nscheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)\n\n# Training Loop\nnum_epochs = 40\ntrain_losses, val_losses = [], []\ntrain_accuracies, val_accuracies = [], []\n\nfor epoch in range(num_epochs):\n    # Training\n    model.train()\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    for images, labels in train_loader:\n        images, labels = images.to(device), labels.to(device)\n        optimizer.zero_grad()\n        outputs = model(images)\n        loss = criterion(outputs, labels)\n        loss.backward()\n        optimizer.step()\n        running_loss += loss.item()\n        _, predicted = torch.max(outputs.data, 1)\n        total += labels.size(0)\n        correct += (predicted == labels).sum().item()\n    train_loss = running_loss / len(train_loader)\n    train_acc = 100 * correct / total\n    train_losses.append(train_loss)\n    train_accuracies.append(train_acc)\n\n    # Validation\n    model.eval()\n    val_loss = 0.0\n    correct = 0\n    total = 0\n    all_preds, all_labels = [], []\n    with torch.no_grad():\n        for images, labels in val_loader:\n            images, labels = images.to(device), labels.to(device)\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            val_loss += loss.item()\n            _, predicted = torch.max(outputs.data, 1)\n            total += labels.size(0)\n            correct += (predicted == labels).sum().item()\n            all_preds.extend(predicted.cpu().numpy())\n            all_labels.extend(labels.cpu().numpy())\n    val_loss = val_loss / len(val_loader)\n    val_acc = 100 * correct / total\n    val_losses.append(val_loss)\n    val_accuracies.append(val_acc)\n\n    print(f'Epoch [{epoch+1}/{num_epochs}] Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%')\n    scheduler.step()\n\n# Save the Model\nmodel_save_path = '/kaggle/working/stego_model.pth'\ntorch.save(model.state_dict(), model_save_path)\n\n# Evaluation Metrics\nprecision, recall, f1, _ = precision_recall_fscore_support(all_labels, all_preds, average='binary')\ncm = confusion_matrix(all_labels, all_preds)\n\n# Generate Report\nreport = f\"\"\"\n# Steganalysis Model Report\n\n## Training Summary\n- **Epochs Trained**: {num_epochs}\n- **Final Training Accuracy**: {train_accuracies[-1]:.2f}%\n- **Final Validation Accuracy**: {val_accuracies[-1]:.2f}%\n- **Final Training Loss**: {train_losses[-1]:.4f}\n- **Final Validation Loss**: {val_losses[-1]:.4f}\n\n## Evaluation Metrics (Validation Set)\n- **Accuracy**: {val_accuracies[-1]:.2f}%\n- **Precision**: {precision:.4f}\n- **Recall**: {recall:.4f}\n- **F1 Score**: {f1:.4f}\n\n## Confusion Matrix\n|                | Predicted Cover | Predicted Stego |\n|----------------|-----------------|-----------------|\n| Actual Cover   | {cm[0,0]}       | {cm[0,1]}       |\n| Actual Stego   | {cm[1,0]}       | {cm[1,1]}       |\n\n## Model Saved\n- **Path**: {model_save_path}\n\"\"\"\n\n# Plot Training and Validation Metrics\nplt.figure(figsize=(12, 5))\nplt.subplot(1, 2, 1)\nplt.plot(train_losses, label='Train Loss')\nplt.plot(val_losses, label='Val Loss')\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.title('Training and Validation Loss')\nplt.legend()\n\nplt.subplot(1, 2, 2)\nplt.plot(train_accuracies, label='Train Accuracy')\nplt.plot(val_accuracies, label='Val Accuracy')\nplt.xlabel('Epoch')\nplt.ylabel('Accuracy (%)')\nplt.title('Training and Validation Accuracy')\nplt.legend()\nplt.savefig('/kaggle/working/training_plots.png')\n\n# Plot Confusion Matrix\nplt.figure(figsize=(6, 5))\nsns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=['Cover', 'Stego'], yticklabels=['Cover', 'Stego'])\nplt.xlabel('Predicted')\nplt.ylabel('Actual')\nplt.title('Confusion Matrix')\nplt.savefig('/kaggle/working/confusion_matrix.png')\n\n# Save Report to File\nreport_file = '/kaggle/working/stego_report.md'\nwith open(report_file, 'w') as f:\n    f.write(report)\n\nprint(f\"Training completed. Model saved at {model_save_path}. Report saved at {report_file}.\")\nprint(report)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:50:37.610882Z","iopub.execute_input":"2025-10-21T05:50:37.611107Z","iopub.status.idle":"2025-10-21T05:54:38.976405Z","shell.execute_reply.started":"2025-10-21T05:50:37.611082Z","shell.execute_reply":"2025-10-21T05:54:38.975698Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#test code for the lsb trained model \nimport os\nimport torch\nimport torch.nn as nn\nfrom torchvision import models, transforms\nfrom PIL import Image\n\n# Define image transformations\ndata_transforms = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.ToTensor(),\n    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n])\n\n# Load the saved model\ndef load_model(model_path):\n    model = models.resnet18(pretrained=False)\n    num_ftrs = model.fc.in_features\n    model.fc = nn.Linear(num_ftrs, 2)  # Binary classification\n    model.load_state_dict(torch.load(model_path))\n    model.eval()\n    return model\n\n# Function to classify a single image\ndef classify_image(model, image_path, device):\n    image = Image.open(image_path).convert('RGB')\n    image = data_transforms(image).unsqueeze(0).to(device)\n    with torch.no_grad():\n        outputs = model(image)\n        _, predicted = torch.max(outputs, 1)\n    return predicted.item(), image_path\n\n# Main function to process directory and output results\ndef analyze_directory(input_dir, model_path):\n    device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n    model = load_model(model_path).to(device)\n    \n    # Get list of images\n    image_files = [f for f in os.listdir(input_dir) if f.endswith(('.png', '.jpg'))]\n    stego_images = []\n    cover_images = []\n    \n    # Classify each image\n    for img_file in image_files:\n        img_path = os.path.join(input_dir, img_file)\n        prediction, _ = classify_image(model, img_path, device)\n        if prediction == 1:  # Stego\n            stego_images.append(img_file)\n        else:  # Cover\n            cover_images.append(img_file)\n    \n    # Print results\n    total_images = len(image_files)\n    print(f\"Total Images Processed: {total_images}\")\n\n    num_stego = len(stego_images)\n    num_cover = len(cover_images)\n    \n    print(f\"Stego Images Detected: {num_stego}\")\n    print(f\"Cover Images Detected: {num_cover}\")\n    print(f\"Stego Detection Rate: {(num_stego / total_images * 100):.2f}%\")\n\n# Example usage\ninput_directory = '/kaggle/input/lsb-stego/lsb_stego/lsb_stego/lsb'  # Replace with your input directory\nmodel_path = '/kaggle/working/stego_model.pth'\nanalyze_directory(input_directory, model_path)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:55:53.762600Z","iopub.execute_input":"2025-10-21T05:55:53.762897Z","iopub.status.idle":"2025-10-21T05:55:59.438025Z","shell.execute_reply.started":"2025-10-21T05:55:53.762868Z","shell.execute_reply":"2025-10-21T05:55:59.437395Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Train-->resnet18+lsb dataset+nsgaIII\n#Install necessary libraries\n!pip install pymoo==0.6.1.1 -q\n!pip install tqdm -q\n\nimport os\nimport time\nimport torch\nimport gc # Import the garbage collector\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torchvision import datasets, transforms, models\nfrom torch.utils.data import DataLoader, SubsetRandomSampler\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom sklearn.metrics import precision_recall_fscore_support, confusion_matrix\nimport seaborn as sns\nfrom collections import Counter\nfrom tqdm.auto import tqdm\nfrom pymoo.algorithms.moo.nsga3 import NSGA3\nfrom pymoo.util.ref_dirs import get_reference_directions\nfrom pymoo.optimize import minimize\nfrom pymoo.core.problem import ElementwiseProblem\nfrom pymoo.core.callback import Callback\n\n# --- 1. Configuration ---\nprint(\"--- 1. CONFIGURATION ---\")\nDATA_DIR = \"/kaggle/working/data1\" # Example Kaggle input path\nSAVE_DIR = \"/kaggle/working/\"\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nos.makedirs(SAVE_DIR, exist_ok=True)\nNUM_EPOCHS_FINAL, PATIENCE = 20, 5\nRANDOM_SEED = 42\nPOP_SIZE, N_GEN, NUM_EPOCHS_OPT = 12, 10, 5\ntorch.manual_seed(RANDOM_SEED)\ntorch.cuda.manual_seed_all(RANDOM_SEED)\nnp.random.seed(RANDOM_SEED)\nprint(f\"Device: {DEVICE}, Data Dir: {DATA_DIR}, Save Dir: {SAVE_DIR}\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 2. Data Preparation ---\n# (This section is unchanged)\nprint(\"--- 2. DATA PREPARATION ---\")\ntransform_train = transforms.Compose([transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])])\nval_dataset = datasets.ImageFolder(DATA_DIR, transform=transform_train) # Use same transform for simplicity\nfull_dataset = val_dataset\nprint(f\"Classes found: {full_dataset.classes}\")\ndataset_size = len(full_dataset)\nindices = list(range(dataset_size))\nnp.random.shuffle(indices)\nsplit = int(np.floor(0.2 * dataset_size))\ntrain_indices, val_indices = indices[split:], indices[:split]\ntrain_sampler = SubsetRandomSampler(train_indices)\nval_sampler = SubsetRandomSampler(val_indices)\nprint(f\"Total: {dataset_size} | Train: {len(train_indices)} | Val: {len(val_indices)}\\n\" + \"-\"*30 + \"\\n\")\n\n\n# --- 3. Model & Helper Functions ---\n# (This section is unchanged)\nprint(\"--- 3. MODEL & HELPER FUNCTIONS ---\")\ndef create_model(dropout_rate=0.5):\n    model = models.resnet18(weights=\"IMAGENET1K_V1\"); [p.requires_grad_(False) for p in model.parameters()]; [p.requires_grad_(True) for p in model.layer4.parameters()]\n    model.fc = nn.Sequential(nn.Linear(model.fc.in_features, 512), nn.ReLU(), nn.Dropout(dropout_rate), nn.Linear(512, 2))\n    return model.to(DEVICE)\ndef train_model(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs, patience):\n    best_val_acc = 0.0; best_model_wts = None; patience_counter = 0; start_time = time.time()\n    for epoch in range(num_epochs):\n        model.train()\n        train_pbar = tqdm(train_loader, desc=f\"Epoch {epoch+1}/{num_epochs} [Train]\")\n        for inputs, labels in train_pbar:\n            inputs, labels = inputs.to(DEVICE), labels.to(DEVICE); optimizer.zero_grad()\n            outputs = model(inputs); loss = criterion(outputs, labels); loss.backward(); optimizer.step()\n        model.eval()\n        val_pbar = tqdm(val_loader, desc=f\"Epoch {epoch+1}/{num_epochs} [Val]\"); val_correct_preds = 0\n        with torch.no_grad():\n            for inputs, labels in val_pbar:\n                inputs, labels = inputs.to(DEVICE), labels.to(DEVICE)\n                outputs = model(inputs); _, preds = torch.max(outputs, 1); val_correct_preds += torch.sum(preds == labels.data)\n        val_acc = val_correct_preds.double() / len(val_loader.sampler)\n        if val_acc > best_val_acc:\n            best_val_acc = val_acc; best_model_wts = model.state_dict().copy(); patience_counter = 0\n        else:\n            patience_counter += 1\n            if patience_counter >= patience: break\n    if best_model_wts: model.load_state_dict(best_model_wts)\n    return model, best_val_acc.item(), time.time() - start_time\ndef evaluate_model(model, loader):\n    model.eval(); all_preds, all_labels = [], []\n    with torch.no_grad():\n        for images, labels in loader:\n            images, labels = images.to(DEVICE), labels.to(DEVICE); outputs = model(images)\n            _, predicted = torch.max(outputs, 1); all_preds.extend(predicted.cpu().numpy()); all_labels.extend(labels.cpu().numpy())\n    accuracy = 100 * np.mean(np.array(all_preds) == np.array(all_labels))\n    precision, recall, f1, _ = precision_recall_fscore_support(all_labels, all_preds, average='binary', pos_label=1, zero_division=0)\n    return accuracy, precision, recall, f1\nprint(\"Helpers defined.\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 4. Hyperparameter Optimization (NSGA-III) ---\nprint(\"--- 4. HYPERPARAMETER OPTIMIZATION (NSGA-III) ---\")\n\nclass HyperparameterOptimization(ElementwiseProblem):\n    def __init__(self, save_dir): # CHANGED: Pass save_dir\n        super().__init__(n_var=4, n_obj=3, n_constr=0, xl=np.array([1e-5, 0.1, 1e-6, 16]), xu=np.array([1e-2, 0.6, 1e-3, 64]))\n        # NEW: Track best accuracy and provide the save directory\n        self.best_acc_so_far = 0.0\n        self.save_dir = save_dir\n\n    def _evaluate(self, x, out, *args, **kwargs):\n        lr, dropout, weight_decay, batch_size = x\n        batch_size = int(round(batch_size))\n        print(f\"\\nEvaluating: LR={lr:.6f}, Dropout={dropout:.3f}, WD={weight_decay:.6f}, BS={batch_size}\")\n        try:\n            train_loader = DataLoader(full_dataset, batch_size=batch_size, sampler=train_sampler, num_workers=0)\n            val_loader = DataLoader(val_dataset, batch_size=batch_size, sampler=val_sampler, num_workers=0)\n            model = create_model(dropout_rate=dropout)\n            criterion = nn.CrossEntropyLoss()\n            optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)\n            scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=NUM_EPOCHS_OPT)\n            model, val_acc, training_time = train_model(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs=NUM_EPOCHS_OPT, patience=3)\n            _, _, _, f1 = evaluate_model(model, val_loader)\n            out[\"F\"] = [-val_acc, -f1, training_time]\n            print(f\"Result: ValAcc={val_acc:.4f}, F1={f1:.4f}, Time={training_time:.2f}s\")\n            \n            # NEW: Check if this is the best model so far and save it\n            if val_acc > self.best_acc_so_far:\n                self.best_acc_so_far = val_acc\n                print(f\"🎉 New best validation accuracy: {val_acc:.4f}. Saving model to 'best_model_during_opt.pth'\")\n                torch.save(model.state_dict(), os.path.join(self.save_dir, \"best_model_during_opt.pth\"))\n\n        except Exception as e:\n            print(f\"An error occurred during evaluation: {e}\")\n            out[\"F\"] = [1.0, 1.0, 1e6]\n        finally:\n            if 'model' in locals(): del model\n            if 'optimizer' in locals(): del optimizer\n            if 'criterion' in locals(): del criterion\n            if 'train_loader' in locals(): del train_loader\n            if 'val_loader' in locals(): del val_loader\n            gc.collect()\n            torch.cuda.empty_cache()\n\n# We no longer need the SaveCallback for hyperparameters, but you can keep it if you want\nref_dirs = get_reference_directions(\"das-dennis\", 3, n_partitions=3)\nalgorithm = NSGA3(pop_size=POP_SIZE, ref_dirs=ref_dirs)\n\n# CHANGED: Instantiate the problem with the save directory\nproblem = HyperparameterOptimization(save_dir=SAVE_DIR)\n\nres = minimize(\n    problem, # Use the problem instance\n    algorithm,\n    termination=('n_gen', N_GEN),\n    seed=RANDOM_SEED,\n    verbose=True,\n    save_history=True\n)\n\nprint(\"\\nOptimization finished.\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 5. Final Analysis and Training ---\n# The rest of the script remains the same. It will still train a final model for 20 epochs\n# which is the recommended approach. The 'best_model_during_opt.pth' serves as a great backup.\n# ... (rest of the script is unchanged) ...\nprint(\"--- 5. ANALYSIS & FINAL TRAINING ---\")\nnp.save(os.path.join(SAVE_DIR, 'nsga3_final_X.npy'), res.X)\nnp.save(os.path.join(SAVE_DIR, 'nsga3_final_F.npy'), res.F)\nsorted_indices = np.lexsort((-res.F[:, 1], -res.F[:, 0]))\nbest_idx = sorted_indices[0]\nbest_hyperparams = res.X[best_idx]\nprint(f\"Best Hyperparameters Found: {best_hyperparams}\")\nbest_lr, best_dropout, best_wd, best_bs = best_hyperparams\nfinal_train_loader = DataLoader(full_dataset, batch_size=int(round(best_bs)), sampler=train_sampler, num_workers=0)\nfinal_val_loader = DataLoader(val_dataset, batch_size=int(round(best_bs)), sampler=val_sampler, num_workers=0)\nfinal_model = create_model(dropout_rate=best_dropout)\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(final_model.parameters(), lr=best_lr, weight_decay=best_wd)\nscheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=NUM_EPOCHS_FINAL)\nfinal_model, best_val_acc, _ = train_model(final_model, final_train_loader, final_val_loader, criterion, optimizer, scheduler, num_epochs=NUM_EPOCHS_FINAL, patience=PATIENCE)\nmodel_save_path = os.path.join(SAVE_DIR, \"final_trained_model.pth\")\ntorch.save({'model_state_dict': final_model.state_dict(), 'best_val_acc': best_val_acc, 'hyperparameters': best_hyperparams,}, model_save_path)\nprint(f\"\\nFinal model saved to {model_save_path}\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 6. Final Evaluation & Visualization ---\nprint(\"--- 6. FINAL EVALUATION ---\")\naccuracy, precision, recall, f1 = evaluate_model(final_model, final_val_loader)\nprint(f\"Accuracy:  {accuracy:.4f}%\\nPrecision: {precision:.4f}\\nRecall:    {recall:.4f}\\nF1-Score:  {f1:.4f}\")\ndef plot_confusion_matrix(model, loader, class_names):\n    model.eval(); all_preds, all_labels = [], []\n    with torch.no_grad():\n        for images, labels in loader:\n            images, labels = images.to(DEVICE), labels.to(DEVICE); outputs = model(images)\n            _, predicted = torch.max(outputs, 1); all_preds.extend(predicted.cpu().numpy()); all_labels.extend(labels.cpu().numpy())\n    cm = confusion_matrix(all_labels, all_preds)\n    plt.figure(figsize=(6, 5)); sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=class_names, yticklabels=class_names)\n    plt.title('Confusion Matrix'); plt.xlabel('Predicted Label'); plt.ylabel('True Label'); plt.show()\nplot_confusion_matrix(final_model, final_val_loader, full_dataset.classes)","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-07-21T07:42:37.770772Z","iopub.execute_input":"2025-07-21T07:42:37.771521Z"},"_kg_hide-input":true,"_kg_hide-output":true,"scrolled":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Testing on alaska dt using the previously trained model\n\nimport os\nimport torch\nimport torch.nn as nn\nfrom torchvision import transforms, models\nfrom PIL import Image\nfrom tqdm.auto import tqdm\n\n# --- 1. CONFIGURATION ---\n# IMPORTANT: Update these paths before running!\nMODEL_PATH = \"/kaggle/input/rasheed-models/final_trained_model.pth\"  # Path to your saved model\nTEST_DIR = \"/kaggle/input/alaska2-image-steganalysis/Cover\"  # Path to the folder with images you want to test\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n# -------------------------\n\n# Define the class names based on your training setup\n# This assumes 'clean' was the first folder alphabetically, and 'stego' was the second.\nCLASS_NAMES = ['clean', 'stego']\n\n\n# --- 2. MODEL DEFINITION ---\n# This function must be identical to the one used for training to ensure the architecture matches the saved weights.\ndef create_model(dropout_rate=0.5):\n    \"\"\"Creates a ResNet-18 model with a custom classifier head.\"\"\"\n    model = models.resnet18(weights=\"IMAGENET1K_V1\") # Using pre-trained weights is fine, they get overwritten by your state_dict\n    \n    # Freeze all layers initially\n    for param in model.parameters():\n        param.requires_grad = False\n    # Unfreeze the final convolutional block (layer4) for fine-tuning\n    for param in model.layer4.parameters():\n        param.requires_grad = True\n\n    # Replace the fully connected layer\n    num_ftrs = model.fc.in_features\n    model.fc = nn.Sequential(\n        nn.Linear(num_ftrs, 512),\n        nn.ReLU(),\n        nn.Dropout(dropout_rate),\n        nn.Linear(512, 2) # 2 classes: clean vs stego\n    )\n    return model.to(DEVICE)\n\n\n# --- 3. PREDICTION SCRIPT ---\ndef test_directory(model_path, test_dir):\n    \"\"\"\n    Loads a model and predicts classes for all images in a directory.\n    \"\"\"\n    if not os.path.exists(model_path):\n        print(f\"Error: Model file not found at {model_path}\")\n        return\n\n    if not os.path.exists(test_dir):\n        print(f\"Error: Test directory not found at {test_dir}\")\n        return\n\n    print(\"--- Starting Prediction ---\")\n    print(f\"Using device: {DEVICE}\")\n\n    # Load the checkpoint\n    checkpoint = torch.load(model_path, map_location=DEVICE, weights_only=False)    \n    # Extract hyperparameters to recreate the model architecture\n    # The key here is the dropout rate\n    hyperparameters = checkpoint['hyperparameters']\n    dropout_rate = hyperparameters[1] \n    \n    # Create model instance and load the trained weights\n    model = create_model(dropout_rate=dropout_rate)\n    model.load_state_dict(checkpoint['model_state_dict'])\n    model.eval() # Set model to evaluation mode\n\n    # Define the image transformations (must be same as validation transforms)\n    transform = transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n    ])\n\n    # Initialize counters\n    prediction_counts = {class_name: 0 for class_name in CLASS_NAMES}\n    image_files = [f for f in os.listdir(test_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp'))]\n    \n    if not image_files:\n        print(\"No image files found in the test directory.\")\n        return\n\n    # Loop through all images and predict\n    for image_name in tqdm(image_files, desc=\"Predicting on test images\"):\n        image_path = os.path.join(test_dir, image_name)\n        try:\n            # Open and preprocess the image\n            image = Image.open(image_path).convert('RGB')\n            image_tensor = transform(image).unsqueeze(0).to(DEVICE) # Add batch dimension and send to device\n\n            # Make prediction\n            with torch.no_grad():\n                outputs = model(image_tensor)\n                _, predicted_idx = torch.max(outputs, 1)\n            \n            # Record the prediction\n            predicted_class_name = CLASS_NAMES[predicted_idx.item()]\n            prediction_counts[predicted_class_name] += 1\n\n        except Exception as e:\n            print(f\"Could not process {image_name}. Error: {e}\")\n\n    # --- 4. REPORT RESULTS ---\n    print(\"\\n--- Prediction Complete ---\")\n    total_images = len(image_files)\n    print(f\"Total images processed: {total_images}\")\n    print(f\"Number of 'clean' images predicted: {prediction_counts['clean']}\")\n    print(f\"Number of 'stego' images predicted: {prediction_counts['stego']}\")\n    print(\"---------------------------\\n\")\n\n\n# --- RUN THE TEST ---\ntest_directory(MODEL_PATH, TEST_DIR)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:56:05.437840Z","iopub.execute_input":"2025-10-21T05:56:05.438362Z","iopub.status.idle":"2025-10-21T05:56:30.128484Z","shell.execute_reply.started":"2025-10-21T05:56:05.438323Z","shell.execute_reply":"2025-10-21T05:56:30.127369Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"##nsga3-jpg test code\nimport os\nimport torch\nimport torch.nn as nn\nimport torchvision.models as models\nimport torchvision.transforms as transforms\nfrom torch.utils.data import Dataset, DataLoader\nfrom PIL import Image\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom sklearn.metrics import confusion_matrix\n\n# Custom Dataset for loading images\nclass ImageDataset(Dataset):\n    def __init__(self, image_dir, transform=None, label_fn=None):\n        self.image_dir = image_dir\n        self.transform = transform\n        self.label_fn = label_fn\n        self.images = [os.path.join(image_dir, img) for img in os.listdir(image_dir) if img.endswith(('.png', '.jpg', '.jpeg'))]\n    \n    def __len__(self):\n        return len(self.images)\n    \n    def __getitem__(self, idx):\n        img_path = self.images[idx]\n        image = Image.open(img_path).convert('RGB')\n        if self.transform:\n            image = self.transform(image)\n        if self.label_fn:\n            label = self.label_fn(img_path)\n            return image, label, img_path\n        return image, img_path\n\n# CNN Model matching the saved model's architecture\nclass StegoClassifier(nn.Module):\n    def __init__(self, dropout=0.5):\n        super(StegoClassifier, self).__init__()\n        self.resnet = models.resnet18(weights=None)\n        num_ftrs = self.resnet.fc.in_features\n        self.resnet.fc = nn.Sequential(\n            nn.Linear(num_ftrs, 512),\n            nn.Dropout(dropout),\n            nn.Linear(512, 2)\n        )\n    \n    def forward(self, x):\n        return self.resnet(x)\n\ndef evaluate_model(image_dir, model_path, dataset_name, label_fn=None, device='cuda' if torch.cuda.is_available() else 'cpu'):\n    # Define transforms\n    transform = transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n    ])\n    \n    # Load dataset\n    dataset = ImageDataset(image_dir, transform=transform, label_fn=label_fn)\n    if len(dataset) == 0:\n        print(f\"No images found in {image_dir}.\")\n        return 0, 0, []\n    \n    dataloader = DataLoader(dataset, batch_size=32, shuffle=False, num_workers=4, pin_memory=True)\n    \n    # Initialize model (dropout set to 0.5; adjust if known from training)\n    model = StegoClassifier(dropout=0.5).to(device)\n    \n    # Load pretrained weights\n    if model_path and os.path.exists(model_path):\n        try:\n            checkpoint = torch.load(model_path, map_location=device)\n            state_dict = checkpoint.get('model_state_dict', checkpoint)\n            \n            # Remap state dictionary keys\n            new_state_dict = {}\n            for key, value in state_dict.items():\n                if 'num_batches_tracked' in key:\n                    continue\n                # Add 'resnet.' prefix to match StegoClassifier\n                if key.startswith('fc.'):\n                    new_key = 'resnet.' + key\n                else:\n                    new_key = 'resnet.' + key\n                new_state_dict[new_key] = value\n            \n            model.load_state_dict(new_state_dict, strict=True)\n            print(f\"Loaded model from {model_path}.\")\n        except Exception as e:\n            print(f\"Error loading {model_path}: {e}\")\n            print(\"Cannot proceed without a valid model.\")\n            return 0, 0, []\n    else:\n        print(f\"Model path {model_path} not found. Cannot proceed.\")\n        return 0, 0, []\n    \n    model.eval()\n    \n    stego_count = 0\n    clean_count = 0\n    all_preds = []\n    all_confidences = []\n    all_labels = [] if label_fn else None\n    \n    try:\n        with torch.no_grad():\n            for batch in dataloader:\n                if label_fn:\n                    images, labels, _ = batch\n                    images, labels = images.to(device, non_blocking=True), labels.to(device)\n                    all_labels.extend(labels.cpu().numpy())\n                else:\n                    images, _ = batch\n                    images = images.to(device, non_blocking=True)\n                outputs = model(images)\n                probabilities = torch.softmax(outputs, dim=1)\n                confidences, predicted = torch.max(probabilities, 1)\n                stego_count += (predicted == 1).sum().item()\n                clean_count += (predicted == 0).sum().item()\n                all_preds.extend(predicted.cpu().numpy())\n                all_confidences.extend(confidences.cpu().numpy())\n    except KeyboardInterrupt:\n        print(\"Inference interrupted. Returning partial results.\")\n    \n    # Print results\n    total_images = stego_count + clean_count\n    print(f\"\\n{dataset_name} Results:\")\n    print(f\"Number of stego images detected: {stego_count}\")\n    print(f\"Number of clean images detected: {clean_count}\")\n    print(f\"Percentage of stego images: {(stego_count / total_images * 100) if total_images > 0 else 0:.2f}%\")\n    print(f\"Average confidence for predictions: {np.mean(all_confidences):.4f}\")\n    \n    # Plot confidence distribution\n    plt.figure(figsize=(8, 4))\n    plt.hist(all_confidences, bins=20, range=(0, 1), color='blue', alpha=0.7)\n    plt.title(f'Confidence Distribution - {dataset_name}')\n    plt.xlabel('Confidence Score')\n    plt.ylabel('Frequency')\n    plt.show()\n    \n    # Plot confusion matrix if labels are available\n    if all_labels:\n        cm = confusion_matrix(all_labels, all_preds)\n        plt.figure(figsize=(6, 4))\n        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=['clean', 'stego'], yticklabels=['clean', 'stego'])\n        plt.xlabel('Predicted')\n        plt.ylabel('True')\n        plt.title('Confusion Matrix')\n        plt.show()\n    \n    return stego_count, clean_count, all_preds\n\n# Example label function (modify based on your naming convention)\ndef label_fn(img_path):\n    # Example: Assume filenames contain 'stego' for stego images, else clean\n    return 1 if 'stego' in os.path.basename(img_path).lower() else 0\n\n# Example usage\nif __name__ == \"__main__\":\n    model_path = \"/kaggle/input/rasheed-models/nsga3.pth\"\n    lsb_test_dir = \"/kaggle/input/lsb-stego/lsb_stego/lsb_stego/lsb\"\n    \n    # Evaluate on LSB test set\n    if os.path.exists(lsb_test_dir):\n        # Set label_fn if you have a way to infer labels (e.g., filename patterns)\n        # Otherwise, use label_fn=None\n        lsb_stego, lsb_clean, lsb_preds = evaluate_model(lsb_test_dir, model_path, \"LSB Test\", label_fn=None)\n    else:\n        print(f\"LSB test directory {lsb_test_dir} not found.\")\n        lsb_stego, lsb_clean, lsb_preds = 0, 0, []","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:56:36.117917Z","iopub.execute_input":"2025-10-21T05:56:36.118441Z","iopub.status.idle":"2025-10-21T05:56:38.697101Z","shell.execute_reply.started":"2025-10-21T05:56:36.118415Z","shell.execute_reply":"2025-10-21T05:56:38.696473Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#resnet18+44k png imagedt+train code\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torchvision import datasets, transforms, models\nfrom torch.utils.data import DataLoader, SubsetRandomSampler\nimport numpy as np\nimport re\nfrom collections import Counter\nfrom tqdm.auto import tqdm\n\n# --- 1. CONFIGURATION ---\ntorch.manual_seed(42); torch.cuda.manual_seed_all(42); np.random.seed(42)\ndata_dir = \"/kaggle/input/stegoimagesdataset/train/train\"\nsave_dir = \"/kaggle/working/steg_model_from_scratch\"\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nos.makedirs(save_dir, exist_ok=True)\nprint(f\"Device: {device}, Data Dir: {data_dir}, Save Dir: {save_dir}\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 2. DATA PREPARATION ---\nprint(\"--- 2. DATA PREPARATION ---\")\nfull_dataset = datasets.ImageFolder(data_dir)\ndef create_disjoint_split(dataset, test_split=0.2):\n    base_name_map = {}\n    for idx, (path, _) in enumerate(dataset.samples):\n        base_name = re.split(r'[._]', os.path.basename(path))[0]\n        if base_name not in base_name_map: base_name_map[base_name] = []\n        base_name_map[base_name].append(idx)\n    unique_base_names = list(base_name_map.keys())\n    np.random.shuffle(unique_base_names)\n    split_idx = int(np.floor(test_split * len(unique_base_names)))\n    val_names, train_names = unique_base_names[:split_idx], unique_base_names[split_idx:]\n    train_indices = [idx for name in train_names for idx in base_name_map[name]]\n    val_indices = [idx for name in val_names for idx in base_name_map[name]]\n    return train_indices, val_indices\n\ntrain_idx, val_idx = create_disjoint_split(full_dataset)\ntrain_sampler = SubsetRandomSampler(train_idx)\nval_sampler = SubsetRandomSampler(val_idx)\n\nclass_counts = Counter(np.array(full_dataset.targets)[train_idx])\nweights = torch.tensor([len(train_idx) / class_counts[i] for i in range(len(class_counts))], dtype=torch.float32).to(device)\nprint(f\"Loss weights: {weights}\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 3. TRANSFORMS & DATALOADERS ---\nprint(\"--- 3. TRANSFORMS & DATALOADERS ---\")\ntransform_train = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.RandomHorizontalFlip(p=0.5),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n])\ntransform_val = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n])\n\nfull_dataset.transform = transform_train\nval_dataset = datasets.ImageFolder(data_dir, transform=transform_val)\ntrain_loader = DataLoader(full_dataset, batch_size=64, sampler=train_sampler, num_workers=2, pin_memory=True)\nval_loader = DataLoader(val_dataset, batch_size=64, sampler=val_sampler, num_workers=2, pin_memory=True)\nprint(\"DataLoaders created.\\n\" + \"-\"*30 + \"\\n\")\n\n# --- 4. MODEL & TRAINING ---\nprint(\"--- 4. MODEL & TRAINING ---\")\n\n# --- MODEL SETUP CHANGED ---\n# 1. Initialize with weights=None to train from scratch\n# 2. Make all parameters trainable by removing the freezing loops\nmodel = models.resnet18(weights=None)\nnum_ftrs = model.fc.in_features\nmodel.fc = nn.Sequential(\n    nn.Linear(num_ftrs, 512), nn.ReLU(),\n    nn.Dropout(0.5), nn.Linear(512, 2)\n)\nmodel = model.to(device)\n# --- END OF CHANGES ---\n\ndef train_model_loop(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs=30, patience=5):\n    best_val_acc = 0.0\n    best_model_wts = model.state_dict()\n    for epoch in range(num_epochs):\n        model.train()\n        train_pbar = tqdm(train_loader, desc=f\"Epoch {epoch+1}/{num_epochs} [Train]\")\n        for images, labels in train_pbar:\n            images, labels = images.to(device), labels.to(device)\n            optimizer.zero_grad()\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            loss.backward()\n            optimizer.step()\n        \n        # Get final training accuracy for the epoch\n        model.eval()\n        correct, total = 0, 0\n        with torch.no_grad():\n            for images, labels in train_loader:\n                images, labels = images.to(device), labels.to(device)\n                outputs = model(images)\n                _, predicted = torch.max(outputs, 1)\n                correct += (predicted == labels).sum().item()\n                total += labels.size(0)\n        train_acc = correct / total\n\n        # Validation\n        correct, total = 0, 0\n        with torch.no_grad():\n            for images, labels in val_loader:\n                images, labels = images.to(device), labels.to(device)\n                outputs = model(images)\n                _, predicted = torch.max(outputs, 1)\n                correct += (predicted == labels).sum().item()\n                total += labels.size(0)\n        val_acc = correct / total\n        \n        print(f\"Epoch {epoch+1}/{num_epochs} -> Train Acc: {train_acc:.4f} | Val Acc: {val_acc:.4f}\")\n        \n        if val_acc > best_val_acc:\n            best_val_acc = val_acc\n            best_model_wts = model.state_dict().copy()\n            print(f\"🚀 New best validation accuracy: {best_val_acc:.4f}. Saving model.\")\n            torch.save({'model_state_dict': best_model_wts}, os.path.join(save_dir, \"steganalyzer_from_scratch.pth\"))\n        \n        scheduler.step()\n    \n    model.load_state_dict(best_model_wts)\n    return model, best_val_acc\n\n# --- OPTIMIZER CHANGED ---\n# Use a slightly higher learning rate, which is common when training from scratch\ncriterion = nn.CrossEntropyLoss(weight=weights)\noptimizer = optim.Adam(model.parameters(), lr=0.001) # Higher LR\nscheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=15, eta_min=1e-6)\n\nmodel, val_acc = train_model_loop(model, train_loader, val_loader, criterion, optimizer, scheduler)\n\nprint(\"\\n--- Training Complete ---\")\nprint(f\"Final best validation accuracy: {val_acc:.4f}\")\nprint(f\"Model saved at {os.path.join(save_dir, 'steganalyzer_from_scratch.pth')}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:03:07.809961Z","iopub.execute_input":"2025-10-21T05:03:07.810700Z","iopub.status.idle":"2025-10-21T05:03:10.248759Z","shell.execute_reply.started":"2025-10-21T05:03:07.810664Z","shell.execute_reply":"2025-10-21T05:03:10.247786Z"},"scrolled":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#test code for the resnet18+png trained model\nimport os\nimport torch\nimport torch.nn as nn\nfrom torchvision import models, transforms\nfrom PIL import Image\nfrom tqdm.auto import tqdm\n\n# --- 1. CONFIGURATION ---\n# IMPORTANT: Update these paths before running\nMODEL_PATH = \"/kaggle/working/steg_model_from_scratch/steganalyzer_from_scratch.pth\"\nTEST_DIR   = \"/kaggle/input/stegoimagesdataset/train/train/stego\"\n\nDEVICE = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nCLASS_NAMES = ['clean', 'stego'] # 'clean' is class 0, 'stego' is 1\n# -------------------------\n\n\n# --- 2. MODEL DEFINITION ---\n# This function must EXACTLY match the architecture of the saved model\ndef create_resnet18_model(dropout_rate=0.5):\n    \"\"\"Creates the ResNet-18 architecture used during training.\"\"\"\n    # Start with an untrained ResNet-18 architecture\n    model = models.resnet18(weights=None)\n    \n    # Re-create the custom fully-connected (fc) layer\n    num_ftrs = model.fc.in_features\n    model.fc = nn.Sequential(\n        nn.Linear(num_ftrs, 512),\n        nn.ReLU(),\n        nn.Dropout(dropout_rate),\n        nn.Linear(512, 2)\n    )\n    return model.to(DEVICE)\n\n\n# --- 3. PREDICTION SCRIPT ---\ndef analyze_directory(model_path, test_dir):\n    \"\"\"\n    Loads a model and predicts the percentage of stego images in a directory.\n    \"\"\"\n    if not os.path.exists(model_path):\n        print(f\"Error: Model file not found at {model_path}\")\n        return\n\n    if not os.path.exists(test_dir):\n        print(f\"Error: Test directory not found at {test_dir}\")\n        return\n\n    print(\"--- Starting Analysis ---\")\n\n    # --- Load Model ---\n    model = create_resnet18_model(dropout_rate=0.5)\n    # The saved file is a dictionary, so we load the 'model_state_dict' key\n    checkpoint = torch.load(model_path, map_location=DEVICE)\n    model.load_state_dict(checkpoint['model_state_dict'])\n    model.eval() # Set model to evaluation mode\n    print(f\"Model loaded successfully from {model_path}\")\n\n    # --- Define Image Transformations ---\n    # These must match the validation transforms from your training script\n    transform = transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n    ])\n\n    # --- Predict on all images ---\n    stego_count = 0\n    image_files = [f for f in os.listdir(test_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))]\n    total_images = len(image_files)\n\n    if total_images == 0:\n        print(\"No valid image files found in the directory.\")\n        return\n\n    for fname in tqdm(image_files, desc=f\"Analyzing images in {os.path.basename(test_dir)}\"):\n        image_path = os.path.join(test_dir, fname)\n        try:\n            image = Image.open(image_path).convert('RGB')\n            tensor = transform(image).unsqueeze(0).to(DEVICE)\n\n            with torch.no_grad():\n                outputs = model(tensor)\n                # Apply softmax to get probabilities\n                probabilities = torch.nn.functional.softmax(outputs, dim=1)\n                # Get the prediction index (0 for 'clean', 1 for 'stego')\n                _, pred_idx = torch.max(outputs, 1)\n\n            if pred_idx.item() == 1: # Class 1 is 'stego'\n                stego_count += 1\n        except Exception as e:\n            print(f\"Skipping {fname} due to error: {e}\")\n    \n    # --- Report Results ---\n    stego_percentage = (stego_count / total_images) * 100 if total_images > 0 else 0\n    \n    print(\"\\n--- Analysis Complete ---\")\n    print(f\"Total images processed: {total_images}\")\n    print(f\"Images predicted as 'stego': {stego_count}\")\n    print(f\"Percentage of stego images: {stego_percentage:.2f}%\")\n    print(\"--------------------------\\n\")\n\n# --- RUN THE ANALYSIS ---\nanalyze_directory(MODEL_PATH, TEST_DIR)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:56:48.118994Z","iopub.execute_input":"2025-10-21T05:56:48.119670Z","iopub.status.idle":"2025-10-21T05:56:48.136011Z","shell.execute_reply.started":"2025-10-21T05:56:48.119640Z","shell.execute_reply":"2025-10-21T05:56:48.135187Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"##png model train code\n\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torchvision import datasets, transforms, models\nfrom torch.utils.data import DataLoader, SubsetRandomSampler\nimport numpy as np\nfrom PIL import Image\nimport matplotlib.pyplot as plt\n\n# Set seeds\ntorch.manual_seed(42)\ntorch.cuda.manual_seed_all(42)\nnp.random.seed(42)\n\n# Config\ndata_dir = \"/kaggle/input/stegoimagesdataset/train/train\"\nsave_dir = \"/kaggle/working/train_stego_pvd\"\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nos.makedirs(save_dir, exist_ok=True)\n\n# Verify dataset integrity\ndef verify_images(data_dir):\n    corrupt_files = []\n    for root, _, files in os.walk(data_dir):\n        for file in files:\n            try:\n                img = Image.open(os.path.join(root, file))\n                img.verify()\n            except Exception:\n                corrupt_files.append(os.path.join(root, file))\n    return corrupt_files\n\nprint(\"Checking dataset integrity...\")\ncorrupt_files = verify_images(data_dir)\nprint(f\"Corrupt files found: {len(corrupt_files)}\")\nif corrupt_files:\n    print(corrupt_files[:5])\nelse:\n    print(\"No corrupt files found.\")\n\n# Verify dataset\ndataset = datasets.ImageFolder(data_dir)\nprint(\"Dataset:\", os.listdir(data_dir))\nprint(\"Classes:\", dataset.class_to_idx)\n\n# Visualize sample images\ndef denormalize(image):\n    image = image.clone()\n    for i in range(3):\n        image[i] = image[i] * 0.229 + 0.485\n    return image\n\ntransform_vis = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n])\ndataset_vis = datasets.ImageFolder(data_dir, transform=transform_vis)\nvis_loader = DataLoader(dataset_vis, batch_size=4, shuffle=True, num_workers=0)\nfor images, labels in vis_loader:\n    plt.figure(figsize=(8, 3))\n    for i in range(4):\n        plt.subplot(1, 4, i+1)\n        img = denormalize(images[i])\n        plt.imshow(img.permute(1, 2, 0).numpy())\n        plt.title(f\"{'stego' if labels[i].item() == 1 else 'cover'}\")\n        plt.axis('off')\n    plt.show()\n    plt.close()\n    break\n\n# Transforms\ntransform_train = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.RandomHorizontalFlip(),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n])\ntransform_val = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),\n])\n\n# Dataset setup\ndataset = datasets.ImageFolder(data_dir, transform=transform_train)\ndataset_val = datasets.ImageFolder(data_dir, transform=transform_val)\ndataset_size = len(dataset)\nindices = list(range(dataset_size))\nnp.random.shuffle(indices)\nsplit = int(np.floor(0.2 * dataset_size))\ntrain_idx, val_idx = indices[split:], indices[:split]\ntrain_sampler = SubsetRandomSampler(train_idx)\nval_sampler = SubsetRandomSampler(val_idx)\ntrain_loader = DataLoader(dataset, batch_size=64, sampler=train_sampler, num_workers=0)\nval_loader = DataLoader(dataset_val, batch_size=64, sampler=val_sampler, num_workers=0)\n\n# Model setup\nmodel = models.resnet18(weights=\"IMAGENET1K_V1\")\nfor param in model.parameters():\n    param.requires_grad = False\nfor param in model.layer2.parameters():\n    param.requires_grad = True\nfor param in model.layer3.parameters():\n    param.requires_grad = True\nfor param in model.layer4.parameters():\n    param.requires_grad = True\nmodel.fc = nn.Sequential(\n    nn.Linear(model.fc.in_features, 512),\n    nn.ReLU(),\n    nn.Dropout(0.5),\n    nn.Linear(512, 2)\n)\nmodel = model.to(device)\n\n# Training function\ndef train_model(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs=30, patience=10):\n    best_val_acc = 0.0\n    best_model_wts = None\n    patience_counter = 0\n    for epoch in range(num_epochs):\n        model.train()\n        running_loss, correct, total = 0.0, 0, 0\n        for images, labels in train_loader:\n            images, labels = images.to(device), labels.to(device)\n            optimizer.zero_grad()\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            loss.backward()\n            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)\n            optimizer.step()\n            running_loss += loss.item() * images.size(0)\n            _, predicted = torch.max(outputs, 1)\n            correct += (predicted == labels).sum().item()\n            total += labels.size(0)\n        train_loss = running_loss / total\n        train_acc = correct / total\n        val_loss, correct, total = 0.0, 0, 0\n        with torch.no_grad():\n            for images, labels in val_loader:\n                images, labels = images.to(device), labels.to(device)\n                outputs = model(images)\n                loss = criterion(outputs, labels)\n                val_loss += loss.item() * images.size(0)\n                _, predicted = torch.max(outputs, 1)\n                correct += (predicted == labels).sum().item()\n                total += labels.size(0)\n        val_loss = val_loss / total\n        val_acc = correct / total\n        scheduler.step()\n        print(f\"Epoch {epoch+1}/{num_epochs} - Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, \"\n              f\"Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}\")\n        if val_acc > best_val_acc:\n            best_val_acc = val_acc\n            best_model_wts = model.state_dict().copy()\n            patience_counter = 0\n        else:\n            patience_counter += 1\n            if patience_counter >= patience:\n                print(f\"Early stopping at epoch {epoch+1}\")\n                break\n    model.load_state_dict(best_model_wts)\n    return model, best_val_acc\n\n# Train\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.0003, weight_decay=1e-4)\nscheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=10)\nmodel, val_acc = train_model(model, train_loader, val_loader, criterion, optimizer, scheduler)\ntorch.save({'model_state_dict': model.state_dict(), 'val_acc': val_acc}, os.path.join(save_dir, \"stego_pvd_trained_resnet18.pth\"))\n\n# Verify\nprint(f\"Trained model saved at {os.path.join(save_dir, 'stego_pvd_trained_resnet18.pth')}\")\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T08:56:24.328207Z","iopub.status.idle":"2025-07-21T08:56:24.328447Z","shell.execute_reply.started":"2025-07-21T08:56:24.328340Z","shell.execute_reply":"2025-07-21T08:56:24.328352Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#test code for png\nimport os\nimport torch\nimport torch.nn as nn\nimport torchvision.models as models\nimport torchvision.transforms as transforms\nfrom torch.utils.data import Dataset, DataLoader\nfrom PIL import Image\nimport numpy as np\nimport matplotlib.pyplot as plt\n\n# Custom Dataset for loading images\nclass ImageDataset(Dataset):\n    def __init__(self, image_dir, transform=None):\n        self.image_dir = image_dir\n        self.transform = transform\n        self.images = [os.path.join(image_dir, img) for img in os.listdir(image_dir) if img.endswith(('.png', '.jpg', '.jpeg'))]\n    \n    def __len__(self):\n        return len(self.images)\n    \n    def __getitem__(self, idx):\n        img_path = self.images[idx]\n        image = Image.open(img_path).convert('RGB')\n        if self.transform:\n            image = self.transform(image)\n        return image, img_path\n\n# CNN Model matching the saved model's architecture\nclass StegoClassifier(nn.Module):\n    def __init__(self):\n        super(StegoClassifier, self).__init__()\n        self.resnet = models.resnet18(weights=None)\n        num_ftrs = self.resnet.fc.in_features\n        self.resnet.fc = nn.Sequential(\n            nn.Linear(num_ftrs, 512),\n            nn.ReLU(),\n            nn.Dropout(0.5),\n            nn.Linear(512, 2)\n        )\n    \n    def forward(self, x):\n        return self.resnet(x)\n\ndef evaluate_model(image_dir, model_path, dataset_name, device='cuda' if torch.cuda.is_available() else 'cpu'):\n    # Define transforms\n    transform = transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n    ])\n    \n    # Load dataset\n    dataset = ImageDataset(image_dir, transform=transform)\n    if len(dataset) == 0:\n        print(f\"No images found in {image_dir}.\")\n        return 0, 0, []\n    \n    dataloader = DataLoader(dataset, batch_size=32, shuffle=False, num_workers=4, pin_memory=True)\n    \n    # Initialize model\n    model = StegoClassifier().to(device)\n    \n    # Load pretrained weights\n    if model_path and os.path.exists(model_path):\n        try:\n            checkpoint = torch.load(model_path, map_location=device)\n            state_dict = checkpoint.get('model_state_dict', checkpoint)\n            \n            # Remap state dictionary keys to match model structure\n            new_state_dict = {}\n            for key, value in state_dict.items():\n                if 'num_batches_tracked' in key:\n                    continue  # Skip non-parameter keys\n                if key.startswith('fc.0.'):\n                    new_key = key.replace('fc.0.', 'resnet.fc.0.')\n                elif key.startswith('fc.3.'):\n                    new_key = key.replace('fc.3.', 'resnet.fc.3.')\n                else:\n                    new_key = 'resnet.' + key\n                new_state_dict[new_key] = value\n            \n            model.load_state_dict(new_state_dict, strict=True)\n            print(f\"Loaded model from {model_path}.\")\n        except Exception as e:\n            print(f\"Error loading {model_path}: {e}\")\n            print(\"Cannot proceed without a valid model.\")\n            return 0, 0, []\n    else:\n        print(f\"Model path {model_path} not found. Cannot proceed.\")\n        return 0, 0, []\n    \n    model.eval()\n    \n    stego_count = 0\n    clean_count = 0\n    all_preds = []\n    all_confidences = []\n    \n    try:\n        with torch.no_grad():\n            for images, _ in dataloader:\n                images = images.to(device, non_blocking=True)\n                outputs = model(images)\n                probabilities = torch.softmax(outputs, dim=1)\n                confidences, predicted = torch.max(probabilities, 1)\n                stego_count += (predicted == 1).sum().item()\n                clean_count += (predicted == 0).sum().item()\n                all_preds.extend(predicted.cpu().numpy())\n                all_confidences.extend(confidences.cpu().numpy())\n    except KeyboardInterrupt:\n        print(\"Inference interrupted. Returning partial results.\")\n    \n    # Print results\n    total_images = stego_count + clean_count\n    print(f\"\\n{dataset_name} Results:\")\n    print(f\"Number of stego images detected: {stego_count}\")\n    print(f\"Number of clean images detected: {clean_count}\")\n    print(f\"Percentage of stego images: {(stego_count / total_images * 100) if total_images > 0 else 0:.2f}%\")\n    print(f\"Average confidence for predictions: {np.mean(all_confidences):.4f}\")\n    \n    # Plot confidence distribution\n    plt.figure(figsize=(8, 4))\n    plt.hist(all_confidences, bins=20, range=(0, 1), color='blue', alpha=0.7)\n    plt.title(f'Confidence Distribution - {dataset_name}')\n    plt.xlabel('Confidence Score')\n    plt.ylabel('Frequency')\n    plt.show()\n    \n    return stego_count, clean_count, all_preds\n\n# Example usage\nif __name__ == \"__main__\":\n    model_path = \"/kaggle/input/rasheed-models/stego_pvd_trained_resnet18.pth\"\n    lsb_test_dir = \"/kaggle/input/stego-pvd-dataset/Stego-pvd-dataset/test/stegoTest\"\n    \n    # Evaluate on LSB test set\n    if os.path.exists(lsb_test_dir):\n        lsb_stego, lsb_clean, lsb_preds = evaluate_model(lsb_test_dir, model_path, \"LSB Test\")\n    else:\n        print(f\"LSB test directory {lsb_test_dir} not found.\")\n        lsb_stego, lsb_clean, lsb_preds = 0, 0, []\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-10-21T05:57:08.955703Z","iopub.execute_input":"2025-10-21T05:57:08.956201Z","iopub.status.idle":"2025-10-21T05:57:17.804639Z","shell.execute_reply.started":"2025-10-21T05:57:08.956175Z","shell.execute_reply":"2025-10-21T05:57:17.803971Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Dataset Aggregation and Organization Script\n\nimport os\nimport shutil\nfrom pathlib import Path\n\n# Define output directory structure\noutput_base_dir = '/kaggle/working/stego_dataset'\nclean_base_dir = os.path.join(output_base_dir, 'clean')\nstego_base_dir = os.path.join(output_base_dir, 'stego')\nsubfolders = ['a', 'b', 'c', 'd', 'e', 'f', 'g']\n\n# Create directories\nfor subfolder in subfolders:\n    os.makedirs(os.path.join(clean_base_dir, subfolder), exist_ok=True)\n    os.makedirs(os.path.join(stego_base_dir, subfolder), exist_ok=True)\n\n# Function to copy images to a specific target directory\ndef copy_images(src_path, dst_dir, num_images, start_idx=0, match_filenames=None):\n    copied = 0\n    image_files = sorted([f for f in os.listdir(src_path) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])\n    \n    # If matching filenames are provided, filter by those\n    if match_filenames:\n        image_files = [f for f in image_files if f in match_filenames]\n    \n    # Select images starting from start_idx\n    for img_file in image_files[start_idx:start_idx + num_images]:\n        if copied >= num_images:\n            break\n        src_file = os.path.join(src_path, img_file)\n        dst_file = os.path.join(dst_dir, img_file)\n        shutil.copy2(src_file, dst_file)\n        copied += 1\n    \n    return copied\n\n# 1. LSB Dataset: 100 cover and 100 stego with matching filenames to clean/a and stego/a\nlsb_cover_dir = '/kaggle/input/lsb-stego/lsb_stego/lsb_stego/cover'\nlsb_stego_dir = '/kaggle/input/lsb-stego/lsb_stego/lsb_stego/lsb'\nlsb_clean_dst = os.path.join(clean_base_dir, 'a')\nlsb_stego_dst = os.path.join(stego_base_dir, 'a')\n\n# Get 100 cover images\ncover_files = sorted([f for f in os.listdir(lsb_cover_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])[:100]\ncopied_cover = copy_images(lsb_cover_dir, lsb_clean_dst, 100, match_filenames=cover_files)\ncopied_stego = copy_images(lsb_stego_dir, lsb_stego_dst, 100, match_filenames=cover_files)\n\nprint(f\"Copied {copied_cover} cover images to {lsb_clean_dst}\")\nprint(f\"Copied {copied_stego} stego images to {lsb_stego_dst}\")\n\n# 2. StegoImagesDataset: 300 clean and 300 stego to clean/b and stego/b\nval_clean_dir = '/kaggle/input/stegoimagesdataset/val/val/clean'\nval_stego_dir = '/kaggle/input/stegoimagesdataset/val/val/stego'\nval_clean_dst = os.path.join(clean_base_dir, 'b')\nval_stego_dst = os.path.join(stego_base_dir, 'b')\n\ncopied_clean = copy_images(val_clean_dir, val_clean_dst, 300)\ncopied_stego = copy_images(val_stego_dir, val_stego_dst, 300)\n\nprint(f\"Copied {copied_clean} clean images to {val_clean_dst}\")\nprint(f\"Copied {copied_stego} stego images to {val_stego_dst}\")\n\n# 3. Stego-PVD Dataset: 300 clean and 300 stego to clean/c and stego/c\npvd_clean_dir = '/kaggle/input/stego-pvd-dataset/Stego-pvd-dataset/train/cleanTrain'\npvd_stego_dir = '/kaggle/input/stego-pvd-dataset/Stego-pvd-dataset/train/stegoTrain'\npvd_clean_dst = os.path.join(clean_base_dir, 'c')\npvd_stego_dst = os.path.join(stego_base_dir, 'c')\n\ncopied_clean = copy_images(pvd_clean_dir, pvd_clean_dst, 300)\ncopied_stego = copy_images(pvd_stego_dir, pvd_stego_dst, 300)\n\nprint(f\"Copied {copied_clean} clean images to {pvd_clean_dst}\")\nprint(f\"Copied {copied_stego} stego images to {pvd_stego_dst}\")\n\n# 4. ALASKA2 Dataset: First 2000 cover and JMiPOD to clean/d and stego/d\nalaska_cover_dir = '/kaggle/input/alaska2-image-steganalysis/Cover'\nalaska_jmipod_dir = '/kaggle/input/alaska2-image-steganalysis/JMiPOD'\nalaska_clean_dst_d = os.path.join(clean_base_dir, 'd')\nalaska_stego_dst_d = os.path.join(stego_base_dir, 'd')\n\ncopied_cover = copy_images(alaska_cover_dir, alaska_clean_dst_d, 2000, start_idx=0)\ncopied_stego = copy_images(alaska_jmipod_dir, alaska_stego_dst_d, 2000, start_idx=0)\n\nprint(f\"Copied {copied_cover} cover images to {alaska_clean_dst_d}\")\nprint(f\"Copied {copied_stego} JMiPOD stego images to {alaska_stego_dst_d}\")\n\n# 5. ALASKA2 Dataset: Third 2000 cover and JUNIWARD to clean/e and stego/e\nalaska_juniward_dir = '/kaggle/input/alaska2-image-steganalysis/JUNIWARD'\nalaska_clean_dst_e = os.path.join(clean_base_dir, 'e')\nalaska_stego_dst_e = os.path.join(stego_base_dir, 'e')\n\ncopied_cover = copy_images(alaska_cover_dir, alaska_clean_dst_e, 2000, start_idx=4000)\ncopied_stego = copy_images(alaska_juniward_dir, alaska_stego_dst_e, 2000, start_idx=4000)\n\nprint(f\"Copied {copied_cover} cover images to {alaska_clean_dst_e}\")\nprint(f\"Copied {copied_stego} JUNIWARD stego images to {alaska_stego_dst_e}\")\n\n# 6. ALASKA2 Dataset: Fourth 2000 cover and UERD to clean/f and stego/f\nalaska_uerd_dir = '/kaggle/input/alaska2-image-steganalysis/UERD'\nalaska_clean_dst_f = os.path.join(clean_base_dir, 'f')\nalaska_stego_dst_f = os.path.join(stego_base_dir, 'f')\n\ncopied_cover = copy_images(alaska_cover_dir, alaska_clean_dst_f, 2000, start_idx=6000)\ncopied_stego = copy_images(alaska_uerd_dir, alaska_stego_dst_f, 2000, start_idx=6000)\n\nprint(f\"Copied {copied_cover} cover images to {alaska_clean_dst_f}\")\nprint(f\"Copied {copied_stego} UERD stego images to {alaska_stego_dst_f}\")\n\n# 7. Stego-Dataset: 400 cover and 400 F5 to clean/g and stego/g\nstego_dataset_cover_dir = '/kaggle/input/stego-dataset/data_Set/train/cover'\nstego_dataset_f5_dir = '/kaggle/input/stego-dataset/data_Set/train/f5'\nstego_clean_dst = os.path.join(clean_base_dir, 'g')\nstego_stego_dst = os.path.join(stego_base_dir, 'g')\n\ncopied_cover = copy_images(stego_dataset_cover_dir, stego_clean_dst, 400, start_idx=0)\ncopied_stego = copy_images(stego_dataset_f5_dir, stego_stego_dst, 400, start_idx=0)\n\nprint(f\"Copied {copied_cover} cover images to {stego_clean_dst}\")\nprint(f\"Copied {copied_stego} F5 stego images to {stego_stego_dst}\")\n\n# Verify total images\ntotal_clean = sum(len(os.listdir(os.path.join(clean_base_dir, sub))) for sub in subfolders)\ntotal_stego = sum(len(os.listdir(os.path.join(stego_base_dir, sub))) for sub in subfolders)\n\nprint(f\"\\nTotal Clean Images Copied: {total_clean}\")\nprint(f\"Total Stego Images Copied: {total_stego}\")\nprint(f\"Images saved in: {output_base_dir}/clean/[a-g] and {output_base_dir}/stego/[a-g]\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T17:45:37.691735Z","iopub.execute_input":"2025-07-21T17:45:37.691937Z","iopub.status.idle":"2025-07-21T17:48:10.469127Z","shell.execute_reply.started":"2025-07-21T17:45:37.691919Z","shell.execute_reply":"2025-07-21T17:48:10.468534Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#resnet18+ using the massive, combined dataset train code\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision import models, transforms\nfrom sklearn.metrics import accuracy_score, precision_recall_fscore_support, confusion_matrix\nimport numpy as np\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport seaborn as sns\n\n# Custom Dataset for Clean and Stego Images\nclass StegoDataset(Dataset):\n    def __init__(self, clean_dirs, stego_dirs, transform=None):\n        self.clean_dirs = clean_dirs\n        self.stego_dirs = stego_dirs\n        self.transform = transform\n        self.all_images = []\n        self.labels = []\n\n        # Load clean images (label 0)\n        for clean_dir in clean_dirs:\n            clean_images = [os.path.join(clean_dir, f) for f in os.listdir(clean_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))]\n            self.all_images.extend(clean_images)\n            self.labels.extend([0] * len(clean_images))\n\n        # Load stego images (label 1)\n        for stego_dir in stego_dirs:\n            stego_images = [os.path.join(stego_dir, f) for f in os.listdir(stego_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))]\n            self.all_images.extend(stego_images)\n            self.labels.extend([1] * len(stego_images))\n\n    def __len__(self):\n        return len(self.all_images)\n\n    def __getitem__(self, idx):\n        img_path = self.all_images[idx]\n        label = self.labels[idx]\n        image = Image.open(img_path).convert('RGB')\n        if self.transform:\n            image = self.transform(image)\n        return image, label\n\n# Data Transforms\ndata_transforms = {\n    'train': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.RandomHorizontalFlip(),\n        transforms.RandomRotation(10),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n    'val': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n}\n\n# Define directory paths\nbase_dir = '/kaggle/working/stego_dataset'\nsubfolders = ['a', 'b', 'c', 'd', 'e', 'f', 'g']\nclean_dirs = [os.path.join(base_dir, 'clean', sub) for sub in subfolders]\nstego_dirs = [os.path.join(base_dir, 'stego', sub) for sub in subfolders]\n\n# Create dataset\ndataset = StegoDataset(clean_dirs, stego_dirs, transform=data_transforms['train'])\n\n# Split dataset into train and validation\ntrain_size = int(0.8 * len(dataset))\nval_size = len(dataset) - train_size\ntrain_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])\nval_dataset.dataset.transform = data_transforms['val']\n\ntrain_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, num_workers=2)\nval_loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=2)\n\n# Load Pretrained ResNet18\nmodel = models.resnet18(pretrained=True)\nnum_ftrs = model.fc.in_features\nmodel.fc = nn.Linear(num_ftrs, 2)  # Binary classification (cover vs stego)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nmodel = model.to(device)\n\n# Loss Function and Optimizer\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.001)\nscheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)\n\n# Training Loop\nnum_epochs = 32\ntrain_losses, val_losses = [], []\ntrain_accuracies, val_accuracies = [], []\n\nprint(\"Starting Training...\")\nprint(\"-\" * 50)\n\nfor epoch in range(num_epochs):\n    # Training\n    model.train()\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    for images, labels in train_loader:\n        images, labels = images.to(device), labels.to(device)\n        optimizer.zero_grad()\n        outputs = model(images)\n        loss = criterion(outputs, labels)\n        loss.backward()\n        optimizer.step()\n        running_loss += loss.item()\n        _, predicted = torch.max(outputs.data, 1)\n        total += labels.size(0)\n        correct += (predicted == labels).sum().item()\n    train_loss = running_loss / len(train_loader)\n    train_acc = 100 * correct / total\n    train_losses.append(train_loss)\n    train_accuracies.append(train_acc)\n\n    # Validation\n    model.eval()\n    val_loss = 0.0\n    correct = 0\n    total = 0\n    all_preds, all_labels = [], []\n    with torch.no_grad():\n        for images, labels in val_loader:\n            images, labels = images.to(device), labels.to(device)\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            val_loss += loss.item()\n            _, predicted = torch.max(outputs.data, 1)\n            total += labels.size(0)\n            correct += (predicted == labels).sum().item()\n            all_preds.extend(predicted.cpu().numpy())\n            all_labels.extend(labels.cpu().numpy())\n    val_loss = val_loss / len(val_loader)\n    val_acc = 100 * correct / total\n    val_losses.append(val_loss)\n    val_accuracies.append(val_acc)\n\n    print(f\"Epoch [{epoch+1}/{num_epochs}]\")\n    print(f\"Train Loss: {train_loss:.4f}, Train Accuracy: {train_acc:.2f}%\")\n    print(f\"Val Loss: {val_loss:.4f}, Val Accuracy: {val_acc:.2f}%\")\n    print(\"-\" * 50)\n\n    scheduler.step()\n\n# Save the Model\nmodel_save_path = '/kaggle/working/stego_model.pth'\ntorch.save(model.state_dict(), model_save_path)\n\n# Evaluation Metrics\nprecision, recall, f1, _ = precision_recall_fscore_support(all_labels, all_preds, average='binary')\ncm = confusion_matrix(all_labels, all_preds)\n\n# Display Final Report\nprint(\"\\nTraining Completed!\")\nprint(\"=\" * 50)\nprint(\"Final Steganalysis Model Report\")\nprint(\"=\" * 50)\nprint(f\"Total Epochs Trained: {num_epochs}\")\nprint(f\"Total Images: {len(dataset)} (Train: {train_size}, Validation: {val_size})\")\nprint(f\"Final Training Accuracy: {train_accuracies[-1]:.2f}%\")\nprint(f\"Final Validation Accuracy: {val_accuracies[-1]:.2f}%\")\nprint(f\"Final Training Loss: {train_losses[-1]:.4f}\")\nprint(f\"Final Validation Loss: {val_losses[-1]:.4f}\")\nprint(\"\\nEvaluation Metrics (Validation Set):\")\nprint(f\"Accuracy: {val_accuracies[-1]:.2f}%\")\nprint(f\"Precision: {precision:.4f}\")\nprint(f\"Recall: {recall:.4f}\")\nprint(f\"F1 Score: {f1:.4f}\")\nprint(\"\\nConfusion Matrix:\")\nprint(f\"{'':<15} | {'Predicted Cover':<15} | {'Predicted Stego':<15}\")\nprint(f\"{'-'*15}-+-{'-'*15}-+-{'-'*15}\")\nprint(f\"{'Actual Cover':<15} | {cm[0,0]:<15} | {cm[0,1]:<15}\")\nprint(f\"{'Actual Stego':<15} | {cm[1,0]:<15} | {cm[1,1]:<15}\")\nprint(f\"\\nModel Saved at: {model_save_path}\")\n\n# Plot Training and Validation Metrics\nplt.figure(figsize=(12, 5))\nplt.subplot(1, 2, 1)\nplt.plot(train_losses, label='Train Loss', color='#1f77b4')\nplt.plot(val_losses, label='Val Loss', color='#ff7f0e')\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.title('Training and Validation Loss')\nplt.legend()\n\nplt.subplot(1, 2, 2)\nplt.plot(train_accuracies, label='Train Accuracy', color='#1f77b4')\nplt.plot(val_accuracies, label='Val Accuracy', color='#ff7f0e')\nplt.xlabel('Epoch')\nplt.ylabel('Accuracy (%)')\nplt.title('Training and Validation Accuracy')\nplt.legend()\nplt.show()\n\n# Plot Confusion Matrix\nplt.figure(figsize=(6, 5))\nsns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=['Cover', 'Stego'], yticklabels=['Cover', 'Stego'])\nplt.xlabel('Predicted')\nplt.ylabel('Actual')\nplt.title('Confusion Matrix')\nplt.show()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T17:49:12.289412Z","iopub.execute_input":"2025-07-21T17:49:12.289872Z","iopub.status.idle":"2025-07-21T18:16:30.214787Z","shell.execute_reply.started":"2025-07-21T17:49:12.289850Z","shell.execute_reply":"2025-07-21T18:16:30.213914Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#resnet18+alaska 2 dt train code\nimport os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import DataLoader, Dataset\nfrom torchvision import models, transforms\nfrom sklearn.metrics import accuracy_score, precision_recall_fscore_support, confusion_matrix\nimport numpy as np\nfrom PIL import Image\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom torch.amp import GradScaler, autocast  # Updated for PyTorch >= 2.0\n\n# Custom Dataset for ALASKA2 Images\nclass StegoDataset(Dataset):\n    def __init__(self, cover_dir, stego_dirs, num_images_per_subset, transform=None):\n        self.cover_dir = cover_dir\n        self.stego_dirs = stego_dirs\n        self.num_images_per_subset = num_images_per_subset\n        self.transform = transform\n        self.all_images = []\n        self.labels = []\n\n        # Load cover images (label 0) from Cover directory\n        cover_files = sorted([f for f in os.listdir(cover_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])\n        \n        # First 10k Cover\n        self.all_images.extend([os.path.join(cover_dir, f) for f in cover_files[0:num_images_per_subset]])\n        self.labels.extend([0] * min(num_images_per_subset, len(cover_files[0:num_images_per_subset])))\n        \n        # Second 10k Cover\n        self.all_images.extend([os.path.join(cover_dir, f) for f in cover_files[25000:25000 + num_images_per_subset]])\n        self.labels.extend([0] * min(num_images_per_subset, len(cover_files[25000:25000 + num_images_per_subset])))\n        \n        # Third 10k Cover\n        self.all_images.extend([os.path.join(cover_dir, f) for f in cover_files[50000:50000 + num_images_per_subset]])\n        self.labels.extend([0] * min(num_images_per_subset, len(cover_files[50000:50000 + num_images_per_subset])))\n\n        # Load stego images (label 1) from JMiPOD, JUNIWARD, UERD\n        for stego_dir, start_idx in zip(stego_dirs, [0, 25000, 50000]):\n            stego_files = sorted([f for f in os.listdir(stego_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg'))])\n            self.all_images.extend([os.path.join(stego_dir, f) for f in stego_files[start_idx:start_idx + num_images_per_subset]])\n            self.labels.extend([1] * min(num_images_per_subset, len(stego_files[start_idx:start_idx + num_images_per_subset])))\n\n        # Verify balance\n        print(f\"Dataset: {len(self.all_images)} images ({sum(self.labels)} stego, {len(self.all_images) - sum(self.labels)} cover)\")\n\n    def __len__(self):\n        return len(self.all_images)\n\n    def __getitem__(self, idx):\n        img_path = self.all_images[idx]\n        label = self.labels[idx]\n        try:\n            image = Image.open(img_path).convert('RGB')\n        except Exception as e:\n            print(f\"Error loading image {img_path}: {e}\")\n            # Return a blank image and label to avoid crashing\n            image = Image.new('RGB', (224, 224), (0, 0, 0))\n            label = 0\n        if self.transform:\n            image = self.transform(image)\n        return image, label\n\n# Data Transforms\ndata_transforms = {\n    'train': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.RandomHorizontalFlip(),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n    'val': transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.ToTensor(),\n        transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])\n    ]),\n}\n\n# Define directory paths\ncover_dir = '/kaggle/input/alaska2-image-steganalysis/Cover'\nstego_dirs = [\n    '/kaggle/input/alaska2-image-steganalysis/JMiPOD',\n    '/kaggle/input/alaska2-image-steganalysis/JUNIWARD',\n    '/kaggle/input/alaska2-image-steganalysis/UERD'\n]\nnum_images_per_subset = 10000  # 10k per subset for speed\n\n# Create dataset\ndataset = StegoDataset(cover_dir, stego_dirs, num_images_per_subset, transform=data_transforms['train'])\n\n# Split dataset into train and validation\ntrain_size = int(0.8 * len(dataset))\nval_size = len(dataset) - train_size\ntrain_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])\nval_dataset.dataset.transform = data_transforms['val']\n\ntrain_loader = DataLoader(train_dataset, batch_size=64, shuffle=True, num_workers=4, pin_memory=True)\nval_loader = DataLoader(val_dataset, batch_size=64, shuffle=False, num_workers=4, pin_memory=True)\n\n# Define Model\nmodel = models.resnet18(pretrained=True)\n# Unfreeze last two layer blocks (layer3 and layer4) and fc layer\nfor param in model.parameters():\n    param.requires_grad = False\nfor param in model.layer3.parameters():\n    param.requires_grad = True\nfor param in model.layer4.parameters():\n    param.requires_grad = True\nfor param in model.fc.parameters():\n    param.requires_grad = True\nnum_ftrs = model.fc.in_features\nmodel.fc = nn.Linear(num_ftrs, 2)  # Binary classification (cover vs stego)\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nmodel = model.to(device)\n\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=0.001)\nscheduler = optim.lr_scheduler.StepLR(optimizer, step_size=8, gamma=0.1)\nscaler = GradScaler('cuda')  # Updated for PyTorch >= 2.0\n\n# Training Loop\nnum_epochs = 18\ntrain_losses, val_losses = [], []\ntrain_accuracies, val_accuracies = [], []\n\nprint(\"Starting Training...\")\nprint(\"-\" * 50)\n\nfor epoch in range(num_epochs):\n    # Training\n    model.train()\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    for images, labels in train_loader:\n        images, labels = images.to(device), labels.to(device)\n        optimizer.zero_grad()\n        with autocast('cuda'):  # Updated for PyTorch >= 2.0\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n        scaler.scale(loss).backward()\n        scaler.step(optimizer)\n        scaler.update()\n        running_loss += loss.item()\n        _, predicted = torch.max(outputs.data, 1)\n        total += labels.size(0)\n        correct += (predicted == labels).sum().item()\n    train_loss = running_loss / len(train_loader)\n    train_acc = 100 * correct / total\n    train_losses.append(train_loss)\n    train_accuracies.append(train_acc)\n\n    # Validation\n    model.eval()\n    val_loss = 0.0\n    correct = 0\n    total = 0\n    all_preds, all_labels = [], []\n    with torch.no_grad():\n        for images, labels in val_loader:\n            images, labels = images.to(device), labels.to(device)\n            with autocast('cuda'):  # Updated for PyTorch >= 2.0\n                outputs = model(images)\n                loss = criterion(outputs, labels)\n            val_loss += loss.item()\n            _, predicted = torch.max(outputs.data, 1)\n            total += labels.size(0)\n            correct += (predicted == labels).sum().item()\n            all_preds.extend(predicted.cpu().numpy())\n            all_labels.extend(labels.cpu().numpy())\n    val_loss = val_loss / len(val_loader)\n    val_acc = 100 * correct / total\n    val_losses.append(val_loss)\n    val_accuracies.append(val_acc)\n\n    print(f\"Epoch [{epoch+1}/{num_epochs}]\")\n    print(f\"Train Loss: {train_loss:.4f}, Train Accuracy: {train_acc:.2f}%\")\n    print(f\"Val Loss: {val_loss:.4f}, Val Accuracy: {val_acc:.2f}%\")\n    print(\"-\" * 50)\n\n    scheduler.step()\n\n# Save the Model\nmodel_save_path = '/kaggle/working/stego_model.pth'\ntorch.save(model.state_dict(), model_save_path)\n\n# Evaluation Metrics\nprecision, recall, f1, _ = precision_recall_fscore_support(all_labels, all_preds, average='binary')\ncm = confusion_matrix(all_labels, all_preds)\n\n# Display Final Report\nprint(\"\\nTraining Completed!\")\nprint(\"=\" * 50)\nprint(\"Final Steganalysis Model Report\")\nprint(\"=\" * 50)\nprint(f\"Total Epochs Trained: {num_epochs}\")\nprint(f\"Total Images: {len(dataset)} (Train: {train_size}, Validation: {val_size})\")\nprint(f\"Final Training Accuracy: {train_accuracies[-1]:.2f}%\")\nprint(f\"Final Validation Accuracy: {val_accuracies[-1]:.2f}%\")\nprint(f\"Final Training Loss: {train_losses[-1]:.4f}\")\nprint(f\"Final Validation Loss: {val_losses[-1]:.4f}\")\nprint(\"\\nEvaluation Metrics (Validation Set):\")\nprint(f\"Accuracy: {val_accuracies[-1]:.2f}%\")\nprint(f\"Precision: {precision:.4f}\")\nprint(f\"Recall: {recall:.4f}\")\nprint(f\"F1 Score: {f1:.4f}\")\nprint(\"\\nConfusion Matrix:\")\nprint(f\"{'':<15} | {'Predicted Cover':<15} | {'Predicted Stego':<15}\")\nprint(f\"{'-'*15}-+-{'-'*15}-+-{'-'*15}\")\nprint(f\"{'Actual Cover':<15} | {cm[0,0]:<15} | {cm[0,1]:<15}\")\nprint(f\"{'Actual Stego':<15} | {cm[1,0]:<15} | {cm[1,1]:<15}\")\nprint(f\"\\nModel Saved at: {model_save_path}\")\n\n# Plot Training and Validation Metrics\nplt.figure(figsize=(12, 5))\nplt.subplot(1, 2, 1)\nplt.plot(train_losses, label='Train Loss', color='#1f77b4')\nplt.plot(val_losses, label='Val Loss', color='#ff7f0e')\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.title('Training and Validation Loss')\nplt.legend()\n\nplt.subplot(1, 2, 2)\nplt.plot(train_accuracies, label='Train Accuracy', color='#1f77b4')\nplt.plot(val_accuracies, label='Val Accuracy', color='#ff7f0e')\nplt.xlabel('Epoch')\nplt.ylabel('Accuracy (%)')\nplt.title('Training and Validation Accuracy')\nplt.legend()\nplt.show()\n\n# Plot Confusion Matrix\nplt.figure(figsize=(6, 5))\nsns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=['Cover', 'Stego'], yticklabels=['Cover', 'Stego'])\nplt.xlabel('Predicted')\nplt.ylabel('Actual')\nplt.title('Confusion Matrix')\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T19:26:58.288687Z","iopub.execute_input":"2025-07-21T19:26:58.289467Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#lr train code\nimport numpy as np\nimport os\nfrom skimage.io import imread\nfrom skimage.color import rgb2gray\nfrom skimage.feature import hog, local_binary_pattern\nfrom PIL import Image\nfrom tqdm import tqdm\nimport pywt\n\n\ndef extract_hog(img):\n    return hog(img, orientations=9, pixels_per_cell=(8, 8),\n               cells_per_block=(2, 2), block_norm='L2-Hys', visualize=False)\n\ndef extract_lbp(img):\n    lbp = local_binary_pattern(img, P=8, R=1, method='uniform')\n    (hist, _) = np.histogram(lbp.ravel(), bins=np.arange(0, 10), density=True)\n    return hist\n\ndef extract_bitplane_stats(img):\n    bit_planes = [(img >> i) & 1 for i in range(8)]\n    return np.array([np.mean(p) for p in bit_planes] + [np.std(p) for p in bit_planes])\n\ndef extract_color_stats(img_rgb):\n    means = np.mean(img_rgb, axis=(0, 1))\n    stds = np.std(img_rgb, axis=(0, 1))\n    return np.concatenate([means, stds])\n\ndef extract_wavelet(img):\n    coeffs = pywt.dwt2(img, 'haar')\n    cA, (cH, cV, cD) = coeffs\n    return np.concatenate([cA.ravel()[:100], cH.ravel()[:100], cV.ravel()[:100], cD.ravel()[:100]])\n\nclean_path = \"/kaggle/input/stegoimagesdataset/test/test/clean\"\nstego_path = \"/kaggle/input/stegoimagesdataset/test/test/stego\"\nclean_images = sorted(os.listdir(clean_path))\nstego_images = sorted(os.listdir(stego_path))\n\nfeatures_clean = []\nfeatures_stego = []\n\nprint(\"Extracting clean image features...\")\nfor img in tqdm(clean_images):\n    feat = extract_all_features(os.path.join(clean_path, img))\n    if feat is not None:\n        features_clean.append(feat)\nfeatures_clean = np.array(features_clean)\nnp.save(\"/kaggle/working/features_clean.npy\", features_clean)\nprint(\"✅ Features saved. Clean:\", features_clean.shape)\n\nprint(\"Extracting stego image features...\")\nfor img in tqdm(stego_images):\n    feat = extract_all_features(os.path.join(stego_path, img))\n    if feat is not None:\n        features_stego.append(feat)\n\nfeatures_stego = np.array(features_stego)\nnp.save(\"/kaggle/working/features_stego.npy\", features_stego)\n\nprint(\"✅ Features saved. Stego:\", features_stego.shape)\n\n\nimport numpy as np\n\n# Load feature file to verify shape\nfeatures_clean = np.load(\"/kaggle/input/features-for-stego/features_clean.npy\")\nprint(\"Feature vector shape:\", features_clean.shape)\nprint(\"Number of features per image:\", features_clean.shape[1])","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#lr test code\nimport os\nimport numpy as np\nimport joblib\nfrom tqdm import tqdm\nfrom joblib import Parallel, delayed\n\nimport pywt\nfrom skimage.color import rgb2gray\nfrom skimage.feature import hog, local_binary_pattern\nfrom skimage.io import imread\nfrom PIL import Image\n\n# ── 1. Feature Extraction Function (same as used in training) ─────────────\ndef extract_all_features(image_path):\n    try:\n        img = imread(image_path)\n        if img is None:\n            return None\n        if img.ndim == 2:\n            gray = img\n            rgb = np.stack((img,) * 3, axis=-1)\n        elif img.shape[2] == 4:\n            img = np.array(Image.open(image_path).convert(\"RGB\"))\n            gray = rgb2gray(img)\n            rgb = img\n        else:\n            gray = rgb2gray(img)\n            rgb = img\n\n        gray = (gray * 255).astype(np.uint8)\n\n        # HOG\n        hog_feat = hog(gray, orientations=9, pixels_per_cell=(8, 8),\n                       cells_per_block=(2, 2), block_norm='L2-Hys', visualize=False)\n\n        # LBP\n        lbp = local_binary_pattern(gray, P=8, R=1, method='uniform')\n        lbp_feat, _ = np.histogram(lbp.ravel(), bins=np.arange(0, 10), density=True)\n\n        # Bit-planes\n        bitplanes = [(gray >> i) & 1 for i in range(8)]\n        bp_feat = np.array([np.mean(b) for b in bitplanes] +\n                           [np.std(b) for b in bitplanes])\n\n        # Color stats\n        means = np.mean(rgb, axis=(0, 1))\n        stds = np.std(rgb, axis=(0, 1))\n        color_feat = np.concatenate([means, stds])\n\n        # Wavelet\n        cA, (cH, cV, cD) = pywt.dwt2(gray, 'haar')\n        wav_feat = np.concatenate([\n            cA.ravel()[:100], cH.ravel()[:100],\n            cV.ravel()[:100], cD.ravel()[:100]\n        ])\n\n        return np.concatenate([hog_feat, lbp_feat, bp_feat, color_feat, wav_feat])\n    except:\n        return None\n\n# ── 2. Load Model ─────────────────────────────────────────────────────────\nmodel_path = \"/kaggle/input/batch-70/alaska_sgd_lr_pipeline1.pkl\"  # ✅ Change if needed\npipeline = joblib.load(model_path)\nprint(f\"✅ Loaded trained model: {model_path}\")\n\n# ── 3. Predict on a Given Directory ───────────────────────────────────────\ndef test_on_directory(folder_path, n_jobs=4):\n    print(f\"\\n🔍 Scanning directory: {folder_path}\")\n    valid_exts = ('.jpg', '.jpeg', '.png', '.bmp', '.tif', '.tiff', '.pgm')\n    image_paths = [os.path.join(folder_path, f)\n                   for f in os.listdir(folder_path)\n                   if f.lower().endswith(valid_exts)]\n    \n    if not image_paths:\n        print(\"⚠️ No valid images found.\")\n        return\n\n    print(f\"📸 Found {len(image_paths)} images\")\n\n    # Extract features\n    features = Parallel(n_jobs=n_jobs, backend=\"threading\")(\n        delayed(extract_all_features)(img_path) for img_path in tqdm(image_paths)\n    )\n    features = [f for f in features if f is not None]\n    print(f\"✅ Extracted features from {len(features)} images\")\n\n    # Predict\n    preds = pipeline.predict(np.vstack(features))\n    stego = np.sum(preds == 1)\n    clean = np.sum(preds == 0)\n\n    print(\"\\n📊 Result Summary:\")\n    print(f\"Total Images       : {len(preds)}\")\n    print(f\"Predicted as Stego : {stego}\")\n    print(f\"Predicted as Clean : {clean}\")\n\n# ── 4. Call with Your Directory Path ──────────────────────────────────────\n# ⚠️ Replace this with your real test directory path\ntest_dir = \"/kaggle/input/stegoimagesdataset/train/train/stego\"  \ntest_on_directory(test_dir)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}