# app.py # Purpose: Streamlit app — MultiTaskOmicsNet inference # Image only → Tumor/Normal + Genomic Profile + Diffusion Generation # Project: Omics-Guided Histopathology Analysis # Author : Ranjith Kumar import os import io import re import numpy as np import pandas as pd from PIL import Image import torch import torch.nn as nn import torch.nn.functional as F from torchvision.models import efficientnet_b0 import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import cv2 import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side import streamlit as st # ────────────────────────────────────────────── # PATHS # ────────────────────────────────────────────── BASE_DIR = os.path.dirname(os.path.abspath(__file__)) MODEL_PATH = os.path.join(BASE_DIR, "models", "multitask_omics_best.pth") DIFF_PATH = os.path.join(BASE_DIR, "diffusion","diffusion_model.pth") SAMPLES_DIR = os.path.join(BASE_DIR, "test_samples") GENOMIC_FEATURES = [ "BRCA1_mutation","TP53_mutation","HER2_amplification", "PIK3CA_mutation","CDH1_mutation","genomic_risk_score" ] GENOMIC_LABELS = ["BRCA1","TP53","HER2","PIK3CA","CDH1","Risk Score"] PCAM_MEAN = np.array([0.701, 0.538, 0.692]) PCAM_STD = np.array([0.235, 0.277, 0.213]) IMAGE_FEAT_DIM = 256 FUSION_DIM = 128 NUM_CLASSES = 2 GENOMIC_DIM = 6 DROPOUT = 0.3 COND_DIM = 64 IMAGE_SIZE = 32 CHANNELS = 3 T_STEPS = 100 # ────────────────────────────────────────────── # MULTI-TASK CLASSIFIER # ────────────────────────────────────────────── class ImageEncoder(nn.Module): def __init__(self): super().__init__() # Load architecture only — weights loaded from our trained .pth file base = efficientnet_b0(weights=None) self.features = base.features self.avgpool = base.avgpool self.projector = nn.Sequential( nn.Dropout(p=DROPOUT), nn.Linear(1280, IMAGE_FEAT_DIM), nn.BatchNorm1d(IMAGE_FEAT_DIM), nn.ReLU(inplace=True), ) def forward(self, x): x = self.features(x) x = self.avgpool(x) x = torch.flatten(x, 1) return self.projector(x) class ClassificationHead(nn.Module): def __init__(self): super().__init__() self.head = nn.Sequential( nn.Linear(IMAGE_FEAT_DIM, FUSION_DIM), nn.BatchNorm1d(FUSION_DIM), nn.ReLU(inplace=True), nn.Dropout(p=DROPOUT), nn.Linear(FUSION_DIM, NUM_CLASSES), ) def forward(self, x): return self.head(x) class GenomicRegressionHead(nn.Module): def __init__(self): super().__init__() self.head = nn.Sequential( nn.Linear(IMAGE_FEAT_DIM, FUSION_DIM), nn.BatchNorm1d(FUSION_DIM), nn.ReLU(inplace=True), nn.Dropout(p=DROPOUT), nn.Linear(FUSION_DIM, GENOMIC_DIM), nn.Sigmoid() ) def forward(self, x): return self.head(x) class MultiTaskOmicsNet(nn.Module): def __init__(self): super().__init__() self.image_encoder = ImageEncoder() self.cls_head = ClassificationHead() self.genomic_head = GenomicRegressionHead() def forward(self, images): features = self.image_encoder(images) cls_logits = self.cls_head(features) genomic_pred = self.genomic_head(features) return cls_logits, genomic_pred # ────────────────────────────────────────────── # DIFFUSION MODEL # ────────────────────────────────────────────── class SinusoidalTimeEmbedding(nn.Module): def __init__(self, dim): super().__init__() self.dim = dim def forward(self, t): device = t.device half = self.dim // 2 freqs = torch.exp( -torch.log(torch.tensor(10000.0)) * torch.arange(half, device=device) / half ) args = t[:, None].float() * freqs[None] return torch.cat([args.sin(), args.cos()], dim=-1) class ResidualBlock(nn.Module): def __init__(self, in_ch, out_ch, time_dim, cond_dim): super().__init__() self.conv1 = nn.Conv2d(in_ch, out_ch, 3, padding=1) self.conv2 = nn.Conv2d(out_ch, out_ch, 3, padding=1) self.bn1 = nn.GroupNorm(8, out_ch) self.bn2 = nn.GroupNorm(8, out_ch) self.time_mlp = nn.Linear(time_dim, out_ch) self.cond_mlp = nn.Linear(cond_dim, out_ch) self.skip = nn.Conv2d(in_ch, out_ch, 1) if in_ch != out_ch else nn.Identity() def forward(self, x, t_emb, c_emb): h = F.silu(self.bn1(self.conv1(x))) h = h + self.time_mlp(t_emb)[:, :, None, None] h = h + self.cond_mlp(c_emb)[:, :, None, None] h = F.silu(self.bn2(self.conv2(h))) return h + self.skip(x) class DownBlock(nn.Module): def __init__(self, in_ch, out_ch, time_dim, cond_dim): super().__init__() self.res = ResidualBlock(in_ch, out_ch, time_dim, cond_dim) self.down = nn.Conv2d(out_ch, out_ch, 3, stride=2, padding=1) def forward(self, x, t_emb, c_emb): x = self.res(x, t_emb, c_emb) return self.down(x), x class UpBlock(nn.Module): def __init__(self, in_ch, skip_ch, out_ch, time_dim, cond_dim): super().__init__() self.up = nn.ConvTranspose2d(in_ch, in_ch, 2, stride=2) self.res = ResidualBlock(in_ch + skip_ch, out_ch, time_dim, cond_dim) def forward(self, x, skip, t_emb, c_emb): x = self.up(x) x = torch.cat([x, skip], dim=1) return self.res(x, t_emb, c_emb) class ConditionalUNet(nn.Module): def __init__(self): super().__init__() TIME_DIM = 128 self.time_embed = nn.Sequential( SinusoidalTimeEmbedding(TIME_DIM), nn.Linear(TIME_DIM, TIME_DIM), nn.SiLU(), nn.Linear(TIME_DIM, TIME_DIM), ) self.cond_embed = nn.Sequential( nn.Linear(GENOMIC_DIM, 32), nn.SiLU(), nn.Linear(32, COND_DIM), ) self.enc1 = DownBlock(CHANNELS, 32, TIME_DIM, COND_DIM) self.enc2 = DownBlock(32, 64, TIME_DIM, COND_DIM) self.enc3 = DownBlock(64, 128, TIME_DIM, COND_DIM) self.bottleneck = ResidualBlock(128, 128, TIME_DIM, COND_DIM) self.dec3 = UpBlock(128, 128, 64, TIME_DIM, COND_DIM) self.dec2 = UpBlock( 64, 64, 32, TIME_DIM, COND_DIM) self.dec1 = UpBlock( 32, 32, 32, TIME_DIM, COND_DIM) self.out = nn.Conv2d(32, CHANNELS, 1) def forward(self, x, t, genomic): t_emb = self.time_embed(t) c_emb = self.cond_embed(genomic) x, s1 = self.enc1(x, t_emb, c_emb) x, s2 = self.enc2(x, t_emb, c_emb) x, s3 = self.enc3(x, t_emb, c_emb) x = self.bottleneck(x, t_emb, c_emb) x = self.dec3(x, s3, t_emb, c_emb) x = self.dec2(x, s2, t_emb, c_emb) x = self.dec1(x, s1, t_emb, c_emb) return self.out(x) # ────────────────────────────────────────────── # GRAD-CAM # ────────────────────────────────────────────── class GradCAM: def __init__(self, model): self.model = model self.gradients = None self.activations = None target = model.image_encoder.features[-1] target.register_forward_hook(self._save_activation) target.register_full_backward_hook(self._save_gradient) def _save_activation(self, module, input, output): self.activations = output.detach() def _save_gradient(self, module, grad_input, grad_output): self.gradients = grad_output[0].detach() def generate(self, img_tensor, class_idx=1): self.model.eval() img_tensor = img_tensor.unsqueeze(0).requires_grad_(True) cls_logits, _ = self.model(img_tensor) self.model.zero_grad() cls_logits[0, class_idx].backward() weights = self.gradients.mean(dim=[2,3], keepdim=True) cam = (weights * self.activations).sum(dim=1, keepdim=True) cam = torch.relu(cam).squeeze().numpy() cam = cv2.resize(cam, (96,96)) cam = (cam - cam.min()) / (cam.max() - cam.min() + 1e-8) return cam # ────────────────────────────────────────────── # UTILITIES # ────────────────────────────────────────────── @st.cache_resource def load_classifier(): model = MultiTaskOmicsNet() model.load_state_dict(torch.load(MODEL_PATH, map_location="cpu")) model.eval() return model @st.cache_resource def load_diffusion(): ckpt = torch.load(DIFF_PATH, map_location="cpu") diff_model = ConditionalUNet() diff_model.load_state_dict(ckpt["model_state_dict"]) diff_model.eval() return diff_model, ckpt["betas"], ckpt["alphas"], ckpt["alpha_bars"] @st.cache_data def get_sample_files(): if not os.path.exists(SAMPLES_DIR): return [] return sorted([f for f in os.listdir(SAMPLES_DIR) if f.endswith((".png",".jpg",".jpeg"))]) def extract_idx(filename): m = re.search(r"idx(\d+)", filename) return int(m.group(1)) if m else -1 def enhance_image(pil_img): img = np.array(pil_img.convert("RGB")) lab = cv2.cvtColor(img, cv2.COLOR_RGB2LAB) l, a, b = cv2.split(lab) clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8)) lab_eq = cv2.merge([clahe.apply(l), a, b]) enhanced = cv2.cvtColor(lab_eq, cv2.COLOR_LAB2RGB) sharpened = cv2.filter2D(enhanced, -1, np.array([[0,-1,0],[-1,5,-1],[0,-1,0]])) return Image.fromarray(sharpened) def preprocess(pil_img): img = pil_img.convert("RGB").resize((96,96)) img_np = np.array(img, dtype=np.float32) / 255.0 norm = (img_np - PCAM_MEAN) / PCAM_STD tensor = torch.tensor(norm.transpose(2,0,1), dtype=torch.float32) return tensor, img_np @torch.no_grad() def generate_synthetic(diff_model, genomic_vec, betas, alphas, alpha_bars, ddim_steps=50, eta=0.0): """ DDIM sampler — produces sharper images than DDPM. ddim_steps: number of inference steps (50 is enough, vs 100 for DDPM) eta=0.0: deterministic sampling (sharpest output) eta=1.0: stochastic (same as DDPM) """ T = T_STEPS g = genomic_vec.unsqueeze(0) x = torch.randn(1, CHANNELS, IMAGE_SIZE, IMAGE_SIZE) # Build DDIM timestep sequence — evenly spaced subset of [0, T] step_size = T // ddim_steps timesteps = list(reversed(range(0, T, step_size))) # e.g. [99,97,95,...,1] for i, t_idx in enumerate(timesteps): t_tensor = torch.tensor([t_idx], dtype=torch.long) pred_noise = diff_model(x, t_tensor, g) alpha_bar_t = alpha_bars[t_idx] # Predict x0 x0_pred = (x - torch.sqrt(1 - alpha_bar_t) * pred_noise) / torch.sqrt(alpha_bar_t) x0_pred = x0_pred.clamp(-1, 1) if i < len(timesteps) - 1: t_prev = timesteps[i + 1] alpha_bar_prev = alpha_bars[t_prev] else: alpha_bar_prev = torch.tensor(1.0) # DDIM update sigma = (eta * torch.sqrt((1 - alpha_bar_prev) / (1 - alpha_bar_t)) * torch.sqrt(1 - alpha_bar_t / alpha_bar_prev)) direction = torch.sqrt(1 - alpha_bar_prev - sigma**2) * pred_noise noise = sigma * torch.randn_like(x) if eta > 0 else 0 x = torch.sqrt(alpha_bar_prev) * x0_pred + direction + noise img = (x.clamp(-1, 1) + 1) / 2.0 img = F.interpolate(img, size=(96, 96), mode="bicubic", align_corners=False) return img.squeeze(0).permute(1, 2, 0).numpy() def make_excel(image_idx, filename, label, confidence, genomic_pred): wb = openpyxl.Workbook() ws = wb.active ws.title = "Prediction Report" hdr_fill = PatternFill("solid", start_color="1F4E79") hdr_font = Font(bold=True, color="FFFFFF", name="Arial", size=11) center = Alignment(horizontal="center", vertical="center") thin = Border(left=Side(style="thin"), right=Side(style="thin"), top=Side(style="thin"), bottom=Side(style="thin")) res_fill = PatternFill("solid", start_color="FFE0E0" if label=="TUMOR" else "E0F0E0") rows = [ ["Field", "Predicted Value"], ["Filename", filename], ["Image Index", image_idx], ["Prediction", label], ["Confidence", f"{confidence:.2%}"], ["─── Predicted Omics ───", "─────────────────"], ["BRCA1 Mutation", round(float(genomic_pred[0]),4)], ["TP53 Mutation", round(float(genomic_pred[1]),4)], ["HER2 Amplification", round(float(genomic_pred[2]),4)], ["PIK3CA Mutation", round(float(genomic_pred[3]),4)], ["CDH1 Mutation", round(float(genomic_pred[4]),4)], ["Genomic Risk Score", round(float(genomic_pred[5]),4)], ["─── Note ───", "Genomic values predicted from image by MultiTaskOmicsNet"], ] for r, row in enumerate(rows, 1): for c, val in enumerate(row, 1): cell = ws.cell(r, c, value=val) cell.alignment = center; cell.border = thin if r == 1: cell.font = hdr_font; cell.fill = hdr_fill else: cell.font = Font(name="Arial", size=10) cell.fill = res_fill ws.column_dimensions["A"].width = 24 ws.column_dimensions["B"].width = 45 buf = io.BytesIO() wb.save(buf); buf.seek(0) return buf # ────────────────────────────────────────────── # MAIN APP # ────────────────────────────────────────────── def main(): st.set_page_config( page_title="OmicsGuidedNet", page_icon="🔬", layout="wide" ) st.markdown("""
Multi-Task Biomarker Discovery — Image → Tumor/Normal + Genomic Profile
Confidence: {confidence:.2%}
Conditional DDPM · 100 denoising timesteps · Conditioned on predicted genomic profile · 32×32 → 96×96
" "Source: PCam test set
", unsafe_allow_html=True ) with col_g: st.image(np.clip(syn_patch,0,1), caption="🧬 Diffusion-generated patch", width=300) st.markdown( ""
"Generated by Conditional DDPM
"
"conditioned on predicted genomic profile
How this works: The MultiTaskOmicsNet first predicts the genomic profile from the image. The diffusion model then uses this predicted profile as a condition to generate a synthetic patch — showing what tissue with that genomic signature looks like.