{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":19991,"databundleVersionId":1117522,"sourceType":"competition"}],"dockerImageVersionId":30919,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install pennylane","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:50.160607Z","iopub.execute_input":"2025-03-19T17:20:50.161026Z","iopub.status.idle":"2025-03-19T17:20:53.688808Z","shell.execute_reply.started":"2025-03-19T17:20:50.160975Z","shell.execute_reply":"2025-03-19T17:20:53.687897Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 1. Import Libraries and Define Global Variables\n# =============================================================================\nimport os\nimport cv2\nimport csv\nimport glob\nimport numpy as np\nimport tensorflow as tf\nfrom tensorflow.keras import layers, models\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\nfrom PIL import Image, ExifTags\nimport pennylane as qml\n\n# Set random seed for reproducibility\nnp.random.seed(42)\ntf.random.set_seed(42)\n\n# Directories for training data and test set\nTRAIN_DIRS = {\n    \"Cover\": \"data/Cover\",         # Clean images\n    \"JMiPOD\": \"data/JMiPOD\",       # Stego images via JMiPOD\n    \"JUNIWARD\": \"data/JUNIWARD\",   # Stego images via JUNIWARD\n    \"UERD\": \"data/UERD\"            # Stego images via UERD\n}\nTEST_DIR = \"data/Test\"             # Test images for prediction","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:53.690411Z","iopub.execute_input":"2025-03-19T17:20:53.690694Z","iopub.status.idle":"2025-03-19T17:20:54.908863Z","shell.execute_reply.started":"2025-03-19T17:20:53.690660Z","shell.execute_reply":"2025-03-19T17:20:54.908046Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 2. Feature Extraction Functions\n# =============================================================================\ndef extract_features(image_path):\n    \"\"\"\n    Extract features from an image:\n      - Pixel values (normalized grayscale image).\n      - Entropy computed from the intensity histogram.\n      - Frequency components using Discrete Cosine Transform (DCT).\n      - Metadata extraction via PIL (if available).\n    \"\"\"\n    # Read image using OpenCV\n    image = cv2.imread(image_path)\n    if image is None:\n        raise ValueError(f\"Image not found or unable to load: {image_path}\")\n    # Convert to grayscale\n    image_gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n    \n    # --- Pixel Values ---\n    pixel_values = image_gray.flatten() / 255.0\n\n    # --- Entropy Calculation ---\n    hist, _ = np.histogram(image_gray, bins=256, range=(0, 256), density=True)\n    hist = hist[hist > 0]\n    entropy = -np.sum(hist * np.log2(hist))\n    \n    # --- Frequency Components using DCT ---\n    dct = cv2.dct(np.float32(image_gray))\n    \n    # --- Metadata Extraction ---\n    try:\n        pil_img = Image.open(image_path)\n        exif_data = pil_img._getexif()\n        metadata = {}\n        if exif_data is not None:\n            for tag, value in exif_data.items():\n                decoded = ExifTags.TAGS.get(tag, tag)\n                metadata[decoded] = value\n        else:\n            metadata = {}\n    except Exception as e:\n        metadata = {}\n    \n    return {\"pixel_values\": pixel_values, \"entropy\": entropy, \"dct\": dct, \"metadata\": metadata}\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:54.910647Z","iopub.execute_input":"2025-03-19T17:20:54.911266Z","iopub.status.idle":"2025-03-19T17:20:54.917209Z","shell.execute_reply.started":"2025-03-19T17:20:54.911241Z","shell.execute_reply":"2025-03-19T17:20:54.916400Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\n# =============================================================================\n# 3. Quantum-Inspired CNN for Stego Detection (Binary Classification)\n# =============================================================================\ndef build_quantum_inspired_cnn(input_shape):\n    \"\"\"\n    Builds a CNN model (using Keras) that mimics quantum-inspired feature extraction.\n    This model is trained to perform binary classification:\n      - 0: Cover (Clean)\n      - 1: Stego (hidden message embedded)\n    \"\"\"\n    model = models.Sequential([\n        layers.Input(shape=input_shape),\n        layers.Conv2D(32, (3, 3), activation='relu'),\n        layers.MaxPooling2D((2, 2)),\n        layers.Conv2D(64, (3, 3), activation='relu'),\n        layers.MaxPooling2D((2, 2)),\n        layers.Flatten(),\n        layers.Dense(64, activation='relu'),\n        layers.Dense(1, activation='sigmoid')  # Output: probability of stego\n    ])\n    model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])\n    return model\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:54.918593Z","iopub.execute_input":"2025-03-19T17:20:54.918896Z","iopub.status.idle":"2025-03-19T17:20:54.945099Z","shell.execute_reply.started":"2025-03-19T17:20:54.918869Z","shell.execute_reply":"2025-03-19T17:20:54.944381Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 4. Variational Quantum Classifier (VQC) for Steganography Algorithm Classification\n# =============================================================================\ndef create_vqc(num_qubits=4, num_layers=2):\n    \"\"\"\n    Creates a Variational Quantum Classifier using Pennylane.\n    This circuit encodes a classical 4-dimensional input into a quantum state,\n    applies variational rotations (with RX) and entangling CNOTs, and finally\n    measures expectation values which are passed to a classical dense layer.\n    \n    The VQC outputs a probability distribution over the three steganography algorithms:\n      - 0: JMiPOD\n      - 1: JUNIWARD\n      - 2: UERD\n    \"\"\"\n    dev = qml.device(\"default.qubit\", wires=num_qubits)\n    \n    @qml.qnode(dev, interface='tf')\n    def circuit(inputs, weights):\n        # Data encoding: use RY rotations\n        for i in range(num_qubits):\n            qml.RY(inputs[i], wires=i)\n        # Variational layers: apply RX and entangle using CNOT in a ring\n        for layer in range(num_layers):\n            for i in range(num_qubits):\n                qml.RX(weights[layer, i], wires=i)\n            for i in range(num_qubits):\n                qml.CNOT(wires=[i, (i + 1) % num_qubits])\n        # Return expectation values for each qubit\n        return [qml.expval(qml.PauliZ(i)) for i in range(num_qubits)]\n    \n    weight_shapes = {\"weights\": (num_layers, num_qubits)}\n    qlayer = qml.qnn.KerasLayer(circuit, weight_shapes, output_dim=num_qubits)\n    \n    # Build the full VQC model\n    model = tf.keras.Sequential([\n        tf.keras.layers.Dense(num_qubits, activation='relu'),\n        qlayer,\n        tf.keras.layers.Dense(3, activation='softmax')  # Three stego algorithm classes\n    ])\n    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])\n    return model\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:54.945886Z","iopub.execute_input":"2025-03-19T17:20:54.946108Z","iopub.status.idle":"2025-03-19T17:20:54.966649Z","shell.execute_reply.started":"2025-03-19T17:20:54.946085Z","shell.execute_reply":"2025-03-19T17:20:54.965957Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 5. Data Loader Functions\n# =============================================================================\ndef load_image(image_path, target_size=(64, 64)):\n    \"\"\"\n    Loads an image in grayscale, resizes it to target_size, and normalizes pixel values.\n    Returns an image array with shape (target_size[0], target_size[1], 1).\n    \"\"\"\n    image = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)\n    if image is None:\n        raise ValueError(f\"Failed to load image: {image_path}\")\n    image_resized = cv2.resize(image, target_size)\n    image_resized = image_resized.astype(\"float32\") / 255.0\n    return np.expand_dims(image_resized, axis=-1)\n\ndef get_file_paths_and_labels():\n    \"\"\"\n    Retrieves file paths and labels for training.\n    - For the binary stego detection: label 0 for Cover images; label 1 for any stego image.\n    - For stego algorithm classification, a separate mapping is provided:\n         0: JMiPOD, 1: JUNIWARD, 2: UERD.\n    Returns two lists: one for binary detection and one for stego algorithm classification.\n    \"\"\"\n    binary_file_paths = []\n    binary_labels = []  # 0: Cover, 1: Stego\n    \n    stego_file_paths = []\n    stego_algo_labels = []  # 0: JMiPOD, 1: JUNIWARD, 2: UERD\n    \n    # Cover images (clean)\n    cover_paths = glob.glob(os.path.join(TRAIN_DIRS[\"Cover\"], \"*.jpg\"))\n    for path in cover_paths:\n        binary_file_paths.append(path)\n        binary_labels.append(0)\n    \n    # Stego images from each algorithm\n    for algo, label in zip([\"JMiPOD\", \"JUNIWARD\", \"UERD\"], [0, 1, 2]):\n        algo_paths = glob.glob(os.path.join(TRAIN_DIRS[algo], \"*.jpg\"))\n        for path in algo_paths:\n            binary_file_paths.append(path)\n            binary_labels.append(1)\n            stego_file_paths.append(path)\n            stego_algo_labels.append(label)\n    \n    return (binary_file_paths, binary_labels), (stego_file_paths, stego_algo_labels)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:54.967497Z","iopub.execute_input":"2025-03-19T17:20:54.967765Z","iopub.status.idle":"2025-03-19T17:20:54.988175Z","shell.execute_reply.started":"2025-03-19T17:20:54.967742Z","shell.execute_reply":"2025-03-19T17:20:54.987335Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 6. Stego Detector Class: Integrates the Two-Stage Pipeline\n# =============================================================================\nclass StegoDetector:\n    def _init_(self):\n        \"\"\"\n        Initializes the detection system by building:\n          - The quantum-inspired CNN for binary stego detection.\n          - The VQC for classifying the steganography algorithm.\n        \"\"\"\n        self.cnn_input_shape = (64, 64, 1)\n        self.cnn_model = build_quantum_inspired_cnn(self.cnn_input_shape)\n        self.vqc_model = create_vqc(num_qubits=4, num_layers=2)\n    \n    def preprocess_image(self, image_path):\n        \"\"\"\n        Loads and preprocesses an image for the CNN.\n        \"\"\"\n        return load_image(image_path, target_size=(self.cnn_input_shape[0], self.cnn_input_shape[1]))\n    \n    def detect_stego(self, image_path):\n        \"\"\"\n        Uses the CNN to detect steganography.\n          Returns: \"Cover\" if clean, \"Stego\" if hidden message is detected.\n        \"\"\"\n        img = self.preprocess_image(image_path)\n        img = np.expand_dims(img, axis=0)\n        pred = self.cnn_model.predict(img)\n        # Threshold of 0.5 (can be tuned)\n        return \"Stego\" if pred[0][0] > 0.5 else \"Cover\"\n    \n    def classify_stego_algorithm(self, image_path):\n        \"\"\"\n        For images detected as stego, use the VQC to classify the algorithm used.\n        A feature vector is constructed using entropy and DCT mean from the image.\n        Returns one of: \"JMiPOD\", \"JUNIWARD\", \"UERD\".\n        \"\"\"\n        features = extract_features(image_path)\n        entropy = features[\"entropy\"]\n        dct_mean = np.mean(features[\"dct\"])\n        # Construct a 4-dimensional feature vector\n        input_vector = np.array([entropy, dct_mean, entropy * dct_mean, entropy + dct_mean], dtype=\"float32\")\n        input_vector = np.expand_dims(input_vector, axis=0)\n        algo_pred = self.vqc_model.predict(input_vector)\n        algo_index = np.argmax(algo_pred, axis=1)[0]\n        algo_labels = {0: \"JMiPOD\", 1: \"JUNIWARD\", 2: \"UERD\"}\n        return algo_labels.get(algo_index, \"Unknown\")\n    \n    def run_detection(self, image_path):\n        \"\"\"\n        Runs the full detection pipeline:\n          1. Detects whether the image is Cover or Stego.\n          2. If Stego, classifies the stego algorithm used.\n        Prints the final result.\n        \"\"\"\n        result = self.detect_stego(image_path)\n        if result == \"Cover\":\n            print(f\"Image: {os.path.basename(image_path)} --> Classified as: COVER (no hidden message)\")\n        else:\n            algo = self.classify_stego_algorithm(image_path)\n            print(f\"Image: {os.path.basename(image_path)} --> Classified as: STEGO (Algorithm: {algo})\")\n    \n    def predict_test_set(self, submission_file=\"sample_submission.csv\"):\n        \"\"\"\n        Processes all images in the Test/ directory and writes predictions in the required\n        CSV format. For each test image:\n          - If detected as Cover, assign label 0.\n          - If detected as Stego, assign label 1.\n        Optionally, you may include the algorithm classification for further analysis.\n        \"\"\"\n        test_files = glob.glob(os.path.join(TEST_DIR, \"*.jpg\"))\n        predictions = []\n        for path in test_files:\n            base_name = os.path.basename(path)\n            detection = self.detect_stego(path)\n            # For competition, only the binary label is needed:\n            label = 0 if detection == \"Cover\" else 1\n            predictions.append((base_name, label))\n        \n        # Write predictions to CSV in the format of sample_submission.csv\n        with open(submission_file, mode=\"w\", newline=\"\") as file:\n            writer = csv.writer(file)\n            writer.writerow([\"filename\", \"label\"])\n            for fname, lbl in predictions:\n                writer.writerow([fname, lbl])\n        print(f\"Submission file saved as {submission_file}\")\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:54.989251Z","iopub.execute_input":"2025-03-19T17:20:54.989566Z","iopub.status.idle":"2025-03-19T17:20:55.010368Z","shell.execute_reply.started":"2025-03-19T17:20:54.989536Z","shell.execute_reply":"2025-03-19T17:20:55.009757Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 7. (Optional) Training Pipeline for the Models\n# =============================================================================\ndef train_models(detector, epochs=5, batch_size=32):\n    \"\"\"\n    Trains the CNN for stego detection and the VQC for stego algorithm classification.\n    Note: This is a simplified training loop. In practice, you may use Keras generators,\n    callbacks, and more sophisticated data augmentation.\n    \"\"\"\n    # Retrieve file paths and labels\n    (binary_paths, binary_labels), (stego_paths, stego_algo_labels) = get_file_paths_and_labels()\n    \n    # Prepare training data for the CNN\n    X_cnn = []\n    for path in binary_paths:\n        try:\n            img = load_image(path, target_size=(64, 64))\n            X_cnn.append(img)\n        except Exception as e:\n            continue\n    X_cnn = np.array(X_cnn)\n    y_cnn = np.array(binary_labels[:len(X_cnn)]).reshape(-1, 1)\n    \n    print(f\"Training CNN on {len(X_cnn)} images...\")\n    detector.cnn_model.fit(X_cnn, y_cnn, epochs=epochs, batch_size=batch_size, validation_split=0.1)\n    \n    # Prepare training data for the VQC (only stego images)\n    X_vqc = []\n    y_vqc = []\n    for path, label in zip(stego_paths, stego_algo_labels):\n        try:\n            features = extract_features(path)\n            entropy = features[\"entropy\"]\n            dct_mean = np.mean(features[\"dct\"])\n            input_vector = [entropy, dct_mean, entropy * dct_mean, entropy + dct_mean]\n            X_vqc.append(input_vector)\n            y_vqc.append(label)\n        except Exception as e:\n            continue\n    X_vqc = np.array(X_vqc, dtype=\"float32\")\n    # Convert labels to one-hot encoding for three classes\n    y_vqc = tf.keras.utils.to_categorical(y_vqc, num_classes=3)\n    \n    print(f\"Training VQC on {len(X_vqc)} stego images...\")\n    detector.vqc_model.fit(X_vqc, y_vqc, epochs=epochs, batch_size=batch_size, validation_split=0.1)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:20:55.011913Z","iopub.execute_input":"2025-03-19T17:20:55.012224Z","iopub.status.idle":"2025-03-19T17:20:55.031252Z","shell.execute_reply.started":"2025-03-19T17:20:55.012193Z","shell.execute_reply":"2025-03-19T17:20:55.030481Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# =============================================================================\n# 8. Main Execution: Train Models and/or Run Predictions\n# =============================================================================\nif __name__ == \"__main__\":\n    # Initialize the detection system\n    detector = StegoDetector()\n    \n    # Uncomment the following line to train both models.\n    # Ensure that the dataset directories exist and contain images.\n    # train_models(detector, epochs=5, batch_size=32)\n    \n    # Example: Run detection on a single image (for demo purposes)\n    test_image = \"/kaggle/input/alaska2-image-steganalysis/JUNIWARD/00008.jpg\"  # Replace with an actual test image path\n    try:\n        detector.run_detection(test_image)\n    except Exception as e:\n        print(f\"Error during detection: {e}\")\n    \n    # Uncomment the following line to predict the entire test set and generate a submission CSV.\n    # detector.predict_test_set(submission_file=\"sample_submission.csv\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-19T17:21:57.553241Z","iopub.execute_input":"2025-03-19T17:21:57.553550Z","iopub.status.idle":"2025-03-19T17:21:57.558664Z","shell.execute_reply.started":"2025-03-19T17:21:57.553526Z","shell.execute_reply":"2025-03-19T17:21:57.557687Z"}},"outputs":[],"execution_count":null}]}