Spaces:

ThaiLearnCoding
/

Coco_caption_classification

Sleeping

Coco_caption_classification / src /utils.py

fix(cv2)

fc15e02 6 months ago

10.4 kB

import torch
from sklearn.metrics import accuracy_score, f1_score
from torchvision.transforms import Compose, Resize, CenterCrop, ToTensor, Normalize
import clip
from PIL import Image

# CLIP normalization for ViT and ResNet
# Reference: https://github.com/openai/CLIP/blob/main/clip/clip.py#L79
clip_normalize = Normalize(
    mean=(0.48145466, 0.4578275, 0.40821073),
    std=(0.26862954, 0.26130258, 0.27577711)
)

def get_image_transform(image_size=224):
    """
    Standard image transformation pipeline for models like ViT or ResNet.
    """
    return Compose([
        Resize(image_size, interpolation=3), # InterpolationMode.BICUBIC
        CenterCrop(image_size),
        ToTensor(),
        clip_normalize
    ])

def evaluate_model(model, dataloader, device):
    model.eval()
    all_preds = []
    all_targets = []
    
    with torch.no_grad():
        for batch in dataloader:
            images, text_candidates, targets, _, _ = batch
            images = images.to(device)
            targets = targets.to(device)
            
            logits = model(images, text_candidates)
            preds = torch.argmax(logits, dim=1)
            
            all_preds.extend(preds.cpu().numpy())
            all_targets.extend(targets.cpu().numpy())
            
    acc = accuracy_score(all_targets, all_preds)
    f1_macro = f1_score(all_targets, all_preds, average="macro")
    f1_micro = f1_score(all_targets, all_preds, average="micro")
    f1_weighted = f1_score(all_targets, all_preds, average="weighted")

    return acc, f1_macro, f1_micro, f1_weighted

def calculate_model_size_params(model):
    # Total parameters
    total_params = sum(p.numel() for p in model.parameters())
    # Trainable parameters
    trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
    
    # Model size in MB
    param_size = 0
    for param in model.parameters():
        param_size += param.nelement() * param.element_size()
    buffer_size = 0
    for buffer in model.buffers():
        buffer_size += buffer.nelement() * buffer.element_size()
        
    size_all_mb = (param_size + buffer_size) / 1024**2
    
    print(f"Total Parameters: {total_params:,}")
    print(f"Trainable Parameters: {trainable_params:,}")
    print('Model size: {:.3f}MB'.format(size_all_mb))
    
    return total_params, trainable_params, size_all_mb

def extract_image_attention(model, preprocess, image_path, text_query, device, model_type="vit"):
    """
    Extract attention map using EigenCAM for ViT or GradCAM for ResNet.
    For simplicity, we visualize what the image encoder focuses on.
    """
    try:
        from pytorch_grad_cam import GradCAM, EigenCAM
        from pytorch_grad_cam.utils.image import show_cam_on_image
        import cv2
        import numpy as np
    except ImportError:
        print("Attention unavailable: install grad-cam and opencv-python-headless.")
        return None, None

    try:
        # If wrapped in our custom classifier, get the base model
        if hasattr(model, 'model'):
            base_model = model.model
        else:
            base_model = model
            
        # Temporarily cast to float32 to avoid GradCAM HalfTensor errors
        prev_dtype = next(base_model.parameters()).dtype
        base_model.float()

        # Load and preprocess image
        original_image = Image.open(image_path).convert("RGB")
        img_tensor = preprocess(original_image).unsqueeze(0).to(device).float()
        text_tokens = clip.tokenize([text_query]).to(device)
        
        # Prepare image for visualization
        rgb_img = np.float32(original_image) / 255
        rgb_img = cv2.resize(rgb_img, (224, 224))
        
        # Determine target layer based on model type
        def reshape_transform(tensor, height=7, width=7):
            # For CLIP ViT: output shape is [50, 1, 768] (Seq_len, Batch, Channels)
            tensor = tensor.permute(1, 0, 2) # [1, 50, 768]
            result = tensor[:, 1:, :].reshape(tensor.size(0), height, width, tensor.size(2)) 
            result = result.transpose(2, 3).transpose(1, 2)
            return result

        if model_type == "vit":
            # For ViT, we use the last normalization layer before the final projection
            target_layers = [base_model.visual.transformer.resblocks[-1].ln_1]
            cam = EigenCAM(model=base_model.visual, target_layers=target_layers, reshape_transform=reshape_transform)
        else:
            # For ResNet (RN50), we use the last bottleneck layer
            target_layers = [base_model.visual.layer4[-1]]
            cam = GradCAM(model=base_model.visual, target_layers=target_layers)
            
        # Generate CAM
        grayscale_cam = cam(input_tensor=img_tensor, targets=None)[0, :]
        grayscale_cam = cv2.resize(grayscale_cam, (rgb_img.shape[1], rgb_img.shape[0]))
        cam_image = show_cam_on_image(rgb_img, grayscale_cam, use_rgb=True)
        
        # Revert model dtype
        base_model.to(prev_dtype)

        return original_image, cam_image
    except Exception as e:
        print(f"CAM extraction failed: {e}")
        return None, None

