Upload evaluation/hierarchy_evaluation.py with huggingface_hub

evaluation/hierarchy_evaluation.py

@@ -0,0 +1,589 @@
+"""
+Hierarchy embedding evaluation for clothing category classification.
+This file evaluates the quality of hierarchy embeddings generated by the hierarchy model
+by calculating intra-class and inter-class similarity metrics, nearest neighbor and centroid-based
+classification accuracies, and generating confusion matrices. It can be used on different datasets
+(local validation, Kagl Marqo) to measure model generalization.
+"""
+
+import torch
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+import seaborn as sns
+from sklearn.metrics.pairwise import cosine_similarity
+from sklearn.metrics import confusion_matrix, classification_report, accuracy_score
+from collections import defaultdict
+import os
+from tqdm import tqdm
+from torch.utils.data import Dataset, DataLoader
+from torchvision import transforms
+from sklearn.model_selection import train_test_split
+from io import BytesIO
+from PIL import Image
+import config
+import warnings
+warnings.filterwarnings('ignore')
+from hierarchy_model import Model, HierarchyExtractor, HierarchyDataset, collate_fn
+
+
+class EmbeddingEvaluator:
+    """
+    Evaluator for hierarchy embeddings generated by the hierarchy model.
+    
+    This class provides methods to evaluate the quality of hierarchy embeddings by computing
+    similarity metrics, classification accuracies, and generating visualizations.
+    """
+    
+    def __init__(self, model_path, directory):
+        """
+        Initialize the embedding evaluator.
+        
+        Args:
+            model_path: Path to the trained hierarchy model checkpoint
+            directory: Directory to save evaluation results and visualizations
+        """
+        self.device = config.device
+        self.directory = directory
+
+        # 1. Load the dataset
+        CSV = config.local_dataset_path
+        print(f"📁 Using dataset with local images: {CSV}")
+        df = pd.read_csv(CSV)
+        
+        print(f"📁 Loaded {len(df)} samples")
+        
+        # 2. Get unique hierarchy classes from the dataset
+        hierarchy_classes = sorted(df[config.hierarchy_column].unique().tolist())
+        print(f"📋 Found {len(hierarchy_classes)} hierarchy classes")
+
+        _, self.val_df = train_test_split(df, test_size=0.2, random_state=42, stratify=df[config.hierarchy_column])
+        
+        # 3. Load the model
+        if os.path.exists(model_path):
+            checkpoint = torch.load(model_path, map_location=self.device)
+            
+            # Use model_config to avoid shadowing the imported config module
+            model_config = checkpoint.get('config', {})
+            saved_hierarchy_classes = checkpoint['hierarchy_classes']
+            
+            # Use the saved hierarchy classes
+            self.hierarchy_classes = saved_hierarchy_classes
+            
+            # Create the hierarchy extractor
+            self.vocab = HierarchyExtractor(saved_hierarchy_classes)
+            
+            # Create the model with the saved configuration
+            self.model = Model(
+                num_hierarchy_classes=len(saved_hierarchy_classes),
+                embed_dim=model_config['embed_dim'],
+                dropout=model_config['dropout']
+            ).to(self.device)   
+                
+            self.model.load_state_dict(checkpoint['model_state'])
+            
+            print(f"✅ Model loaded with:")
+            print(f"📋 Hierarchy classes: {len(saved_hierarchy_classes)}")
+            print(f"🎯 Embed dim: {model_config['embed_dim']}")
+            print(f"💧 Dropout: {model_config['dropout']}")
+            print(f"📅 Epoch: {checkpoint.get('epoch', 'unknown')}")
+                
+        else:
+            raise FileNotFoundError(f"Model file {model_path} not found")
+        
+        self.model.eval()
+    
+    def create_dataloader(self, dataframe, batch_size=16):
+        """
+        Create a DataLoader for the hierarchy dataset.
+        
+        Args:
+            dataframe: DataFrame containing the dataset
+            batch_size: Batch size for the DataLoader
+            
+        Returns:
+            DataLoader instance
+        """
+        dataset = HierarchyDataset(dataframe, image_size=224)
+        
+        dataloader = DataLoader(
+            dataset, 
+            batch_size=batch_size, 
+            shuffle=False, 
+            collate_fn=lambda batch: collate_fn(batch, self.vocab), 
+            num_workers=0
+        )
+        
+        return dataloader
+    
+    def extract_embeddings(self, dataloader, embedding_type='text'):
+        """
+        Extract embeddings from the model for a given dataloader.
+        
+        Args:
+            dataloader: DataLoader containing images, texts, and hierarchy labels
+            embedding_type: Type of embeddings to extract ('text' or 'image')
+            
+        Returns:
+            Tuple of (embeddings array, labels list, texts list)
+        """
+        all_embeddings = []
+        all_labels = []
+        all_texts = []
+        
+        with torch.no_grad():
+            for batch in tqdm(dataloader, desc=f"Extracting {embedding_type} embeddings"):
+                images = batch['image'].to(self.device)
+                hierarchy_indices = batch['hierarchy_indices'].to(self.device)
+                hierarchy_labels = batch['hierarchy']
+                
+                # Forward pass
+                out = self.model(image=images, hierarchy_indices=hierarchy_indices)
+                embeddings = out['z_txt'] if embedding_type == 'text' else out['z_img'] if embedding_type == 'image' else out['z_txt']
+                
+                all_embeddings.append(embeddings.cpu().numpy())
+                all_labels.extend(hierarchy_labels)
+                all_texts.extend(hierarchy_labels)
+        
+        return np.vstack(all_embeddings), all_labels, all_texts
+    
+    def compute_similarity_metrics(self, embeddings, labels):
+        """
+        Compute intra-class and inter-class similarity metrics.
+        
+        Args:
+            embeddings: Array of embeddings [N, embed_dim]
+            labels: List of labels for each embedding
+            
+        Returns:
+            Dictionary containing similarity metrics, accuracies, and separation scores
+        """
+        similarities = cosine_similarity(embeddings)
+        
+        # Group embeddings by hierarchy
+        hierarchy_groups = defaultdict(list)
+        for i, hierarchy in enumerate(labels):
+            hierarchy_groups[hierarchy].append(i)
+        
+        # Calculate intra-class similarities (same hierarchy)
+        intra_class_similarities = []
+        for hierarchy, indices in hierarchy_groups.items():
+            if len(indices) > 1:
+                for i in range(len(indices)):
+                    for j in range(i+1, len(indices)):
+                        sim = similarities[indices[i], indices[j]]
+                        intra_class_similarities.append(sim)
+        
+                
+        # Calculate inter-class similarities (different hierarchies)
+        inter_class_similarities = []
+        hierarchies = list(hierarchy_groups.keys())
+        for i in range(len(hierarchies)):
+            for j in range(i+1, len(hierarchies)):
+                hierarchy1_indices = hierarchy_groups[hierarchies[i]]
+                hierarchy2_indices = hierarchy_groups[hierarchies[j]]
+                
+                for idx1 in hierarchy1_indices:
+                    for idx2 in hierarchy2_indices:
+                        sim = similarities[idx1, idx2]
+                        inter_class_similarities.append(sim)
+        
+        # Calculate classification accuracy using nearest neighbor in embedding space
+        nn_accuracy = self.compute_embedding_accuracy(embeddings, labels, similarities)
+        
+        # Calculate classification accuracy using centroids
+        centroid_accuracy = self.compute_centroid_accuracy(embeddings, labels)
+        
+        return {
+            'intra_class_similarities': intra_class_similarities,
+            'inter_class_similarities': inter_class_similarities,
+            'intra_class_mean': np.mean(intra_class_similarities) if intra_class_similarities else 0,
+            'inter_class_mean': np.mean(inter_class_similarities) if inter_class_similarities else 0,
+            'separation_score': np.mean(intra_class_similarities) - np.mean(inter_class_similarities) if intra_class_similarities and inter_class_similarities else 0,
+            'accuracy': nn_accuracy,
+            'centroid_accuracy': centroid_accuracy
+        }
+    
+    def compute_embedding_accuracy(self, embeddings, labels, similarities):
+        """
+        Compute classification accuracy using nearest neighbor in embedding space.
+        
+        Args:
+            embeddings: Array of embeddings [N, embed_dim]
+            labels: List of true labels
+            similarities: Pre-computed similarity matrix [N, N]
+            
+        Returns:
+            Accuracy score (float between 0 and 1)
+        """
+        correct_predictions = 0
+        total_predictions = len(labels)
+        
+        for i in range(len(embeddings)):
+            true_label = labels[i]
+            
+            # Find the most similar embedding (excluding itself)
+            similarities_row = similarities[i].copy()
+            similarities_row[i] = -1  # Exclude self-similarity
+            nearest_neighbor_idx = np.argmax(similarities_row)
+            predicted_label = labels[nearest_neighbor_idx]
+            
+            if predicted_label == true_label:
+                correct_predictions += 1
+        
+        return correct_predictions / total_predictions if total_predictions > 0 else 0
+    
+    def compute_centroid_accuracy(self, embeddings, labels):
+        """
+        Compute classification accuracy using hierarchy centroids.
+        
+        Each hierarchy class is represented by its centroid (mean embedding), and each
+        embedding is classified to the nearest centroid.
+        
+        Args:
+            embeddings: Array of embeddings [N, embed_dim]
+            labels: List of true labels
+            
+        Returns:
+            Accuracy score (float between 0 and 1)
+        """
+        # Create centroids for each hierarchy
+        unique_hierarchies = list(set(labels))
+        centroids = {}
+        
+        for hierarchy in unique_hierarchies:
+            hierarchy_indices = [i for i, label in enumerate(labels) if label == hierarchy]
+            hierarchy_embeddings = embeddings[hierarchy_indices]
+            centroids[hierarchy] = np.mean(hierarchy_embeddings, axis=0)
+        
+        # Classify each embedding to nearest centroid
+        correct_predictions = 0
+        total_predictions = len(labels)
+        
+        for i, embedding in enumerate(embeddings):
+            true_label = labels[i]
+            
+            # Find closest centroid
+            best_similarity = -1
+            predicted_label = None
+            
+            for hierarchy, centroid in centroids.items():
+                similarity = cosine_similarity([embedding], [centroid])[0][0]
+                if similarity > best_similarity:
+                    best_similarity = similarity
+                    predicted_label = hierarchy
+            
+            if predicted_label == true_label:
+                correct_predictions += 1
+        
+        return correct_predictions / total_predictions if total_predictions > 0 else 0
+    
+    def predict_hierarchy_from_embeddings(self, embeddings, labels):
+        """
+        Predict hierarchy from embeddings using centroid-based classification.
+        
+        Args:
+            embeddings: Array of embeddings [N, embed_dim]
+            labels: List of labels used to compute centroids
+            
+        Returns:
+            List of predicted hierarchy labels
+        """
+        # Create hierarchy centroids from training data
+        unique_hierarchies = list(set(labels))
+        centroids = {}
+        
+        for hierarchy in unique_hierarchies:
+            hierarchy_indices = [i for i, label in enumerate(labels) if label == hierarchy]
+            hierarchy_embeddings = embeddings[hierarchy_indices]
+            centroids[hierarchy] = np.mean(hierarchy_embeddings, axis=0)
+        
+        # Predict hierarchy for all embeddings
+        predictions = []
+        
+        for i, embedding in enumerate(embeddings):
+            # Find closest centroid
+            best_similarity = -1
+            predicted_hierarchy = None
+            
+            for hierarchy, centroid in centroids.items():
+                similarity = cosine_similarity([embedding], [centroid])[0][0]
+                if similarity > best_similarity:
+                    best_similarity = similarity
+                    predicted_hierarchy = hierarchy
+            
+            predictions.append(predicted_hierarchy)
+        
+        return predictions
+    
+    def create_confusion_matrix(self, true_labels, predicted_labels, title="Confusion Matrix"):
+        """
+        Create and plot a confusion matrix.
+        
+        Args:
+            true_labels: List of true labels
+            predicted_labels: List of predicted labels
+            title: Title for the confusion matrix plot
+            
+        Returns:
+            Tuple of (figure, accuracy, confusion_matrix)
+        """
+        # Get unique labels
+        unique_labels = sorted(list(set(true_labels + predicted_labels)))
+        
+        # Create confusion matrix
+        cm = confusion_matrix(true_labels, predicted_labels, labels=unique_labels)
+        
+        # Calculate accuracy
+        accuracy = accuracy_score(true_labels, predicted_labels)
+        
+        # Plot confusion matrix
+        plt.figure(figsize=(12, 10))
+        sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', 
+                   xticklabels=unique_labels, yticklabels=unique_labels)
+        plt.title(f'{title}\nAccuracy: {accuracy:.3f} ({accuracy*100:.1f}%)')
+        plt.ylabel('True Hierarchy')
+        plt.xlabel('Predicted Hierarchy')
+        plt.xticks(rotation=45)
+        plt.yticks(rotation=0)
+        plt.tight_layout()
+        
+        return plt.gcf(), accuracy, cm
+    
+    def evaluate_classification_performance(self, embeddings, labels, embedding_type="Embeddings"):
+        """
+        Evaluate classification performance and create confusion matrix.
+        
+        Args:
+            embeddings: Array of embeddings [N, embed_dim]
+            labels: List of true labels
+            embedding_type: Type of embeddings for display purposes
+            
+        Returns:
+            Dictionary containing accuracy, predictions, confusion matrix, and classification report
+        """
+        # Predict hierarchy
+        predictions = self.predict_hierarchy_from_embeddings(embeddings, labels)
+        
+        # Calculate accuracy
+        accuracy = accuracy_score(labels, predictions)
+        
+        # Create confusion matrix
+        fig, acc, cm = self.create_confusion_matrix(labels, predictions, 
+                                                   f"{embedding_type} - Hierarchy Classification")
+        
+        # Generate classification report
+        unique_labels = sorted(list(set(labels)))
+        report = classification_report(labels, predictions, labels=unique_labels, 
+                                     target_names=unique_labels, output_dict=True)
+        
+        return {
+            'accuracy': accuracy,
+            'predictions': predictions,
+            'confusion_matrix': cm,
+            'classification_report': report,
+            'figure': fig
+        }
+    
+    def evaluate_dataset(self, dataframe, dataset_name="Dataset"):
+        """
+        Evaluate embeddings on a given dataset.
+        
+        This method extracts embeddings for text and image, computes similarity metrics,
+        evaluates classification performance, and saves confusion matrices.
+        
+        Args:
+            dataframe: DataFrame containing the dataset
+            dataset_name: Name of the dataset for display purposes
+            
+        Returns:
+            Dictionary containing evaluation results for text and image embeddings
+        """
+        print(f"\n{'='*60}")
+        print(f"Evaluating {dataset_name}")
+        print(f"{'='*60}")
+        
+        # Create dataloader exactly as during training
+        dataloader = self.create_dataloader(dataframe, batch_size=16)
+        
+        results = {}
+        
+        # Evaluate text embeddings
+        text_embeddings, text_labels, texts = self.extract_embeddings(dataloader, 'text')
+        text_metrics = self.compute_similarity_metrics(text_embeddings, text_labels)
+        text_classification = self.evaluate_classification_performance(text_embeddings, text_labels, "Text Embeddings")
+        text_metrics.update(text_classification)
+        results['text'] = text_metrics
+        
+        # Evaluate image embeddings
+        image_embeddings, image_labels, _ = self.extract_embeddings(dataloader, 'image')
+        image_metrics = self.compute_similarity_metrics(image_embeddings, image_labels)
+        image_classification = self.evaluate_classification_performance(image_embeddings, image_labels, "Image Embeddings")
+        image_metrics.update(image_classification)
+        results['image'] = image_metrics
+        
+        # Evaluate hierarchy embeddings
+        hierarchy_embeddings, hierarchy_labels, _ = self.extract_embeddings(dataloader, 'category2')
+        hierarchy_metrics = self.compute_similarity_metrics(hierarchy_embeddings, hierarchy_labels)
+        hierarchy_classification = self.evaluate_classification_performance(hierarchy_embeddings, hierarchy_labels, "hierarchy Embeddings")
+        hierarchy_metrics.update(hierarchy_classification)
+        results['hierarchy'] = hierarchy_metrics
+        
+        # Print results
+        print(f"\n{dataset_name} Results:")
+        print("-" * 40)
+        for emb_type, metrics in results.items():
+            print(f"{emb_type.capitalize()} Embeddings:")
+            print(f"  Intra-class similarity (same hierarchy): {metrics['intra_class_mean']:.4f}")
+            print(f"  Inter-class similarity (diff hierarchy): {metrics['inter_class_mean']:.4f}")
+            print(f"  Separation score: {metrics['separation_score']:.4f}")
+            print(f"  Nearest Neighbor Accuracy: {metrics['accuracy']:.4f} ({metrics['accuracy']*100:.1f}%)")
+            print(f"  Centroid Accuracy: {metrics['centroid_accuracy']:.4f} ({metrics['centroid_accuracy']*100:.1f}%)")
+            
+            # Classification report summary
+            report = metrics['classification_report']
+            print(f"  📊 Classification Performance:")
+            print(f"    • Macro Avg F1-Score: {report['macro avg']['f1-score']:.4f}")
+            print(f"    • Weighted Avg F1-Score: {report['weighted avg']['f1-score']:.4f}")
+            print(f"    • Support: {report['macro avg']['support']:.0f} samples")
+            print()
+        
+        # Create visualizations
+        os.makedirs(f'{self.directory}', exist_ok=True)
+        
+        # Confusion matrices
+        results['text']['figure'].savefig(f'{self.directory}/{dataset_name.lower()}_text_confusion_matrix.png', dpi=300, bbox_inches='tight')
+        plt.close(results['text']['figure'])
+        
+        results['image']['figure'].savefig(f'{self.directory}/{dataset_name.lower()}_image_confusion_matrix.png', dpi=300, bbox_inches='tight')
+        plt.close(results['image']['figure'])
+        
+        results['hierarchy']['figure'].savefig(f'{self.directory}/{dataset_name.lower()}_hierarchy_confusion_matrix.png', dpi=300, bbox_inches='tight')
+        plt.close(results['hierarchy']['figure'])
+        
+        return results
+
+class KaglDataset(Dataset):
+    def __init__(self, dataframe):
+        self.dataframe = dataframe
+        # Use VALIDATION transforms (no augmentation)
+        self.transform = transforms.Compose([
+            transforms.Resize((224, 224)),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+        ])
+
+    def __len__(self):
+        return len(self.dataframe)
+
+    def __getitem__(self, idx):
+        row = self.dataframe.iloc[idx]
+        
+        # Handle image
+        image_data = row['image_url']
+        image = Image.open(BytesIO(image_data['bytes'])).convert("RGB")
+        image = self.transform(image)
+
+        # Get text and hierarchy
+        description = row['text']  
+        hierarchy = row['hierarchy']
+
+        return image, description, hierarchy
+
+def load_Kagl_marqo_dataset(evaluator):
+    """Load and prepare Kagl KAGL dataset"""
+    from datasets import load_dataset
+    print("Loading Kagl KAGL dataset...")
+
+    # Load the dataset
+    dataset = load_dataset("Marqo/KAGL")
+    df = dataset["data"].to_pandas()
+    print(f"✅ Dataset Kagl loaded")
+    print(f"📊 Before filtering: {len(df)} samples")
+    print(f"📋 Available columns: {list(df.columns)}")
+
+    # Check available categories and map them to our hierarchy
+    print(f"🎨 Available categories: {sorted(df['category2'].unique())}")
+    # Apply mapping
+    df['hierarchy'] = df['category2'].str.lower()
+    df['hierarchy'] = df['hierarchy'].replace('bags', 'bag').replace('topwear', 'top').replace('flip flops', 'shoes').replace('sandal', 'shoes')
+
+    # Filter to only include valid hierarchies that exist in our model
+    valid_hierarchies = df['hierarchy'].dropna().unique()
+    print(f"🎯 Valid hierarchies found: {sorted(valid_hierarchies)}")
+    print(f"🎯 Model hierarchies: {sorted(evaluator.hierarchy_classes)}")
+    
+    # Filter to only include hierarchies that exist in our model
+    df = df[df['hierarchy'].isin(evaluator.hierarchy_classes)]
+    print(f"📊 After filtering to model hierarchies: {len(df)} samples")
+    
+    if len(df) == 0:
+        print("❌ No samples left after hierarchy filtering.")
+        return pd.DataFrame()
+    
+    # Ensure we have text and image data
+    df = df.dropna(subset=['text', 'image'])
+    print(f"📊 After removing missing text/image: {len(df)} samples")
+    
+    # Show sample of text data to verify quality
+    print(f"📝 Sample texts:")
+    for i, (text, hierarchy) in enumerate(zip(df['text'].head(3), df['hierarchy'].head(3))):
+        print(f"  {i+1}. [{hierarchy}] {text[:100]}...")
+    
+    print(f"📊 After sampling: {len(df)} samples")
+    print(f"📊 Samples per hierarchy:")
+    for hierarchy in sorted(df['hierarchy'].unique()):
+        count = len(df[df['hierarchy'] == hierarchy])
+        print(f"  {hierarchy}: {count} samples")
+    
+    # Create formatted dataset with proper column names
+    Kagl_formatted = pd.DataFrame({
+        'image_url': df['image'],
+        'text': df['text'],
+        'hierarchy': df['hierarchy']
+    })
+    
+    print(f"📊 Final dataset size: {len(Kagl_formatted)} samples")
+    return Kagl_formatted
+
+if __name__ == "__main__":
+    device = config.device
+    model_path = config.hierarchy_model_path
+    directory = config.evaluation_directory
+
+    print(f"🚀 Starting evaluation with {model_path}")
+    
+    evaluator = EmbeddingEvaluator(model_path, directory)
+    
+    print(f"📊 Final hierarchy classes after initialization: {len(evaluator.vocab.hierarchy_classes)} classes")
+    
+    # Evaluate on validation dataset (same subset as during training)
+    print("\n" + "="*60)
+    print("EVALUATING VALIDATION DATASET")
+    print("="*60)
+    val_results = evaluator.evaluate_dataset(evaluator.val_df, "Validation Dataset")
+    
+    print("\n" + "="*60)
+    print("EVALUATING Kagl MARQO DATASET") 
+    print("="*60)
+    df_Kagl_marqo = load_Kagl_marqo_dataset(evaluator)
+    Kagl_results = evaluator.evaluate_dataset(df_Kagl_marqo, "Kagl Marqo Dataset")
+    
+    # Compare results
+    print(f"\n{'='*60}")
+    print("FINAL EVALUATION SUMMARY")
+    print(f"{'='*60}")
+    
+    print("\n🔍 VALIDATION DATASET RESULTS:")
+    print(f"Text      - Separation: {val_results['text']['separation_score']:.4f} | NN Acc: {val_results['text']['accuracy']*100:.1f}% | Centroid Acc: {val_results['text']['centroid_accuracy']*100:.1f}%")
+    print(f"Image     - Separation: {val_results['image']['separation_score']:.4f} | NN Acc: {val_results['image']['accuracy']*100:.1f}% | Centroid Acc: {val_results['image']['centroid_accuracy']*100:.1f}%")
+    print(f"hierarchy     - Separation: {val_results['hierarchy']['separation_score']:.4f} | NN Acc: {val_results['hierarchy']['accuracy']*100:.1f}% | Centroid Acc: {val_results['hierarchy']['centroid_accuracy']*100:.1f}%")
+    
+    print("\n🌐 Kagl MARQO DATASET RESULTS:")
+    print(f"Text      - Separation: {Kagl_results['text']['separation_score']:.4f} | NN Acc: {Kagl_results['text']['accuracy']*100:.1f}% | Centroid Acc: {Kagl_results['text']['centroid_accuracy']*100:.1f}%")
+    print(f"Image     - Separation: {Kagl_results['image']['separation_score']:.4f} | NN Acc: {Kagl_results['image']['accuracy']*100:.1f}% | Centroid Acc: {Kagl_results['image']['centroid_accuracy']*100:.1f}%")
+    print(f"Hierarchy     - Separation: {Kagl_results['hierarchy']['separation_score']:.4f} | NN Acc: {Kagl_results['hierarchy']['accuracy']*100:.1f}% | Centroid Acc: {Kagl_results['hierarchy']['centroid_accuracy']*100:.1f}%")
+    
+    
+    print(f"\n✅ Evaluation completed! Check 'improved_model_evaluation/' for visualization files.")
+    print(f"📊 Final hierarchy classes used: {len(evaluator.vocab.hierarchy_classes)} classes")