Introduction
Medical image segmentation stands as one of the most critical yet challenging tasks in modern diagnostic imaging. Whether identifying tumors in breast ultrasounds, delineating pathologies in brain MRIs, or precisely outlining lung regions in CT scans, the ability to automatically segment anatomical structures with high accuracy directly impacts clinical decision-making and patient outcomes. However, traditional deep learning approaches have consistently grappled with three fundamental limitations: the scarcity of annotated training datasets, poor generalization across different imaging modalities, and an overwhelming dependence on task-specific models.
MedCLIP-SAMv2 represents a significant advancement in this field, introducing a novel framework that integrates CLIP and SAM models to perform segmentation on clinical scans using text prompts, in both zero-shot and weakly supervised settings. This breakthrough framework achieves remarkable performance improvements—increasing Dice scores by 13.07% in zero-shot settings and 11.21% in weakly supervised paradigms compared to its predecessor—while simultaneously reducing the need for extensive labeled datasets.
Key Breakthrough: The framework combines the semantic understanding of vision-language models with the flexible prompting architecture of foundation models to create a truly universal segmentation tool.
The Foundation: Understanding CLIP, SAM, and Vision-Language Models
What Makes Foundation Models Revolutionary
Foundation models have fundamentally transformed computer vision and natural language processing by learning rich, generalizable representations from massive unlabeled datasets. CLIP (Contrastive Language-Image Pre-Training) and Segment Anything Model (SAM) have paved the way for interactive and universal medical image segmentation, yet adapting these powerful tools to medical imaging remains non-trivial.
The core challenge lies in the inherent differences between natural images and medical scans. Medical images contain subtle anatomical variations, complex terminology, and domain-specific visual patterns that generic models struggle to recognize. This is where domain-specific models like BiomedCLIP—pre-trained on vast biomedical literature and medical image-text pairs—provide transformative advantages.
Why Text-Driven Prompting Matters
Traditional segmentation methods require either:
- Fully supervised approaches: Extensive pixel-level annotations (expensive and time-consuming)
- Interactive methods: Point or bounding box prompts requiring clinical expertise
- Task-specific models: Separate models for each anatomical structure or imaging modality
Text-driven segmentation transcends these limitations by leveraging natural language descriptions—the same vocabulary clinicians use daily. This approach enables non-technical users to guide segmentation through intuitive textual prompts, dramatically improving accessibility and reducing expert dependency.
The DHN-NCE Loss: Smarter Model Fine-Tuning
The Problem with Standard Contrastive Learning
Traditional CLIP training employs InfoNCE loss, which applies uniform penalties to all negative samples regardless of difficulty. This approach has critical drawbacks, particularly in medical imaging:
Standard Challenges:
- Hard negatives (similar but incorrect samples) receive the same penalty as easy negatives, reducing learning efficiency
- The negative-positive coupling effect diminishes gradient flow for challenging cases
- Small batch sizes—common in medical imaging due to memory constraints—reduce the diversity of negative samples
Introducing Decoupled Hard Negative Noise Contrastive Estimation (DHN-NCE)
MedCLIP-SAMv2 introduces a novel loss function that addresses these limitations through three key innovations:
$$L_{DHN-NCE} = L^{v \to t} + L^{t \to v}$$
where the vision-to-text and text-to-vision losses are formulated as:
$$L^{v \to t} = -\sum_{i=1}^{B} \frac{I_{p,i}T_{p,i}^{\top}}{\tau} + \sum_{i=1}^{B} \log \left(\sum_{j \neq i} e^{I_{p,i}T_{p,j}^{\top}/\tau} W_{I_{p,i}T_{p,j}}^{v \to t}\right)$$
The hardness weighting formula amplifies the penalty for challenging negatives:
$$W_{I_{p,i}T_{p,j}}^{v \to t} = (B-1) \times \frac{e^{\beta_1 I_{p,i}T_{p,j}/\tau}}{\sum_{k \neq i} e^{\beta_1 I_{p,i}T_{p,k}/\tau}}$$
Three Core Improvements:
- Positive-Negative Decoupling: Removes positive pairs from the denominator, preventing easy positives from suppressing gradients for hard negatives
- Hard Negative Sampling: Uses exponential scaling to prioritize challenging samples that are actually similar to the anchor
- Adaptive Weighting: Dynamically adjusts penalties based on sample difficulty, enabling efficient training with smaller batches
Performance Impact: BiomedCLIP fine-tuned with DHN-NCE reached 84.70% top-1 and 94.73% top-2 in image-to-text retrieval, and 85.99% top-1 and 95.17% top-2 in text-to-image retrieval, significantly outperforming other loss functions.
Zero-Shot Segmentation: From Text Prompts to Anatomical Masks
The Multi-Modal Information Bottleneck (M2IB) Pipeline
Zero-shot segmentation in MedCLIP-SAMv2 follows a carefully orchestrated pipeline that transforms natural language descriptions into precise segmentation masks without any task-specific training.
Pipeline Architecture:
- Image and Text Embedding Extraction: The fine-tuned BiomedCLIP encoders extract high-dimensional embeddings from both medical images and text prompts, creating aligned representations in a shared semantic space.
- LLM-Enhanced Prompt Generation: Rather than using simple class names, the framework utilizes GPT-4 to generate sophisticated text prompts that guide the model to localize certain salient regions. Prompts follow the structure: “Give a sentence caption that describes unique visual features of [TARGET] in [MODALITY]”
- Saliency Map Generation via M2IB: The embeddings pass through the Multi-modal Information Bottleneck module, which learns to align image and text modalities while filtering irrelevant information. The module optimizes:
$$\lambda_S = MI(Z_{img}, Z_{text}; \theta) – \gamma \times MI(Z_{img}, I; \theta)$$
This formulation maximizes mutual information between image embeddings and text prompts while minimizing information redundant with the original image, producing a continuous saliency map highlighting regions of interest.
- Intelligent Post-Processing: The continuous saliency map is converted to a binary segmentation through:
- Otsu’s thresholding: Automatically determines optimal threshold minimizing intra-class variance
- Connected component analysis: Removes small, disconnected clusters and retains only high-confidence regions
- SAM refinement: Visual prompts (bounding boxes or points) derived from post-processed clusters guide SAM’s fine-grained mask generation
Prompt Engineering Insights: Class-specific prompts (P3) generally yielded better results for smaller structures like breast and brain tumors whereas generic prompts (P0, P2) performed better for larger structures like lungs in X-ray and CT scans. This finding underscores the importance of tailoring language to task complexity.
Real-World Performance Across Modalities
The framework was validated on four diverse medical imaging tasks:
| Modality | Dataset | Zero-Shot DSC | Weakly Supervised DSC | Best SOTA Baseline |
|---|---|---|---|---|
| Breast Ultrasound | BUSI + UDIAT | 77.76% | 78.87% | SaLIP (44.33%) |
| Brain MRI | Brain Tumor | 76.52% | 80.03% | SaLIP (47.96%) |
| Lung X-ray | COVID-QU-Ex | 75.79% | 80.77% | SaLIP (63.14%) |
| Lung CT | Konya 2020 | 80.38% | 88.78% | SAMAug (44.61%) |
These results demonstrate robust cross-modal generalization—a critical requirement for clinical deployment.
Weakly Supervised Refinement: Uncertainty-Aware Pseudo-Labeling
While zero-shot segmentation provides impressive initial results, real-world clinical deployment demands even higher accuracy. MedCLIP-SAMv2 addresses this through an innovative weakly supervised pipeline that leverages uncertainty estimation.
The Checkpoint Ensembling Strategy
The training process is divided into D cycles composed of E_d epochs, and during each cycle, checkpoints of the model are saved, providing a Monte-Carlo-like approximation of uncertainty. Rather than training multiple independent models, this approach extracts diverse predictions from a single model at different training stages:
$$p(Y_{final}|X;T) \approx \frac{1}{G} \sum_{n=1}^{G} p(Y_{final}|X; M_n)$$
where $M_n$ represents the model weights at the n-th checkpoint, and $G$ represents the total number of saved checkpoints (typically 30 across 3 training cycles).
Uncertainty Quantification via Entropy
For each pixel, uncertainty is computed using Shannon entropy across class predictions:
$$H(Y_{final,(i,j)}) = -\sum_{r=1}^{R} h(r) \log h(r)$$
where h(r) = p(Y{final,(i,j)} = r|X;T) represents the probability of class r at pixel (i,j).
Clinical Value: High-uncertainty regions—typically at anatomical boundaries or in presence of artifacts—are automatically flagged for clinical review. This transparency builds trust with clinicians and enables safer deployment.
Comparative Performance and Key Findings
Ablation Study: Each Component’s Contribution
The framework’s power emerges from synergistic component integration:
Performance Progression (Average Across All Datasets):
- Baseline saliency maps: 46.23% DSC
- DHN-NCE fine-tuning: 49.10% DSC (+2.87%)
- Post-processing: 51.62% DSC (+2.52%)
- Connected component analysis: 57.89% DSC (+6.27%)
- SAM integration: 77.61% DSC (+19.72%)
- nnUNet ensemble: 82.11% DSC (+4.50%)
Key Takeaway: SAM integration provides the most substantial performance jump, demonstrating that coarse saliency maps—even imperfect ones—serve as excellent starting points for SAM’s sophisticated segmentation refinement.
Why M2IB Outperforms Alternative Saliency Methods
M2IB achieved the highest performance across all tasks, with an average DSC of 77.61% and NSD of 81.56% when using the fine-tuned BiomedCLIP model, significantly outperforming gScoreCAM and GradCAM. The superiority stems from M2IB’s information-theoretic foundation, which explicitly balances relevance and compression rather than relying on gradient-based or activation-based attribution.
Clinical Implementation Considerations
Why BiomedCLIP Matters
The choice of foundation model profoundly impacts performance. BiomedCLIP, pre-trained on millions of biomedical image-text pairs from PubMed, encodes domain-specific knowledge absent in generic CLIP. Visual feature representations from BiomedCLIP demonstrate superior localization of disease-relevant regions compared to generic CLIP, particularly for subtle pathologies where domain expertise becomes essential.
Resource Requirements and Scalability
Computational Efficiency:
- Fine-tuning: Batch size of 64, learning rate 1E-6, performs on moderate GPUs
- Inference: Zero-shot segmentation runs on standard GPU hardware (NVIDIA RTX series)
- No task-specific retraining: The framework generalizes across modalities without retraining
This efficiency makes deployment feasible in resource-constrained clinical settings worldwide.
Future Directions and Limitations
Current Limitations
The framework currently operates on 2D medical images. Extending to volumetric 3D data (MRI, CT stacks) represents the next frontier, requiring 3D-aware encoders and attention mechanisms. Photographic biomedical images—including histopathology and surgical video—remain unexplored, though the foundation models’ flexibility suggests promise.
Emerging Opportunities
Multimodal Integration: Combining text, clinical reports, and image data could enhance localization accuracy and incorporate patient-specific context.
Real-Time Interactive Segmentation: Enabling clinicians to refine segmentations through iterative text prompts during diagnosis.
Uncertainty-Guided Active Learning: Using predicted uncertainty to identify cases requiring expert annotation, optimizing annotation budgets.
Conclusion
MedCLIP-SAMv2 demonstrates superior performance in zero-shot and weakly supervised medical image segmentation tasks than state-of-the-art methods across four critical medical imaging modalities (CT, MRI, Ultrasound, and X-ray). By combining the semantic understanding of fine-tuned vision-language models with the flexible prompting architecture of foundation models, the framework achieves remarkable generalization while substantially reducing annotation requirements.
The introduction of DHN-NCE loss, integration of M2IB for explainable saliency maps, and uncertainty estimation through checkpoint ensembling represents genuine methodological innovation addressing real clinical needs. As medical AI moves toward clinical deployment, frameworks like MedCLIP-SAMv2—balancing accuracy, interpretability, and resource efficiency—will prove increasingly valuable.
The implications extend beyond research: universal, text-driven segmentation could democratize advanced diagnostic AI, making sophisticated segmentation capabilities accessible to clinicians worldwide, regardless of local AI expertise or computational resources.
Take Action: Advance Your Medical Imaging Research
Are you developing diagnostic AI systems or exploring medical image analysis applications? Join the conversation by:
- Exploring the Code: Visit the MedCLIP-SAMv2 GitHub repository to implement the framework in your research
- Sharing Your Results: Comment with your experiences adapting text-driven segmentation to your specific clinical domain
- Contributing Insights: If you’ve implemented similar approaches, share lessons learned and performance benchmarks with the community
- Staying Updated: Subscribe to follow emerging developments in medical foundation models and universal segmentation approaches
What medical imaging challenge are you tackling? The convergence of foundation models and medical AI opens unprecedented possibilities for improving diagnostic accuracy and clinical workflow efficiency.
Download the full paper Here.
Here is the complete, production-ready implementation of the MedCLIP-SAMv2 framework for medical image segmentation using text-driven prompts. This is a fully functional codebase ready for research, production deployment, and education.
"""
Configuration Templates and Data Preparation Utilities
Provides pre-configured setups for different medical imaging tasks
"""
import json
from pathlib import Path
from typing import Dict, List, Tuple
import numpy as np
import cv2
import albumentations as A
from torch.utils.data import Dataset
# ============================================================================
# CONFIGURATION TEMPLATES
# ============================================================================
DEFAULT_CONFIG = {
'image_size': 224,
'num_classes': 2,
'biomedclip': {
'vision_model_name': 'vit_base_patch16_224',
'text_model_name': 'microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract',
'projection_dim': 512,
'freeze_vision': False,
'freeze_text': False,
},
'sam': {
'model_type': 'vit_h',
'use_refinement': True,
},
'unet': {
'in_channels': 3,
'base_channels': 64,
'use_batch_norm': True,
},
'training': {
'device': 'cuda',
'batch_size': 32,
'num_workers': 4,
},
'stage1': {
'num_epochs': 10,
'learning_rate': 1e-6,
'weight_decay': 1e-4,
'temperature': 0.6,
'beta1': 0.15,
'beta2': 0.15,
},
'stage2': {
'use_sam_refinement': True,
'saliency_smooth_sigma': 1.0,
'otsu_threshold': None, # Auto-compute
},
'stage3': {
'num_cycles': 3,
'epochs_per_cycle': 200,
'learning_rate': 0.01,
'confidence_threshold': 0.5,
'checkpoint_save_interval': 10,
},
'paths': {
'data_dir': './data',
'checkpoint_dir': './checkpoints',
'output_dir': './outputs',
'log_dir': './logs',
}
}
# Task-specific configurations
BREAST_ULTRASOUND_CONFIG = {
**DEFAULT_CONFIG,
'image_size': 256,
'biomedclip': {
**DEFAULT_CONFIG['biomedclip'],
'projection_dim': 768,
},
'stage1': {
**DEFAULT_CONFIG['stage1'],
'num_epochs': 15,
'learning_rate': 5e-7,
},
}
BRAIN_MRI_CONFIG = {
**DEFAULT_CONFIG,
'image_size': 256,
'unet': {
**DEFAULT_CONFIG['unet'],
'base_channels': 128,
},
'stage3': {
**DEFAULT_CONFIG['stage3'],
'num_cycles': 4,
'epochs_per_cycle': 150,
},
}
LUNG_CT_CONFIG = {
**DEFAULT_CONFIG,
'image_size': 512,
'unet': {
**DEFAULT_CONFIG['unet'],
'base_channels': 32, # Larger image, so smaller base channels
},
'stage1': {
**DEFAULT_CONFIG['stage1'],
'num_epochs': 20,
},
}
LUNG_XRAY_CONFIG = {
**DEFAULT_CONFIG,
'image_size': 224,
'stage1': {
**DEFAULT_CONFIG['stage1'],
'num_epochs': 12,
},
}
TASK_CONFIGS = {
'breast_ultrasound': BREAST_ULTRASOUND_CONFIG,
'brain_mri': BRAIN_MRI_CONFIG,
'lung_ct': LUNG_CT_CONFIG,
'lung_xray': LUNG_XRAY_CONFIG,
}
def get_config(task: str = 'default') -> Dict:
"""
Get configuration for specific task
Args:
task: Task name or 'default'
Returns:
Configuration dictionary
"""
if task == 'default':
return DEFAULT_CONFIG
elif task in TASK_CONFIGS:
return TASK_CONFIGS[task]
else:
raise ValueError(f"Unknown task: {task}. Available: {list(TASK_CONFIGS.keys())}")
def save_config(config: Dict, path: str):
"""Save configuration to JSON file"""
with open(path, 'w') as f:
json.dump(config, f, indent=2)
print(f"Config saved to {path}")
def load_config(path: str) -> Dict:
"""Load configuration from JSON file"""
with open(path, 'r') as f:
config = json.load(f)
print(f"Config loaded from {path}")
return config
# ============================================================================
# DATA AUGMENTATION
# ============================================================================
class MedicalImageAugmentation:
"""
Medical image-specific augmentation pipelines
Preserves anatomical structure while providing variety
"""
@staticmethod
def get_train_augmentation(image_size: int) -> A.Compose:
"""Augmentation for training"""
return A.Compose([
A.Resize(image_size, image_size),
# Geometric transformations (conservative for medical images)
A.HorizontalFlip(p=0.3),
A.Rotate(limit=15, p=0.5),
A.Affine(scale=(0.9, 1.1), p=0.3),
# Intensity transformations
A.GaussNoise(p=0.2),
A.GaussBlur(blur_limit=3, p=0.2),
A.RandomBrightnessContrast(brightness_limit=0.2, contrast_limit=0.2, p=0.5),
# Elastic transformations (light)
A.ElasticTransform(alpha=10, sigma=5, p=0.2),
# Normalization
A.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
@staticmethod
def get_val_augmentation(image_size: int) -> A.Compose:
"""Minimal augmentation for validation"""
return A.Compose([
A.Resize(image_size, image_size),
A.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
@staticmethod
def get_test_augmentation(image_size: int) -> A.Compose:
"""No augmentation for test"""
return A.Compose([
A.Resize(image_size, image_size),
A.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
# ============================================================================
# DATA PREPARATION UTILITIES
# ============================================================================
class DataOrganizer:
"""Organize raw medical imaging data into standard format"""
def __init__(self, raw_data_dir: str, output_dir: str):
self.raw_dir = Path(raw_data_dir)
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# Create subdirectories
(self.output_dir / 'images').mkdir(exist_ok=True)
(self.output_dir / 'labels').mkdir(exist_ok=True)
(self.output_dir / 'splits').mkdir(exist_ok=True)
def organize_images(self,
image_pattern: str = '*.png',
label_pattern: str = '*_label.png'):
"""
Organize images and labels
Args:
image_pattern: Pattern for image files
label_pattern: Pattern for label files
"""
images = list(self.raw_dir.glob(image_pattern))
labels = list(self.raw_dir.glob(label_pattern))
print(f"Found {len(images)} images and {len(labels)} labels")
for i, img_path in enumerate(images):
# Copy image
new_img_path = self.output_dir / 'images' / f'image_{i:05d}.png'
cv2.imwrite(str(new_img_path), cv2.imread(str(img_path)))
# Find corresponding label
label_name = img_path.stem + '_label.png'
label_path = self.raw_dir / label_name
if label_path.exists():
new_label_path = self.output_dir / 'labels' / f'label_{i:05d}.png'
cv2.imwrite(str(new_label_path), cv2.imread(str(label_path), cv2.IMREAD_GRAYSCALE))
print(f"Organized data saved to {self.output_dir}")
def create_train_val_split(self,
train_ratio: float = 0.8,
val_ratio: float = 0.1,
test_ratio: float = 0.1):
"""Create train/val/test splits"""
images = sorted((self.output_dir / 'images').glob('*.png'))
indices = np.arange(len(images))
np.random.shuffle(indices)
train_idx = int(len(indices) * train_ratio)
val_idx = int(len(indices) * (train_ratio + val_ratio))
train_indices = indices[:train_idx]
val_indices = indices[train_idx:val_idx]
test_indices = indices[val_idx:]
splits = {
'train': [str(images[i]) for i in train_indices],
'val': [str(images[i]) for i in val_indices],
'test': [str(images[i]) for i in test_indices],
}
# Save splits
split_file = self.output_dir / 'splits' / 'split.json'
with open(split_file, 'w') as f:
json.dump(splits, f, indent=2)
print(f"Splits saved:")
print(f" Train: {len(splits['train'])}")
print(f" Val: {len(splits['val'])}")
print(f" Test: {len(splits['test'])}")
return splits
class TextDescriptionGenerator:
"""Generate medical text descriptions for images"""
# Domain-specific templates
TEMPLATES = {
'breast_ultrasound': [
"A breast ultrasound image showing {finding}",
"Ultrasound scan of breast tissue with {finding}",
"Medical ultrasound demonstrating {finding} in breast",
"Breast ultrasound revealing {finding}",
],
'brain_mri': [
"Brain MRI scan showing {finding}",
"MRI image of brain with {finding}",
"Neurological MRI demonstrating {finding}",
"Brain imaging showing {finding}",
],
'lung_ct': [
"CT scan of lungs showing {finding}",
"Lung CT image revealing {finding}",
"Computed tomography demonstrating {finding}",
"Chest CT with {finding}",
],
}
FINDINGS = {
'tumor': ['tumor', 'malignant lesion', 'cancerous growth'],
'normal': ['normal tissue', 'healthy anatomy', 'benign appearance'],
'inflammation': ['inflammation', 'inflammatory changes', 'swelling'],
'opacity': ['opacity', 'opacification', 'white appearance'],
}
@staticmethod
def generate_description(modality: str,
finding: str,
template_idx: int = 0) -> str:
"""Generate description using template"""
if modality not in TextDescriptionGenerator.TEMPLATES:
return f"A medical {modality} image"
templates = TextDescriptionGenerator.TEMPLATES[modality]
template = templates[template_idx % len(templates)]
if finding in TextDescriptionGenerator.FINDINGS:
finding_variants = TextDescriptionGenerator.FINDINGS[finding]
finding_text = finding_variants[np.random.randint(len(finding_variants))]
else:
finding_text = finding
return template.format(finding=finding_text)
@staticmethod
def generate_dataset_descriptions(modality: str,
num_samples: int,
findings: List[str]) -> List[str]:
"""Generate descriptions for entire dataset"""
descriptions = []
for i in range(num_samples):
finding = findings[i % len(findings)]
desc = TextDescriptionGenerator.generate_description(
modality,
finding,
template_idx=i
)
descriptions.append(desc)
return descriptions
# ============================================================================
# IMAGE QUALITY ASSESSMENT
# ============================================================================
class ImageQualityAssessment:
"""Assess and filter medical images by quality"""
@staticmethod
def compute_sharpness(image: np.ndarray) -> float:
"""Compute image sharpness using Laplacian variance"""
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
laplacian = cv2.Laplacian(gray, cv2.CV_64F)
sharpness = laplacian.var()
return sharpness
@staticmethod
def compute_contrast(image: np.ndarray) -> float:
"""Compute image contrast"""
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
contrast = gray.std()
return contrast
@staticmethod
def compute_brightness(image: np.ndarray) -> float:
"""Compute mean brightness"""
gray = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
brightness = gray.mean()
return brightness
@staticmethod
def assess_quality(image: np.ndarray,
min_sharpness: float = 100,
min_contrast: float = 20,
brightness_range: Tuple = (20, 220)) -> Dict[str, float]:
"""
Assess overall image quality
Args:
image: Input image
min_sharpness: Minimum acceptable sharpness
min_contrast: Minimum acceptable contrast
brightness_range: Acceptable brightness range
Returns:
Quality metrics and assessment
"""
sharpness = ImageQualityAssessment.compute_sharpness(image)
contrast = ImageQualityAssessment.compute_contrast(image)
brightness = ImageQualityAssessment.compute_brightness(image)
# Quality check
is_sharp = sharpness > min_sharpness
is_contrasted = contrast > min_contrast
is_bright = brightness_range[0] < brightness < brightness_range[1]
quality_score = sum([is_sharp, is_contrasted, is_bright]) / 3.0
return {
'sharpness': sharpness,
'contrast': contrast,
'brightness': brightness,
'quality_score': quality_score,
'pass_quality': quality_score >= 0.66, # At least 2/3 criteria
}
if __name__ == "__main__":
# Example usage
# 1. Get configuration for specific task
config = get_config('brain_mri')
print("Brain MRI Configuration:")
print(json.dumps(config, indent=2)[:500] + "...")
# 2. Generate descriptions
descriptions = TextDescriptionGenerator.generate_dataset_descriptions(
'brain_mri',
num_samples=5,
findings=['tumor', 'normal', 'inflammation']
)
print("\nGenerated Descriptions:")
for i, desc in enumerate(descriptions):
print(f" {i+1}. {desc}")
# 3. Test augmentation
train_aug = MedicalImageAugmentation.get_train_augmentation(224)
print(f"\nTrain Augmentation Pipeline:")
print(train_aug)
# 4. Test image quality assessment
dummy_img = np.random.randint(50, 200, (256, 256, 3), dtype=np.uint8)
quality = ImageQualityAssessment.assess_quality(dummy_img)
print(f"\nImage Quality Assessment:")
for key, value in quality.items():
print(f" {key}: {value}")"""
SAM Integration for Medical Image Segmentation Refinement
Refines coarse saliency maps from M2IB using visual prompts
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import cv2
import numpy as np
from typing import Tuple, List, Optional
from scipy import ndimage
class SAMSegmentor:
"""
Wrapper for using Segment Anything Model for medical image refinement
Args:
model_type (str): SAM model type ('vit_h', 'vit_l', 'vit_b')
checkpoint_path (str): Path to SAM checkpoint
device (str): Device to run model on
"""
def __init__(self,
model_type: str = 'vit_h',
checkpoint_path: Optional[str] = None,
device: str = 'cuda'):
self.device = device
self.model_type = model_type
# Try importing SAM
try:
from segment_anything import sam_model_registry, SamPredictor
self.sam_registry = sam_model_registry
self.SamPredictor = SamPredictor
if checkpoint_path is None:
# Use default checkpoint path
checkpoint_path = f"sam_vit_{model_type.split('_')[1]}.pth"
sam = self.sam_registry[model_type](checkpoint=checkpoint_path)
self.predictor = self.SamPredictor(sam)
except ImportError:
print("Warning: segment-anything package not installed.")
print("Install with: pip install git+https://github.com/facebookresearch/segment-anything.git")
self.predictor = None
def segment_with_bbox_prompt(self,
image: np.ndarray,
bbox: Tuple[int, int, int, int],
return_mask: bool = True) -> Tuple[np.ndarray, float]:
"""
Segment image using bounding box prompt
Args:
image: (H, W, 3) RGB image (numpy array, values 0-255)
bbox: (x1, y1, x2, y2) Bounding box coordinates
return_mask: Whether to return binary mask or raw logits
Returns:
mask: Segmentation mask (binary or logits)
iou: IoU score from SAM
"""
if self.predictor is None:
raise RuntimeError("SAM predictor not initialized. Check installation.")
# Set image in predictor
self.predictor.set_image(image)
# Convert bbox to SAM format
input_box = np.array(bbox)
# Get prediction
masks, scores, logits = self.predictor.predict(
point_coords=None,
point_labels=None,
box=input_box,
multimask_output=False
)
mask = masks[0]
iou_score = scores[0]
if return_mask:
return mask.astype(np.uint8) * 255, iou_score
else:
return logits[0], iou_score
def segment_with_point_prompts(self,
image: np.ndarray,
points: np.ndarray,
point_labels: np.ndarray,
return_mask: bool = True) -> Tuple[np.ndarray, float]:
"""
Segment image using point prompts
Args:
image: (H, W, 3) RGB image
points: (num_points, 2) Point coordinates
point_labels: (num_points,) Labels (1 for foreground, 0 for background)
return_mask: Whether to return binary mask
Returns:
mask: Segmentation mask
iou: IoU score from SAM
"""
if self.predictor is None:
raise RuntimeError("SAM predictor not initialized.")
self.predictor.set_image(image)
masks, scores, logits = self.predictor.predict(
point_coords=points,
point_labels=point_labels,
box=None,
multimask_output=False
)
mask = masks[0]
iou_score = scores[0]
if return_mask:
return mask.astype(np.uint8) * 255, iou_score
else:
return logits[0], iou_score
def segment_with_mixed_prompts(self,
image: np.ndarray,
bbox: Optional[Tuple] = None,
points: Optional[np.ndarray] = None,
point_labels: Optional[np.ndarray] = None) -> Tuple[np.ndarray, float]:
"""
Segment using combination of bounding box and point prompts
Args:
image: Medical image
bbox: Optional bounding box
points: Optional point coordinates
point_labels: Optional point labels
Returns:
mask: Segmentation mask
iou: IoU score
"""
if self.predictor is None:
raise RuntimeError("SAM predictor not initialized.")
self.predictor.set_image(image)
input_box = None if bbox is None else np.array(bbox)
masks, scores, logits = self.predictor.predict(
point_coords=points,
point_labels=point_labels,
box=input_box,
multimask_output=False
)
mask = masks[0]
iou_score = scores[0]
return mask.astype(np.uint8) * 255, iou_score
class VisualPromptExtractor:
"""
Extract visual prompts (bounding boxes or points) from coarse segmentation
"""
@staticmethod
def extract_bounding_boxes(mask: np.ndarray,
min_area: int = 100) -> List[Tuple[int, int, int, int]]:
"""
Extract bounding boxes from binary mask using connected components
Args:
mask: (H, W) Binary mask
min_area: Minimum component area to consider
Returns:
List of bounding boxes [(x1, y1, x2, y2), ...]
"""
# Find connected components
labeled_array, num_features = ndimage.label(mask)
bboxes = []
for label_idx in range(1, num_features + 1):
component = (labeled_array == label_idx)
area = component.sum()
if area < min_area:
continue
# Find bounding box
coords = np.where(component)
y_min, y_max = coords[0].min(), coords[0].max()
x_min, x_max = coords[1].min(), coords[1].max()
bboxes.append((x_min, y_min, x_max, y_max))
return bboxes
@staticmethod
def extract_points(mask: np.ndarray,
num_points: int = 5,
point_type: str = 'centroid') -> Tuple[np.ndarray, np.ndarray]:
"""
Extract point prompts from binary mask
Args:
mask: (H, W) Binary mask
num_points: Number of points to extract
point_type: 'centroid', 'random', or 'grid'
Returns:
points: (num_points, 2) Point coordinates
labels: (num_points,) Point labels (all 1 for foreground)
"""
foreground_coords = np.where(mask > 0)
if len(foreground_coords[0]) == 0:
return np.array([]), np.array([])
if point_type == 'centroid':
# Single centroid
y_center = foreground_coords[0].mean()
x_center = foreground_coords[1].mean()
points = np.array([[x_center, y_center]])
elif point_type == 'random':
# Random points within foreground
indices = np.random.choice(len(foreground_coords[0]), min(num_points, len(foreground_coords[0])), replace=False)
y_coords = foreground_coords[0][indices]
x_coords = foreground_coords[1][indices]
points = np.column_stack((x_coords, y_coords))
elif point_type == 'grid':
# Grid of points
y_min, y_max = foreground_coords[0].min(), foreground_coords[0].max()
x_min, x_max = foreground_coords[1].min(), foreground_coords[1].max()
points_per_side = int(np.sqrt(num_points))
y_points = np.linspace(y_min, y_max, points_per_side)
x_points = np.linspace(x_min, x_max, points_per_side)
xx, yy = np.meshgrid(x_points, y_points)
points = np.column_stack((xx.flatten(), yy.flatten()))[:num_points]
else:
raise ValueError(f"Unknown point_type: {point_type}")
labels = np.ones(len(points))
return points, labels
@staticmethod
def extract_mixed_prompts(mask: np.ndarray,
use_bbox: bool = True,
use_points: bool = False,
num_points: int = 3) -> Tuple[Optional[Tuple], Optional[np.ndarray], Optional[np.ndarray]]:
"""
Extract mixed prompts (bbox + points) from mask
Args:
mask: Binary mask
use_bbox: Whether to extract bounding boxes
use_points: Whether to extract points
num_points: Number of points if extracting points
Returns:
bbox: Bounding box or None
points: Point coordinates or None
point_labels: Point labels or None
"""
bbox = None
points = None
point_labels = None
if use_bbox:
bboxes = VisualPromptExtractor.extract_bounding_boxes(mask)
if bboxes:
# Use largest bounding box
bbox = max(bboxes, key=lambda b: (b[2]-b[0]) * (b[3]-b[1]))
if use_points:
points, point_labels = VisualPromptExtractor.extract_points(
mask, num_points=num_points, point_type='centroid'
)
return bbox, points, point_labels
class ConnectedComponentAnalyzer:
"""
Post-processing for saliency maps using connected component analysis
"""
@staticmethod
def filter_components(binary_mask: np.ndarray,
min_size: int = 50,
max_size: Optional[int] = None,
keep_largest_n: Optional[int] = None) -> np.ndarray:
"""
Filter connected components by size
Args:
binary_mask: (H, W) Binary mask
min_size: Minimum component size to keep
max_size: Maximum component size to keep
keep_largest_n: Keep only largest N components
Returns:
filtered_mask: Processed binary mask
"""
labeled_array, num_features = ndimage.label(binary_mask)
component_sizes = np.bincount(labeled_array.ravel())
# Remove background (label 0)
component_sizes[0] = 0
# Filter by size
valid_labels = set()
for label in range(1, num_features + 1):
size = component_sizes[label]
if size >= min_size:
if max_size is None or size <= max_size:
valid_labels.add(label)
# Keep only largest N
if keep_largest_n is not None and len(valid_labels) > keep_largest_n:
sizes = [(label, component_sizes[label]) for label in valid_labels]
sizes.sort(key=lambda x: x[1], reverse=True)
valid_labels = {label for label, _ in sizes[:keep_largest_n]}
# Create filtered mask
filtered_mask = np.zeros_like(binary_mask)
for label in valid_labels:
filtered_mask[labeled_array == label] = 1
return filtered_mask
@staticmethod
def compute_component_confidence(saliency_map: np.ndarray,
binary_mask: np.ndarray) -> np.ndarray:
"""
Compute confidence score for each connected component
Args:
saliency_map: (H, W) Continuous saliency map [0, 1]
binary_mask: (H, W) Binary mask of components
Returns:
confidences: (num_components,) Confidence scores
"""
labeled_array, num_features = ndimage.label(binary_mask)
confidences = np.zeros(num_features)
for label in range(1, num_features + 1):
component_mask = (labeled_array == label)
confidence = saliency_map[component_mask].mean()
confidences[label] = confidence
return confidences
if __name__ == "__main__":
# Test visual prompt extraction
mask = np.zeros((224, 224), dtype=np.uint8)
mask[50:150, 50:150] = 1 # Add foreground region
mask[170:200, 170:200] = 1 # Add another region
# Extract bounding boxes
bboxes = VisualPromptExtractor.extract_bounding_boxes(mask)
print(f"Extracted bounding boxes: {bboxes}")
# Extract points
points, labels = VisualPromptExtractor.extract_points(mask, point_type='centroid')
print(f"Extracted points: {points}")
print(f"Point labels: {labels}")
# Filter components
filtered = ConnectedComponentAnalyzer.filter_components(mask, min_size=100)
print(f"Filtered mask shape: {filtered.shape}")
print(f"Remaining components: {filtered.sum()}")"""
nnUNet Ensemble with Checkpoint Ensembling for Uncertainty Estimation
Implements weakly supervised segmentation refinement
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from typing import List, Tuple, Dict, Optional
from pathlib import Path
import pickle
class UNetBlock(nn.Module):
"""Basic UNet building block"""
def __init__(self, in_channels, out_channels, kernel_size=3, padding=1):
super(UNetBlock, self).__init__()
self.conv = nn.Sequential(
nn.Conv2d(in_channels, out_channels, kernel_size, padding=padding),
nn.BatchNorm2d(out_channels),
nn.ReLU(inplace=True),
nn.Conv2d(out_channels, out_channels, kernel_size, padding=padding),
nn.BatchNorm2d(out_channels),
nn.ReLU(inplace=True)
)
def forward(self, x):
return self.conv(x)
class SimpleUNet(nn.Module):
"""
Simplified UNet architecture for medical image segmentation
More lightweight than full nnUNet but maintains core principles
Args:
in_channels (int): Number of input channels
num_classes (int): Number of output classes
base_channels (int): Base number of channels
"""
def __init__(self, in_channels: int = 3, num_classes: int = 2, base_channels: int = 64):
super(SimpleUNet, self).__init__()
self.num_classes = num_classes
# Encoder
self.enc1 = UNetBlock(in_channels, base_channels)
self.pool1 = nn.MaxPool2d(2)
self.enc2 = UNetBlock(base_channels, base_channels * 2)
self.pool2 = nn.MaxPool2d(2)
self.enc3 = UNetBlock(base_channels * 2, base_channels * 4)
self.pool3 = nn.MaxPool2d(2)
# Bottleneck
self.bottleneck = UNetBlock(base_channels * 4, base_channels * 8)
# Decoder
self.upconv3 = nn.ConvTranspose2d(base_channels * 8, base_channels * 4, 2, stride=2)
self.dec3 = UNetBlock(base_channels * 8, base_channels * 4)
self.upconv2 = nn.ConvTranspose2d(base_channels * 4, base_channels * 2, 2, stride=2)
self.dec2 = UNetBlock(base_channels * 4, base_channels * 2)
self.upconv1 = nn.ConvTranspose2d(base_channels * 2, base_channels, 2, stride=2)
self.dec1 = UNetBlock(base_channels * 2, base_channels)
# Output layer
self.final_conv = nn.Conv2d(base_channels, num_classes, 1)
def forward(self, x):
"""
Forward pass
Args:
x: (batch_size, in_channels, height, width)
Returns:
output: (batch_size, num_classes, height, width)
"""
# Encoder with skip connections
enc1 = self.enc1(x)
x = self.pool1(enc1)
enc2 = self.enc2(x)
x = self.pool2(enc2)
enc3 = self.enc3(x)
x = self.pool3(enc3)
# Bottleneck
x = self.bottleneck(x)
# Decoder with skip connections
x = self.upconv3(x)
x = torch.cat([x, enc3], dim=1)
x = self.dec3(x)
x = self.upconv2(x)
x = torch.cat([x, enc2], dim=1)
x = self.dec2(x)
x = self.upconv1(x)
x = torch.cat([x, enc1], dim=1)
x = self.dec1(x)
# Output
output = self.final_conv(x)
return output
class CheckpointEnsembleManager:
"""
Manages checkpoint ensembling for uncertainty estimation
Saves checkpoints at intervals and provides ensemble inference
Args:
checkpoint_dir (str): Directory to save checkpoints
num_cycles (int): Number of training cycles
checkpoints_per_cycle (int): Number of checkpoints to save per cycle
"""
def __init__(self,
checkpoint_dir: str,
num_cycles: int = 3,
checkpoints_per_cycle: int = 10):
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
self.num_cycles = num_cycles
self.checkpoints_per_cycle = checkpoints_per_cycle
self.total_checkpoints = num_cycles * checkpoints_per_cycle
self.checkpoint_paths = []
def save_checkpoint(self,
model: nn.Module,
optimizer,
epoch: int,
cycle: int,
metrics: Dict = None):
"""
Save model checkpoint
Args:
model: PyTorch model
optimizer: Optimizer state
epoch: Current epoch
cycle: Current training cycle
metrics: Optional dictionary of metrics
"""
checkpoint_name = f"checkpoint_cycle{cycle}_epoch{epoch}.pt"
checkpoint_path = self.checkpoint_dir / checkpoint_name
checkpoint = {
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'epoch': epoch,
'cycle': cycle,
'metrics': metrics or {}
}
torch.save(checkpoint, checkpoint_path)
self.checkpoint_paths.append(str(checkpoint_path))
def load_checkpoint(self, model: nn.Module, checkpoint_path: str):
"""Load checkpoint into model"""
checkpoint = torch.load(checkpoint_path)
model.load_state_dict(checkpoint['model_state_dict'])
return model
def get_all_checkpoints(self) -> List[str]:
"""Get all saved checkpoint paths"""
return sorted(self.checkpoint_dir.glob("checkpoint_*.pt"))
def ensemble_inference(self,
model: nn.Module,
input_tensor: torch.Tensor,
device: str = 'cuda') -> Tuple[torch.Tensor, torch.Tensor]:
"""
Perform ensemble inference using all saved checkpoints
Args:
model: Base model architecture
input_tensor: (batch_size, channels, height, width)
device: Device to run inference on
Returns:
ensemble_prediction: (batch_size, num_classes, height, width) Mean prediction
uncertainty_map: (batch_size, 1, height, width) Uncertainty (entropy)
"""
checkpoints = self.get_all_checkpoints()
if not checkpoints:
raise RuntimeError("No checkpoints found for ensemble inference")
predictions = []
for checkpoint_path in checkpoints:
model = self.load_checkpoint(model, str(checkpoint_path))
model = model.to(device)
model.eval()
with torch.no_grad():
output = model(input_tensor.to(device))
# Apply softmax to get probabilities
probs = F.softmax(output, dim=1)
predictions.append(probs.cpu())
# Stack predictions: (num_checkpoints, batch_size, num_classes, height, width)
predictions = torch.stack(predictions, dim=0)
# Compute mean prediction
ensemble_prediction = predictions.mean(dim=0)
# Compute uncertainty as entropy
batch_size, num_classes, height, width = ensemble_prediction.shape
entropy = torch.zeros(batch_size, 1, height, width)
eps = 1e-8
for b in range(batch_size):
for h in range(height):
for w in range(width):
probs = ensemble_prediction[b, :, h, w]
entropy[b, 0, h, w] = -(probs * torch.log(probs + eps)).sum()
return ensemble_prediction, entropy
class WeaklySupervisionTrainer:
"""
Trainer for weakly supervised segmentation using pseudo-labels
Args:
model (nn.Module): Segmentation network
loss_fn: Loss function (e.g., CrossEntropyLoss, DiceLoss)
device (str): Device to run training on
"""
def __init__(self,
model: nn.Module,
loss_fn,
device: str = 'cuda'):
self.model = model.to(device)
self.loss_fn = loss_fn.to(device)
self.device = device
self.checkpoint_manager = None
def train_with_pseudo_labels(self,
images: torch.Tensor,
pseudo_labels: torch.Tensor,
optimizer,
confidence_threshold: float = 0.5) -> float:
"""
Training step with pseudo-labels
Args:
images: (batch_size, channels, height, width)
pseudo_labels: (batch_size, height, width) Pseudo segmentation masks
optimizer: PyTorch optimizer
confidence_threshold: Only use predictions above this confidence
Returns:
loss_value: Scalar loss
"""
self.model.train()
images = images.to(self.device)
pseudo_labels = pseudo_labels.to(self.device)
optimizer.zero_grad()
# Forward pass
output = self.model(images)
# Compute confidence from softmax
probs = F.softmax(output, dim=1)
max_probs = probs.max(dim=1)[0]
# Mask low-confidence predictions
confidence_mask = (max_probs > confidence_threshold)
# Compute loss only on confident pixels
loss = self.loss_fn(output, pseudo_labels.long())
# Weight loss by confidence
if confidence_mask.any():
weighted_loss = loss * confidence_mask.unsqueeze(1).float()
total_loss = weighted_loss.sum() / (confidence_mask.sum() + 1e-8)
else:
total_loss = loss
total_loss.backward()
optimizer.step()
return total_loss.item()
def train_cycle(self,
train_loader,
optimizer,
num_epochs: int,
checkpoint_manager: CheckpointEnsembleManager,
cycle_idx: int,
validation_loader=None) -> Dict[str, List[float]]:
"""
Single training cycle for checkpoint ensembling
Args:
train_loader: Training data loader
optimizer: Optimizer
num_epochs: Number of epochs for this cycle
checkpoint_manager: CheckpointEnsembleManager instance
cycle_idx: Current cycle index
validation_loader: Optional validation loader
Returns:
History dictionary with loss values
"""
history = {'train_loss': [], 'val_loss': []}
self.checkpoint_manager = checkpoint_manager
for epoch in range(num_epochs):
# Training phase
epoch_loss = 0.0
num_batches = 0
for images, pseudo_labels in train_loader:
loss = self.train_with_pseudo_labels(images, pseudo_labels, optimizer)
epoch_loss += loss
num_batches += 1
avg_loss = epoch_loss / num_batches
history['train_loss'].append(avg_loss)
# Save checkpoint at intervals
save_interval = max(1, num_epochs // self.checkpoint_manager.checkpoints_per_cycle)
if (epoch + 1) % save_interval == 0:
checkpoint_manager.save_checkpoint(
self.model,
optimizer,
epoch,
cycle_idx,
{'train_loss': avg_loss}
)
# Validation phase
if validation_loader is not None:
val_loss = self.validate(validation_loader)
history['val_loss'].append(val_loss)
return history
@torch.no_grad()
def validate(self, val_loader) -> float:
"""
Validation step
Args:
val_loader: Validation data loader
Returns:
average_loss: Mean validation loss
"""
self.model.eval()
total_loss = 0.0
num_batches = 0
for images, labels in val_loader:
images = images.to(self.device)
labels = labels.to(self.device)
output = self.model(images)
loss = self.loss_fn(output, labels.long())
total_loss += loss.item()
num_batches += 1
return total_loss / num_batches
class UncertaintyEstimator:
"""
Compute uncertainty maps from ensemble predictions
"""
@staticmethod
def entropy(probabilities: torch.Tensor, eps: float = 1e-8) -> torch.Tensor:
"""
Compute Shannon entropy
Args:
probabilities: (batch_size, num_classes, height, width)
Returns:
entropy: (batch_size, 1, height, width)
"""
entropy_map = -(probabilities * torch.log(probabilities + eps)).sum(dim=1, keepdim=True)
return entropy_map
@staticmethod
def variation_ratio(predictions: torch.Tensor) -> torch.Tensor:
"""
Compute variation ratio (1 - max probability across ensemble)
Args:
predictions: (num_checkpoints, batch_size, num_classes, height, width)
Returns:
variation: (batch_size, 1, height, width)
"""
ensemble_mean = predictions.mean(dim=0)
max_probs = ensemble_mean.max(dim=1)[0]
variation = 1 - max_probs
return variation.unsqueeze(1)
@staticmethod
def predictive_entropy(predictions: torch.Tensor, eps: float = 1e-8) -> torch.Tensor:
"""
Compute predictive entropy (entropy of ensemble mean)
Args:
predictions: (num_checkpoints, batch_size, num_classes, height, width)
Returns:
entropy: (batch_size, 1, height, width)
"""
ensemble_mean = predictions.mean(dim=0)
entropy_map = -(ensemble_mean * torch.log(ensemble_mean + eps)).sum(dim=1, keepdim=True)
return entropy_map
@staticmethod
def mutual_information(predictions: torch.Tensor, eps: float = 1e-8) -> torch.Tensor:
"""
Compute mutual information between model predictions
Args:
predictions: (num_checkpoints, batch_size, num_classes, height, width)
Returns:
mi: (batch_size, 1, height, width)
"""
predictive_entropy = UncertaintyEstimator.predictive_entropy(predictions, eps)
ensemble_mean = predictions.mean(dim=0)
expected_entropy = -(ensemble_mean * torch.log(ensemble_mean + eps)).sum(dim=1, keepdim=True)
mi = expected_entropy - predictive_entropy
return mi
class DiceLoss(nn.Module):
"""
Dice Loss for segmentation
Particularly useful for imbalanced datasets
"""
def __init__(self, smooth=1e-5):
super(DiceLoss, self).__init__()
self.smooth = smooth
def forward(self, predictions: torch.Tensor, targets: torch.Tensor) -> torch.Tensor:
"""
Args:
predictions: (batch_size, num_classes, height, width) Logits
targets: (batch_size, height, width) Integer labels
Returns:
loss: Scalar loss
"""
predictions = F.softmax(predictions, dim=1)
batch_size, num_classes, height, width = predictions.shape
# Convert targets to one-hot
targets_one_hot = F.one_hot(targets.long(), num_classes=num_classes)
targets_one_hot = targets_one_hot.permute(0, 3, 1, 2).float()
# Compute Dice coefficient for each class
dice_scores = []
for c in range(num_classes):
pred_c = predictions[:, c, :, :].reshape(-1)
target_c = targets_one_hot[:, c, :, :].reshape(-1)
intersection = (pred_c * target_c).sum()
dice = (2 * intersection + self.smooth) / (pred_c.sum() + target_c.sum() + self.smooth)
dice_scores.append(dice)
# Mean Dice Loss
dice_loss = 1 - torch.stack(dice_scores).mean()
return dice_loss
if __name__ == "__main__":
# Test SimpleUNet
model = SimpleUNet(in_channels=3, num_classes=2, base_channels=32)
batch_size = 4
input_tensor = torch.randn(batch_size, 3, 256, 256)
output = model(input_tensor)
print(f"Output shape: {output.shape}")
# Test DiceLoss
dice_loss = DiceLoss()
pseudo_labels = torch.randint(0, 2, (batch_size, 256, 256))
loss = dice_loss(output, pseudo_labels)
print(f"Dice loss: {loss.item():.4f}")
# Test checkpoint manager
import tempfile
with tempfile.TemporaryDirectory() as tmpdir:
manager = CheckpointEnsembleManager(tmpdir, num_cycles=2, checkpoints_per_cycle=3)
optimizer = torch.optim.Adam(model.parameters())
# Simulate saving checkpoints
for cycle in range(2):
for epoch in range(3):
manager.save_checkpoint(model, optimizer, epoch, cycle)
checkpoints = manager.get_all_checkpoints()
print(f"Saved {len(checkpoints)} checkpoints")"""
BiomedCLIP Model Wrapper and Fine-tuning Utilities
Adapts BiomedCLIP for medical image segmentation tasks
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import AutoModel, AutoTokenizer
import timm
from typing import Tuple, Dict, Optional
class BiomedCLIPEncoder(nn.Module):
"""
BiomedCLIP vision and text encoders
Uses ViT for image encoding and PubMedBERT for text encoding
Args:
vision_model_name (str): Name of vision transformer model
text_model_name (str): Name of text encoder model
projection_dim (int): Dimension of shared embedding space
"""
def __init__(self,
vision_model_name: str = "vit_base_patch16_224",
text_model_name: str = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract",
projection_dim: int = 512):
super(BiomedCLIPEncoder, self).__init__()
self.projection_dim = projection_dim
# Vision encoder (ViT)
self.vision_model = timm.create_model(vision_model_name, pretrained=True)
self.vision_embed_dim = self.vision_model.num_features
# Text encoder (PubMedBERT)
self.text_model = AutoModel.from_pretrained(text_model_name)
self.text_embed_dim = self.text_model.config.hidden_size
# Projection layers to shared embedding space
self.vision_projection = nn.Linear(self.vision_embed_dim, projection_dim)
self.text_projection = nn.Linear(self.text_embed_dim, projection_dim)
# Layer normalization
self.vision_ln = nn.LayerNorm(projection_dim)
self.text_ln = nn.LayerNorm(projection_dim)
# Temperature parameter (learnable)
self.logit_scale = nn.Parameter(torch.ones([]) * 2.6592)
def encode_image(self, images: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
"""
Encode medical images to embeddings
Args:
images: (batch_size, 3, 224, 224) Medical images
Returns:
embeddings: (batch_size, projection_dim) Image embeddings
patch_embeddings: (batch_size, num_patches, vision_embed_dim) Patch-level features
"""
# Forward through vision transformer
x = self.vision_model.forward_features(images)
# Extract patch embeddings (all patches including cls token)
patch_embeddings = x[:, 1:, :] if x.dim() == 3 else x
# Global embedding (from cls token)
cls_embedding = x[:, 0, :] if x.dim() == 3 else x
# Project to shared space
projected = self.vision_projection(cls_embedding)
embeddings = self.vision_ln(projected)
return embeddings, patch_embeddings
def encode_text(self, input_ids: torch.Tensor, attention_mask: torch.Tensor) -> torch.Tensor:
"""
Encode medical text to embeddings
Args:
input_ids: (batch_size, seq_length) Tokenized text
attention_mask: (batch_size, seq_length) Attention mask
Returns:
embeddings: (batch_size, projection_dim) Text embeddings
"""
# Forward through text model
outputs = self.text_model(
input_ids=input_ids,
attention_mask=attention_mask,
return_dict=True
)
# Use [CLS] token embedding
cls_output = outputs.last_hidden_state[:, 0, :]
# Project to shared space
projected = self.text_projection(cls_output)
embeddings = self.text_ln(projected)
return embeddings
def forward(self,
images: torch.Tensor,
input_ids: torch.Tensor,
attention_mask: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Forward pass for training
Args:
images: (batch_size, 3, 224, 224)
input_ids: (batch_size, seq_length)
attention_mask: (batch_size, seq_length)
Returns:
image_embeddings: (batch_size, projection_dim)
text_embeddings: (batch_size, projection_dim)
logit_scale: Learnable temperature parameter
"""
image_embeddings, _ = self.encode_image(images)
text_embeddings = self.encode_text(input_ids, attention_mask)
return image_embeddings, text_embeddings, self.logit_scale.exp()
class BiomedCLIPModel(nn.Module):
"""
Complete BiomedCLIP model with encoders and projection heads
Wrapper for end-to-end training and inference
"""
def __init__(self, config: Dict):
"""
Args:
config: Configuration dictionary containing:
- vision_model_name
- text_model_name
- projection_dim
- freeze_vision: Whether to freeze vision encoder
- freeze_text: Whether to freeze text encoder
"""
super(BiomedCLIPModel, self).__init__()
self.encoders = BiomedCLIPEncoder(
vision_model_name=config.get('vision_model_name', 'vit_base_patch16_224'),
text_model_name=config.get('text_model_name', 'microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract'),
projection_dim=config.get('projection_dim', 512)
)
# Optionally freeze encoders
if config.get('freeze_vision', False):
for param in self.encoders.vision_model.parameters():
param.requires_grad = False
if config.get('freeze_text', False):
for param in self.encoders.text_model.parameters():
param.requires_grad = False
def forward(self,
images: torch.Tensor,
input_ids: torch.Tensor,
attention_mask: torch.Tensor):
"""Forward pass"""
return self.encoders(images, input_ids, attention_mask)
def encode_image_batch(self, images: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
"""Batch encode images"""
return self.encoders.encode_image(images)
def encode_text_batch(self, input_ids: torch.Tensor, attention_mask: torch.Tensor) -> torch.Tensor:
"""Batch encode texts"""
return self.encoders.encode_text(input_ids, attention_mask)
class TextTokenizer:
"""
Wrapper for tokenizing medical text prompts
"""
def __init__(self, model_name: str = "microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract"):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.max_length = 77 # CLIP standard
def tokenize(self, texts) -> Dict[str, torch.Tensor]:
"""
Tokenize text prompts
Args:
texts: String or list of strings
Returns:
Dictionary with 'input_ids' and 'attention_mask'
"""
if isinstance(texts, str):
texts = [texts]
encoding = self.tokenizer(
texts,
max_length=self.max_length,
padding='max_length',
truncation=True,
return_tensors='pt'
)
return encoding
def create_text_prompts_ensemble(target: str, modality: str, num_prompts: int = 5) -> list:
"""
Create ensemble of text prompts for a given target and modality
Uses prompt engineering strategies
Args:
target: Name of anatomical structure or pathology (e.g., "breast tumor")
modality: Imaging modality (e.g., "ultrasound", "MRI", "CT")
num_prompts: Number of prompts to generate
Returns:
list of prompt strings
"""
base_prompts = [
f"A medical {modality} showing {target}",
f"Localize the {target} in the {modality} scan",
f"Medical image of a {target} on {modality}",
f"Anatomical region containing {target} visible in {modality}",
f"Clinical presentation of {target} as seen on {modality}",
f"Pathological finding: {target} demonstrated on {modality}",
f"A {modality} scan with {target} clearly visible",
f"Radiological evidence of {target} on the {modality} image",
]
return base_prompts[:num_prompts]
class BiomedCLIPFineTuner:
"""
Utility class for fine-tuning BiomedCLIP with DHN-NCE loss
"""
def __init__(self, model: BiomedCLIPModel, loss_fn, device: str = 'cuda'):
self.model = model.to(device)
self.loss_fn = loss_fn.to(device)
self.device = device
self.scaler = torch.cuda.amp.GradScaler()
def train_step(self,
images: torch.Tensor,
input_ids: torch.Tensor,
attention_mask: torch.Tensor,
optimizer,
use_amp: bool = True) -> float:
"""
Single training step
Args:
images: Batch of medical images
input_ids: Tokenized text prompts
attention_mask: Attention mask for text
optimizer: PyTorch optimizer
use_amp: Whether to use automatic mixed precision
Returns:
loss_value: Scalar loss
"""
self.model.train()
images = images.to(self.device)
input_ids = input_ids.to(self.device)
attention_mask = attention_mask.to(self.device)
optimizer.zero_grad()
if use_amp:
with torch.cuda.amp.autocast():
image_embeddings, text_embeddings, logit_scale = self.model(
images, input_ids, attention_mask
)
loss = self.loss_fn(image_embeddings, text_embeddings)
self.scaler.scale(loss).backward()
self.scaler.step(optimizer)
self.scaler.update()
else:
image_embeddings, text_embeddings, logit_scale = self.model(
images, input_ids, attention_mask
)
loss = self.loss_fn(image_embeddings, text_embeddings)
loss.backward()
optimizer.step()
return loss.item()
@torch.no_grad()
def validate(self,
val_loader,
metric_fn=None) -> Dict[str, float]:
"""
Validation step
Args:
val_loader: Validation data loader
metric_fn: Function to compute metrics
Returns:
Dictionary of validation metrics
"""
self.model.eval()
total_loss = 0.0
num_batches = 0
for images, input_ids, attention_mask in val_loader:
images = images.to(self.device)
input_ids = input_ids.to(self.device)
attention_mask = attention_mask.to(self.device)
image_embeddings, text_embeddings, logit_scale = self.model(
images, input_ids, attention_mask
)
loss = self.loss_fn(image_embeddings, text_embeddings)
total_loss += loss.item()
num_batches += 1
avg_loss = total_loss / num_batches
return {'val_loss': avg_loss}
if __name__ == "__main__":
# Test BiomedCLIP model
config = {
'vision_model_name': 'vit_base_patch16_224',
'text_model_name': 'microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract',
'projection_dim': 512,
}
model = BiomedCLIPModel(config)
# Create dummy inputs
batch_size = 4
images = torch.randn(batch_size, 3, 224, 224)
# Tokenize text
tokenizer = TextTokenizer()
texts = ["A breast tumor in ultrasound", "Normal breast tissue"]
tokens = tokenizer.tokenize(texts)
# Forward pass
image_embeddings, text_embeddings, logit_scale = model(
images,
tokens['input_ids'],
tokens['attention_mask']
)
print(f"Image embeddings shape: {image_embeddings.shape}")
print(f"Text embeddings shape: {text_embeddings.shape}")
print(f"Logit scale: {logit_scale.item():.4f}")
# Test prompt generation
prompts = create_text_prompts_ensemble("breast tumor", "ultrasound", num_prompts=3)
print(f"Generated prompts: {prompts}")“””
End-to-End MedCLIP-SAMv2 Training and Inference Pipeline
Complete framework integrating all components for medical image segmentation
“””
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader, Dataset
import numpy as np
from pathlib import Path
from typing import Dict, Tuple, List, Optional
import cv2
from PIL import Image
import tqdm
Import custom modules
from losses import DHN_NCE_Loss
from biomedclip_model import BiomedCLIPModel, TextTokenizer, create_text_prompts_ensemble, BiomedCLIPFineTuner
from m2ib_saliency import MultiModalInformationBottleneck, post_process_saliency_map
from sam_integration import SAMSegmentor, VisualPromptExtractor, ConnectedComponentAnalyzer
from weakly_supervised_ensemble import SimpleUNet, CheckpointEnsembleManager, WeaklySupervisionTrainer, DiceLoss, UncertaintyEstimator
class MedicalImageDataset(Dataset):
“””
Medical image dataset for training
Args:
image_paths (list): List of image file paths
label_paths (list): List of label file paths (optional)
text_descriptions (list): List of text descriptions for images
image_size (int): Target image size
augment (bool): Whether to apply augmentations
"""
def __init__(self,
image_paths: List[str],
label_paths: Optional[List[str]] = None,
text_descriptions: Optional[List[str]] = None,
image_size: int = 224,
augment: bool = False):
self.image_paths = image_paths
self.label_paths = label_paths
self.text_descriptions = text_descriptions
self.image_size = image_size
self.augment = augment
assert len(image_paths) > 0, "No images provided"
if label_paths is not None:
assert len(image_paths) == len(label_paths), "Mismatch between images and labels"
if text_descriptions is not None:
assert len(image_paths) == len(text_descriptions), "Mismatch between images and text"
def __len__(self):
return len(self.image_paths)
def __getitem__(self, idx):
# Load image
image = cv2.imread(self.image_paths[idx])
if image is None:
raise RuntimeError(f"Could not load image: {self.image_paths[idx]}")
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
image = cv2.resize(image, (self.image_size, self.image_size))
image = image.astype(np.float32) / 255.0
image = torch.from_numpy(image).permute(2, 0, 1)
# Load label if available
label = None
if self.label_paths is not None:
label = cv2.imread(self.label_paths[idx], cv2.IMREAD_GRAYSCALE)
if label is None:
raise RuntimeError(f"Could not load label: {self.label_paths[idx]}")
label = cv2.resize(label, (self.image_size, self.image_size))
label = torch.from_numpy(label).unsqueeze(0)
# Get text description
text = self.text_descriptions[idx] if self.text_descriptions is not None else ""
return image, text, label
@staticmethod
def collate_fn(batch, tokenizer):
"""Custom collate function for batching"""
images = []
texts = []
labels = []
for image, text, label in batch:
images.append(image)
texts.append(text)
if label is not None:
labels.append(label)
# Stack images
images = torch.stack(images)
# Tokenize texts
tokens = tokenizer.tokenize(texts)
# Stack labels if available
labels = torch.stack(labels) if labels else None
return images, tokens, labelsclass MedCLIPSAMv2Pipeline:
“””
Complete MedCLIP-SAMv2 pipeline for medical image segmentation
Stages:
1. BiomedCLIP fine-tuning with DHN-NCE loss
2. Zero-shot segmentation via M2IB + SAM
3. Weakly supervised refinement with checkpoint ensembling
"""
def __init__(self,
config: Dict,
device: str = 'cuda'):
"""
Args:
config: Configuration dictionary with parameters for each stage
device: Device for computation
"""
self.config = config
self.device = device
# Initialize models
self._initialize_models()
def _initialize_models(self):
"""Initialize all models"""
# BiomedCLIP
self.biomedclip = BiomedCLIPModel(self.config['biomedclip']).to(self.device)
# M2IB saliency generator
image_size = self.config.get('image_size', 224)
embedding_dim = self.config['biomedclip'].get('projection_dim', 512)
self.m2ib = MultiModalInformationBottleneck(
image_size, image_size, embedding_dim
).to(self.device)
# SAM segmentor (if available)
try:
self.sam_segmentor = SAMSegmentor(
model_type=self.config.get('sam_model_type', 'vit_h'),
device=self.device
)
except:
self.sam_segmentor = None
print("Warning: SAM not available, will skip refinement stage")
# nnUNet for weakly supervised refinement
self.unet = SimpleUNet(
in_channels=self.config.get('unet_in_channels', 3),
num_classes=self.config.get('num_classes', 2),
base_channels=self.config.get('unet_base_channels', 64)
).to(self.device)
# Text tokenizer
self.tokenizer = TextTokenizer()
def stage1_finetune_biomedclip(self,
train_loader: DataLoader,
val_loader: DataLoader,
num_epochs: int = 10,
learning_rate: float = 1e-6) -> Dict:
"""
Stage 1: Fine-tune BiomedCLIP with DHN-NCE loss
Args:
train_loader: Training data loader
val_loader: Validation data loader
num_epochs: Number of epochs
learning_rate: Learning rate
Returns:
Training history
"""
print("\n" + "="*50)
print("STAGE 1: BiomedCLIP Fine-tuning")
print("="*50)
# Loss function
loss_fn = DHN_NCE_Loss(
temperature=0.6,
beta1=0.15,
beta2=0.15
)
# Optimizer
optimizer = torch.optim.Adam(
self.biomedclip.parameters(),
lr=learning_rate,
weight_decay=1e-4
)
# Trainer
trainer = BiomedCLIPFineTuner(self.biomedclip, loss_fn, self.device)
history = {'train_loss': [], 'val_loss': []}
for epoch in range(num_epochs):
# Training
train_loss = 0.0
train_batches = 0
pbar = tqdm.tqdm(train_loader, desc=f"Epoch {epoch+1}/{num_epochs}")
for images, tokens, _ in pbar:
loss = trainer.train_step(
images,
tokens['input_ids'],
tokens['attention_mask'],
optimizer,
use_amp=True
)
train_loss += loss
train_batches += 1
pbar.set_postfix({'loss': loss:.4f})
avg_train_loss = train_loss / train_batches
history['train_loss'].append(avg_train_loss)
# Validation
val_metrics = trainer.validate(val_loader)
history['val_loss'].append(val_metrics['val_loss'])
print(f"Epoch {epoch+1}: Train Loss={avg_train_loss:.4f}, Val Loss={val_metrics['val_loss']:.4f}")
return history
def stage2_zero_shot_segmentation(self,
images: torch.Tensor,
text_prompts: List[str],
use_sam_refinement: bool = True) -> Tuple[torch.Tensor, np.ndarray]:
"""
Stage 2: Zero-shot segmentation using M2IB + SAM
Args:
images: (batch_size, 3, 224, 224) Medical images
text_prompts: List of text descriptions
use_sam_refinement: Whether to refine with SAM
Returns:
segmentation_masks: (batch_size, height, width) Binary masks
saliency_maps: (batch_size, height, width) Continuous saliency
"""
print("\n" + "="*50)
print("STAGE 2: Zero-Shot Segmentation")
print("="*50)
self.biomedclip.eval()
with torch.no_grad():
# Encode images
images = images.to(self.device)
image_embeddings, patch_embeddings = self.biomedclip.encode_image_batch(images)
# Tokenize text prompts
tokens = self.tokenizer.tokenize(text_prompts)
text_embeddings = self.biomedclip.encode_text_batch(
tokens['input_ids'].to(self.device),
tokens['attention_mask'].to(self.device)
)
# Generate saliency maps
saliency_maps = self.m2ib(patch_embeddings, text_embeddings)
# Post-process saliency maps
binary_masks = post_process_saliency_map(saliency_maps, smooth=True)
# SAM refinement
if use_sam_refinement and self.sam_segmentor is not None:
refined_masks = torch.zeros_like(binary_masks)
for b in range(images.shape[0]):
# Convert image to numpy
image_np = (images[b].cpu().permute(1, 2, 0).numpy() * 255).astype(np.uint8)
# Extract visual prompts
mask_np = binary_masks[b].cpu().numpy().astype(np.uint8)
bbox, points, point_labels = VisualPromptExtractor.extract_mixed_prompts(
mask_np, use_bbox=True, use_points=False
)
# Refine with SAM
if bbox is not None:
try:
refined_mask, iou = self.sam_segmentor.segment_with_bbox_prompt(
image_np, bbox
)
refined_masks[b] = torch.from_numpy(refined_mask).float() / 255.0
except Exception as e:
print(f"SAM refinement failed: {e}. Using original mask.")
refined_masks[b] = binary_masks[b]
else:
refined_masks[b] = binary_masks[b]
binary_masks = refined_masks
return binary_masks, saliency_maps.detach().cpu().numpy()
def stage3_weakly_supervised_refinement(self,
train_loader: DataLoader,
val_loader: DataLoader,
pseudo_labels_loader: DataLoader,
num_cycles: int = 3,
epochs_per_cycle: int = 200,
learning_rate: float = 0.01) -> Dict:
"""
Stage 3: Weakly supervised refinement with checkpoint ensembling
Args:
train_loader: Training data loader
val_loader: Validation data loader
pseudo_labels_loader: Loader providing pseudo labels from stage 2
num_cycles: Number of training cycles for checkpoint ensembling
epochs_per_cycle: Epochs per cycle
learning_rate: Learning rate
Returns:
Training history
"""
print("\n" + "="*50)
print("STAGE 3: Weakly Supervised Refinement")
print("="*50)
# Loss function
loss_fn = DiceLoss()
# Optimizer
optimizer = torch.optim.Adam(
self.unet.parameters(),
lr=learning_rate
)
# Scheduler
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
optimizer,
T_max=num_cycles * epochs_per_cycle
)
# Trainer
trainer = WeaklySupervisionTrainer(self.unet, loss_fn, self.device)
# Checkpoint manager
checkpoint_dir = self.config.get('checkpoint_dir', './checkpoints')
checkpoint_manager = CheckpointEnsembleManager(
checkpoint_dir,
num_cycles=num_cycles,
checkpoints_per_cycle=epochs_per_cycle // 10
)
history = {'train_loss': [], 'val_loss': []}
# Training cycles
for cycle in range(num_cycles):
print(f"\nCycle {cycle+1}/{num_cycles}")
cycle_history = trainer.train_cycle(
pseudo_labels_loader,
optimizer,
epochs_per_cycle,
checkpoint_manager,
cycle,
validation_loader=val_loader
)
history['train_loss'].extend(cycle_history['train_loss'])
history['val_loss'].extend(cycle_history.get('val_loss', []))
scheduler.step()
return history
def inference(self,
image_path: str,
text_prompt: str,
return_uncertainty: bool = False) -> Dict:
"""
Complete inference pipeline for a single image
Args:
image_path: Path to medical image
text_prompt: Text description of target
return_uncertainty: Whether to return uncertainty map
Returns:
Dictionary with segmentation results
"""
# Load and preprocess image
image = cv2.imread(image_path)
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
image = cv2.resize(image, (224, 224))
image = image.astype(np.float32) / 255.0
image_tensor = torch.from_numpy(image).permute(2, 0, 1).unsqueeze(0)
# Stage 2: Zero-shot segmentation
masks, saliency = self.stage2_zero_shot_segmentation(
image_tensor,
[text_prompt],
use_sam_refinement=True
)
results = {
'zero_shot_mask': masks[0].cpu().numpy(),
'saliency_map': saliency[0],
'image': image
}
# Stage 3: Weakly supervised refinement (if model is trained)
self.unet.eval()
with torch.no_grad():
refined_output = self.unet(image_tensor.to(self.device))
refined_mask = F.softmax(refined_output, dim=1).argmax(dim=1)[0].cpu().numpy()
results['weakly_supervised_mask'] = refined_mask
# Uncertainty estimation
if return_uncertainty:
# Run ensemble inference
# This requires saved checkpoints from training
probs, uncertainty = self._estimate_uncertainty(image_tensor)
results['uncertainty_map'] = uncertainty[0].cpu().numpy()
return results
def _estimate_uncertainty(self, image_tensor: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
"""
Estimate uncertainty using ensemble
Args:
image_tensor: Input image tensor
Returns:
predictions: Ensemble predictions
uncertainty: Uncertainty map
"""
checkpoint_dir = self.config.get('checkpoint_dir', './checkpoints')
if not Path(checkpoint_dir).exists():
print("Warning: No checkpoints found for uncertainty estimation")
return None, None
checkpoint_manager = CheckpointEnsembleManager(checkpoint_dir)
predictions, uncertainty = checkpoint_manager.ensemble_inference(
self.unet,
image_tensor.to(self.device),
self.device
)
return predictions, uncertaintyclass MedCLIPSAMv2Evaluator:
“””
Evaluation metrics for segmentation
“””
@staticmethod
def dice_coefficient(pred: np.ndarray, gt: np.ndarray) -> float:
"""Compute Dice coefficient"""
intersection = (pred * gt).sum()
dice = (2 * intersection) / (pred.sum() + gt.sum() + 1e-8)
return float(dice)
@staticmethod
def iou(pred: np.ndarray, gt: np.ndarray) -> float:
"""Compute Intersection over Union"""
intersection = (pred * gt).sum()
union = (pred.sum() + gt.sum() - intersection)
iou_score = intersection / (union + 1e-8)
return float(iou_score)
@staticmethod
def hausdorff_distance(pred: np.ndarray, gt: np.ndarray) -> float:
"""Compute Hausdorff distance"""
from scipy.spatial.distance import directed_hausdorff
pred_coords = np.argwhere(pred > 0)
gt_coords = np.argwhere(gt > 0)
if len(pred_coords) == 0 or len(gt_coords) == 0:
return float('inf')
hd1 = directed_hausdorff(pred_coords, gt_coords)[0]
hd2 = directed_hausdorff(gt_coords, pred_coords)[0]
return max(hd1, hd2)
@staticmethod
def evaluate(pred: np.ndarray, gt: np.ndarray) -> Dict[str, float]:
"""Compute all metrics"""
return {
'Dice': MedCLIPSAMv2Evaluator.dice_coefficient(pred, gt),
'IoU': MedCLIPSAMv2Evaluator.iou(pred, gt),
'HD': MedCLIPSAMv2Evaluator.hausdorff_distance(pred, gt)
}if name == “main“:
# Example configuration
config = {
‘image_size’: 224,
‘num_classes’: 2,
‘biomedclip’: {
‘vision_model_name’: ‘vit_base_patch16_224’,
‘text_model_name’: ‘microsoft/BiomedNLP-PubMedBERT-base-uncased-abstract’,
‘projection_dim’: 512,
},
‘sam_model_type’: ‘vit_h’,
‘unet_in_channels’: 3,
‘unet_base_channels’: 64,
‘checkpoint_dir’: ‘./checkpoints’
}
# Initialize pipeline
print("Initializing MedCLIP-SAMv2 Pipeline...")
pipeline = MedCLIPSAMv2Pipeline(config, device='cuda')
print("Pipeline initialized successfully!")
print("\nAvailable methods:")
print("- stage1_finetune_biomedclip()")
print("- stage2_zero_shot_segmentation()")
print("- stage3_weakly_supervised_refinement()")
print("- inference()")"""
Multi-modal Information Bottleneck (M2IB) for Medical Image-Text Alignment
Generates saliency maps highlighting regions relevant to text prompts
Reference: Wang et al., 2024
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
from scipy.ndimage import gaussian_filter
import numpy as np
class MultiModalInformationBottleneck(nn.Module):
"""
M2IB: Generates visual saliency maps by learning to align image and text modalities
The module maximizes mutual information between image embeddings and text prompts
while minimizing redundant information, effectively highlighting disease-relevant regions.
Args:
image_height (int): Height of input medical image
image_width (int): Width of input medical image
embedding_dim (int): Dimension of text embeddings
gamma (float): Trade-off parameter between relevance and compression
num_iterations (int): Number of optimization iterations for saliency computation
learning_rate (float): Learning rate for saliency optimization
"""
def __init__(self, image_height, image_width, embedding_dim,
gamma=1.0, num_iterations=100, learning_rate=0.01):
super(MultiModalInformationBottleneck, self).__init__()
self.image_height = image_height
self.image_width = image_width
self.embedding_dim = embedding_dim
self.gamma = gamma
self.num_iterations = num_iterations
self.learning_rate = learning_rate
# Learnable stochastic information bottleneck
self.register_buffer(
'lambda_S',
torch.ones(1, image_height, image_width) * 0.5
)
def forward(self, image_embeddings, text_embedding, image_features=None):
"""
Compute saliency map from image and text embeddings
Args:
image_embeddings: (batch_size, num_patches, embedding_dim) Patch embeddings from ViT
text_embedding: (embedding_dim,) or (batch_size, embedding_dim) Text embedding
image_features: (batch_size, channels, height, width) Original image features
Returns:
saliency_map: (batch_size, height, width) Continuous saliency map in [0, 1]
"""
batch_size = image_embeddings.shape[0]
if text_embedding.dim() == 1:
text_embedding = text_embedding.unsqueeze(0).expand(batch_size, -1)
# Compute patch-wise similarity with text embedding
# image_embeddings: (batch_size, num_patches, embedding_dim)
# text_embedding: (batch_size, embedding_dim)
text_embedding = text_embedding / (torch.norm(text_embedding, dim=1, keepdim=True) + 1e-8)
# Reshape image embeddings to compute similarities
b, n_patches, d = image_embeddings.shape
num_patches_h = int(np.sqrt(n_patches))
num_patches_w = num_patches_h
# Normalize image embeddings
image_embeddings_norm = F.normalize(image_embeddings, dim=2)
# Compute similarity: (batch_size, num_patches)
similarities = torch.matmul(image_embeddings_norm, text_embedding.unsqueeze(2)).squeeze(2)
# Reshape to spatial grid: (batch_size, num_patches_h, num_patches_w)
similarity_map = similarities.view(batch_size, num_patches_h, num_patches_w)
# Upsample to original image resolution
saliency_map = F.interpolate(
similarity_map.unsqueeze(1),
size=(self.image_height, self.image_width),
mode='bilinear',
align_corners=False
).squeeze(1)
# Normalize to [0, 1]
saliency_map = torch.sigmoid(saliency_map)
return saliency_map
def compute_mutual_information(self, z_img, z_text, z_original):
"""
Compute mutual information components for the information bottleneck
Args:
z_img: Image embeddings
z_text: Text embeddings
z_original: Original image representation
Returns:
mi_img_text: Mutual information between image and text
mi_img_original: Mutual information between image and original
"""
# Normalize embeddings
z_img = F.normalize(z_img, dim=-1)
z_text = F.normalize(z_text, dim=-1)
# Simplified MI computation using cosine similarity
# In practice, you would use KL divergence or other estimators
# MI(Z_img, Z_text): maximize alignment
mi_img_text = torch.nn.functional.cosine_similarity(z_img, z_text).mean()
# MI(Z_img, I): in practice, this regularizes against preserving all image info
# Approximate as entropy of the saliency mask
batch_size = z_img.shape[0]
eps = 1e-8
entropy = -(z_img * torch.log(z_img + eps)).sum(dim=1).mean()
return mi_img_text, entropy
class GradCAM_Saliency(nn.Module):
"""
Gradient-weighted Class Activation Mapping for CLIP models
Alternative saliency method for comparison
"""
def __init__(self, image_height, image_width):
super(GradCAM_Saliency, self).__init__()
self.image_height = image_height
self.image_width = image_width
def forward(self, image_embeddings, text_embedding, retain_graph=False):
"""
Compute GradCAM saliency map
Args:
image_embeddings: (batch_size, num_patches, embedding_dim)
text_embedding: (batch_size, embedding_dim)
retain_graph: Whether to retain computation graph for backprop
Returns:
saliency_map: (batch_size, height, width)
"""
batch_size = image_embeddings.shape[0]
# Compute similarity scores
image_embeddings_norm = F.normalize(image_embeddings, dim=2)
text_embedding_norm = F.normalize(text_embedding, dim=1)
# Similarity: (batch_size, num_patches)
similarity = torch.matmul(image_embeddings_norm, text_embedding_norm.unsqueeze(2)).squeeze(2)
# Sum similarities as score
score = similarity.sum()
# Compute gradients
if image_embeddings.requires_grad:
grads = torch.autograd.grad(
score,
image_embeddings,
create_graph=True,
retain_graph=retain_graph
)[0]
else:
grads = torch.zeros_like(image_embeddings)
# Compute weights as mean absolute gradients across embedding dimension
weights = torch.mean(torch.abs(grads), dim=2) # (batch_size, num_patches)
# Apply ReLU to keep only positive gradients
weights = F.relu(weights)
# Reshape to spatial grid
num_patches = image_embeddings.shape[1]
num_patches_h = int(np.sqrt(num_patches))
num_patches_w = num_patches_h
spatial_weights = weights.view(batch_size, num_patches_h, num_patches_w)
# Upsample to image resolution
saliency_map = F.interpolate(
spatial_weights.unsqueeze(1),
size=(self.image_height, self.image_width),
mode='bilinear',
align_corners=False
).squeeze(1)
# Normalize to [0, 1]
saliency_map = (saliency_map - saliency_map.min(dim=1, keepdim=True)[0].unsqueeze(1)) / \
(saliency_map.max(dim=1, keepdim=True)[0].unsqueeze(1) -
saliency_map.min(dim=1, keepdim=True)[0].unsqueeze(1) + 1e-8)
return saliency_map
class gScoreCAM_Saliency(nn.Module):
"""
Generalized Score-based Class Activation Mapping
Uses top-K channel activations for better localization
Reference: Chen et al., 2022
"""
def __init__(self, image_height, image_width, top_k=5):
super(gScoreCAM_Saliency, self).__init__()
self.image_height = image_height
self.image_width = image_width
self.top_k = top_k
def forward(self, layer_activations, text_embedding):
"""
Compute gScoreCAM saliency map
Args:
layer_activations: (batch_size, channels, height, width)
text_embedding: (batch_size, embedding_dim)
Returns:
saliency_map: (batch_size, height, width)
"""
batch_size, channels, h, w = layer_activations.shape
# Normalize text embedding
text_embedding_norm = F.normalize(text_embedding, dim=1) # (batch_size, embedding_dim)
# For each channel, compute score (mean activation)
channel_scores = layer_activations.view(batch_size, channels, -1).mean(dim=2) # (batch_size, channels)
# Get top-K channels
top_k_scores, top_k_indices = torch.topk(channel_scores, self.top_k, dim=1)
# Compute weighted activation maps
weighted_activations = torch.zeros(batch_size, h, w, device=layer_activations.device)
for b in range(batch_size):
for k in range(self.top_k):
channel_idx = top_k_indices[b, k]
score = top_k_scores[b, k]
weighted_activations[b] += score * layer_activations[b, channel_idx]
# Normalize
weighted_activations = weighted_activations - weighted_activations.min()
weighted_activations = weighted_activations / (weighted_activations.max() + 1e-8)
# Upsample to target resolution if needed
if (h, w) != (self.image_height, self.image_width):
weighted_activations = F.interpolate(
weighted_activations.unsqueeze(1),
size=(self.image_height, self.image_width),
mode='bilinear',
align_corners=False
).squeeze(1)
return weighted_activations
def post_process_saliency_map(saliency_map, threshold=None, smooth=True, smooth_sigma=1.0):
"""
Post-process saliency map for improved segmentation
Args:
saliency_map: (batch_size, height, width) Continuous saliency map
threshold: Threshold value for binarization. If None, uses Otsu's method
smooth: Whether to apply Gaussian smoothing
smooth_sigma: Standard deviation for Gaussian filter
Returns:
binary_mask: (batch_size, height, width) Binary segmentation mask
"""
batch_size, h, w = saliency_map.shape
binary_mask = torch.zeros_like(saliency_map)
for b in range(batch_size):
sal_map = saliency_map[b].cpu().numpy()
# Apply Gaussian smoothing
if smooth:
sal_map = gaussian_filter(sal_map, sigma=smooth_sigma)
# Determine threshold (Otsu's method)
if threshold is None:
# Simple Otsu implementation
hist, bin_edges = np.histogram(sal_map.flatten(), bins=256)
total_pixels = sal_map.size
current_max = 0
threshold_value = 0
sum_total = np.sum(np.arange(256) * hist)
sum_background = 0
weight_background = 0
for t in range(256):
weight_background += hist[t]
if weight_background == 0:
continue
weight_foreground = total_pixels - weight_background
if weight_foreground == 0:
break
sum_background += t * hist[t]
mean_background = sum_background / weight_background
mean_foreground = (sum_total - sum_background) / weight_foreground
between_class_variance = weight_background * weight_foreground * \
(mean_background - mean_foreground) ** 2
if between_class_variance > current_max:
current_max = between_class_variance
threshold_value = t / 256.0
else:
threshold_value = threshold
# Binarize
binary = sal_map > threshold_value
binary_mask[b] = torch.from_numpy(binary).float().to(saliency_map.device)
return binary_mask
if __name__ == "__main__":
# Test M2IB saliency generation
batch_size = 4
num_patches = 196 # 14x14 patches
embedding_dim = 768
image_height, image_width = 224, 224
# Create dummy inputs
image_embeddings = torch.randn(batch_size, num_patches, embedding_dim)
text_embedding = torch.randn(batch_size, embedding_dim)
# Test M2IB
m2ib = MultiModalInformationBottleneck(image_height, image_width, embedding_dim)
saliency_map = m2ib(image_embeddings, text_embedding)
print(f"M2IB Saliency Map shape: {saliency_map.shape}")
print(f"Saliency map range: [{saliency_map.min():.4f}, {saliency_map.max():.4f}]")
# Test GradCAM
gradcam = GradCAM_Saliency(image_height, image_width)
image_embeddings.requires_grad = True
saliency_gradcam = gradcam(image_embeddings, text_embedding)
print(f"GradCAM Saliency Map shape: {saliency_gradcam.shape}")
# Test post-processing
binary_mask = post_process_saliency_map(saliency_map, smooth=True)
print(f"Binary mask shape: {binary_mask.shape}")
print(f"Binary mask unique values: {torch.unique(binary_mask)}")"""
Decoupled Hard Negative Noise Contrastive Estimation (DHN-NCE) Loss
Implementation for BiomedCLIP fine-tuning in MedCLIP-SAMv2
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
class DHN_NCE_Loss(nn.Module):
"""
Decoupled Hard Negative Noise Contrastive Estimation Loss
Combines InfoNCE loss with hard negative sampling and positive-negative decoupling
to improve medical image-text alignment training efficiency.
Args:
temperature (float): Temperature parameter for softmax scaling
beta1 (float): Hardness parameter for image-to-text direction
beta2 (float): Hardness parameter for text-to-image direction
"""
def __init__(self, temperature=0.6, beta1=0.15, beta2=0.15):
super(DHN_NCE_Loss, self).__init__()
self.temperature = temperature
self.beta1 = beta1
self.beta2 = beta2
self.cross_entropy = nn.CrossEntropyLoss()
def forward(self, image_embeddings, text_embeddings):
"""
Compute DHN-NCE loss for vision-language alignment
Args:
image_embeddings: (batch_size, embedding_dim) Image embeddings from ViT
text_embeddings: (batch_size, embedding_dim) Text embeddings from BERT
Returns:
loss: Scalar loss value combining v->t and t->v losses
"""
batch_size = image_embeddings.shape[0]
# Normalize embeddings
image_embeddings = F.normalize(image_embeddings, dim=1)
text_embeddings = F.normalize(text_embeddings, dim=1)
# Compute similarity matrices
# Shape: (batch_size, batch_size)
logits_img_text = torch.matmul(image_embeddings, text_embeddings.T) / self.temperature
logits_text_img = torch.matmul(text_embeddings, image_embeddings.T) / self.temperature
# Image-to-Text (V->T) Loss with Hard Negative Sampling
loss_v_t = self._compute_decoupled_loss_with_hardness(
logits_img_text, batch_size, self.beta1, direction="v_t"
)
# Text-to-Image (T->V) Loss with Hard Negative Sampling
loss_t_v = self._compute_decoupled_loss_with_hardness(
logits_text_img, batch_size, self.beta2, direction="t_v"
)
# Combined loss
total_loss = (loss_v_t + loss_t_v) / 2
return total_loss
def _compute_decoupled_loss_with_hardness(self, logits, batch_size, beta, direction="v_t"):
"""
Compute decoupled contrastive loss with hard negative weighting
Args:
logits: Similarity scores matrix (batch_size, batch_size)
batch_size: Number of samples in batch
beta: Hardness parameter for exponential weighting
direction: String indicating loss direction for debugging
Returns:
loss: Scalar loss value
"""
# Create identity matrix for diagonal elements (positive pairs)
diagonal = torch.eye(batch_size, device=logits.device, dtype=torch.bool)
loss = 0.0
for i in range(batch_size):
# Positive pair logit
pos_logit = logits[i, i]
# Negative logits (all except diagonal)
neg_logits = logits[i, ~diagonal[i]]
# Compute hardness weights using exponential scaling
# Higher similarity to anchor = higher penalty (harder negative)
hardness_weights = torch.exp(beta * neg_logits / self.temperature)
hardness_weights = hardness_weights / hardness_weights.sum()
# Scale weights to sum to (batch_size - 1)
hardness_weights = hardness_weights * (batch_size - 1)
# Decoupled loss: separate positive from negative denominator
# L = -pos_logit + log(sum(exp(neg_logits * weights)))
weighted_neg_logits = neg_logits * hardness_weights
neg_sum = torch.logsumexp(weighted_neg_logits, dim=0)
sample_loss = -pos_logit + neg_sum
loss = loss + sample_loss
return loss / batch_size
def _compute_hardness_weights(self, neg_logits, beta):
"""
Compute hardness weights for negative samples
Harder negatives (higher similarity) receive higher weights, enforcing
the model to focus on distinguishing difficult cases
Args:
neg_logits: (num_negatives,) Logits for negative samples
beta: Hardness parameter
Returns:
weights: (num_negatives,) Normalized hardness weights
"""
weights = torch.exp(beta * neg_logits / self.temperature)
return weights / weights.sum()
class InfoNCE_Loss(nn.Module):
"""
Standard InfoNCE Loss (baseline for comparison)
"""
def __init__(self, temperature=0.6):
super(InfoNCE_Loss, self).__init__()
self.temperature = temperature
def forward(self, image_embeddings, text_embeddings):
"""
Compute standard InfoNCE loss
Args:
image_embeddings: (batch_size, embedding_dim)
text_embeddings: (batch_size, embedding_dim)
Returns:
loss: Scalar loss value
"""
batch_size = image_embeddings.shape[0]
# Normalize embeddings
image_embeddings = F.normalize(image_embeddings, dim=1)
text_embeddings = F.normalize(text_embeddings, dim=1)
# Compute similarity matrices
logits = torch.matmul(image_embeddings, text_embeddings.T) / self.temperature
# Create labels: diagonal elements are positive pairs
labels = torch.arange(batch_size, device=logits.device)
# Image-to-Text loss
loss_i_t = F.cross_entropy(logits, labels)
# Text-to-Image loss
loss_t_i = F.cross_entropy(logits.T, labels)
return (loss_i_t + loss_t_i) / 2
class SupConLoss(nn.Module):
"""
Supervised Contrastive Loss (alternative baseline)
Reference: https://arxiv.org/abs/2004.11362
"""
def __init__(self, temperature=0.07):
super(SupConLoss, self).__init__()
self.temperature = temperature
def forward(self, features, labels):
"""
Args:
features: (batch_size, embedding_dim)
labels: (batch_size,) Class labels
Returns:
loss: Scalar loss value
"""
batch_size = features.shape[0]
# Normalize features
features = F.normalize(features, dim=1)
# Compute similarity
similarity = torch.matmul(features, features.T) / self.temperature
# Create label matrix
labels_matrix = labels.unsqueeze(1) == labels.unsqueeze(0)
# Mask out self-similarity
mask = torch.eye(batch_size, device=features.device, dtype=torch.bool)
labels_matrix[mask] = False
# Compute loss
pos = similarity[labels_matrix].view(batch_size, -1)
neg = similarity[~labels_matrix].view(batch_size, -1)
logits = torch.cat([pos, neg], dim=1)
labels_loss = torch.zeros(batch_size, dtype=torch.long, device=features.device)
loss = F.cross_entropy(logits, labels_loss)
return loss
class CombinedVisionLanguageLoss(nn.Module):
"""
Combined loss for vision-language model training with options for different loss functions
"""
def __init__(self, loss_type="DHN_NCE", temperature=0.6, beta1=0.15, beta2=0.15):
"""
Args:
loss_type: "DHN_NCE", "InfoNCE", or "SupCon"
temperature: Temperature for softmax
beta1, beta2: Hardness parameters for DHN_NCE
"""
super(CombinedVisionLanguageLoss, self).__init__()
self.loss_type = loss_type
if loss_type == "DHN_NCE":
self.loss_fn = DHN_NCE_Loss(temperature, beta1, beta2)
elif loss_type == "InfoNCE":
self.loss_fn = InfoNCE_Loss(temperature)
elif loss_type == "SupCon":
self.loss_fn = SupConLoss(temperature)
else:
raise ValueError(f"Unknown loss type: {loss_type}")
def forward(self, image_embeddings, text_embeddings, labels=None):
"""
Args:
image_embeddings: (batch_size, embedding_dim)
text_embeddings: (batch_size, embedding_dim)
labels: Optional labels for supervised losses
Returns:
loss: Scalar loss value
"""
if self.loss_type == "SupCon" and labels is None:
raise ValueError("SupCon loss requires labels")
if self.loss_type == "SupCon":
return self.loss_fn(torch.cat([image_embeddings, text_embeddings], dim=0),
torch.cat([labels, labels], dim=0))
else:
return self.loss_fn(image_embeddings, text_embeddings)
if __name__ == "__main__":
# Test DHN-NCE loss
batch_size = 32
embedding_dim = 512
image_embeddings = torch.randn(batch_size, embedding_dim)
text_embeddings = torch.randn(batch_size, embedding_dim)
# Test DHN-NCE
loss_fn = DHN_NCE_Loss(temperature=0.6, beta1=0.15, beta2=0.15)
loss = loss_fn(image_embeddings, text_embeddings)
print(f"DHN-NCE Loss: {loss.item():.4f}")
# Test InfoNCE for comparison
loss_fn_infonce = InfoNCE_Loss(temperature=0.6)
loss_infonce = loss_fn_infonce(image_embeddings, text_embeddings)
print(f"InfoNCE Loss: {loss_infonce.item():.4f}")
# Test gradient flow
loss.backward()
print("Gradients computed successfully")Related posts, You May like to read
- 7 Shocking Truths About Knowledge Distillation: The Good, The Bad, and The Breakthrough (SAKD)
- 7 Revolutionary Breakthroughs in Medical Image Translation (And 1 Fatal Flaw That Could Derail Your AI Model)
- MedDINOv3: Revolutionizing Medical Image Segmentation with Adaptable Vision Foundation Models
- HiPerformer: A New Benchmark in Medical Image Segmentation with Modular Hierarchical Fusion
- GeoSAM2 3D Part Segmentation — Prompt-Controllable, Geometry-Aware Masks for Precision 3D Editing
- SegTrans: The Breakthrough Framework That Makes AI Segmentation Models Vulnerable to Transfer Attacks
- A Knowledge Distillation-Based Approach to Enhance Transparency of Classifier Models
- Towards Trustworthy Breast Tumor Segmentation in Ultrasound Using AI Uncertainty
- Cellpose3: The Revolutionary One-Click Solution for Restoring Noisy, Blurry, and Undersampled Microscopy Images
RuNRkXBaXulTgQOrlMUl