import os

def plot_prediction_visualizations(subset_data, vit_zs_model, rn50_zs_model, preprocess_vit, preprocess_rn50, device, image_dir='../coco_subset_images/images'):
    """
    Plot and visualize attention maps using GradCAM / EigenCAM.
    """
    import matplotlib.pyplot as plt
    print("\n--- Visualizing Predictions ---")
    if subset_data and len(subset_data) > 0:
        fig, axes = plt.subplots(2, 4, figsize=(16, 8))
        fig.suptitle("Prediction Visualizations: Original Image vs Attention Map\nTop: ViT (EigenCAM) | Bottom: RN50 (GradCAM)", fontsize=16)
        
        # Select a sample image from subset_data
        sample = subset_data[0]
        img_path = os.path.join(image_dir, f"{sample['image_id']}.jpg")
        text_query = sample['choices'][sample['ground_truth_idx']]
        
        # Helper to plot
        def plot_cam(ax_orig, ax_cam, orig_img, cam_img, title):
            if orig_img is not None and cam_img is not None:
                ax_orig.imshow(orig_img)
                ax_orig.set_title("Original")
                ax_orig.axis('off')
                ax_cam.imshow(cam_img)
                ax_cam.set_title(title)
                ax_cam.axis('off')
            else:
                ax_orig.axis('off')
                ax_cam.axis('off')

        # ViT EigenCAM
        vit_orig, vit_cam = extract_image_attention(vit_zs_model, preprocess_vit, img_path, text_query, device, model_type="vit")
        plot_cam(axes[0, 0], axes[0, 1], vit_orig, vit_cam, "ViT EigenCAM")
        
        # ResNet50 GradCAM
        rn50_orig, rn50_cam = extract_image_attention(rn50_zs_model, preprocess_rn50, img_path, text_query, device, model_type="rn50")
        plot_cam(axes[1, 0], axes[1, 1], rn50_orig, rn50_cam, "RN50 GradCAM")
        
        # Visualize another sample
        if len(subset_data) > 1:
            sample2 = subset_data[-1]
            img_path2 = os.path.join(image_dir, f"{sample2['image_id']}.jpg")
            text_query2 = sample2['choices'][sample2['ground_truth_idx']]
            
            vit_orig2, vit_cam2 = extract_image_attention(vit_zs_model, preprocess_vit, img_path2, text_query2, device, model_type="vit")
            plot_cam(axes[0, 2], axes[0, 3], vit_orig2, vit_cam2, "ViT EigenCAM (Sample 2)")
            
            rn50_orig2, rn50_cam2 = extract_image_attention(rn50_zs_model, preprocess_rn50, img_path2, text_query2, device, model_type="rn50")
            plot_cam(axes[1, 2], axes[1, 3], rn50_orig2, rn50_cam2, "RN50 GradCAM (Sample 2)")
            
        plt.tight_layout()
        plt.show()
    else:
        print("Data not available for visualization.")

