{"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":"tpu1vmV38","dataSources":[{"sourceId":75232,"databundleVersionId":8331367,"sourceType":"competition"}],"dockerImageVersionId":30918,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\n\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Preprocessing of Hyperspectral Images","metadata":{}},{"cell_type":"code","source":"!pip install rasterio\n!pip install spectral\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:26:48.216760Z","iopub.execute_input":"2025-04-15T18:26:48.217244Z","iopub.status.idle":"2025-04-15T18:26:59.270005Z","shell.execute_reply.started":"2025-04-15T18:26:48.217217Z","shell.execute_reply":"2025-04-15T18:26:59.268945Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport numpy as np\nimport rasterio\nimport cv2\nfrom glob import glob\nfrom tqdm import tqdm\nfrom sklearn.model_selection import train_test_split\n\n# Define dataset paths\nraw_dataset_path = \"/kaggle/input/beyond-visible-spectrum-ai-for-agriculture-2024/archive\"  \nprocessed_dataset_path = \"/kaggle/working/processed_images\"\nos.makedirs(processed_dataset_path, exist_ok=True)\n\n# Standard shape for hyperspectral processing\nTARGET_SIZE = (128, 128)  # Resize to 128x128\nSELECTED_BANDS = 100  # Number of spectral bands to keep\n\n# Function to process a single hyperspectral image\ndef process_hyperspectral_image(image_path):\n    with rasterio.open(image_path) as img:\n        hyperspectral_data = img.read()  # Shape: (Bands, Height, Width)\n    \n    # Normalize each band separately\n    hyperspectral_data = hyperspectral_data.astype(np.float32)\n    for i in range(hyperspectral_data.shape[0]):  # Iterate over bands\n        min_val = hyperspectral_data[i].min()\n        max_val = hyperspectral_data[i].max()\n        if max_val > min_val:\n            hyperspectral_data[i] = (hyperspectral_data[i] - min_val) / (max_val - min_val)\n\n    # Perform band selection (Take first 50 bands or use PCA if needed)\n    if hyperspectral_data.shape[0] > SELECTED_BANDS:\n        hyperspectral_data = hyperspectral_data[:SELECTED_BANDS, :, :]\n\n    # Resize each band separately\n    resized_bands = np.zeros((SELECTED_BANDS, TARGET_SIZE[0], TARGET_SIZE[1]))\n    for i in range(SELECTED_BANDS):\n        resized_bands[i] = cv2.resize(hyperspectral_data[i], TARGET_SIZE, interpolation=cv2.INTER_CUBIC)\n    \n    return resized_bands  # Shape: (50, 128, 128)\n\n# Process entire dataset\nall_images = []\nall_labels = []\n\nfor category in os.listdir(os.path.join(raw_dataset_path, \"train\")):  # Using \"train\" folder as full dataset\n    category_path = os.path.join(raw_dataset_path, \"train\", category)\n    os.makedirs(os.path.join(processed_dataset_path, category), exist_ok=True)\n\n    for img_path in tqdm(glob(os.path.join(category_path, \"*.tif\")), desc=f\"Processing {category}\"):\n        processed_image = process_hyperspectral_image(img_path)\n        \n        # Save as .npy file\n        save_path = os.path.join(processed_dataset_path, category, os.path.basename(img_path).replace(\".tif\", \".npy\"))\n        np.save(save_path, processed_image)\n\n        all_images.append(save_path)\n        all_labels.append(category)\n\nprint(f\"✅ Processed {len(all_images)} images! Stored in {processed_dataset_path}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:26:59.271650Z","iopub.execute_input":"2025-04-15T18:26:59.272003Z","iopub.status.idle":"2025-04-15T18:27:33.851539Z","shell.execute_reply.started":"2025-04-15T18:26:59.271969Z","shell.execute_reply":"2025-04-15T18:27:33.850622Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Split dataset into train (80%), val (10%), and test (10%)\ntrain_imgs, temp_imgs, train_labels, temp_labels = train_test_split(all_images, all_labels, test_size=0.2, stratify=all_labels, random_state=42)\nval_imgs, test_imgs, val_labels, test_labels = train_test_split(temp_imgs, temp_labels, test_size=0.5, stratify=temp_labels, random_state=42)\n\n# Create new directories\nsplit_dataset_path = \"/kaggle/working/split_dataset\"\nfor split in [\"train\", \"val\", \"test\"]:\n    for category in os.listdir(processed_dataset_path):\n        os.makedirs(os.path.join(split_dataset_path, split, category), exist_ok=True)\n\n# Function to move files to split directories\nimport shutil\n\ndef move_files(image_paths, labels, split_name):\n    for img_path, label in zip(image_paths, labels):\n        dest_path = os.path.join(split_dataset_path, split_name, label, os.path.basename(img_path))\n        shutil.move(img_path, dest_path)\n\n# Move processed files to split directories\nmove_files(train_imgs, train_labels, \"train\")\nmove_files(val_imgs, val_labels, \"val\")\nmove_files(test_imgs, test_labels, \"test\")\n\nprint(f\"✅ Dataset split completed! Train: {len(train_imgs)}, Val: {len(val_imgs)}, Test: {len(test_imgs)}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:27:33.852791Z","iopub.execute_input":"2025-04-15T18:27:33.853229Z","iopub.status.idle":"2025-04-15T18:27:33.887792Z","shell.execute_reply.started":"2025-04-15T18:27:33.853206Z","shell.execute_reply":"2025-04-15T18:27:33.887170Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Making RGB Images","metadata":{}},{"cell_type":"code","source":"import os\nimport numpy as np\nimport cv2\nfrom glob import glob\nfrom tqdm import tqdm\nimport shutil\n\n# Define source and destination directories for split data\nsplit_dataset_path = \"/kaggle/working/split_dataset\"\nrgb_split_dataset_path = \"/kaggle/working/rgb_split_dataset\"\nos.makedirs(rgb_split_dataset_path, exist_ok=True)\n\n# Create corresponding folder structure for RGB images\nfor split in [\"train\", \"val\", \"test\"]:\n    split_path = os.path.join(split_dataset_path, split)\n    rgb_split_path = os.path.join(rgb_split_dataset_path, split)\n    os.makedirs(rgb_split_path, exist_ok=True)\n    \n    # For every category folder in this split:\n    for category in os.listdir(split_path):\n        category_dir = os.path.join(split_path, category)\n        if os.path.isdir(category_dir):\n            rgb_category_dir = os.path.join(rgb_split_path, category)\n            os.makedirs(rgb_category_dir, exist_ok=True)\n\ndef hsi_to_rgb(hsi_image, r_idx=42, g_idx=15, b_idx=0):\n    \"\"\"\n    Convert hyperspectral image (shape: (channels, H, W)) to a pseudo-RGB image using band selection.\n    \n    The three chosen indices are arbitrary and should be tailored to your sensor.\n    \"\"\"\n    # Ensure hsi_image is a numpy array\n    if isinstance(hsi_image, (list, tuple)):\n        hsi_image = np.array(hsi_image)\n    \n    # hsi_image is assumed to be normalized to [0,1]\n    # Extract the three bands (shape: (H, W))\n    try:\n        red_band   = hsi_image[r_idx]\n        green_band = hsi_image[g_idx]\n        blue_band  = hsi_image[b_idx]\n    except IndexError:\n        raise ValueError(\"Chosen band indices are out of bounds for the image with shape {}\".format(hsi_image.shape))\n    \n    # Stack into RGB (H, W, 3)\n    rgb = np.stack([red_band, green_band, blue_band], axis=-1)\n    \n    # Optional: clip or re-normalize if needed\n    rgb = (rgb - rgb.min()) / (rgb.max() - rgb.min() + 1e-8)  # Back to [0,1]\n    rgb = (rgb * 255).astype(np.uint8)  # Convert to 8-bit image\n    return rgb\n\n# Loop over the split directories and convert/save RGB images\nfor split in [\"train\", \"val\", \"test\"]:\n    split_path = os.path.join(split_dataset_path, split)\n    rgb_split_path = os.path.join(rgb_split_dataset_path, split)\n    for category in os.listdir(split_path):\n        category_dir = os.path.join(split_path, category)\n        if not os.path.isdir(category_dir):\n            continue\n        rgb_category_dir = os.path.join(rgb_split_path, category)\n        for file in tqdm(os.listdir(category_dir), desc=f\"Converting {split}/{category}\"):\n            if file.endswith(\".npy\"):\n                hsi_path = os.path.join(category_dir, file)\n                # Load the hyperspectral image (shape: (50, 128, 128))\n                hsi_img = np.load(hsi_path)\n                # Convert to RGB\n                rgb_img = hsi_to_rgb(hsi_img)\n                # Save as PNG file\n                rgb_save_path = os.path.join(rgb_category_dir, file.replace(\".npy\", \".png\"))\n                cv2.imwrite(rgb_save_path, cv2.cvtColor(rgb_img, cv2.COLOR_RGB2BGR))  # OpenCV expects BGR order\n\nprint(\"✅ RGB conversion completed!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:50:58.486174Z","iopub.execute_input":"2025-04-15T13:50:58.486485Z","iopub.status.idle":"2025-04-15T13:51:00.812987Z","shell.execute_reply.started":"2025-04-15T13:50:58.486459Z","shell.execute_reply":"2025-04-15T13:51:00.812090Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nfrom torch.utils.data import Dataset, DataLoader\nfrom PIL import Image\n\nclass DualModalityDataset(Dataset):\n    def __init__(self, hsi_root, rgb_root, transform_hsi=None, transform_rgb=None):\n        \"\"\"\n        Args:\n            hsi_root (str): Root directory of hyperspectral images (e.g., /kaggle/working/split_dataset/train)\n            rgb_root (str): Root directory of RGB images (e.g., /kaggle/working/rgb_split_dataset/train)\n            transform_hsi (callable, optional): Transformation to apply to the hyperspectral tensor.\n            transform_rgb (callable, optional): Transformation to apply to the RGB image (PIL/Image tensor).\n        \"\"\"\n        self.samples = []\n        self.transform_hsi = transform_hsi\n        self.transform_rgb = transform_rgb\n        \n        # Walk over each category folder in hsi_root\n        for category in sorted(os.listdir(hsi_root)):\n            hsi_category_path = os.path.join(hsi_root, category)\n            rgb_category_path = os.path.join(rgb_root, category)\n            if not os.path.isdir(hsi_category_path):\n                continue\n            # List all files in the HSI folder\n            for filename in os.listdir(hsi_category_path):\n                if filename.endswith(\".npy\"):\n                    hsi_path = os.path.join(hsi_category_path, filename)\n                    # Assume matching RGB file has same base filename with .png extension\n                    rgb_path = os.path.join(rgb_category_path, filename.replace(\".npy\", \".png\"))\n                    # Only add if the RGB image exists\n                    if os.path.exists(rgb_path):\n                        self.samples.append((hsi_path, rgb_path, category))\n        \n        # Create a mapping from category to an index\n        categories = sorted({s[2] for s in self.samples})\n        self.cat2idx = {cat: idx for idx, cat in enumerate(categories)}\n\n    def __len__(self):\n        return len(self.samples)\n\n    def __getitem__(self, idx):\n        hsi_path, rgb_path, category = self.samples[idx]\n        # Load hyperspectral image (stored as .npy, shape (50, 128, 128))\n        hsi_img = np.load(hsi_path)\n        # Convert to torch tensor (float32)\n        hsi_tensor = torch.tensor(hsi_img, dtype=torch.float32)\n        if self.transform_hsi:\n            hsi_tensor = self.transform_hsi(hsi_tensor)\n        \n        # Load RGB image using PIL (to allow use of torchvision transforms)\n        rgb_img = Image.open(rgb_path).convert(\"RGB\")\n        if self.transform_rgb:\n            rgb_img = self.transform_rgb(rgb_img)\n        else:\n            # By default, convert the image to tensor in [0, 1]\n            from torchvision.transforms import ToTensor\n            rgb_img = ToTensor()(rgb_img)\n        \n        label = self.cat2idx[category]\n        return hsi_tensor, rgb_img, label\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:17.020799Z","iopub.execute_input":"2025-04-15T13:54:17.021140Z","iopub.status.idle":"2025-04-15T13:54:17.029303Z","shell.execute_reply.started":"2025-04-15T13:54:17.021117Z","shell.execute_reply":"2025-04-15T13:54:17.028651Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nfrom torch.utils.data import DataLoader\nimport torchvision.transforms as T\nimport torch\nimport torchvision.transforms.functional as TF\n\n# Custom resize transform for hyperspectral data (if tensor input)\n\n#transform_hsi = TensorResize((224, 224))  # Match both HSI and RGB resolutions\ntransform_rgb = T.Compose([\n    T.Resize((128, 128)),\n    T.RandomHorizontalFlip(p=0.5),\n    T.RandomVerticalFlip(p=0.5),  # Random vertical flip\n    T.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.05),\n    T.RandomRotation(30),  # Random rotation by up to 30 degrees\n    T.RandomAffine(degrees=0, translate=(0.1, 0.1)),  # Random affine transformation\n    T.RandomCrop(128, padding=8),  # Random crop with padding\n    T.RandomPerspective(distortion_scale=0.2, p=0.5),  # Random perspective transformation\n    T.GaussianBlur(kernel_size=5),  # Apply Gaussian blur\n    T.ToTensor(),\n    T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])  # Normalization for pre-trained models\n])\n\n\n\n# Define root directories for each split\nhsi_train_root = os.path.join(split_dataset_path, \"train\")\nrgb_train_root = os.path.join(rgb_split_dataset_path, \"train\")\n\nhsi_val_root = os.path.join(split_dataset_path, \"val\")\nrgb_val_root = os.path.join(rgb_split_dataset_path, \"val\")\n\nhsi_test_root = os.path.join(split_dataset_path, \"test\")\nrgb_test_root = os.path.join(rgb_split_dataset_path, \"test\")\n\n# Instantiate datasets\ntrain_dataset = DualModalityDataset(hsi_root=hsi_train_root, rgb_root=rgb_train_root,\n                                    transform_hsi=transform_hsi, transform_rgb=transform_rgb)\n\nval_dataset = DualModalityDataset(hsi_root=hsi_val_root, rgb_root=rgb_val_root,\n                                  transform_hsi=transform_hsi, transform_rgb=transform_rgb)\n\ntest_dataset = DualModalityDataset(hsi_root=hsi_test_root, rgb_root=rgb_test_root,\n                                   transform_hsi=transform_hsi, transform_rgb=transform_rgb)\n\n# Create DataLoaders\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)\ntest_loader = DataLoader(test_dataset, batch_size=32, shuffle=False, num_workers=2)\n\n# Quick sanity check\nfor hsi_batch, rgb_batch, labels in val_loader:\n    print(\"Val HSI batch shape:\", hsi_batch.shape)\n    print(\"Val RGB batch shape:\", rgb_batch.shape)\n    print(\"Val Labels:\", labels)\n    break\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:20.197218Z","iopub.execute_input":"2025-04-15T13:54:20.197501Z","iopub.status.idle":"2025-04-15T13:54:27.286052Z","shell.execute_reply.started":"2025-04-15T13:54:20.197479Z","shell.execute_reply":"2025-04-15T13:54:27.285155Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### Doing SSL","metadata":{}},{"cell_type":"code","source":"import os\nimport shutil\n\n# Paths\nsplit_dataset_path = \"/kaggle/working/split_dataset\"  # Where train/val/test exist\nssl_dataset_path = \"/kaggle/working/ssl_dataset\"  # New dataset for SSL\n\n# Create SSL dataset directory\nos.makedirs(ssl_dataset_path, exist_ok=True)\n\n# Merge all splits into SSL dataset\nfor split in [\"train\", \"val\", \"test\"]:\n    split_path = os.path.join(split_dataset_path, split)\n    if not os.path.exists(split_path):\n        continue  # Skip if path doesn't exist\n\n    for category in os.listdir(split_path):  # Loop over classes (Health, Rust, Other)\n        category_path = os.path.join(ssl_dataset_path, category)\n        os.makedirs(category_path, exist_ok=True)  # Ensure class folder exists\n\n        # Move all images to the SSL dataset\n        for file in os.listdir(os.path.join(split_path, category)):\n            src = os.path.join(split_path, category, file)\n            dest = os.path.join(category_path, file)\n            shutil.move(src, dest)\n\nprint(f\"✅ SSL dataset is ready in {ssl_dataset_path}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:09:03.215814Z","iopub.execute_input":"2025-04-15T13:09:03.216620Z","iopub.status.idle":"2025-04-15T13:09:03.241253Z","shell.execute_reply.started":"2025-04-15T13:09:03.216584Z","shell.execute_reply":"2025-04-15T13:09:03.240370Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### Data Augmentation for SSL","metadata":{}},{"cell_type":"code","source":"import numpy as np\nimport albumentations as A\nimport cv2\nimport os\nfrom glob import glob\n\n# Define augmentations\naugmentations = A.Compose([\n    A.RandomCrop(height=128, width=128, always_apply=True),  # Crop to 128x128\n    A.HorizontalFlip(p=0.5),  # Flip 50% of the time\n    A.VerticalFlip(p=0.5),  # Flip 50% of the time\n])\n\n# Function to apply augmentation\ndef apply_augmentations(img_path):\n    img = np.load(img_path)  # Load .npy file\n    img = np.float32(img)  # Ensure correct dtype for Albumentations\n    \n    # Normalize if needed (optional, depending on your dataset)\n    img = cv2.normalize(img, None, 0, 255, cv2.NORM_MINMAX)  # Scale to 0-255\n    img = img.astype(np.uint8)  # Convert to uint8 if necessary\n\n    # Albumentations expects HWC format, so ensure the correct shape\n    if img.ndim == 2:  \n        img = np.expand_dims(img, axis=-1)  # Convert grayscale to HWC\n    elif img.shape[0] < img.shape[-1]:  # If channels are first, swap axes\n        img = np.transpose(img, (1, 2, 0))\n\n    augmented = augmentations(image=img)[\"image\"]  # Apply augmentations\n    \n    return augmented\n\n# Apply augmentations to SSL dataset\nfor class_folder in os.listdir(ssl_dataset_path):\n    class_path = os.path.join(ssl_dataset_path, class_folder)\n    if not os.path.isdir(class_path):  # Skip non-folder items\n        continue\n\n    for img_path in glob(os.path.join(class_path, \"*.npy\")):\n        augmented_img = apply_augmentations(img_path)\n        np.save(img_path.replace(\".npy\", \"_aug.npy\"), augmented_img)  # Save augmented image\n\nprint(\"✅ Augmentations applied to SSL dataset!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:09:09.029228Z","iopub.execute_input":"2025-04-15T13:09:09.029542Z","iopub.status.idle":"2025-04-15T13:09:16.852928Z","shell.execute_reply.started":"2025-04-15T13:09:09.029516Z","shell.execute_reply":"2025-04-15T13:09:16.851961Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"### SSL Training with SimCLR","metadata":{}},{"cell_type":"markdown","source":"### AHAM Addition","metadata":{}},{"cell_type":"markdown","source":"NTXent loss - Loss function for SimCLR encoder","metadata":{}},{"cell_type":"code","source":"import os\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nimport numpy as np\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as T\nimport torchvision.transforms.functional as TF\nfrom timm.models import create_model  \n\n# -------------------------\n# GPU/Memory Setup\n# -------------------------\ntorch.cuda.empty_cache()\ntorch.backends.cuda.matmul.allow_tf32 = True\ntorch.backends.cudnn.benchmark = True\ntorch.backends.cuda.matmul.allow_fp16_reduced_precision_reduction = True\nos.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'max_split_size_mb:64'\n\ndevice = torch.device(\"cuda:1\" if torch.cuda.is_available() else \"cpu\")\nprint(f\"Using device: {device}\")\n\n# -------------------------\n# Define AHAM Module\n# -------------------------\nclass AHAM(nn.Module):\n    \"\"\"\n    Adaptive Hyperspectral Attention Module (AHAM).\n    Expects input x of shape [B, P, D] where:\n      B = batch size,\n      P = number of tokens/patches,\n      D = embedding dimension.\n    Returns a weighted average over tokens (i.e. a refined feature vector)\n    and the attention weights (for visualization).\n    \"\"\"\n    def __init__(self, embed_dim, reduction=16):\n        super(AHAM, self).__init__()\n        self.avg_pool = nn.AdaptiveAvgPool1d(1)\n        self.fc1 = nn.Linear(embed_dim, embed_dim // reduction)\n        self.fc2 = nn.Linear(embed_dim // reduction, embed_dim)\n        self.sigmoid = nn.Sigmoid()\n\n    def forward(self, x):\n        # x: [B, P, D]\n        attention = self.avg_pool(x.transpose(1, 2)).squeeze(-1)  # [B, D]\n        attention = self.fc1(attention)\n        attention = nn.functional.relu(attention)\n        attention = self.fc2(attention)\n        attention = self.sigmoid(attention)  # [B, D]\n        x_weighted = (x * attention.unsqueeze(1)).mean(dim=1)  # [B, D]\n        return x_weighted, attention","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:09:18.235847Z","iopub.execute_input":"2025-04-15T13:09:18.236142Z","iopub.status.idle":"2025-04-15T13:09:21.730496Z","shell.execute_reply.started":"2025-04-15T13:09:18.236118Z","shell.execute_reply":"2025-04-15T13:09:21.729353Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -------------------------\n# Define NT-Xent Loss (Contrastive Loss)\n# -------------------------\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\nclass NTXentLossFull(nn.Module):\n    \"\"\"\n    A full NT-Xent loss where each sample is compared against all in the batch\n    (2B in total, minus itself).\n    This is the canonical approach in SimCLR.\n    \"\"\"\n    def __init__(self, temperature=0.5):\n        super().__init__()\n        self.temperature = temperature\n\n    def forward(self, z_i, z_j):\n        \"\"\"\n        z_i: [B, D]\n        z_j: [B, D]\n        Returns scalar contrastive loss.\n        \"\"\"\n        # If the inputs are 1D, unsqueeze them to add the batch dimension.\n        # if z_i.dim() == 1:\n        #     z_i = z_i.unsqueeze(0)\n        # if z_j.dim() == 1:\n        #     z_j = z_j.unsqueeze(0)\n        \n        # z_i = z_i.float()\n        # z_j = z_j.float()\n        z_i = F.normalize(z_i, dim=1)\n        z_j = F.normalize(z_j, dim=1)\n\n        # Concatenate along batch dimension\n        z = torch.cat([z_i, z_j], dim=0)  # [2B, D]\n\n        # Compute similarity matrix\n        sim_matrix = torch.matmul(z, z.T)  # [2B, 2B]\n\n        # Create labels for positives:\n        # For sample i in [0..B-1], the positive is i+B\n        # For sample i in [B..2B-1], the positive is i-B\n        batch_size = z_i.size(0)\n        labels = torch.arange(batch_size, device=z.device)\n        labels = torch.cat([labels + batch_size, labels], dim=0)  # [2B]\n\n        # Mask out self-similarity by setting the diagonal to large negative\n        mask = torch.eye(2 * batch_size, device=z.device).bool()\n        sim_matrix.masked_fill_(mask, -9e15)\n\n        # Scale by temperature\n        sim_matrix /= self.temperature\n\n        # Use cross-entropy over the row to find the positive sample.\n        loss = F.cross_entropy(sim_matrix, labels)\n        return loss\n\n\n\ncriterion = NTXentLossFull()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:09:38.627273Z","iopub.execute_input":"2025-04-15T13:09:38.627649Z","iopub.status.idle":"2025-04-15T13:09:38.634260Z","shell.execute_reply.started":"2025-04-15T13:09:38.627615Z","shell.execute_reply":"2025-04-15T13:09:38.633338Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -------------------------\n# Define the SimCLR Model with AHAM Integrated\n# -------------------------\nclass SimCLR(nn.Module):\n    def __init__(self, backbone=\"vit_base_patch16_224\", feature_dim=128, input_channels=100):\n        \"\"\"\n        backbone: name of the backbone ViT model.\n        feature_dim: Final embedding dimension for contrastive learning.\n        input_channels: Number of channels in the hyperspectral input.\n        \"\"\"\n        super(SimCLR, self).__init__()\n        # Create the ViT model without classifier.\n        # Remove 'output_cls_token' because it is not supported.\n        self.encoder = create_model(backbone, pretrained=False, num_classes=0)\n        \n        # Modify patch embedding projection layer to accept input_channels\n        new_conv_layer = nn.Conv2d(\n            input_channels,\n            self.encoder.patch_embed.proj.out_channels,\n            kernel_size=self.encoder.patch_embed.proj.kernel_size,\n            stride=self.encoder.patch_embed.proj.stride,\n            padding=self.encoder.patch_embed.proj.padding,\n            bias=False\n        )\n        self.encoder.patch_embed.proj = new_conv_layer\n\n        # Integrate AHAM.\n        self.aham = AHAM(embed_dim=self.encoder.num_features)\n        # Project refined features to contrastive embedding space.\n        self.projector = nn.Sequential(\n            nn.Linear(self.encoder.num_features, 512),\n            nn.ReLU(),\n            nn.Linear(512, feature_dim)\n        )\n\n    def forward(self, x, return_attention=False):\n        # Get token embeddings from the encoder; x shape: [B, input_channels, 224, 224]\n        tokens = self.encoder.forward_features(x)  # Output shape: [B, P, D]\n        x_weighted, attention = self.aham(tokens)      # [B, D] and [B, D]\n        projections = self.projector(x_weighted)\n        if return_attention:\n            return projections, attention\n        return projections\n\nhsi_model = SimCLR(input_channels=100).to(device)\nhsi_model.train()\nprint(\"✅ SimCLR with AHAM is ready!\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:09:43.118798Z","iopub.execute_input":"2025-04-15T13:09:43.119087Z","iopub.status.idle":"2025-04-15T13:09:45.035748Z","shell.execute_reply.started":"2025-04-15T13:09:43.119065Z","shell.execute_reply":"2025-04-15T13:09:45.034951Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -------------------------\n# Define Custom Transforms for the SSL Dataset\n# -------------------------\nclass RandomHorizontalFlipTensor:\n    \"\"\"Randomly horizontally flips a tensor image with probability p.\"\"\"\n    def __init__(self, p=0.5):\n        self.p = p\n    def __call__(self, img):\n        if torch.rand(1).item() < self.p:\n            return torch.flip(img, dims=[-1])\n        return img\n\nclass RandomVerticalFlipTensor:\n    \"\"\"Randomly vertically flips a tensor image with probability p.\"\"\"\n    def __init__(self, p=0.5):\n        self.p = p\n    def __call__(self, img):\n        if torch.rand(1).item() < self.p:\n            return torch.flip(img, dims=[-2])\n        return img\n\nclass RandomRotationTensor:\n    \"\"\"Randomly rotates a tensor image by an angle between -max_angle and max_angle degrees.\"\"\"\n    def __init__(self, max_angle=30):\n        self.max_angle = max_angle\n    def __call__(self, img):\n        angle = torch.empty(1).uniform_(-self.max_angle, self.max_angle).item()\n        return TF.rotate(img, angle)\n\nclass AddGaussianNoise:\n    \"\"\"Adds Gaussian noise to a tensor image.\"\"\"\n    def __init__(self, mean=0.0, std=0.1):\n        self.mean = mean\n        self.std = std\n    def __call__(self, img):\n        noise = torch.randn_like(img) * self.std + self.mean\n        return img + noise\n\nclass RandomContrastTensor:\n    \"\"\"Randomly adjusts contrast for a tensor image by a factor chosen uniformly.\"\"\"\n    def __init__(self, lower=0.7, upper=1.3):\n        self.lower = lower\n        self.upper = upper\n    def __call__(self, img):\n        factor = torch.empty(1).uniform_(self.lower, self.upper).item()\n        mean = torch.mean(img, dim=[1,2], keepdim=True)\n        return (img - mean) * factor + mean\n\nclass TensorResize:\n    \"\"\"Resizes a tensor image to the target size.\"\"\"\n    def __init__(self, size):\n        self.size = size\n    def __call__(self, img):\n        return torch.nn.functional.interpolate(\n            img.unsqueeze(0), size=self.size, mode='bilinear', align_corners=False\n        ).squeeze(0)\n\n# Compose an augmentation pipeline with many augmentations.\ntransform = T.Compose([\n    TensorResize((224, 224)),\n    RandomHorizontalFlipTensor(p=0.5),\n    RandomVerticalFlipTensor(p=0.5),\n    RandomRotationTensor(max_angle=30),\n    AddGaussianNoise(mean=0.0, std=0.05),\n    RandomContrastTensor(lower=0.8, upper=1.2)\n])\n\n# -------------------------\n# Define the SSL Dataset for Contrastive Learning\n# -------------------------\nclass SSLDataset(Dataset):\n    def __init__(self, root_dir, transform=None, target_channels=100):\n        self.root_dir = root_dir\n        self.transform = transform\n        self.target_channels = target_channels\n        self.image_paths = [\n            os.path.join(root, f)\n            for root, _, files in os.walk(root_dir)\n            for f in files if f.endswith(\".npy\")\n        ]\n        print(f\"Found {len(self.image_paths)} hyperspectral images in {root_dir}\")\n\n    def __len__(self):\n        return len(self.image_paths)\n\n    def __getitem__(self, idx):\n        img_path = self.image_paths[idx]\n        img = np.load(img_path)  # Load hyperspectral image (H, W, C)\n        if len(img.shape) == 2:\n            raise ValueError(f\"Image at {img_path} is grayscale (expected HxWxC).\")\n        C, H, W = img.shape[2], img.shape[0], img.shape[1]\n        if C < self.target_channels:\n            img_padded = np.zeros((H, W, self.target_channels), dtype=np.float32)\n            img_padded[:, :, :C] = img\n            img = img_padded\n        elif C > self.target_channels:\n            img = img[:, :, :self.target_channels]\n        img = np.transpose(img, (2, 0, 1)).astype(np.float32)\n        img_min, img_max = img.min(), img.max()\n        if img_max > img_min:\n            img = (img - img_min) / (img_max - img_min)\n        else:\n            img = np.zeros_like(img)\n        img_tensor = torch.tensor(img, dtype=torch.float32)\n        # Generate two different augmented views.\n        if self.transform:\n            view1 = self.transform(img_tensor)\n            view2 = self.transform(img_tensor)\n        else:\n            view1, view2 = img_tensor, img_tensor\n        return view1, view2\n\n# -------------------------\n# Instantiate the Dataset and DataLoader\n# -------------------------\n# Update this path to point to your hyperspectral .npy files.\n#ssl_dataset_path = \"/path/to/hsi_dataset\"\nssl_dataset = SSLDataset(ssl_dataset_path, transform)\nssl_loader = DataLoader(ssl_dataset, batch_size=32, shuffle=True, num_workers=2)\n\n# -------------------------\n# Define the Optimizer\n# -------------------------\noptimizer = optim.Adam(hsi_model.parameters(), lr=3e-4)\n\n# -------------------------\n# Training Loop for Self-Supervised Learning (with AHAM integrated)\n# -------------------------\nnum_epochs = 10\nfor epoch in range(num_epochs):\n    hsi_model.train()\n    epoch_loss = 0.0\n    for images1, images2 in ssl_loader:\n        images1, images2 = images1.to(device), images2.to(device)\n        z_i, z_j = hsi_model(images1), hsi_model(images2)\n        loss = criterion(z_i, z_j)\n        optimizer.zero_grad()\n        loss.backward()\n        optimizer.step()\n        epoch_loss += loss.item() * images1.size(0)\n    avg_loss = epoch_loss / len(ssl_dataset)\n    print(f\"Epoch [{epoch+1}/{num_epochs}], Loss: {avg_loss:.4f}\")\n\nprint(\"✅ Self-Supervised Learning (SSL) Training Complete!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:10:07.792469Z","iopub.execute_input":"2025-04-15T13:10:07.792795Z","iopub.status.idle":"2025-04-15T13:47:20.225408Z","shell.execute_reply.started":"2025-04-15T13:10:07.792769Z","shell.execute_reply":"2025-04-15T13:47:20.224096Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Fine Tuning for Crop Health Classification","metadata":{}},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torchvision.models as models\n\nclass DualStreamFusionModel(nn.Module):\n    def __init__(self, hsi_ssl_model, num_classes=3):\n        \"\"\"\n        hsi_ssl_model: your already-trained SimCLR-style model for hyperspectral data.\n        num_classes: number of crop health classes.\n        \"\"\"\n        super(DualStreamFusionModel, self).__init__()\n\n        # HSI Branch: use your trained SSL model.\n        # Freeze its weights so that only the new fusion head is trained.\n        self.hsi_encoder = hsi_ssl_model  \n        for param in self.hsi_encoder.parameters():\n            param.requires_grad = False\n        \n        # RGB branch: use a pretrained ResNet18.\n        self.rgb_encoder = models.resnet18(pretrained=True)\n        # Remove the default classifier so that we only obtain features.\n        self.rgb_encoder.fc = nn.Identity()  # Outputs a 512-D feature vector\n        \n        # Fusion Head: Combine the 128-D (HSI) and 512-D (RGB) features.\n        self.classifier = nn.Sequential(\n            nn.Linear(128 + 512, 256),\n            nn.ReLU(),\n            nn.Dropout(0.3),\n            nn.Linear(256, num_classes)\n        )\n    \n    def forward(self, hsi, rgb):\n        # print(\"FusionModel.forward() called with:\")\n        # print(\" - hsi type:\", type(hsi), \"shape:\", hsi.shape)\n        # print(\" - rgb type:\", type(rgb), \"shape:\", rgb.shape)\n        hsi_features = self.hsi_encoder(hsi)\n        rgb_features = self.rgb_encoder(rgb)\n        #transform_hsi = TensorResize((224, 224))\n        fused_features = torch.cat([hsi_features, rgb_features], dim=1)\n        logits = self.classifier(fused_features)\n        return logits\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:39.654687Z","iopub.execute_input":"2025-04-15T13:54:39.655009Z","iopub.status.idle":"2025-04-15T13:54:39.661796Z","shell.execute_reply.started":"2025-04-15T13:54:39.654986Z","shell.execute_reply":"2025-04-15T13:54:39.660921Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Assume hsi_ssl_model is your trained SSL model with SimCLR architecture.\n# If using the same 'device' as in training:\n#device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nhsi_model.to(device)\nhsi_model.eval()  # Set to evaluation mode\n\n# Create the fusion model.\nnum_classes = 3  # Adjust based on your dataset.\nfusion_model = DualStreamFusionModel(hsi_model, num_classes=num_classes).to(device)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:49:24.138229Z","iopub.execute_input":"2025-04-15T13:49:24.138508Z","iopub.status.idle":"2025-04-15T13:49:24.797404Z","shell.execute_reply.started":"2025-04-15T13:49:24.138487Z","shell.execute_reply":"2025-04-15T13:49:24.796521Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport shutil\n\ndef delete_all_in_working(dir_path=\"/kaggle/working\"):\n    # List all items in the directory.\n    for filename in os.listdir(dir_path):\n        file_path = os.path.join(dir_path, filename)\n        try:\n            if os.path.isfile(file_path) or os.path.islink(file_path):\n                os.unlink(file_path)  # Remove file or link\n            elif os.path.isdir(file_path):\n                shutil.rmtree(file_path)  # Remove directory and its contents\n            print(f\"Deleted: {file_path}\")\n        except Exception as e:\n            print(f\"Failed to delete {file_path}. Reason: {e}\")\n\n# Call the function (will delete everything under /kaggle/working)\ndelete_all_in_working()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:49:29.246241Z","iopub.execute_input":"2025-04-15T13:49:29.246576Z","iopub.status.idle":"2025-04-15T13:49:29.251802Z","shell.execute_reply.started":"2025-04-15T13:49:29.246528Z","shell.execute_reply":"2025-04-15T13:49:29.251101Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch.optim as optim\n\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(fusion_model.parameters(), lr=1e-4)  # Only parameters in classifier & rgb_encoder will update.\n\n# num_epochs = 10\n\n# for epoch in range(num_epochs):\n#     fusion_model.train()\n#     running_loss = 0.0\n#     total_samples = 0\n#     correct_predictions = 0\n    \n#     for hsi_imgs, rgb_imgs, labels in train_loader:\n#         hsi_imgs = hsi_imgs.to(device)   # (B, 50, H, W)\n#         rgb_imgs = rgb_imgs.to(device)   # (B, 3, H, W)\n#         labels = labels.to(device)\n        \n#         optimizer.zero_grad()\n#         outputs = fusion_model(hsi_imgs, rgb_imgs)\n#         loss = criterion(outputs, labels)\n#         loss.backward()\n#         optimizer.step()\n        \n#         running_loss += loss.item() * hsi_imgs.size(0)\n#         total_samples += hsi_imgs.size(0)\n#         _, preds = torch.max(outputs, 1)\n#         correct_predictions += (preds == labels).sum().item()\n    \n#     epoch_loss = running_loss / total_samples\n#     epoch_acc = correct_predictions / total_samples\n#     print(f\"Epoch [{epoch+1}/{num_epochs}], Loss: {epoch_loss:.4f}, Accuracy: {epoch_acc:.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:47.992159Z","iopub.execute_input":"2025-04-15T13:54:47.992473Z","iopub.status.idle":"2025-04-15T13:54:48.001198Z","shell.execute_reply.started":"2025-04-15T13:54:47.992447Z","shell.execute_reply":"2025-04-15T13:54:48.000334Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\ndef evaluate_model(model, dataloader, criterion, device):\n    model.eval()\n    running_loss = 0.0\n    correct_predictions = 0\n    total_samples = 0\n\n    with torch.no_grad():\n        for hsi_imgs, rgb_imgs, labels in dataloader:\n            hsi_imgs = hsi_imgs.to(device)\n            rgb_imgs = rgb_imgs.to(device)\n            labels = labels.to(device)\n\n            outputs = model(hsi_imgs, rgb_imgs)\n            loss = criterion(outputs, labels)\n\n            running_loss += loss.item() * hsi_imgs.size(0)\n            _, preds = torch.max(outputs, 1)\n            correct_predictions += (preds == labels).sum().item()\n            total_samples += hsi_imgs.size(0)\n\n    avg_loss = running_loss / total_samples\n    accuracy = correct_predictions / total_samples\n    return avg_loss, accuracy\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:51.112153Z","iopub.execute_input":"2025-04-15T13:54:51.112483Z","iopub.status.idle":"2025-04-15T13:54:51.118168Z","shell.execute_reply.started":"2025-04-15T13:54:51.112453Z","shell.execute_reply":"2025-04-15T13:54:51.117281Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"val_losses, val_accuracies = [], []\noptimizer = optim.Adam(fusion_model.parameters(), lr=1e-4, weight_decay=1e-5)\nnum_epochs = 20\nfor epoch in range(num_epochs):\n    fusion_model.train()\n    running_loss = 0.0\n    total_samples = 0\n    correct_predictions = 0\n\n    for hsi_imgs, rgb_imgs, labels in train_loader:\n        hsi_imgs = hsi_imgs.to(device)\n        rgb_imgs = rgb_imgs.to(device)\n        labels = labels.to(device)\n\n        optimizer.zero_grad()\n        outputs = fusion_model(hsi_imgs, rgb_imgs)\n        loss = criterion(outputs, labels)\n        loss.backward()\n        optimizer.step()\n\n        running_loss += loss.item() * hsi_imgs.size(0)\n        total_samples += hsi_imgs.size(0)\n        _, preds = torch.max(outputs, 1)\n        correct_predictions += (preds == labels).sum().item()\n\n    epoch_loss = running_loss / total_samples\n    epoch_acc = correct_predictions / total_samples\n\n    val_loss, val_acc = evaluate_model(fusion_model, val_loader, criterion, device)\n    val_losses.append(val_loss)\n    val_accuracies.append(val_acc)\n\n    print(f\"Epoch [{epoch+1}/{num_epochs}], Train Loss: {epoch_loss:.4f}, \"\n          f\"Train Acc: {epoch_acc:.4f}, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T13:54:58.849017Z","iopub.execute_input":"2025-04-15T13:54:58.849324Z","iopub.status.idle":"2025-04-15T14:13:56.897880Z","shell.execute_reply.started":"2025-04-15T13:54:58.849299Z","shell.execute_reply":"2025-04-15T14:13:56.896801Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from sklearn.metrics import confusion_matrix, classification_report\nimport seaborn as sns\nimport matplotlib.pyplot as plt\nimport numpy as np\n\ndef evaluate_on_test(model, test_loader, criterion, device, class_names=None):\n    model.eval()\n    running_loss = 0.0\n    correct_predictions = 0\n    total_samples = 0\n\n    all_labels = []\n    all_preds = []\n\n    with torch.no_grad():\n        for hsi_imgs, rgb_imgs, labels in test_loader:\n            hsi_imgs = hsi_imgs.to(device)\n            rgb_imgs = rgb_imgs.to(device)\n            labels = labels.to(device)\n\n            outputs = model(hsi_imgs, rgb_imgs)\n            loss = criterion(outputs, labels)\n\n            running_loss += loss.item() * hsi_imgs.size(0)\n            _, preds = torch.max(outputs, 1)\n            correct_predictions += (preds == labels).sum().item()\n            total_samples += hsi_imgs.size(0)\n\n            all_labels.extend(labels.cpu().numpy())\n            all_preds.extend(preds.cpu().numpy())\n\n    avg_loss = running_loss / total_samples\n    accuracy = correct_predictions / total_samples\n\n    print(f\"\\nTest Loss: {avg_loss:.4f}, Test Accuracy: {accuracy:.4f}\")\n\n    # Confusion Matrix\n    cm = confusion_matrix(all_labels, all_preds)\n    plt.figure(figsize=(8, 6))\n    sns.heatmap(cm, annot=True, fmt=\"d\", cmap=\"Blues\",\n                xticklabels=class_names if class_names else 'auto',\n                yticklabels=class_names if class_names else 'auto')\n    plt.title(\"Confusion Matrix\")\n    plt.xlabel(\"Predicted\")\n    plt.ylabel(\"True\")\n    plt.show()\n\n    # Classification Report\n    print(\"\\nClassification Report:\")\n    print(classification_report(all_labels, all_preds, target_names=class_names if class_names else None))\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T14:13:56.899439Z","iopub.execute_input":"2025-04-15T14:13:56.899742Z","iopub.status.idle":"2025-04-15T14:13:57.204096Z","shell.execute_reply.started":"2025-04-15T14:13:56.899715Z","shell.execute_reply":"2025-04-15T14:13:57.203447Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class_names = [\"Health\",\"Other\",\"Rust\"]\nevaluate_on_test(fusion_model, test_loader, criterion, device, class_names)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T14:13:57.205424Z","iopub.execute_input":"2025-04-15T14:13:57.205907Z","iopub.status.idle":"2025-04-15T14:14:05.786060Z","shell.execute_reply.started":"2025-04-15T14:13:57.205873Z","shell.execute_reply":"2025-04-15T14:14:05.785216Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\nepochs = range(1, num_epochs + 1)\n\n# Plot accuracy\nplt.figure(figsize=(10, 6))\nplt.plot(epochs, val_accuracies, label='Validation Accuracy', marker='o')\nplt.plot(epochs, [acc for acc in val_accuracies], label='Test Accuracy Snapshot (same as val)', linestyle='--', marker='x')\nplt.plot(epochs, [None if i >= len(val_accuracies) else val_accuracies[i] for i in range(num_epochs)],\n         label='(Simulated) Train Accuracy Line', linestyle=':', color='gray')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:24:47.630696Z","iopub.execute_input":"2025-04-15T18:24:47.631245Z","iopub.status.idle":"2025-04-15T18:24:47.724859Z","shell.execute_reply.started":"2025-04-15T18:24:47.631191Z","shell.execute_reply":"2025-04-15T18:24:47.723186Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Hybrid-SSRN-ViT Architecture-For-Hyperspectral-Image-Classification","metadata":{}},{"cell_type":"code","source":"import os\nimport numpy as np\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader\nfrom glob import glob\nimport cv2\nimport rasterio\nfrom tqdm import tqdm\nfrom sklearn.model_selection import train_test_split\nimport shutil","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:43:04.877548Z","iopub.execute_input":"2025-04-15T18:43:04.877908Z","iopub.status.idle":"2025-04-15T18:43:06.717748Z","shell.execute_reply.started":"2025-04-15T18:43:04.877875Z","shell.execute_reply":"2025-04-15T18:43:06.716983Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"processed_dataset_path = \"/kaggle/working/processed_images\"  # where processed npy files were saved\nsplit_dataset_path = \"/kaggle/working/split_dataset\"   ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:43:14.927735Z","iopub.execute_input":"2025-04-15T18:43:14.928298Z","iopub.status.idle":"2025-04-15T18:43:14.932075Z","shell.execute_reply.started":"2025-04-15T18:43:14.928264Z","shell.execute_reply":"2025-04-15T18:43:14.931209Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport glob\nimport torch\nimport numpy as np\nimport torchvision.transforms as T\nimport torchvision.transforms.functional as TF\n\n# -------------------------------\n# Custom Transform Definitions\n# -------------------------------\n\nclass TensorResize:\n    \"\"\"Resizes a tensor image to the target size.\"\"\"\n    def __init__(self, size):\n        self.size = size  # e.g. (128, 128) or (224, 224)\n    def __call__(self, img):\n        # img is expected to be a tensor of shape (C, H, W)\n        return torch.nn.functional.interpolate(\n            img.unsqueeze(0), size=self.size, mode='bilinear', align_corners=False\n        ).squeeze(0)\n\nclass RandomHorizontalFlipTensor:\n    \"\"\"Randomly horizontally flips a tensor image with probability p.\"\"\"\n    def __init__(self, p=0.5):\n        self.p = p\n    def __call__(self, img):\n        if torch.rand(1).item() < self.p:\n            return torch.flip(img, dims=[-1])\n        return img\n\nclass RandomVerticalFlipTensor:\n    \"\"\"Randomly vertically flips a tensor image with probability p.\"\"\"\n    def __init__(self, p=0.5):\n        self.p = p\n    def __call__(self, img):\n        if torch.rand(1).item() < self.p:\n            return torch.flip(img, dims=[-2])\n        return img\n\nclass RandomRotationTensor:\n    \"\"\"Randomly rotates a tensor image by an angle between -max_angle and max_angle degrees.\"\"\"\n    def __init__(self, max_angle=30):\n        self.max_angle = max_angle\n    def __call__(self, img):\n        angle = torch.empty(1).uniform_(-self.max_angle, self.max_angle).item()\n        return TF.rotate(img, angle)\n\nclass AddGaussianNoise:\n    \"\"\"Adds Gaussian noise to a tensor image.\"\"\"\n    def __init__(self, mean=0.0, std=0.1):\n        self.mean = mean\n        self.std = std\n    def __call__(self, img):\n        noise = torch.randn_like(img) * self.std + self.mean\n        return img + noise\n\nclass RandomContrastTensor:\n    \"\"\"Randomly adjusts contrast for a tensor image by a factor chosen uniformly.\"\"\"\n    def __init__(self, lower=0.7, upper=1.3):\n        self.lower = lower\n        self.upper = upper\n    def __call__(self, img):\n        factor = torch.empty(1).uniform_(self.lower, self.upper).item()\n        mean = torch.mean(img, dim=[1,2], keepdim=True)\n        return (img - mean) * factor + mean\n\n# -------------------------------\n# AgricultureHSIDataset Definition\n# -------------------------------\n\nclass AgricultureHSIDataset(torch.utils.data.Dataset):\n    def __init__(self, root_dir, transform=None):\n        \"\"\"\n        Args:\n            root_dir (str): Directory with images in subdirectories per category.\n            transform (callable, optional): Optional transform to be applied on a tensor sample.\n        \"\"\"\n        # If no transform is provided, use the default augmentation pipeline.\n        if transform is None:\n            transform = T.Compose([\n                TensorResize((224, 224)),\n                RandomHorizontalFlipTensor(p=0.5),\n                RandomVerticalFlipTensor(p=0.5),\n                RandomRotationTensor(max_angle=30),\n                AddGaussianNoise(mean=0.0, std=0.05),\n                RandomContrastTensor(lower=0.8, upper=1.2)\n            ])\n        self.transform = transform\n        self.root_dir = root_dir\n        # Build list of files along with labels\n        self.image_paths = []\n        self.labels = []\n        # Mapping for categories to integers; update keys if folder names differ\n        self.label_mapping = {\"Health\": 0, \"Rust\": 1, \"Other\": 2}\n        categories = os.listdir(root_dir)\n        for cat in categories:\n            cat_path = os.path.join(root_dir, cat)\n            if os.path.isdir(cat_path):\n                # Expecting npy files in each category folder.\n                for f in glob.glob(os.path.join(cat_path, \"*.npy\")):\n                    self.image_paths.append(f)\n                    self.labels.append(self.label_mapping[cat])\n                    \n    def __len__(self):\n        return len(self.image_paths)\n    \n    def __getitem__(self, idx):\n        # Load the processed npy file. Each image is expected to be stored as (bands, H, W)\n        image = np.load(self.image_paths[idx]).astype(np.float32)\n        label = self.labels[idx]\n        # Convert numpy array to torch tensor\n        image = torch.tensor(image)\n        # Apply transformations if available\n        if self.transform:\n            image = self.transform(image)\n        return image, torch.tensor(label)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:50:38.887283Z","iopub.execute_input":"2025-04-15T18:50:38.887569Z","iopub.status.idle":"2025-04-15T18:50:38.901715Z","shell.execute_reply.started":"2025-04-15T18:50:38.887549Z","shell.execute_reply":"2025-04-15T18:50:38.900796Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"batch_size = 16\n\ntrain_dataset = AgricultureHSIDataset(os.path.join(split_dataset_path, \"train\"))\nval_dataset   = AgricultureHSIDataset(os.path.join(split_dataset_path, \"val\"))\ntest_dataset  = AgricultureHSIDataset(os.path.join(split_dataset_path, \"test\"))\n\ntrain_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4)\nval_loader   = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=4)\ntest_loader  = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=4)\n\nprint(f\"Train samples: {len(train_dataset)}, Val samples: {len(val_dataset)}, Test samples: {len(test_dataset)}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:50:42.382504Z","iopub.execute_input":"2025-04-15T18:50:42.382816Z","iopub.status.idle":"2025-04-15T18:50:42.393548Z","shell.execute_reply.started":"2025-04-15T18:50:42.382792Z","shell.execute_reply":"2025-04-15T18:50:42.392532Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"SELECTED_BANDS = 100  # Ensure this matches the band selection used during processing\n\nclass SSRN(nn.Module):\n    def __init__(self, in_channels, num_features=64):\n        super(SSRN, self).__init__()\n        self.conv1 = nn.Conv2d(in_channels, num_features, kernel_size=3, padding=1)\n        self.bn1   = nn.BatchNorm2d(num_features)\n        self.relu  = nn.ReLU()\n        # Residual block\n        self.res_block = nn.Sequential(\n            nn.Conv2d(num_features, num_features, kernel_size=3, padding=1),\n            nn.BatchNorm2d(num_features),\n            nn.ReLU(),\n            nn.Conv2d(num_features, num_features, kernel_size=3, padding=1),\n            nn.BatchNorm2d(num_features)\n        )\n\n    def forward(self, x):\n        out = self.relu(self.bn1(self.conv1(x)))\n        res = self.res_block(out)\n        out = self.relu(out + res)\n        return out\n\nclass ViTHead(nn.Module):\n    def __init__(self, in_features, num_classes, embed_dim=256, num_heads=4, num_layers=2):\n        super(ViTHead, self).__init__()\n        # A convolutional patch embedding layer\n        self.patch_embed = nn.Conv2d(in_features, embed_dim, kernel_size=3, stride=2, padding=1)\n        # Transformer encoder block\n        encoder_layer = nn.TransformerEncoderLayer(d_model=embed_dim, nhead=num_heads)\n        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)\n        self.classifier = nn.Linear(embed_dim, num_classes)\n\n    def forward(self, x):\n        # x shape: (batch_size, channels, H, W)\n        x = self.patch_embed(x)   # Now: (batch_size, embed_dim, H', W')\n        batch_size, embed_dim, H_new, W_new = x.shape\n        # Flatten spatial dimensions into patches\n        x = x.flatten(2).transpose(1, 2)  # (batch_size, num_patches, embed_dim)\n        x = self.transformer_encoder(x)\n        # Aggregate features across patches (simple average)\n        x = x.mean(dim=1)\n        logits = self.classifier(x)\n        return logits\n\n# Full hybrid SSRN-ViT model combining the spectral-spatial extractor and ViT head.\nclass SSRN_ViT(nn.Module):\n    def __init__(self, in_channels, num_classes):\n        super(SSRN_ViT, self).__init__()\n        self.ssrn = SSRN(in_channels)\n        # The SSRN is configured to output 64 channels.\n        self.vit_head = ViTHead(in_features=64, num_classes=num_classes)\n\n    def forward(self, x):\n        features = self.ssrn(x)\n        logits = self.vit_head(features)\n        return logits","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:54:12.492918Z","iopub.execute_input":"2025-04-15T18:54:12.493370Z","iopub.status.idle":"2025-04-15T18:54:12.505337Z","shell.execute_reply.started":"2025-04-15T18:54:12.493333Z","shell.execute_reply":"2025-04-15T18:54:12.504217Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device('cuda:1' if torch.cuda.is_available() else 'cpu')\nprint(device)\nmodel = SSRN_ViT(in_channels=SELECTED_BANDS, num_classes=3)\nmodel = model.to(device)\nprint(model)\n\n# %% [code]\n# Setup loss, optimizer and training parameters\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.001)\nnum_epochs = 50\n\n# Define training and evaluation loops\ndef train_epoch(model, loader, optimizer, criterion, device):\n    model.train()\n    running_loss = 0.0\n    for images, labels in tqdm(loader, desc=\"Training\", leave=False):\n        images = images.to(device)\n        labels = 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() * images.size(0)\n    return running_loss / len(loader.dataset)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:56:18.958219Z","iopub.execute_input":"2025-04-15T18:56:18.958615Z","iopub.status.idle":"2025-04-15T18:56:19.008607Z","shell.execute_reply.started":"2025-04-15T18:56:18.958585Z","shell.execute_reply":"2025-04-15T18:56:19.007635Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device('cuda:1' if torch.cuda.is_available() else 'cpu')\nmodel = SSRN_ViT(in_channels=SELECTED_BANDS, num_classes=3)\nmodel = model.to(device)\nprint(model)\n\n# %% [code]\n# Setup loss, optimizer and training parameters\ncriterion = nn.CrossEntropyLoss()\noptimizer = optim.Adam(model.parameters(), lr=0.001)\nnum_epochs = 50\n\n# Define training and evaluation loops\ndef train_epoch(model, loader, optimizer, criterion, device):\n    model.train()\n    running_loss = 0.0\n    for images, labels in tqdm(loader, desc=\"Training\", leave=False):\n        images = images.to(device)\n        labels = 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() * images.size(0)\n    return running_loss / len(loader.dataset)\n\ndef eval_epoch(model, loader, criterion, device):\n    model.eval()\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    with torch.no_grad():\n        for images, labels in tqdm(loader, desc=\"Evaluating\", leave=False):\n            images = images.to(device)\n            labels = labels.to(device)\n            outputs = model(images)\n            loss = criterion(outputs, labels)\n            running_loss += loss.item() * images.size(0)\n            _, preds = torch.max(outputs, 1)\n            total += labels.size(0)\n            correct += (preds == labels).sum().item()\n    return running_loss / len(loader.dataset), 100 * correct / total","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:56:28.588363Z","iopub.execute_input":"2025-04-15T18:56:28.588707Z","iopub.status.idle":"2025-04-15T18:56:28.620620Z","shell.execute_reply.started":"2025-04-15T18:56:28.588681Z","shell.execute_reply":"2025-04-15T18:56:28.619786Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for epoch in range(num_epochs):\n    train_loss = train_epoch(model, train_loader, optimizer, criterion, device)\n    val_loss, val_acc = eval_epoch(model, val_loader, criterion, device)\n    print(f\"Epoch [{epoch+1}/{num_epochs}] - Train Loss: {train_loss:.4f}, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:56:32.047569Z","iopub.execute_input":"2025-04-15T18:56:32.047896Z","iopub.status.idle":"2025-04-15T18:56:43.308803Z","shell.execute_reply.started":"2025-04-15T18:56:32.047871Z","shell.execute_reply":"2025-04-15T18:56:43.307428Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"test_loss, test_acc = eval_epoch(model, test_loader, criterion, device)\nprint(f\"Test Loss: {test_loss:.4f}, Test Accuracy: {test_acc:.2f}%\")\n\n# %% [code]\n# Save the trained model\nmodel_save_path = \"/kaggle/working/ssrn_vit.pth\"\ntorch.save(model.state_dict(), model_save_path)\nprint(\"Model saved as\", model_save_path)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-15T18:55:53.737379Z","iopub.execute_input":"2025-04-15T18:55:53.737700Z","iopub.status.idle":"2025-04-15T18:55:58.303814Z","shell.execute_reply.started":"2025-04-15T18:55:53.737674Z","shell.execute_reply":"2025-04-15T18:55:58.302428Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}