import argparse import csv import os import random import torch import torch.nn as nn from PIL import Image from torch.utils.data import Dataset, DataLoader, random_split DATA_DIR = "./data" CSV_PATH = os.path.join(DATA_DIR, "dataset.csv") IMAGES_DIR = os.path.join(DATA_DIR, "images") CHECKPOINT_DIR = "./checkpoints" CHECKPOINT_PATH = os.path.join(CHECKPOINT_DIR, "fusion_model.pth") CONFIDENCE_THRESHOLD = 0.75 CSV_COLUMNS = ["image_path", "source", "symptoms", "diagnosis", "labels"] def load_label_list(): if not os.path.exists(CSV_PATH): return [] labels = set() with open(CSV_PATH, newline="", encoding="utf-8") as f: for row in csv.DictReader(f): d = row.get("diagnosis", "").strip().lower() if d: labels.add(d) return sorted(labels) def prepare_data(): from datasets import load_dataset os.makedirs(IMAGES_DIR, exist_ok=True) file_exists = os.path.exists(CSV_PATH) if not file_exists: with open(CSV_PATH, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(CSV_COLUMNS) rows_written = 0 sources_done = [] # ── IU-Xray (image + question + report) ── print("Downloading IU-Xray from Hugging Face...") iuxray = load_dataset("ayyuce/Indiana_University_Chest_X-ray_Collection", split="train") written = 0 for i, example in enumerate(iuxray): symptoms = (example.get("question") or "").strip() diagnosis = (example.get("report") or "").strip() image = example.get("image") if not symptoms or not diagnosis or image is None: continue image_path = os.path.join(IMAGES_DIR, f"iu_xray_{i}.jpg") image.convert("RGB").save(image_path) with open(CSV_PATH, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow([image_path, "iu_xray", symptoms, diagnosis, ""]) written += 1 print(f" IU-Xray: {written} rows") rows_written += written sources_done.append(f"iu_xray ({written})") # ── NIH Chest X-ray (image + disease labels) ── print("Downloading NIH Chest X-ray from Hugging Face...") nih = load_dataset("g-ronimo/NIH-Chest-X-ray-dataset_resized300px", split="train", streaming=True) label_names = [ "No Finding", "Atelectasis", "Cardiomegaly", "Effusion", "Infiltration", "Mass", "Nodule", "Pneumonia", "Pneumothorax", "Consolidation", "Edema", "Emphysema", "Fibrosis", "Pleural_Thickening", "Hernia" ] written = 0 for i, example in enumerate(nih): if written >= 3000: break image = example.get("image") label_indices = example.get("labels", []) if image is None or not label_indices: continue label_str = "|".join(label_names[idx] for idx in label_indices) primary_diagnosis = label_names[label_indices[0]] image_path = os.path.join(IMAGES_DIR, f"nih_{i}.jpg") image.convert("RGB").save(image_path) with open(CSV_PATH, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow([image_path, "nih", "", primary_diagnosis, label_str]) written += 1 if written % 500 == 0: print(f" NIH progress: {written}...") print(f" NIH: {written} rows") rows_written += written sources_done.append(f"nih ({written})") print(f"Done. Total: {rows_written} rows written to {CSV_PATH}") print(f"Sources: {', '.join(sources_done)}") def add_data(image_path, symptoms, diagnosis, labels=""): os.makedirs(DATA_DIR, exist_ok=True) file_exists = os.path.exists(CSV_PATH) with open(CSV_PATH, "a", newline="", encoding="utf-8") as f: writer = csv.writer(f) if not file_exists: writer.writerow(CSV_COLUMNS) writer.writerow([image_path, "user", symptoms, diagnosis, labels]) print(f"Added 1 row to {CSV_PATH}: diagnosis='{diagnosis}'") class FusionDataset(Dataset): def __init__(self, csv_path, label_list): self.rows = [] with open(csv_path, newline="", encoding="utf-8") as f: for row in csv.DictReader(f): d = row.get("diagnosis", "").strip().lower() if d: self.rows.append(row) self.label_list = label_list def __len__(self): return len(self.rows) def __getitem__(self, idx): row = self.rows[idx] image = Image.open(row["image_path"]).convert("RGB") symptoms = row.get("symptoms", "").strip() label_idx = self.label_list.index(row["diagnosis"].strip().lower()) return image, symptoms, label_idx def collate_fn(batch): images = [item[0] for item in batch] symptoms = [item[1] for item in batch] labels = torch.tensor([item[2] for item in batch], dtype=torch.long) return images, symptoms, labels class DiagnosisFusionModel(nn.Module): def __init__(self, num_conditions): super().__init__() from transformers import CLIPModel, CLIPProcessor, AutoTokenizer, AutoModel self.image_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32") self.image_encoder = CLIPModel.from_pretrained("openai/clip-vit-base-patch32") self.symptom_tokenizer = AutoTokenizer.from_pretrained("emilyalsentzer/Bio_ClinicalBERT") self.symptom_encoder = AutoModel.from_pretrained("emilyalsentzer/Bio_ClinicalBERT") for param in self.image_encoder.parameters(): param.requires_grad = False for param in self.symptom_encoder.parameters(): param.requires_grad = False self.classifier = nn.Sequential( nn.Linear(512 + 768, 256), nn.ReLU(), nn.Dropout(0.2), nn.Linear(256, num_conditions), ) def encode_images(self, images): inputs = self.image_processor(images=images, return_tensors="pt") with torch.no_grad(): return self.image_encoder.get_image_features(**inputs) def encode_symptoms(self, symptom_texts): inputs = self.symptom_tokenizer( symptom_texts, return_tensors="pt", padding=True, truncation=True, max_length=64 ) with torch.no_grad(): outputs = self.symptom_encoder(**inputs) return outputs.last_hidden_state.mean(dim=1) def forward(self, images, symptom_texts): image_vecs = self.encode_images(images) symptom_vecs = self.encode_symptoms(symptom_texts) combined = torch.cat([image_vecs, symptom_vecs], dim=-1) return self.classifier(combined) def train(epochs, batch_size, lr, val_split, use_amp, grad_accum): from rich.console import Console from rich.table import Table from rich.progress import Progress, BarColumn, TextColumn, TimeElapsedColumn _console = Console() has_gpu = torch.cuda.is_available() use_amp = use_amp and has_gpu scaler = torch.cuda.amp.GradScaler() if use_amp else None label_list = load_label_list() if not label_list: _console.print("[red]No data found. Run --mode prepare-data or --mode add-data first.[/red]") return dataset = FusionDataset(CSV_PATH, label_list) val_size = max(int(val_split * len(dataset)), 1) train_size = len(dataset) - val_size train_subset, val_subset = random_split(dataset, [train_size, val_size]) train_loader = DataLoader(train_subset, batch_size=batch_size, shuffle=True, collate_fn=collate_fn) val_loader = DataLoader(val_subset, batch_size=batch_size, shuffle=False, collate_fn=collate_fn) _console.print(f"[bold cyan]Training Setup[/bold cyan]") _console.print(f" Classes: {len(label_list)}") _console.print(f" Train/Val: {len(train_subset)}/{len(val_subset)}") _console.print(f" Batch size: {batch_size} Grad accum: {grad_accum}") _console.print(f" Device: {'GPU' if has_gpu else 'CPU'} AMP: {'ON' if use_amp else 'OFF'}") model = DiagnosisFusionModel(num_conditions=len(label_list)) if has_gpu: model = model.cuda() optimizer = torch.optim.AdamW(model.classifier.parameters(), lr=lr) loss_fn = nn.CrossEntropyLoss() for epoch in range(epochs): _console.print(f"\n[bold yellow]Epoch {epoch + 1}/{epochs}[/bold yellow]") _console.print("-" * 40) # ── Train ── model.train() train_loss = 0.0 optimizer.zero_grad() train_progress = Progress( TextColumn("[cyan] Train[/cyan]"), BarColumn(), TextColumn("{task.completed}/{task.total}"), TextColumn("[green]{task.fields[loss]:.4f}[/green]"), TimeElapsedColumn(), transient=True, ) with train_progress: task = train_progress.add_task("", total=len(train_loader), loss=0.0) for i, (images, symptoms, labels) in enumerate(train_loader): if has_gpu: labels = labels.cuda() with torch.amp.autocast("cuda", enabled=use_amp): logits = model(images, symptoms) loss = loss_fn(logits, labels) loss = loss / grad_accum if use_amp: scaler.scale(loss).backward() else: loss.backward() if (i + 1) % grad_accum == 0 or (i + 1) == len(train_loader): if use_amp: scaler.step(optimizer) scaler.update() else: optimizer.step() optimizer.zero_grad() train_loss += loss.item() * grad_accum train_progress.update(task, advance=1, loss=loss.item() * grad_accum) avg_train_loss = train_loss / len(train_loader) # ── Validation ── model.eval() val_loss = 0.0 with torch.no_grad(): for images, symptoms, labels in val_loader: if has_gpu: labels = labels.cuda() logits = model(images, symptoms) loss = loss_fn(logits, labels) val_loss += loss.item() avg_val_loss = val_loss / len(val_loader) table = Table(show_header=False, box=None) table.add_column("Metric", style="cyan") table.add_column("Value", style="green") table.add_row("Train loss", f"{avg_train_loss:.4f}") table.add_row("Val loss", f"{avg_val_loss:.4f}") _console.print(table) os.makedirs(CHECKPOINT_DIR, exist_ok=True) torch.save({"model_state": model.classifier.state_dict(), "label_list": label_list}, CHECKPOINT_PATH) _console.print(f"[green]Saved checkpoint to {CHECKPOINT_PATH}[/green]") def test(batch_size): if not os.path.exists(CHECKPOINT_PATH): print("No checkpoint found. Run --mode train first.") return checkpoint = torch.load(CHECKPOINT_PATH, weights_only=False) label_list = checkpoint["label_list"] model = DiagnosisFusionModel(num_conditions=len(label_list)) model.classifier.load_state_dict(checkpoint["model_state"]) model.eval() dataset = FusionDataset(CSV_PATH, label_list) test_size = max(int(0.2 * len(dataset)), 1) _, test_subset = random_split(dataset, [len(dataset) - test_size, test_size]) loader = DataLoader(test_subset, batch_size=batch_size, shuffle=False, collate_fn=collate_fn) correct = 0 inconclusive = 0 total = 0 with torch.no_grad(): for images, symptoms, labels in loader: logits = model(images, symptoms) probs = torch.softmax(logits, dim=-1) confidence, predicted = torch.max(probs, dim=-1) for i in range(len(labels)): total += 1 if confidence[i].item() < CONFIDENCE_THRESHOLD: inconclusive += 1 elif predicted[i].item() == labels[i].item(): correct += 1 print(f"Tested on {total} held-out examples") print(f"Correct (above confidence threshold): {correct} ({100 * correct / total:.1f}%)") print(f"Flagged as inconclusive / needs follow-up: {inconclusive} ({100 * inconclusive / total:.1f}%)") def info(): if not os.path.exists(CSV_PATH): print("No dataset.csv found. Run --mode prepare-data first.") return sources = {} total = 0 with open(CSV_PATH, newline="", encoding="utf-8") as f: for row in csv.DictReader(f): src = row.get("source", "unknown") sources[src] = sources.get(src, 0) + 1 total += 1 print(f"Dataset: {CSV_PATH}") print(f"Total rows: {total}") for src, count in sorted(sources.items()): print(f" {src}: {count}") print(f"Images dir: {IMAGES_DIR}") img_count = len([x for x in os.listdir(IMAGES_DIR) if os.path.isfile(os.path.join(IMAGES_DIR, x))]) if os.path.exists(IMAGES_DIR) else 0 print(f"Images: {img_count}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Train/test the medical image+symptom fusion model") parser.add_argument("--mode", required=True, choices=["prepare-data", "add-data", "train", "test", "info"]) parser.add_argument("--image", help="Path to an image file (for --mode add-data)") parser.add_argument("--symptoms", help="Symptom description text (for --mode add-data)") parser.add_argument("--diagnosis", help="Diagnosis label (for --mode add-data)") parser.add_argument("--epochs", type=int, default=5) parser.add_argument("--batch_size", type=int, default=8) parser.add_argument("--lr", type=float, default=1e-3) parser.add_argument("--val_split", type=float, default=0.15, help="Fraction of data for validation") parser.add_argument("--use_amp", action="store_true", help="Enable mixed precision (GPU only)") parser.add_argument("--grad_accum", type=int, default=1, help="Gradient accumulation steps") args = parser.parse_args() if args.mode == "prepare-data": prepare_data() elif args.mode == "add-data": if not (args.image and args.symptoms and args.diagnosis): print("--mode add-data requires --image, --symptoms, and --diagnosis") else: add_data(args.image, args.symptoms, args.diagnosis) elif args.mode == "train": train(args.epochs, args.batch_size, args.lr, args.val_split, args.use_amp, args.grad_accum) elif args.mode == "test": test(args.batch_size) elif args.mode == "info": info()