{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":100385,"databundleVersionId":12076007,"sourceType":"competition"}],"dockerImageVersionId":31040,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"from torch.utils.data import Dataset\nimport numpy as np\nimport os\nfrom glob import glob\nimport torch\n\nclass HyperspectralDataset(Dataset):\n    def __init__(self, root_dir, transform=None):\n        self.samples = glob(os.path.join(root_dir, \"*\", \"*.npy\"))\n        self.transform = transform\n\n    def __len__(self):\n        return len(self.samples)\n\n    def __getitem__(self, idx):\n        file_path = self.samples[idx]\n        image = np.load(file_path).astype(np.float32)  # shape (H, W, C)\n        image = image / (image.max() + 1e-8)\n\n        label_str = os.path.basename(os.path.dirname(file_path))\n        label = int(label_str)\n        image = np.transpose(image, (2, 0, 1))  # (C, H, W)\n\n        image = torch.from_numpy(image)\n        if self.transform:\n            image = self.transform(image)\n\n        return image, torch.tensor(label, dtype=torch.long)\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## CVAE model","metadata":{}},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\nclass CVAE(nn.Module):\n    def __init__(self, img_channels=250, condition_dim=10, latent_dim=128, hidden_dims=None):\n        super(CVAE, self).__init__()\n        self.img_channels = img_channels\n        self.latent_dim = latent_dim\n        self.condition_dim = condition_dim\n\n        if hidden_dims is None:\n            hidden_dims = [32, 64, 128, 256]\n\n        self.condition_embed = nn.Linear(condition_dim, 64)\n\n        # Encoder\n        encoder_layers = []\n        in_channels = img_channels + 1  # 1 for condition broadcast\n        for h_dim in hidden_dims:\n            encoder_layers.append(nn.Conv2d(in_channels, h_dim, kernel_size=3, stride=2, padding=1))\n            encoder_layers.append(nn.ReLU())\n            in_channels = h_dim\n        self.encoder = nn.Sequential(*encoder_layers)\n\n        self.flatten = nn.Flatten()\n        self.fc_mu = nn.Linear(hidden_dims[-1]*8*8, latent_dim)\n        self.fc_logvar = nn.Linear(hidden_dims[-1]*8*8, latent_dim)\n\n\n        # Decoder\n        self.decoder_input = nn.Linear(latent_dim + 64, hidden_dims[-1] * 4 * 4)\n\n        hidden_dims.reverse()\n        decoder_layers = []\n        \n        for i in range(len(hidden_dims) - 1):\n            decoder_layers.append(nn.ConvTranspose2d(hidden_dims[i], hidden_dims[i + 1],\n                                                     kernel_size=4, stride=2, padding=1))\n            decoder_layers.append(nn.ReLU())\n        \n        # 32 → 16 → 125\n        decoder_layers.append(nn.ConvTranspose2d(hidden_dims[-1], 64, kernel_size=4, stride=2, padding=1))\n        decoder_layers.append(nn.ReLU())\n        \n        decoder_layers.append(nn.ConvTranspose2d(64, img_channels, kernel_size=4, stride=2, padding=1))\n        decoder_layers.append(nn.Sigmoid())\n        self.decoder = nn.Sequential(*decoder_layers)\n\n#\n    def encode(self, x, c):\n        B, _, H, W = x.shape\n        c_broadcast = c.argmax(dim=1).view(B, 1, 1, 1).float().expand(-1, 1, H, W)\n        x_cond = torch.cat([x, c_broadcast], dim=1)\n        x_enc = self.encoder(x_cond)\n        x_flat = self.flatten(x_enc)\n        mu = self.fc_mu(x_flat)\n        logvar = self.fc_logvar(x_flat)\n        logvar = torch.clamp(logvar, min=-10, max=10)\n\n        return mu, logvar\n\n    def reparameterize(self, mu, logvar):\n        std = torch.exp(0.5 * logvar)\n        eps = torch.randn_like(std)\n        return mu + eps * std\n\n    def decode(self, z, c):\n        c_embed = self.condition_embed(c)\n        zc = torch.cat([z, c_embed], dim=1)\n        x = self.decoder_input(zc)\n        x = x.view(x.size(0), -1, 4, 4)\n        x = self.decoder(x)\n        return x\n\n    def forward(self, x, c):\n        mu, logvar = self.encode(x, c)\n        z = self.reparameterize(mu, logvar)\n        recon = self.decode(z, c)\n        return recon, mu, logvar\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Loss Function","metadata":{}},{"cell_type":"code","source":"def vae_loss(recon_x, x, mu, logvar):\n    recon_loss = F.mse_loss(recon_x, x, reduction='mean')\n    kld_loss = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp()) / x.size(0)\n    return recon_loss + kld_loss\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Training Loop","metadata":{}},{"cell_type":"code","source":"from torch.utils.data import DataLoader\nimport torch.optim as optim\nfrom tqdm import tqdm\n\ndef train_cvae(model, dataloader, device, num_epochs=30, lr=1e-3):\n    optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)\n    model.to(device)\n\n    for epoch in range(num_epochs):\n        model.train()\n        running_loss = 0.0\n        for x, labels in tqdm(dataloader, desc=f\"Epoch {epoch+1}/{num_epochs}\"):\n            x = x.to(device)\n            c = F.one_hot(labels, num_classes=10).float().to(device)\n\n            recon, mu, logvar = model(x, c)\n            loss = vae_loss(recon, x, mu, logvar)\n\n            optimizer.zero_grad()\n            loss.backward()\n            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)\n\n            optimizer.step()\n            running_loss += loss.item()\n\n        avg_loss = running_loss / len(dataloader)\n        print(f\"Epoch [{epoch+1}/{num_epochs}] - Loss: {avg_loss:.4f}\")\n     ","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Usage","metadata":{}},{"cell_type":"code","source":"# Setup\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\ndataset = HyperspectralDataset(root_dir=\"/kaggle/input/beyond-visible-spectrum-ai-for-agriculture-2025p2/Train\")\ndataloader = DataLoader(dataset, batch_size=16, shuffle=True, num_workers=2)\n\nmodel = CVAE(img_channels=125)\ntrain_cvae(model, dataloader, device, num_epochs=10)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Generation","metadata":{}},{"cell_type":"code","source":"def generate_samples(model, disease_level, num_samples=50, latent_dim=128, device='cpu'):\n    model.eval()\n    with torch.no_grad():\n        z = torch.randn(num_samples, latent_dim).to(device)\n        labels = torch.full((num_samples,), disease_level, dtype=torch.long).to(device)\n        c = F.one_hot(labels, num_classes=10).float()\n        samples = model.decode(z, c)\n        return samples.cpu().numpy()  # (B, C, H, W)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Submission","metadata":{}},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"code","source":"import torch\n\n# Normalized SRFs for 125 bands\nSRF_GREEN = torch.tensor([\n    0.0000,0.0000,0.0000,0.0000,0.0001,0.0002,0.0005,0.0008,0.0014,0.0024,0.0041,\n    0.0069,0.0113,0.0180,0.0279,0.0414,0.0583,0.0783,0.1008,0.1252,0.1507,0.1766,\n    0.2023,0.2271,0.2505,0.2721,0.2913,0.3079,0.3216,0.3324,0.3404,0.3459,0.3495,\n    0.3516,0.3528,0.3533,0.3535,0.3536,0.3538,0.3539,0.3541,0.3542,0.3542,0.3541,\n    0.3535,0.3520,0.3491,0.3443,0.3373,0.3277,0.3152,0.2997,0.2811,0.2595,0.2349,\n    0.2076,0.1778,0.1462,0.1140,0.0823,0.0524,0.0259,0.0037,0.0003,0.0000,0.0000,\n    0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,\n]).float()\nSRF_RED = torch.tensor([\n    0.0000,0.0000,0.0000,0.0000,0.0001,0.0002,0.0003,0.0006,0.0012,0.0024,0.0047,\n    0.0087,0.0154,0.0255,0.0395,0.0575,0.0786,0.1020,0.1265,0.1505,0.1732,0.1940,\n    0.2121,0.2269,0.2381,0.2454,0.2491,0.2494,0.2466,0.2409,0.2326,0.2219,0.2093,\n    0.1952,0.1799,0.1639,0.1476,0.1314,0.1157,0.1008,0.0870,0.0744,0.0629,0.0525,\n    0.0430,0.0344,0.0266,0.0195,0.0129,0.0070,0.0018,0.0003,0.0000,0.0000,0.0000,\n    0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,\n]).float()\nSRF_NIR = torch.tensor([\n    0.0000,0.0000,0.0000,0.0000,0.0000,0.0001,0.0002,0.0003,0.0006,0.0011,0.0022,\n    0.0041,0.0073,0.0125,0.0204,0.0317,0.0470,0.0666,0.0905,0.1185,0.1500,0.1841,\n    0.2196,0.2554,0.2900,0.3219,0.3495,0.3715,0.3870,0.3950,0.3950,0.3872,0.3721,\n    0.3503,0.3228,0.2912,0.2573,0.2228,0.1888,0.1563,0.1261,0.0990,0.0755,0.0557,\n    0.0395,0.0265,0.0162,0.0082,0.0023,0.0003,0.0000,0.0000,0.0000,0.0000,0.0000,\n    0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,0.0000,\n]).float()\n\n# Normalize SRFs\nSRF_TABLE = {\n    'green': SRF_GREEN / SRF_GREEN.sum(),\n    'red'  : SRF_RED / SRF_RED.sum(),\n    'nir'  : SRF_NIR / SRF_NIR.sum(),\n}\n\n\nimport torch.nn.functional as F\n\ndef interpolate_srf(srf, target_len):\n    srf = srf.view(1, 1, -1)  # (1, 1, original_len)\n    srf = F.interpolate(srf, size=target_len, mode='linear', align_corners=False)\n    return srf.view(-1)\n\ndef hs_to_rgb(hs_img: torch.Tensor) -> torch.Tensor:\n    \"\"\"\n    Project a hyperspectral image (C, H, W) to RGB using interpolated SRFs.\n    Returns an RGB image of shape (3, H, W)\n    \"\"\"\n    assert hs_img.dim() == 3, \"Expected (C, H, W)\"\n    C, H, W = hs_img.shape\n    device = hs_img.device\n\n    rgb = []\n    for band, srf in zip(['green', 'red', 'nir'], [SRF_GREEN, SRF_RED, SRF_NIR]):\n        w = interpolate_srf(srf, C).to(device)\n        w = w / (w.sum() + 1e-8)  # normalize\n        w = w.view(C, 1, 1)\n        channel = (hs_img * w).sum(0)\n        rgb.append(channel)\n\n    return torch.stack(rgb)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from torchvision.models import inception_v3, Inception_V3_Weights\nimport torch.nn as nn\n\nclass InceptionPool3(nn.Module):\n    def __init__(self, device):\n        super().__init__()\n        weights = Inception_V3_Weights.IMAGENET1K_V1\n        net = inception_v3(weights=weights, aux_logits=True, transform_input=False).to(device)\n        net.eval()\n\n        net.AuxLogits = nn.Identity()  # Remove aux\n        self.stem_and_blocks = nn.Sequential(*list(net.children())[:-2])  # Cut off classifier\n\n    def forward(self, x):  # x: (B, 3, H, W)\n        with torch.no_grad():\n            x = F.interpolate(x, size=(299, 299), mode='bilinear', align_corners=False)\n            x = self.stem_and_blocks(x)\n            x = F.adaptive_avg_pool2d(x, output_size=1)\n            return x.view(x.size(0), -1)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## FID calculation","metadata":{}},{"cell_type":"code","source":"from scipy.linalg import sqrtm\nimport numpy as np\n\ndef compute_fid(mu1, sigma1, mu2, sigma2, eps=1e-6):\n    \"\"\"Compute Fréchet Inception Distance.\"\"\"\n    diff = mu1 - mu2\n    covmean, _ = sqrtm(sigma1 @ sigma2, disp=False)\n    if not np.isfinite(covmean).all():\n        covmean = sqrtm((sigma1 + eps * np.eye(sigma1.shape[0])) @ \n                        (sigma2 + eps * np.eye(sigma2.shape[0])))\n    if np.iscomplexobj(covmean):\n        covmean = covmean.real\n    return diff.dot(diff) + np.trace(sigma1 + sigma2 - 2.0 * covmean)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport pandas as pd\nfrom tqdm import tqdm\nimport torch\nimport numpy as np\n\ndef generate_submission(model, fid_model, real_eval_dir, output_dir, device, num_levels=10, num_samples=50):\n    model.eval()\n    fid_model.eval()\n    latent_dim = model.latent_dim\n    fid_scores = []\n\n    os.makedirs(output_dir, exist_ok=True)\n\n    for level in tqdm(range(num_levels), desc=\"Generating submission\"):\n        z = torch.randn(num_samples, latent_dim).to(device)\n        labels = torch.full((num_samples,), level, dtype=torch.long).to(device)\n        c = F.one_hot(labels, num_classes=num_levels).float().to(device)\n\n        with torch.no_grad():\n            fake_hsi = model.decode(z, c).cpu()\n\n        # Save generated .npy files (optional)\n        for i in range(num_samples):\n            out_path = os.path.join(output_dir, f\"gen_{level}_{i}.npy\")\n            np.save(out_path, fake_hsi[i].numpy())\n\n        # Convert generated HSI to RGB\n        fake_rgb = torch.stack([hs_to_rgb(img) for img in fake_hsi]).to(device)\n\n        # Load and convert real HSI to RGB\n        real_rgb = []\n        real_folder = os.path.join(real_eval_dir, str(level))\n        for f in sorted(os.listdir(real_folder))[:num_samples]:\n            real = torch.tensor(np.load(os.path.join(real_folder, f))).float()\n            real_rgb.append(hs_to_rgb(real))\n        real_rgb = torch.stack(real_rgb).to(device)\n\n        # Compute Inception features\n        f_fake = fid_model(fake_rgb).cpu().numpy()\n        f_real = fid_model(real_rgb).cpu().numpy()\n\n        mu_fake, sigma_fake = f_fake.mean(0), np.cov(f_fake, rowvar=False)\n        mu_real, sigma_real = f_real.mean(0), np.cov(f_real, rowvar=False)\n\n        fid = compute_fid(mu_fake, sigma_fake, mu_real, sigma_real)\n        fid_scores.append(fid)\n\n    # Pad to 20 rows as required (duplicate second set for now)\n    full_fid_scores = fid_scores + fid_scores  # IDs 1–10 and 11–20\n\n    submission_df = pd.DataFrame({\n        'ID': list(range(1, 21)),\n        'Prediction': full_fid_scores\n    })\n    submission_df.to_csv(os.path.join(output_dir, \"submission.csv\"), index=False)\n    return submission_df\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"","metadata":{}},{"cell_type":"code","source":"!ls ../input/beyond-visible-spectrum-ai-for-agriculture-2025p2/evaluation\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nfid_model = InceptionPool3(device)\n\nsubmission = generate_submission(\n    model=model,  \n    fid_model=fid_model,\n    real_eval_dir=\"../input/beyond-visible-spectrum-ai-for-agriculture-2025p2/evaluation\",  # replace with actual path\n    output_dir=\"./generated_submission\",\n    device=device\n)\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!ls generated_submission","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"##!rm generated_submission/*npy","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!head generated_submission/submission.csv","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}