def plot_failure_cases(model, dataloader, device, image_dir, num_cases=4):
    """
    Find and plot failure cases from the dataloader.
    """
    model.eval()
    failures = []
    
    with torch.no_grad():
        for batch in dataloader:
            images, text_candidates, targets, image_ids, _ = batch
            images = images.to(device)
            targets = targets.to(device)
            
            logits = model(images, text_candidates)
            preds = torch.argmax(logits, dim=1)
            
            for i in range(len(preds)):
                if preds[i] != targets[i]:
                    img_path = os.path.join(image_dir, f"{image_ids[i]}.jpg")
                    # Dataloader gives text_candidates as a list of tuples, where text_candidates[j] is the j-th candidate for all items in batch
                    # So text_candidates[j][i] is the j-th candidate for the i-th item in the batch.
                    gt_idx = targets[i].item()
                    pred_idx = preds[i].item()
                    gt_text = text_candidates[gt_idx][i]
                    pred_text = text_candidates[pred_idx][i]
                    
                    failures.append((img_path, gt_text, pred_text))
                    
                    if len(failures) >= num_cases:
                        break
            if len(failures) >= num_cases:
                break
                
    if len(failures) == 0:
        print("No failure cases found!")
        return
        
    num_cases = min(num_cases, len(failures))
    fig, axes = plt.subplots(1, num_cases, figsize=(6 * num_cases, 6))
    if num_cases == 1:
        axes = [axes]
        
    import textwrap
    for idx, (img_path, gt_text, pred_text) in enumerate(failures):
        ax = axes[idx]
        img = Image.open(img_path).convert("RGB")
        ax.imshow(img)
        ax.axis('off')

        wrapped_gt = "\\n".join(textwrap.wrap(gt_text, width=40))
        wrapped_pred = "\\n".join(textwrap.wrap(pred_text, width=40))

        ax.text(
            0.0,
            1.12,
            f"GT: {wrapped_gt}",
            transform=ax.transAxes,
            ha='left',
            va='bottom',
            fontsize=10,
            color='green',
            clip_on=False
        )
        ax.text(
            0.0,
            1.02,
            f"Pred: {wrapped_pred}",
            transform=ax.transAxes,
            ha='left',
            va='bottom',
            fontsize=10,
            color='red',
            clip_on=False
        )

    fig.tight_layout()
    fig.subplots_adjust(top=0.82)
    plt.show()

	import torch
	from sklearn.metrics import accuracy_score, f1_score
	from torchvision.transforms import Compose, Resize, CenterCrop, ToTensor, Normalize
	import clip
	from PIL import Image

	# CLIP normalization for ViT and ResNet
	# Reference: https://github.com/openai/CLIP/blob/main/clip/clip.py#L79
	clip_normalize = Normalize(
	mean=(0.48145466, 0.4578275, 0.40821073),
	std=(0.26862954, 0.26130258, 0.27577711)
	)

	def get_image_transform(image_size=224):
	"""
	Standard image transformation pipeline for models like ViT or ResNet.
	"""
	return Compose([
	Resize(image_size, interpolation=3), # InterpolationMode.BICUBIC
	CenterCrop(image_size),
	ToTensor(),
	clip_normalize
	])

	def evaluate_model(model, dataloader, device):
	model.eval()
	all_preds = []
	all_targets = []

	with torch.no_grad():
	for batch in dataloader:
	images, text_candidates, targets, _, _ = batch
	images = images.to(device)
	targets = targets.to(device)

	logits = model(images, text_candidates)
	preds = torch.argmax(logits, dim=1)

	all_preds.extend(preds.cpu().numpy())
	all_targets.extend(targets.cpu().numpy())

	acc = accuracy_score(all_targets, all_preds)
	f1_macro = f1_score(all_targets, all_preds, average="macro")
	f1_micro = f1_score(all_targets, all_preds, average="micro")
	f1_weighted = f1_score(all_targets, all_preds, average="weighted")

	return acc, f1_macro, f1_micro, f1_weighted

	def calculate_model_size_params(model):
	# Total parameters
	total_params = sum(p.numel() for p in model.parameters())
	# Trainable parameters
	trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)

	# Model size in MB
	param_size = 0
	for param in model.parameters():
	param_size += param.nelement() * param.element_size()
	buffer_size = 0
	for buffer in model.buffers():
	buffer_size += buffer.nelement() * buffer.element_size()

	size_all_mb = (param_size + buffer_size) / 1024**2

	print(f"Total Parameters: {total_params:,}")
	print(f"Trainable Parameters: {trainable_params:,}")
	print('Model size: {:.3f}MB'.format(size_all_mb))

	return total_params, trainable_params, size_all_mb

	def extract_image_attention(model, preprocess, image_path, text_query, device, model_type="vit"):
	"""
	Extract attention map using EigenCAM for ViT or GradCAM for ResNet.
	For simplicity, we visualize what the image encoder focuses on.
	"""
	try:
	from pytorch_grad_cam import GradCAM, EigenCAM
	from pytorch_grad_cam.utils.image import show_cam_on_image
	import cv2
	import numpy as np
	except ImportError:
	print("Attention unavailable: install grad-cam and opencv-python-headless.")
	return None, None

	try:
	# If wrapped in our custom classifier, get the base model
	if hasattr(model, 'model'):
	base_model = model.model
	else:
	base_model = model

	# Temporarily cast to float32 to avoid GradCAM HalfTensor errors
	prev_dtype = next(base_model.parameters()).dtype
	base_model.float()

	# Load and preprocess image
	original_image = Image.open(image_path).convert("RGB")
	img_tensor = preprocess(original_image).unsqueeze(0).to(device).float()
	text_tokens = clip.tokenize([text_query]).to(device)

	# Prepare image for visualization
	rgb_img = np.float32(original_image) / 255
	rgb_img = cv2.resize(rgb_img, (224, 224))

	# Determine target layer based on model type
	def reshape_transform(tensor, height=7, width=7):
	# For CLIP ViT: output shape is [50, 1, 768] (Seq_len, Batch, Channels)
	tensor = tensor.permute(1, 0, 2) # [1, 50, 768]
	result = tensor[:, 1:, :].reshape(tensor.size(0), height, width, tensor.size(2))
	result = result.transpose(2, 3).transpose(1, 2)
	return result

	if model_type == "vit":
	# For ViT, we use the last normalization layer before the final projection
	target_layers = [base_model.visual.transformer.resblocks[-1].ln_1]
	cam = EigenCAM(model=base_model.visual, target_layers=target_layers, reshape_transform=reshape_transform)
	else:
	# For ResNet (RN50), we use the last bottleneck layer
	target_layers = [base_model.visual.layer4[-1]]
	cam = GradCAM(model=base_model.visual, target_layers=target_layers)

	# Generate CAM
	grayscale_cam = cam(input_tensor=img_tensor, targets=None)[0, :]
	grayscale_cam = cv2.resize(grayscale_cam, (rgb_img.shape[1], rgb_img.shape[0]))
	cam_image = show_cam_on_image(rgb_img, grayscale_cam, use_rgb=True)

	# Revert model dtype
	base_model.to(prev_dtype)

	return original_image, cam_image
	except Exception as e:
	print(f"CAM extraction failed: {e}")
	return None, None

	import os

	def plot_prediction_visualizations(subset_data, vit_zs_model, rn50_zs_model, preprocess_vit, preprocess_rn50, device, image_dir='../coco_subset_images/images'):
	"""
	Plot and visualize attention maps using GradCAM / EigenCAM.
	"""
	import matplotlib.pyplot as plt
	print("\n--- Visualizing Predictions ---")
	if subset_data and len(subset_data) > 0:
	fig, axes = plt.subplots(2, 4, figsize=(16, 8))
	fig.suptitle("Prediction Visualizations: Original Image vs Attention Map\nTop: ViT (EigenCAM) \| Bottom: RN50 (GradCAM)", fontsize=16)

	# Select a sample image from subset_data
	sample = subset_data[0]
	img_path = os.path.join(image_dir, f"{sample['image_id']}.jpg")
	text_query = sample['choices'][sample['ground_truth_idx']]

	# Helper to plot
	def plot_cam(ax_orig, ax_cam, orig_img, cam_img, title):
	if orig_img is not None and cam_img is not None:
	ax_orig.imshow(orig_img)
	ax_orig.set_title("Original")
	ax_orig.axis('off')
	ax_cam.imshow(cam_img)
	ax_cam.set_title(title)
	ax_cam.axis('off')
	else:
	ax_orig.axis('off')
	ax_cam.axis('off')

	# ViT EigenCAM
	vit_orig, vit_cam = extract_image_attention(vit_zs_model, preprocess_vit, img_path, text_query, device, model_type="vit")
	plot_cam(axes[0, 0], axes[0, 1], vit_orig, vit_cam, "ViT EigenCAM")

	# ResNet50 GradCAM
	rn50_orig, rn50_cam = extract_image_attention(rn50_zs_model, preprocess_rn50, img_path, text_query, device, model_type="rn50")
	plot_cam(axes[1, 0], axes[1, 1], rn50_orig, rn50_cam, "RN50 GradCAM")

	# Visualize another sample
	if len(subset_data) > 1:
	sample2 = subset_data[-1]
	img_path2 = os.path.join(image_dir, f"{sample2['image_id']}.jpg")
	text_query2 = sample2['choices'][sample2['ground_truth_idx']]

	vit_orig2, vit_cam2 = extract_image_attention(vit_zs_model, preprocess_vit, img_path2, text_query2, device, model_type="vit")
	plot_cam(axes[0, 2], axes[0, 3], vit_orig2, vit_cam2, "ViT EigenCAM (Sample 2)")

	rn50_orig2, rn50_cam2 = extract_image_attention(rn50_zs_model, preprocess_rn50, img_path2, text_query2, device, model_type="rn50")
	plot_cam(axes[1, 2], axes[1, 3], rn50_orig2, rn50_cam2, "RN50 GradCAM (Sample 2)")

	plt.tight_layout()
	plt.show()
	else:
	print("Data not available for visualization.")

	def plot_failure_cases(model, dataloader, device, image_dir, num_cases=4):
	"""
	Find and plot failure cases from the dataloader.
	"""
	model.eval()
	failures = []

	with torch.no_grad():
	for batch in dataloader:
	images, text_candidates, targets, image_ids, _ = batch
	images = images.to(device)
	targets = targets.to(device)

	logits = model(images, text_candidates)
	preds = torch.argmax(logits, dim=1)

	for i in range(len(preds)):
	if preds[i] != targets[i]:
	img_path = os.path.join(image_dir, f"{image_ids[i]}.jpg")
	# Dataloader gives text_candidates as a list of tuples, where text_candidates[j] is the j-th candidate for all items in batch
	# So text_candidates[j][i] is the j-th candidate for the i-th item in the batch.
	gt_idx = targets[i].item()
	pred_idx = preds[i].item()
	gt_text = text_candidates[gt_idx][i]
	pred_text = text_candidates[pred_idx][i]

	failures.append((img_path, gt_text, pred_text))

	if len(failures) >= num_cases:
	break
	if len(failures) >= num_cases:
	break

	if len(failures) == 0:
	print("No failure cases found!")
	return

	num_cases = min(num_cases, len(failures))
	fig, axes = plt.subplots(1, num_cases, figsize=(6 * num_cases, 6))
	if num_cases == 1:
	axes = [axes]

	import textwrap
	for idx, (img_path, gt_text, pred_text) in enumerate(failures):
	ax = axes[idx]
	img = Image.open(img_path).convert("RGB")
	ax.imshow(img)
	ax.axis('off')

	wrapped_gt = "\\n".join(textwrap.wrap(gt_text, width=40))
	wrapped_pred = "\\n".join(textwrap.wrap(pred_text, width=40))

	ax.text(
	0.0,
	1.12,
	f"GT: {wrapped_gt}",
	transform=ax.transAxes,
	ha='left',
	va='bottom',
	fontsize=10,
	color='green',
	clip_on=False
	)
	ax.text(
	0.0,
	1.02,
	f"Pred: {wrapped_pred}",
	transform=ax.transAxes,
	ha='left',
	va='bottom',
	fontsize=10,
	color='red',
	clip_on=False
	)

	fig.tight_layout()
	fig.subplots_adjust(top=0.82)
	plt.show()