First commit

README copy.md

@@ -0,0 +1,13 @@
+---
+title: Prompt Pilot
+emoji: 📊
+colorFrom: gray
+colorTo: gray
+sdk: gradio
+sdk_version: 5.34.2
+app_file: app.py
+pinned: false
+short_description: test
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+"""
+InternVL3 Prompt Engineering Application
+Entry point for the modular InternVL3 image analysis application.
+"""
+
+from frontend.gradio_app import GradioApp
+
+if __name__ == "__main__":
+    # Create and launch the application
+    app = GradioApp()
+    app.launch()

backend/__init__.py

@@ -0,0 +1,27 @@
+from .config import ConfigManager
+from .models import ModelManager, InternVLModel, BaseModel
+from .inference import InferenceEngine
+from .utils import (
+    build_transform, 
+    load_image, 
+    extract_file_dict, 
+    validate_data, 
+    extract_binary_output,
+    create_accuracy_table,
+    save_dataframe_to_csv
+)
+
+__all__ = [
+    'ConfigManager',
+    'ModelManager', 
+    'InternVLModel', 
+    'BaseModel',
+    'InferenceEngine',
+    'build_transform',
+    'load_image',
+    'extract_file_dict',
+    'validate_data',
+    'extract_binary_output',
+    'create_accuracy_table',
+    'save_dataframe_to_csv'
+] 

backend/config/__init__.py

@@ -0,0 +1,3 @@
+from .config_manager import ConfigManager
+
+__all__ = ['ConfigManager'] 

backend/config/config_manager.py

@@ -0,0 +1,120 @@
+import yaml
+import os
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+
+class ConfigManager:
+    """Manages configuration loading and access for the application."""
+    
+    def __init__(self, config_path: Optional[str] = None):
+        """
+        Initialize the configuration manager.
+        
+        Args:
+            config_path: Path to the configuration file. If None, uses default path.
+        """
+        if config_path is None:
+            # Default to config/models.yaml relative to project root
+            project_root = Path(__file__).parent.parent.parent
+            config_path = project_root / "config" / "models.yaml"
+        
+        self.config_path = Path(config_path)
+        self._config = None
+        self.load_config()
+    
+    def load_config(self) -> None:
+        """Load configuration from YAML file."""
+        try:
+            with open(self.config_path, 'r', encoding='utf-8') as file:
+                self._config = yaml.safe_load(file)
+            print(f"✅ Configuration loaded from {self.config_path}")
+        except FileNotFoundError:
+            raise FileNotFoundError(f"Configuration file not found: {self.config_path}")
+        except yaml.YAMLError as e:
+            raise ValueError(f"Invalid YAML in configuration file: {e}")
+    
+    def reload_config(self) -> None:
+        """Reload configuration from file."""
+        self.load_config()
+    
+    @property
+    def config(self) -> Dict[str, Any]:
+        """Get the full configuration dictionary."""
+        if self._config is None:
+            self.load_config()
+        return self._config
+    
+    def get_available_models(self) -> Dict[str, str]:
+        """Get a dictionary of available model names and their IDs."""
+        models = self.config.get('models', {})
+        return {name: model_config['model_id'] for name, model_config in models.items()}
+    
+    def get_model_config(self, model_name: str) -> Dict[str, Any]:
+        """
+        Get configuration for a specific model.
+        
+        Args:
+            model_name: Name of the model (e.g., 'InternVL3-8B')
+            
+        Returns:
+            Model configuration dictionary
+            
+        Raises:
+            KeyError: If model name is not found
+        """
+        models = self.config.get('models', {})
+        if model_name not in models:
+            available = list(models.keys())
+            raise KeyError(f"Model '{model_name}' not found. Available models: {available}")
+        
+        return models[model_name]
+    
+    def get_supported_quantizations(self, model_name: str) -> List[str]:
+        """Get supported quantization methods for a model."""
+        model_config = self.get_model_config(model_name)
+        return model_config.get('supported_quantizations', [])
+    
+    def get_default_quantization(self, model_name: str) -> str:
+        """Get the default quantization method for a model."""
+        model_config = self.get_model_config(model_name)
+        return model_config.get('default_quantization', 'non-quantized(fp16)')
+    
+    def get_default_model(self) -> str:
+        """Get the default model name."""
+        return self.config.get('default_model', 'InternVL3-8B')
+    
+    def validate_model_and_quantization(self, model_name: str, quantization: str) -> bool:
+        """
+        Validate if a quantization method is supported for a model.
+        
+        Args:
+            model_name: Name of the model
+            quantization: Quantization method
+            
+        Returns:
+            True if valid, False otherwise
+        """
+        try:
+            supported = self.get_supported_quantizations(model_name)
+            return quantization in supported
+        except KeyError:
+            return False
+    
+    def apply_environment_settings(self) -> None:
+        """Apply environment settings to the current process."""
+        # Set CUDA memory allocation
+        os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'expandable_segments:True'
+    
+    def get_model_description(self, model_name: str) -> str:
+        """Get description for a model."""
+        model_config = self.get_model_config(model_name)
+        return model_config.get('description', 'No description available')
+    
+    def __str__(self) -> str:
+        """String representation of the configuration manager."""
+        return f"ConfigManager(config_path={self.config_path})"
+    
+    def __repr__(self) -> str:
+        """Detailed string representation."""
+        models = list(self.get_available_models().keys())
+        return f"ConfigManager(config_path={self.config_path}, models={models})" 

backend/inference/__init__.py

@@ -0,0 +1,3 @@
+from .inference_engine import InferenceEngine
+
+__all__ = ['InferenceEngine'] 

backend/inference/inference_engine.py

@@ -0,0 +1,543 @@
+import pandas as pd
+import threading
+import time
+import os
+from pathlib import Path
+from typing import Dict, List, Tuple, Union, Any, Optional, Callable
+import gradio as gr
+from ..models.model_manager import ModelManager
+from ..utils.data_processing import extract_file_dict, validate_data, extract_binary_output
+from ..config.config_manager import ConfigManager
+from ..utils.metrics import create_accuracy_table
+from datetime import datetime
+import boto3
+
+
+class InferenceEngine:
+    """Engine for handling batch inference and processing control."""
+    
+    def __init__(self, model_manager: ModelManager, config_manager: ConfigManager):
+        """
+        Initialize the inference engine.
+        
+        Args:
+            model_manager: Model manager instance
+            config_manager: Configuration manager instance
+        """
+        self.model_manager = model_manager
+        self.config_manager = config_manager
+        self.processing_lock = threading.Lock()
+        self.stop_processing = False
+        self.full_df = None  # Store full dataframe with image paths
+    
+    def set_stop_flag(self) -> str:
+        """Set the global stop flag to interrupt processing."""
+        with self.processing_lock:
+            self.stop_processing = True
+        print("🛑 Stop signal received. Processing will halt after current image...")
+        return "🛑 Stopping process... Please wait for current image to complete."
+    
+    def reset_stop_flag(self) -> None:
+        """Reset the global stop flag before starting new processing."""
+        with self.processing_lock:
+            self.stop_processing = False
+    
+    def check_stop_flag(self) -> bool:
+        """Check if processing should be stopped."""
+        with self.processing_lock:
+            return self.stop_processing
+    
+    def _should_load_model(self, model_selection: str, quantization_type: str) -> bool:
+        """
+        Check if we need to load the model.
+        
+        Args:
+            model_selection: Selected model name
+            quantization_type: Selected quantization type
+            
+        Returns:
+            True if model needs to be loaded, False otherwise
+        """
+        # If no model is loaded, we need to load
+        if not self.model_manager.current_model or not self.model_manager.current_model.is_model_loaded():
+            return True
+        
+        # If different model is selected, we need to load
+        if self.model_manager.current_model_name != model_selection:
+            return True
+            
+        # If same model but different quantization, we need to reload
+        if self.model_manager.current_model.current_quantization != quantization_type:
+            return True
+            
+        return False
+    
+    def _ensure_correct_model_loaded(self, model_selection: str, quantization_type: str, progress: gr.Progress()) -> None:
+        """
+        Ensure the correct model with correct quantization is loaded.
+        
+        Args:
+            model_selection: Selected model name
+            quantization_type: Selected quantization type
+            progress: Gradio progress object
+        """
+        if self._should_load_model(model_selection, quantization_type):
+            progress(0, desc=f"🚀 Loading {model_selection} ({quantization_type})...")
+            print(f"🚀 Loading {model_selection} with {quantization_type}...")
+            success = self.model_manager.load_model(model_selection, quantization_type)
+            if not success:
+                raise Exception(f"Failed to load model {model_selection} with {quantization_type}")
+        else:
+            print(f"✅ Correct model already loaded: {model_selection} with {quantization_type}")
+    
+    def process_folder_input(
+        self, 
+        folder_path: List[Path], 
+        prompt: str, 
+        quantization_type: str, 
+        model_selection: str,
+        progress: gr.Progress()
+    ) -> Tuple[Any, ...]:
+        """
+        Process input folder with images and optional CSV.
+        
+        Args:
+            folder_path: List of Path objects from Gradio
+            prompt: Text prompt for inference
+            quantization_type: Model quantization type
+            model_selection: Selected model name
+            progress: Gradio progress object
+            
+        Returns:
+            Tuple of UI update states and results
+        """
+        # Reset stop flag at the beginning of processing
+        self.reset_stop_flag()
+        
+        # Extract file dictionary
+        file_dict = extract_file_dict(folder_path)
+        
+        # Print all file names for debug
+        for fname in file_dict:
+            print(fname)
+        
+        validation_result, message = validate_data(file_dict)
+        
+        # Handle different validation results
+        if validation_result == False:
+            return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), message, gr.update(visible=False), ""
+        elif validation_result in ["no_csv", "multiple_csv"]:
+            return self._process_without_csv(file_dict, prompt, quantization_type, model_selection, progress)
+        else:
+            return self._process_with_csv(file_dict, prompt, quantization_type, model_selection, progress)
+    
+    def _process_without_csv(
+        self, 
+        file_dict: Dict[str, Path], 
+        prompt: str, 
+        quantization_type: str, 
+        model_selection: str, 
+        progress: gr.Progress()
+    ) -> Tuple[Any, ...]:
+        """Process images without CSV file."""
+        image_exts = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff']
+        image_file_dict = {fname: file_dict[fname] for fname in file_dict 
+                          if any(fname.lower().endswith(ext) for ext in image_exts)}
+        
+        filtered_rows = []
+        total_images = len(image_file_dict)
+        
+        if total_images == 0:
+            return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), "No image files found.", gr.update(visible=False), ""
+        
+        # Ensure correct model is loaded
+        self._ensure_correct_model_loaded(model_selection, quantization_type, progress)
+        
+        # Initialize progress
+        progress(0, desc=f"🚀 Starting to process {total_images} images...")
+        print(f"Starting to process {total_images} images with {model_selection}...")
+        
+        for idx, (img_name, img_path) in enumerate(image_file_dict.items()):
+            # Check stop flag before processing each image
+            if self.check_stop_flag():
+                print(f"🛑 Processing stopped by user at image {idx + 1}/{total_images}")
+                # Add remaining images as "Not processed" entries
+                for remaining_idx, (remaining_name, remaining_path) in enumerate(list(image_file_dict.items())[idx:]):
+                    filtered_rows.append({
+                        'S.No': idx + remaining_idx + 1,
+                        'Image Name': remaining_name,
+                        'Ground Truth': '',
+                        'Binary Output': 'Not processed (stopped)',
+                        'Model Output': 'Processing stopped by user',
+                        'Image Path': str(remaining_path)
+                    })
+                
+                display_df = pd.DataFrame(filtered_rows)[['S.No', 'Image Name', 'Ground Truth', 'Binary Output', 'Model Output']]
+                self.full_df = pd.DataFrame(filtered_rows)
+                final_message = f"🛑 Processing stopped by user. Completed {idx}/{total_images} images."
+                print(final_message)
+                return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), display_df, gr.update(visible=False), final_message
+            
+            try:
+                # Update progress with current image info
+                current_progress = idx / total_images
+                progress_msg = f"🔄 Processing image {idx + 1}/{total_images}: {img_name[:30]}..." if len(img_name) > 30 else f"🔄 Processing image {idx + 1}/{total_images}: {img_name}"
+                progress(current_progress, desc=progress_msg)
+                print(progress_msg)
+                
+                # Use model inference
+                model_output = self.model_manager.inference(str(img_path), prompt) if prompt else "No prompt provided"
+                
+                # Extract binary output (no ground truth available for file-based processing)
+                binary_output = extract_binary_output(model_output, "", [])
+                
+                filtered_rows.append({
+                    'S.No': idx + 1,
+                    'Image Name': img_name,
+                    'Ground Truth': '',  # Empty for manual input
+                    'Binary Output': binary_output,
+                    'Model Output': model_output,
+                    'Image Path': str(img_path)
+                })
+                
+                # Update progress after successful processing
+                current_progress = (idx + 1) / total_images
+                progress_msg = f"✅ Completed {idx + 1}/{total_images} images"
+                progress(current_progress, desc=progress_msg)
+                print(f"Successfully processed image {idx + 1} of {total_images}")
+                
+            except Exception as e:
+                print(f"Error processing image {idx + 1} of {total_images}: {str(e)}")
+                filtered_rows.append({
+                    'S.No': idx + 1,
+                    'Image Name': img_name,
+                    'Ground Truth': '',
+                    'Binary Output': 'Enter the output manually',  # Default for errors
+                    'Model Output': f"Error: {str(e)}",
+                    'Image Path': str(img_path)
+                })
+                
+                # Update progress even for errors
+                current_progress = (idx + 1) / total_images
+                progress_msg = f"⚠️ Processed {idx + 1}/{total_images} images (with errors)"
+                progress(current_progress, desc=progress_msg)
+        
+        # Check if processing was completed or stopped
+        if self.check_stop_flag():
+            final_message = f"🛑 Processing stopped by user. Completed {len(filtered_rows)}/{total_images} images."
+        else:
+            final_message = f"🎉 Successfully completed processing all {total_images} images!"
+        
+        display_df = pd.DataFrame(filtered_rows)[['S.No', 'Image Name', 'Ground Truth', 'Binary Output', 'Model Output']]
+        # Save the full dataframe (with Image Path) for preview
+        self.full_df = pd.DataFrame(filtered_rows)
+        self.save_results_to_s3(display_df)
+
+        print(final_message)
+        
+        # Make the table editable for ground truth input
+        return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), display_df, gr.update(visible=False), final_message
+    
+    def _process_with_csv(
+        self, 
+        file_dict: Dict[str, Path], 
+        prompt: str, 
+        quantization_type: str, 
+        model_selection: str, 
+        progress: gr.Progress()
+    ) -> Tuple[Any, ...]:
+        """Process images with CSV file."""
+        csv_files = [fname for fname in file_dict if fname.lower().endswith('.csv')]
+        csv_file = file_dict[csv_files[0]]
+        df = pd.read_csv(csv_file)
+        
+        # Collect all ground truth values for unique keyword extraction
+        all_ground_truths = [str(row['Ground Truth']) for idx, row in df.iterrows() 
+                            if pd.notna(row['Ground Truth']) and str(row['Ground Truth']).strip()]
+        
+        # Find image files
+        image_exts = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff']
+        image_file_dict = {fname: file_dict[fname] for fname in file_dict 
+                          if any(fname.lower().endswith(ext) for ext in image_exts)}
+        
+        # Only keep rows where image file exists
+        filtered_rows = []
+        matching_images = [row for idx, row in df.iterrows() if row['Image Name'] in image_file_dict]
+        total_images = len(matching_images)
+        
+        if total_images == 0:
+            return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), "No matching images found for entries in CSV.", gr.update(visible=False), ""
+        
+        # Ensure correct model is loaded
+        self._ensure_correct_model_loaded(model_selection, quantization_type, progress)
+        
+        # Initialize progress
+        progress(0, desc=f"🚀 Starting to process {total_images} images...")
+        print(f"Starting to process {total_images} images with {model_selection}...")
+        processed_count = 0
+        
+        for idx, row in df.iterrows():
+            img_name = row['Image Name']
+            if img_name in image_file_dict:
+                # Check stop flag before processing each image
+                if self.check_stop_flag():
+                    print(f"🛑 Processing stopped by user at image {processed_count + 1}/{total_images}")
+                    # Add remaining unprocessed images
+                    for remaining_idx, remaining_row in df.iloc[idx:].iterrows():
+                        if remaining_row['Image Name'] in image_file_dict:
+                            filtered_rows.append({
+                                'S.No': len(filtered_rows) + 1,
+                                'Image Name': remaining_row['Image Name'],
+                                'Ground Truth': remaining_row['Ground Truth'],
+                                'Binary Output': 'Not processed (stopped)',
+                                'Model Output': 'Processing stopped by user',
+                                'Image Path': str(image_file_dict[remaining_row['Image Name']])
+                            })
+                    
+                    display_df = pd.DataFrame(filtered_rows)[['S.No', 'Image Name', 'Ground Truth', 'Binary Output', 'Model Output']]
+                    self.full_df = pd.DataFrame(filtered_rows)
+                    final_message = f"🛑 Processing stopped by user. Completed {processed_count}/{total_images} images."
+                    print(final_message)
+                    return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), display_df, gr.update(visible=False), final_message
+                
+                try:
+                    processed_count += 1
+                    # Update progress with current image info
+                    current_progress = (processed_count - 1) / total_images
+                    progress_msg = f"🔄 Processing image {processed_count}/{total_images}: {img_name[:30]}..." if len(img_name) > 30 else f"🔄 Processing image {processed_count}/{total_images}: {img_name}"
+                    progress(current_progress, desc=progress_msg)
+                    print(progress_msg)
+                    
+                    # Use model inference
+                    model_output = self.model_manager.inference(str(image_file_dict[img_name]), prompt)
+                    
+                    # Extract binary output using ground truth and all ground truths for keyword extraction
+                    ground_truth = str(row['Ground Truth']) if pd.notna(row['Ground Truth']) else ""
+                    binary_output = extract_binary_output(model_output, ground_truth, all_ground_truths)
+                    
+                    filtered_rows.append({
+                        'S.No': len(filtered_rows) + 1,
+                        'Image Name': img_name,
+                        'Ground Truth': row['Ground Truth'],
+                        'Binary Output': binary_output,
+                        'Model Output': model_output,
+                        'Image Path': str(image_file_dict[img_name])
+                    })
+                    
+                    # Update progress after successful processing
+                    current_progress = processed_count / total_images
+                    progress_msg = f"✅ Completed {processed_count}/{total_images} images"
+                    progress(current_progress, desc=progress_msg)
+                    print(f"Successfully processed image {processed_count} of {total_images}")
+                    
+                except Exception as e:
+                    print(f"Error processing image {processed_count} of {total_images}: {str(e)}")
+                    filtered_rows.append({
+                        'S.No': len(filtered_rows) + 1,
+                        'Image Name': img_name,
+                        'Ground Truth': row['Ground Truth'],
+                        'Binary Output': 'Enter the output manually',  # Default for errors
+                        'Model Output': f"Error: {str(e)}",
+                        'Image Path': str(image_file_dict[img_name])
+                    })
+                    
+                    # Update progress even for errors
+                    current_progress = processed_count / total_images
+                    progress_msg = f"⚠️ Processed {processed_count}/{total_images} images (with errors)"
+                    progress(current_progress, desc=progress_msg)
+        
+        # Check if processing was completed or stopped
+        if self.check_stop_flag():
+            final_message = f"🛑 Processing stopped by user. Completed {len([r for r in filtered_rows if 'stopped' not in r['Model Output']])}/{total_images} images."
+        else:
+            final_message = f"🎉 Successfully completed processing all {total_images} images!"
+        
+        display_df = pd.DataFrame(filtered_rows)[['S.No', 'Image Name', 'Ground Truth', 'Binary Output', 'Model Output']]
+        # Save the full dataframe (with Image Path) for preview
+        self.full_df = pd.DataFrame(filtered_rows)
+
+        self.save_results_to_s3(display_df)
+
+        print(final_message)
+        
+        return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), display_df, gr.update(visible=False), final_message
+    
+    def rerun_with_new_prompt(
+        self, 
+        df: pd.DataFrame, 
+        new_prompt: str, 
+        quantization_type: str, 
+        model_selection: str, 
+        progress: gr.Progress()
+    ) -> Tuple[Any, ...]:
+        """Rerun processing with new prompt and clear accuracy data."""
+        if df is None or not new_prompt.strip():
+            return df, None, None, None, gr.update(visible=False), gr.update(visible=False), "⚠️ Please provide a valid prompt"
+        
+        # Reset stop flag at the beginning of reprocessing
+        self.reset_stop_flag()
+        
+        updated_df = df.copy()
+        total_images = len(updated_df)
+        
+        # Collect all ground truth values for unique keyword extraction
+        all_ground_truths = [str(row['Ground Truth']) for idx, row in updated_df.iterrows() 
+                            if pd.notna(row['Ground Truth']) and str(row['Ground Truth']).strip()]
+        
+        # Get the full dataframe with image paths
+        if self.full_df is None:
+            return df, None, None, None, gr.update(visible=False), gr.update(visible=False), "⚠️ No image data available"
+        
+        # Create a copy of the full dataframe to update
+        updated_full_df = self.full_df.copy()
+        
+        # Ensure correct model is loaded
+        self._ensure_correct_model_loaded(model_selection, quantization_type, progress)
+        
+        # Initialize progress
+        progress(0, desc=f"🚀 Starting to reprocess {total_images} images with new prompt...")
+        print(f"🚀 Starting to reprocess {total_images} images with new prompt...")
+        
+        for i in range(len(updated_df)):
+            # Check stop flag before processing each image
+            if self.check_stop_flag():
+                print(f"🛑 Reprocessing stopped by user at image {i + 1}/{total_images}")
+                # Mark remaining images as not reprocessed in both dataframes
+                for j in range(i, len(updated_df)):
+                    updated_df.iloc[j, updated_df.columns.get_loc("Model Output")] = "Reprocessing stopped by user"
+                    updated_df.iloc[j, updated_df.columns.get_loc("Binary Output")] = "Not reprocessed (stopped)"
+                    # Also update the full dataframe
+                    if j < len(updated_full_df):
+                        updated_full_df.iloc[j, updated_full_df.columns.get_loc("Model Output")] = "Reprocessing stopped by user"
+                        updated_full_df.iloc[j, updated_full_df.columns.get_loc("Binary Output")] = "Not reprocessed (stopped)"
+                
+                # Update the full_df reference
+                self.full_df = updated_full_df
+                
+                final_message = f"🛑 Reprocessing stopped by user. Completed {i}/{total_images} images."
+                print(final_message)
+                return updated_df, None, None, None, gr.update(visible=False), gr.update(visible=False), final_message
+            
+            try:
+                # Get image path from full_df
+                image_path = self.full_df.iloc[i]['Image Path']
+                image_name = updated_df.iloc[i]['Image Name']
+                ground_truth = str(updated_df.iloc[i]['Ground Truth']) if pd.notna(updated_df.iloc[i]['Ground Truth']) else ""
+                
+                # Update progress with current image info
+                current_progress = i / total_images
+                progress_msg = f"🔄 Reprocessing image {i + 1}/{total_images}: {image_name[:30]}..." if len(image_name) > 30 else f"🔄 Reprocessing image {i + 1}/{total_images}: {image_name}"
+                progress(current_progress, desc=progress_msg)
+                print(progress_msg)
+                
+                # Use model inference with new prompt
+                model_output = self.model_manager.inference(image_path, new_prompt)
+                
+                # Update both the display dataframe and the full dataframe
+                updated_df.iloc[i, updated_df.columns.get_loc("Model Output")] = model_output
+                updated_full_df.iloc[i, updated_full_df.columns.get_loc("Model Output")] = model_output
+                
+                # Extract binary output using ground truth and all ground truths for keyword extraction
+                binary_output = extract_binary_output(model_output, ground_truth, all_ground_truths)
+                updated_df.iloc[i, updated_df.columns.get_loc("Binary Output")] = binary_output
+                updated_full_df.iloc[i, updated_full_df.columns.get_loc("Binary Output")] = binary_output
+                
+                # Update progress after successful processing
+                current_progress = (i + 1) / total_images
+                progress_msg = f"✅ Completed {i + 1}/{total_images} images"
+                progress(current_progress, desc=progress_msg)
+                print(f"✅ Successfully reprocessed image {i + 1}/{total_images}")
+                
+            except Exception as e:
+                print(f"❌ Error reprocessing image {i + 1}/{total_images}: {str(e)}")
+                error_message = f"Error: {str(e)}"
+                
+                # Update both dataframes with error information
+                updated_df.iloc[i, updated_df.columns.get_loc("Model Output")] = error_message
+                updated_df.iloc[i, updated_df.columns.get_loc("Binary Output")] = "Enter the output manually"
+                updated_full_df.iloc[i, updated_full_df.columns.get_loc("Model Output")] = error_message
+                updated_full_df.iloc[i, updated_full_df.columns.get_loc("Binary Output")] = "Enter the output manually"
+                
+                # Update progress even for errors
+                current_progress = (i + 1) / total_images
+                progress_msg = f"⚠️ Processed {i + 1}/{total_images} images (with errors)"
+                progress(current_progress, desc=progress_msg)
+        
+        # Update the full_df reference with the updated data
+        self.full_df = updated_full_df
+        
+        # Check if reprocessing was completed or stopped
+        if self.check_stop_flag():
+            final_message = f"🛑 Reprocessing stopped by user. Completed reprocessing for some images."
+        else:
+            final_message = f"🎉 Successfully completed reprocessing all {total_images} images with new prompt! Click 'Generate Metrics' to see accuracy data."
+            self.save_results_to_s3(updated_full_df)
+
+        print(final_message)
+        
+        # Return updated dataframe and clear accuracy data (hide section 3)
+        return updated_df, None, None, None, gr.update(visible=False), gr.update(visible=False), final_message 
+    
+    def save_results_to_s3(self, df):
+        """Save results to S3 bucket."""
+        try:
+            s3_bucket = os.getenv('AWS_BUCKET')
+            prefix = os.getenv('AWS_PREFIX')
+            s3_path = f"{prefix}/{datetime.now().date()}"
+            date_time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+            csv_file_name = f'{date_time}_model_output.csv'
+            
+            # create accuracy table
+            metrics_df, _, cm_values = create_accuracy_table(df)
+            # save metrics_df to text file
+            
+            text_file_name = f'{date_time}_evaluation_metrics.txt'
+            # save metrics_df to text file
+            with open(text_file_name, 'w') as f:
+                f.write(metrics_df.to_string() + '\n\n')
+                f.write(cm_values.to_string())
+
+            # save df to csv
+            df.to_csv(csv_file_name, index=False)
+
+            # upload files to s3
+            status = self.upload_file(text_file_name, s3_bucket, f"{s3_path}/{text_file_name}")
+            print(f"Status of uploading {text_file_name} to {s3_bucket}/{s3_path}/{text_file_name}: {status}")
+            status = self.upload_file(csv_file_name, s3_bucket, f"{s3_path}/{csv_file_name}")
+            print(f"Status of uploading {csv_file_name} to {s3_bucket}/{s3_path}/{csv_file_name}: {status}")
+
+            # delete files from local
+            os.remove(text_file_name)
+            os.remove(csv_file_name)
+            print(f"Deleted {text_file_name} and {csv_file_name}")
+        except Exception as e:
+            print(f"Error saving results to s3: {e}")
+            if "No valid data" in str(e) or "Need at least 2 different" in str(e):
+                df.to_csv(csv_file_name, index=False)
+                status = self.upload_file(csv_file_name, s3_bucket, f"{s3_path}/{csv_file_name}")
+                print(f"Status of uploading only csv file to {s3_bucket}/{s3_path}/{csv_file_name}: {status}")
+                os.remove(csv_file_name)
+                print(f"Deleted {csv_file_name}")
+
+    def upload_file(self,file_name, bucket, object_name=None):
+        """Upload a file to an S3 bucket
+
+        :param file_name: File to upload
+        :param bucket: Bucket to upload to
+        :param object_name: S3 object name. If not specified then file_name is used
+        :return: True if file was uploaded, else False
+        """
+        access_key = os.getenv('AWS_ACCESS_KEY_ID')
+        secret_key = os.getenv('AWS_SECRET_ACCESS_KEY')
+        # If S3 object_name was not specified, use file_name
+        if object_name is None:
+            object_name = os.path.basename(file_name)
+
+        # Upload the file
+        s3_client = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)
+        try:
+            response = s3_client.upload_file(file_name, bucket, object_name)
+        except Exception as e:
+            print(f"Error uploading {file_name} to s3: {e}")
+            return False
+        return True

backend/models/__init__.py

@@ -0,0 +1,6 @@
+from .base_model import BaseModel
+from .model_manager import ModelManager
+from .internvl import InternVLModel
+from .qwen import QwenModel
+
+__all__ = ['BaseModel', 'ModelManager', 'InternVLModel', 'QwenModel'] 

backend/models/base_model.py

@@ -0,0 +1,109 @@
+from abc import ABC, abstractmethod
+from typing import Dict, Any, Optional, List
+import torch
+from transformers import AutoModel, AutoTokenizer
+
+class BaseModel(ABC):
+    """Abstract base class for all vision-language models."""
+    
+    def __init__(self, model_name: str, model_config: Dict[str, Any]):
+        """
+        Initialize the base model.
+        
+        Args:
+            model_name: Name of the model
+            model_config: Configuration dictionary for the model
+        """
+        self.model_name = model_name
+        self.model_config = model_config
+        self.model_id = model_config['model_id']
+        self.model = None
+        self.tokenizer = None
+        self.current_quantization = None
+        self.is_loaded = False
+    
+    @abstractmethod
+    def load_model(self, quantization_type: str, **kwargs) -> bool:
+        """
+        Load the model with specified quantization.
+        
+        Args:
+            quantization_type: Type of quantization to use
+            **kwargs: Additional arguments for model loading
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        pass
+    
+    @abstractmethod
+    def unload_model(self) -> None:
+        """Unload the model from memory."""
+        pass
+    
+    @abstractmethod
+    def inference(self, image_path: str, prompt: str, **kwargs) -> str:
+        """
+        Perform inference on an image with a text prompt.
+        
+        Args:
+            image_path: Path to the image file
+            prompt: Text prompt for the model
+            **kwargs: Additional inference parameters
+            
+        Returns:
+            Model's text response
+        """
+        pass
+    
+    def is_model_loaded(self) -> bool:
+        """Check if model is currently loaded."""
+        return self.is_loaded
+    
+    def get_model_info(self) -> Dict[str, Any]:
+        """Get information about the model."""
+        return {
+            'name': self.model_name,
+            'model_id': self.model_id,
+            'description': self.model_config.get('description', ''),
+            'min_gpu_memory_gb': self.model_config.get('min_gpu_memory_gb', 0),
+            'recommended_gpu_memory_gb': self.model_config.get('recommended_gpu_memory_gb', 0),
+            'supported_quantizations': self.model_config.get('supported_quantizations', []),
+            'default_quantization': self.model_config.get('default_quantization', ''),
+            'is_loaded': self.is_loaded,
+            'current_quantization': self.current_quantization
+        }
+    
+    def get_supported_quantizations(self) -> List[str]:
+        """Get list of supported quantization methods."""
+        return self.model_config.get('supported_quantizations', [])
+    
+    def get_memory_requirements(self) -> Dict[str, int]:
+        """Get memory requirements for the model."""
+        return {
+            'min_gpu_memory_gb': self.model_config.get('min_gpu_memory_gb', 0),
+            'recommended_gpu_memory_gb': self.model_config.get('recommended_gpu_memory_gb', 0)
+        }
+    
+    def validate_quantization(self, quantization_type: str) -> bool:
+        """
+        Validate if the quantization type is supported.
+        
+        Args:
+            quantization_type: Quantization type to validate
+            
+        Returns:
+            True if supported, False otherwise
+        """
+        supported = self.get_supported_quantizations()
+        return quantization_type in supported
+    
+    def __str__(self) -> str:
+        """String representation of the model."""
+        status = "loaded" if self.is_loaded else "not loaded"
+        quant = f" ({self.current_quantization})" if self.current_quantization else ""
+        return f"{self.model_name}{quant} - {status}"
+    
+    def __repr__(self) -> str:
+        """Detailed string representation."""
+        return f"BaseModel(name={self.model_name}, loaded={self.is_loaded}, quantization={self.current_quantization})" 

backend/models/internvl/__init__.py

@@ -0,0 +1,3 @@
+from .internvl_model import InternVLModel
+
+__all__ = ['InternVLModel'] 

backend/models/internvl/internvl_model.py

@@ -0,0 +1,363 @@
+import torch
+import gc
+import os
+from typing import Dict, Any, Optional, Callable
+from transformers import AutoModel, AutoTokenizer, AutoConfig
+from ..base_model import BaseModel
+from ...utils.image_processing import load_image
+from ...config.config_manager import ConfigManager
+
+
+class InternVLModel(BaseModel):
+    """InternVL3 model implementation."""
+    
+    def __init__(self, model_name: str, model_config: Dict[str, Any], config_manager: ConfigManager):
+        """
+        Initialize the InternVL model.
+        
+        Args:
+            model_name: Name of the model
+            model_config: Configuration dictionary for the model
+            config_manager: Configuration manager instance
+        """
+        super().__init__(model_name, model_config)
+        self.config_manager = config_manager
+        
+        # Set environment variable for CUDA memory allocation
+        os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'expandable_segments:True'
+    
+    def check_model_exists_locally(self) -> bool:
+        """Check if model exists locally in Hugging Face cache."""
+        try:
+            from transformers.utils import cached_file
+            cached_file(self.model_id, "config.json", local_files_only=True)
+            return True
+        except:
+            return False
+    
+    def download_model_with_progress(self, progress_callback: Optional[Callable] = None) -> bool:
+        """
+        Download model with progress tracking.
+        
+        Args:
+            progress_callback: Callback function for progress updates
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            if progress_callback:
+                progress_callback("📥 Downloading tokenizer...")
+            
+            # Download tokenizer first (smaller)
+            tokenizer = AutoTokenizer.from_pretrained(
+                self.model_id, 
+                trust_remote_code=True, 
+                use_fast=False
+            )
+            
+            if progress_callback:
+                progress_callback("📥 Downloading model weights... This may take several minutes...")
+            
+            # Download model config and weights
+            config = AutoConfig.from_pretrained(self.model_id, trust_remote_code=True)
+            
+            if progress_callback:
+                progress_callback("✅ Model downloaded successfully!")
+            
+            return True
+        except Exception as e:
+            if progress_callback:
+                progress_callback(f"❌ Download failed: {str(e)}")
+            return False
+    
+    def split_model(self) -> Dict[str, int]:
+        """
+        Distribute LLM layers across GPUs, keeping vision encoder on GPU 0.
+        
+        Returns:
+            Device map dictionary
+        """
+        device_map = {}
+        world_size = torch.cuda.device_count()
+        
+        if world_size < 2:
+            return "auto"  # let transformers decide
+        
+        cfg = AutoConfig.from_pretrained(self.model_id, trust_remote_code=True)
+        num_layers = cfg.llm_config.num_hidden_layers  # type: ignore[attr-defined]
+        
+        # More aggressive distribution - treat GPU 0 as 0.3 GPU capacity due to vision model
+        effective_gpus = world_size - 0.7  # More conservative for GPU 0
+        layers_per_gpu = num_layers / effective_gpus
+        
+        # Calculate layer distribution
+        gpu_layers = []
+        for i in range(world_size):
+            if i == 0:
+                # GPU 0 gets fewer layers due to vision model
+                gpu_layers.append(max(1, int(layers_per_gpu * 0.3)))
+            else:
+                gpu_layers.append(int(layers_per_gpu))
+        
+        # Adjust if total doesn't match num_layers
+        total_assigned = sum(gpu_layers)
+        diff = num_layers - total_assigned
+        if diff > 0:
+            # Add remaining layers to non-zero GPUs
+            for i in range(1, min(world_size, diff + 1)):
+                gpu_layers[i] += 1
+        elif diff < 0:
+            # Remove excess layers from GPU 0
+            gpu_layers[0] = max(1, gpu_layers[0] + diff)
+        
+        # Assign layers to devices
+        layer_cnt = 0
+        for gpu_id, num_layers_on_gpu in enumerate(gpu_layers):
+            for _ in range(num_layers_on_gpu):
+                if layer_cnt < num_layers:
+                    device_map[f'language_model.model.layers.{layer_cnt}'] = gpu_id
+                    layer_cnt += 1
+        
+        # Distribute other components more evenly across GPUs
+        last_gpu = world_size - 1
+        
+        # Vision model must stay on GPU 0
+        device_map['vision_model'] = 0
+        device_map['mlp1'] = 0
+        
+        # Distribute language model components across GPUs
+        device_map['language_model.model.tok_embeddings'] = 0
+        device_map['language_model.model.embed_tokens'] = 0
+        device_map['language_model.model.norm'] = last_gpu  # Move to last GPU
+        device_map['language_model.model.rotary_emb'] = 1 if world_size > 1 else 0  # Move to GPU 1
+        device_map['language_model.output'] = last_gpu  # Move to last GPU
+        device_map['language_model.lm_head'] = last_gpu  # Move to last GPU
+        
+        # Keep the last layer on the same GPU as output layers for compatibility
+        device_map[f'language_model.model.layers.{num_layers - 1}'] = last_gpu
+        
+        print(f"Layer distribution: {gpu_layers}")
+        print(f"Total layers: {num_layers}, Assigned: {sum(gpu_layers)}")
+        
+        return device_map
+    
+    def load_model(self, quantization_type: str, progress_callback: Optional[Callable] = None) -> bool:
+        """
+        Load the model with specified quantization.
+        
+        Args:
+            quantization_type: Type of quantization to use
+            progress_callback: Callback function for progress updates
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        if not self.validate_quantization(quantization_type):
+            raise ValueError(f"Quantization type '{quantization_type}' not supported for {self.model_name}")
+        
+        # If model is already loaded with the same quantization, return
+        if (self.model is not None and self.tokenizer is not None and
+            self.current_quantization == quantization_type):
+            if progress_callback:
+                progress_callback(f"✅ {self.model_name} already loaded!")
+            return True
+        
+        print(f"Loading {self.model_name} with {quantization_type} quantization...")
+        if progress_callback:
+            progress_callback(f"🔄 Loading {self.model_name} with {quantization_type} quantization...")
+        
+        try:
+            # Check if model exists locally
+            model_exists = self.check_model_exists_locally()
+            if not model_exists:
+                if progress_callback:
+                    progress_callback(f"📥 {self.model_name} not found locally. Starting download...")
+                print(f"Model {self.model_name} not found locally. Starting download...")
+                success = self.download_model_with_progress(progress_callback)
+                if not success:
+                    raise Exception(f"Failed to download {self.model_name}")
+            else:
+                if progress_callback:
+                    progress_callback(f"✅ {self.model_name} found locally.")
+            
+            # Clear existing model if any
+            if self.model is not None:
+                self.unload_model()
+            
+            # Print memory before loading
+            self._print_gpu_memory("before loading")
+            
+            if progress_callback:
+                progress_callback(f"🚀 Loading {self.model_name} tokenizer...")
+            
+            # Load tokenizer
+            self.tokenizer = AutoTokenizer.from_pretrained(
+                self.model_id, 
+                trust_remote_code=True, 
+                use_fast=False
+            )
+            
+            # Load model based on quantization type
+            if "non-quantized" in quantization_type:
+                if progress_callback:
+                    progress_callback(f"🚀 Loading {self.model_name} model in 16-bit precision...")
+                
+                device_map = self.split_model()
+                print(f"Device map for multi-GPU: {device_map}")
+                
+                # Try loading with custom device_map, fallback to "auto" if it fails
+                # Some InternVL models (e.g., InternVL3_5) don't support custom device_map
+                # due to missing 'all_tied_weights_keys' attribute
+                try:
+                    self.model = AutoModel.from_pretrained(
+                        self.model_id,
+                        torch_dtype=torch.bfloat16,
+                        low_cpu_mem_usage=True,
+                        use_flash_attn=True,
+                        trust_remote_code=True,
+                        device_map=device_map,
+                    ).eval()
+                except (AttributeError, TypeError, RuntimeError, ValueError) as e:
+                    error_str = str(e).lower()
+                    # Check for device_map related errors, especially all_tied_weights_keys
+                    # This is a known issue with some InternVL models that don't expose
+                    # the all_tied_weights_keys attribute required for custom device_map
+                    if ("all_tied_weights_keys" in error_str or 
+                        "tied_weights" in error_str or
+                        ("device_map" in error_str and "attribute" in error_str)):
+                        print(f"⚠️ Custom device_map failed ({str(e)}), falling back to 'auto' device_map...")
+                        if progress_callback:
+                            progress_callback(f"⚠️ Using automatic device mapping...")
+                        self.model = AutoModel.from_pretrained(
+                            self.model_id,
+                            torch_dtype=torch.bfloat16,
+                            low_cpu_mem_usage=True,
+                            use_flash_attn=True,
+                            trust_remote_code=True,
+                            device_map="auto",
+                        ).eval()
+                    else:
+                        # Re-raise if it's a different error
+                        raise
+            else:  # quantized (8bit)
+                if progress_callback:
+                    progress_callback(f"🚀 Loading {self.model_name} model with 8-bit quantization...")
+                
+                print("Loading with 8-bit quantization to reduce memory usage...")
+                self.model = AutoModel.from_pretrained(
+                    self.model_id,
+                    torch_dtype=torch.bfloat16,
+                    load_in_8bit=True,
+                    low_cpu_mem_usage=True,
+                    use_flash_attn=True,
+                    trust_remote_code=True,
+                    device_map="auto"  # Let transformers handle device mapping for quantized model
+                ).eval()
+            
+            # Verify model and tokenizer are properly loaded
+            if self.model is None:
+                raise Exception(f"Model failed to load for {self.model_name}")
+            if self.tokenizer is None:
+                raise Exception(f"Tokenizer failed to load for {self.model_name}")
+            
+            self.current_quantization = quantization_type
+            self.is_loaded = True
+            
+            success_msg = f"✅ {self.model_name} loaded successfully with {quantization_type} quantization!"
+            print(success_msg)
+            if progress_callback:
+                progress_callback(success_msg)
+            
+            # Print GPU memory usage after loading
+            self._print_gpu_memory("after loading")
+            
+            return True
+            
+        except Exception as e:
+            error_msg = f"Failed to load model {self.model_name}: {str(e)}"
+            print(error_msg)
+            if progress_callback:
+                progress_callback(f"❌ {error_msg}")
+            
+            # Reset on failure
+            self.unload_model()
+            raise Exception(error_msg)
+    
+    def unload_model(self) -> None:
+        """Unload the model from memory."""
+        if self.model is not None:
+            print("🧹 Clearing model from memory...")
+            del self.model
+            self.model = None
+            
+        if self.tokenizer is not None:
+            del self.tokenizer
+            self.tokenizer = None
+            
+        self.current_quantization = None
+        self.is_loaded = False
+        
+        # Clear GPU cache
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+        
+        # Force garbage collection
+        gc.collect()
+        
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()  # Clear again after gc
+        
+        print("✅ Model unloaded successfully")
+    
+    def inference(self, image_path: str, prompt: str, **kwargs) -> str:
+        """
+        Perform inference on an image with a text prompt.
+        
+        Args:
+            image_path: Path to the image file
+            prompt: Text prompt for the model
+            **kwargs: Additional inference parameters
+            
+        Returns:
+            Model's text response
+        """
+        if not self.is_loaded:
+            raise RuntimeError(f"Model {self.model_name} is not loaded. Call load_model() first.")
+        
+        try:
+            # Load and preprocess image using default settings from original app.py
+            pixel_values = load_image(image_path, input_size=448, max_num=12).to(torch.bfloat16)
+            
+            # Move pixel_values to the same device as the model
+            if torch.cuda.is_available():
+                # Get the device of the first model parameter
+                model_device = next(self.model.parameters()).device
+                pixel_values = pixel_values.to(model_device)
+            else:
+                # Fallback to CPU if no CUDA available
+                pixel_values = pixel_values.cpu()
+            
+            # Prepare prompt
+            formatted_prompt = f"<image>\n{prompt}" if prompt else "<image>\n"
+            
+            # Generation configuration - using same settings as original app.py
+            gen_cfg = dict(max_new_tokens=1024, do_sample=True)
+            
+            # Perform inference
+            response = self.model.chat(self.tokenizer, pixel_values, formatted_prompt, gen_cfg)
+            return response
+            
+        except Exception as e:
+            error_msg = f"Error processing image: {str(e)}"
+            print(error_msg)
+            return error_msg
+    
+    def _print_gpu_memory(self, stage: str) -> None:
+        """Print GPU memory usage for debugging."""
+        if torch.cuda.is_available():
+            print(f"Memory {stage}:")
+            for i in range(torch.cuda.device_count()):
+                allocated = torch.cuda.memory_allocated(i) / 1024**3
+                reserved = torch.cuda.memory_reserved(i) / 1024**3
+                print(f"GPU {i}: Allocated {allocated:.2f} GB, Reserved {reserved:.2f} GB") 

backend/models/model_manager.py

@@ -0,0 +1,248 @@
+import threading
+from typing import Dict, Any, Optional, Callable
+from .base_model import BaseModel
+from .internvl import InternVLModel
+from .qwen import QwenModel
+from ..config.config_manager import ConfigManager
+
+
+class ModelManager:
+    """Manager class for handling multiple vision-language models."""
+    
+    def __init__(self, config_manager: ConfigManager):
+        """
+        Initialize the model manager.
+        
+        Args:
+            config_manager: Configuration manager instance
+        """
+        self.config_manager = config_manager
+        self.models: Dict[str, BaseModel] = {}
+        self.current_model: Optional[BaseModel] = None
+        self.current_model_name: Optional[str] = None
+        self.loading_lock = threading.Lock()
+        
+        # Apply environment settings
+        self.config_manager.apply_environment_settings()
+        
+        # Initialize models but don't load them yet
+        self._initialize_models()
+    
+    def _get_model_class(self, model_config: Dict[str, Any]) -> type:
+        """
+        Determine the appropriate model class based on model configuration.
+        
+        Args:
+            model_config: Model configuration dictionary
+            
+        Returns:
+            Model class to instantiate
+        """
+        model_type = model_config.get('model_type', 'internvl').lower()
+        model_id = model_config.get('model_id', '').lower()
+        
+        # Determine model type based on model_id or explicit model_type
+        if 'qwen' in model_id or model_type == 'qwen':
+            return QwenModel
+        elif 'internvl' in model_id or model_type == 'internvl':
+            return InternVLModel
+        else:
+            # Default to InternVL for backward compatibility
+            print(f"⚠️ Unknown model type for {model_config.get('name', 'unknown')}, defaulting to InternVL")
+            return InternVLModel
+    
+    def _initialize_models(self) -> None:
+        """Initialize model instances without loading them."""
+        available_models = self.config_manager.get_available_models()
+        
+        for model_name, model_id in available_models.items():
+            model_config = self.config_manager.get_model_config(model_name)
+            
+            # Determine the appropriate model class
+            model_class = self._get_model_class(model_config)
+            
+            # Create model instance
+            self.models[model_name] = model_class(
+                model_name=model_name,
+                model_config=model_config,
+                config_manager=self.config_manager
+            )
+            
+            print(f"✅ Initialized {model_class.__name__}: {model_name}")
+    
+    def get_available_models(self) -> list[str]:
+        """Get list of available model names."""
+        return list(self.models.keys())
+    
+    def get_model_info(self, model_name: str) -> Dict[str, Any]:
+        """
+        Get information about a specific model.
+        
+        Args:
+            model_name: Name of the model
+            
+        Returns:
+            Model information dictionary
+        """
+        if model_name not in self.models:
+            raise KeyError(f"Model '{model_name}' not available")
+        
+        return self.models[model_name].get_model_info()
+    
+    def get_all_models_info(self) -> Dict[str, Dict[str, Any]]:
+        """Get information about all available models."""
+        return {name: model.get_model_info() for name, model in self.models.items()}
+    
+    def load_model(
+        self, 
+        model_name: str, 
+        quantization_type: str, 
+        progress_callback: Optional[Callable] = None
+    ) -> bool:
+        """
+        Load a specific model with given quantization.
+        
+        Args:
+            model_name: Name of the model to load
+            quantization_type: Type of quantization to use
+            progress_callback: Callback function for progress updates
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        with self.loading_lock:
+            if model_name not in self.models:
+                raise KeyError(f"Model '{model_name}' not available")
+            
+            model = self.models[model_name]
+            
+            # Check if this model is already loaded with the same quantization
+            if (self.current_model == model and 
+                model.is_model_loaded() and 
+                model.current_quantization == quantization_type):
+                if progress_callback:
+                    progress_callback(f"✅ {model_name} already loaded with {quantization_type}!")
+                return True
+            
+            # Unload current model if different
+            if (self.current_model and 
+                self.current_model != model and 
+                self.current_model.is_model_loaded()):
+                if progress_callback:
+                    progress_callback(f"🔄 Unloading {self.current_model_name}...")
+                self.current_model.unload_model()
+            
+            # Load the requested model
+            try:
+                success = model.load_model(quantization_type, progress_callback)
+                if success:
+                    self.current_model = model
+                    self.current_model_name = model_name
+                    print(f"✅ Successfully loaded {model_name} with {quantization_type}")
+                    return True
+                else:
+                    if progress_callback:
+                        progress_callback(f"❌ Failed to load {model_name}")
+                    return False
+            except Exception as e:
+                error_msg = f"Error loading {model_name}: {str(e)}"
+                print(error_msg)
+                if progress_callback:
+                    progress_callback(f"❌ {error_msg}")
+                return False
+    
+    def unload_current_model(self) -> None:
+        """Unload the currently loaded model."""
+        with self.loading_lock:
+            if self.current_model and self.current_model.is_model_loaded():
+                print(f"🔄 Unloading {self.current_model_name}...")
+                self.current_model.unload_model()
+                self.current_model = None
+                self.current_model_name = None
+                print("✅ Model unloaded successfully")
+            else:
+                print("ℹ️ No model currently loaded")
+    
+    def inference(self, image_path: str, prompt: str, **kwargs) -> str:
+        """
+        Perform inference using the currently loaded model.
+        
+        Args:
+            image_path: Path to the image file
+            prompt: Text prompt for the model
+            **kwargs: Additional inference parameters
+            
+        Returns:
+            Model's text response
+        """
+        if not self.current_model or not self.current_model.is_model_loaded():
+            raise RuntimeError("No model is currently loaded. Load a model first.")
+        
+        return self.current_model.inference(image_path, prompt, **kwargs)
+    
+    def get_current_model_status(self) -> str:
+        """Get status string for the currently loaded model."""
+        if not self.current_model or not self.current_model.is_model_loaded():
+            return "❌ No model loaded"
+        
+        quantization = self.current_model.current_quantization or "Unknown"
+        model_class = self.current_model.__class__.__name__
+        return f"✅ {self.current_model_name} ({model_class}) loaded with {quantization}"
+    
+    def get_supported_quantizations(self, model_name: str) -> list[str]:
+        """Get supported quantization methods for a model."""
+        if model_name not in self.models:
+            raise KeyError(f"Model '{model_name}' not available")
+        
+        return self.models[model_name].get_supported_quantizations()
+    
+    def validate_model_and_quantization(self, model_name: str, quantization_type: str) -> bool:
+        """
+        Validate if a model and quantization combination is valid.
+        
+        Args:
+            model_name: Name of the model
+            quantization_type: Type of quantization
+            
+        Returns:
+            True if valid, False otherwise
+        """
+        if model_name not in self.models:
+            return False
+        
+        return self.models[model_name].validate_quantization(quantization_type)
+    
+    def get_model_memory_requirements(self, model_name: str) -> Dict[str, int]:
+        """Get memory requirements for a specific model."""
+        if model_name not in self.models:
+            raise KeyError(f"Model '{model_name}' not available")
+        
+        return self.models[model_name].get_memory_requirements()
+    
+    def preload_default_model(self) -> bool:
+        """
+        Preload the default model specified in configuration.
+        
+        Returns:
+            True if successful, False otherwise
+        """
+        default_model = self.config_manager.get_default_model()
+        default_quantization = self.config_manager.get_default_quantization(default_model)
+        
+        print(f"🚀 Preloading default model: {default_model} with {default_quantization}")
+        
+        try:
+            return self.load_model(default_model, default_quantization)
+        except Exception as e:
+            print(f"⚠️ Failed to preload default model: {str(e)}")
+            return False
+    
+    def __str__(self) -> str:
+        """String representation of the model manager."""
+        loaded_info = f"Current: {self.current_model_name}" if self.current_model_name else "None loaded"
+        return f"ModelManager({len(self.models)} models available, {loaded_info})"
+    
+    def __repr__(self) -> str:
+        """Detailed string representation."""
+        models_list = list(self.models.keys())
+        return f"ModelManager(models={models_list}, current={self.current_model_name})"

backend/models/qwen/__init__.py

@@ -0,0 +1,3 @@
+from .qwen_model import QwenModel
+
+__all__ = ['QwenModel'] 

backend/models/qwen/qwen_model.py

@@ -0,0 +1,273 @@
+import torch
+import gc
+import os
+from typing import Dict, Any, Optional, Callable
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from ..base_model import BaseModel
+from ...config.config_manager import ConfigManager
+
+
+class QwenModel(BaseModel):
+    """Qwen2.5 model implementation."""
+    
+    def __init__(self, model_name: str, model_config: Dict[str, Any], config_manager: ConfigManager):
+        """
+        Initialize the Qwen model.
+        
+        Args:
+            model_name: Name of the model
+            model_config: Configuration dictionary for the model
+            config_manager: Configuration manager instance
+        """
+        super().__init__(model_name, model_config)
+        self.config_manager = config_manager
+        
+        # Set environment variable for CUDA memory allocation
+        os.environ['PYTORCH_CUDA_ALLOC_CONF'] = 'expandable_segments:True'
+    
+    def check_model_exists_locally(self) -> bool:
+        """Check if model exists locally in Hugging Face cache."""
+        try:
+            from transformers.utils import cached_file
+            cached_file(self.model_id, "config.json", local_files_only=True)
+            return True
+        except:
+            return False
+    
+    def download_model_with_progress(self, progress_callback: Optional[Callable] = None) -> bool:
+        """
+        Download model with progress tracking.
+        
+        Args:
+            progress_callback: Callback function for progress updates
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        try:
+            if progress_callback:
+                progress_callback("📥 Downloading tokenizer...")
+            
+            # Download tokenizer first (smaller)
+            tokenizer = AutoTokenizer.from_pretrained(self.model_id)
+            
+            if progress_callback:
+                progress_callback("📥 Downloading model weights... This may take several minutes...")
+            
+            # Download model config and weights by trying to load config
+            model = AutoModelForCausalLM.from_pretrained(
+                self.model_id,
+                torch_dtype="auto",
+                device_map="cpu",  # Just download, don't load to GPU yet
+                low_cpu_mem_usage=True
+            )
+            
+            # Clean up the test loading
+            del model
+            
+            if progress_callback:
+                progress_callback("✅ Model downloaded successfully!")
+            
+            return True
+        except Exception as e:
+            if progress_callback:
+                progress_callback(f"❌ Download failed: {str(e)}")
+            return False
+    
+    def load_model(self, quantization_type: str, progress_callback: Optional[Callable] = None) -> bool:
+        """
+        Load the model with specified quantization.
+        
+        Args:
+            quantization_type: Type of quantization to use
+            progress_callback: Callback function for progress updates
+            
+        Returns:
+            True if successful, False otherwise
+        """
+        if not self.validate_quantization(quantization_type):
+            raise ValueError(f"Quantization type '{quantization_type}' not supported for {self.model_name}")
+        
+        # If model is already loaded with the same quantization, return
+        if (self.model is not None and self.tokenizer is not None and
+            self.current_quantization == quantization_type):
+            if progress_callback:
+                progress_callback(f"✅ {self.model_name} already loaded!")
+            return True
+        
+        print(f"Loading {self.model_name} with {quantization_type} quantization...")
+        if progress_callback:
+            progress_callback(f"🔄 Loading {self.model_name} with {quantization_type} quantization...")
+        
+        try:
+            # Check if model exists locally
+            model_exists = self.check_model_exists_locally()
+            if not model_exists:
+                if progress_callback:
+                    progress_callback(f"📥 {self.model_name} not found locally. Starting download...")
+                print(f"Model {self.model_name} not found locally. Starting download...")
+                success = self.download_model_with_progress(progress_callback)
+                if not success:
+                    raise Exception(f"Failed to download {self.model_name}")
+            else:
+                if progress_callback:
+                    progress_callback(f"✅ {self.model_name} found locally.")
+            
+            # Clear existing model if any
+            if self.model is not None:
+                self.unload_model()
+            
+            # Print memory before loading
+            self._print_gpu_memory("before loading")
+            
+            if progress_callback:
+                progress_callback(f"🚀 Loading {self.model_name} tokenizer...")
+            
+            # Load tokenizer
+            self.tokenizer = AutoTokenizer.from_pretrained(self.model_id)
+            
+            # Load model based on quantization type
+            if progress_callback:
+                progress_callback(f"🚀 Loading {self.model_name} model...")
+            
+            if "non-quantized" in quantization_type:
+                # Load with auto dtype and device mapping
+                self.model = AutoModelForCausalLM.from_pretrained(
+                    self.model_id,
+                    torch_dtype="auto",
+                    device_map="auto",
+                    low_cpu_mem_usage=True
+                )
+            else:  # quantized (8bit)
+                print("Loading with 8-bit quantization to reduce memory usage...")
+                self.model = AutoModelForCausalLM.from_pretrained(
+                    self.model_id,
+                    torch_dtype="auto",
+                    load_in_8bit=True,
+                    device_map="auto",
+                    low_cpu_mem_usage=True
+                )
+            
+            # Verify model and tokenizer are properly loaded
+            if self.model is None:
+                raise Exception(f"Model failed to load for {self.model_name}")
+            if self.tokenizer is None:
+                raise Exception(f"Tokenizer failed to load for {self.model_name}")
+            
+            self.current_quantization = quantization_type
+            self.is_loaded = True
+            
+            success_msg = f"✅ {self.model_name} loaded successfully with {quantization_type} quantization!"
+            print(success_msg)
+            if progress_callback:
+                progress_callback(success_msg)
+            
+            # Print GPU memory usage after loading
+            self._print_gpu_memory("after loading")
+            
+            return True
+            
+        except Exception as e:
+            error_msg = f"Failed to load model {self.model_name}: {str(e)}"
+            print(error_msg)
+            if progress_callback:
+                progress_callback(f"❌ {error_msg}")
+            
+            # Reset on failure
+            self.unload_model()
+            raise Exception(error_msg)
+    
+    def unload_model(self) -> None:
+        """Unload the model from memory."""
+        if self.model is not None:
+            print("🧹 Clearing model from memory...")
+            del self.model
+            self.model = None
+            
+        if self.tokenizer is not None:
+            del self.tokenizer
+            self.tokenizer = None
+            
+        self.current_quantization = None
+        self.is_loaded = False
+        
+        # Clear GPU cache
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+        
+        # Force garbage collection
+        gc.collect()
+        
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()  # Clear again after gc
+        
+        print("✅ Model unloaded successfully")
+    
+    def inference(self, image_path: str, prompt: str, **kwargs) -> str:
+        """
+        Perform inference with a text prompt.
+        Note: Qwen2.5 is a text-only model, so image_path is ignored.
+        
+        Args:
+            image_path: Path to the image file (ignored for text-only models)
+            prompt: Text prompt for the model
+            **kwargs: Additional inference parameters
+            
+        Returns:
+            Model's text response
+        """
+        if not self.is_loaded:
+            raise RuntimeError(f"Model {self.model_name} is not loaded. Call load_model() first.")
+        
+        if not prompt or not prompt.strip():
+            return "Error: No prompt provided"
+        
+        try:
+            # Prepare messages for chat format
+            messages = [
+                {"role": "system", "content": "You are Qwen, created by Alibaba Cloud. You are a helpful assistant."},
+                {"role": "user", "content": prompt}
+            ]
+            
+            # Apply chat template
+            text = self.tokenizer.apply_chat_template(
+                messages,
+                tokenize=False,
+                add_generation_prompt=True
+            )
+            
+            # Tokenize input
+            model_inputs = self.tokenizer([text], return_tensors="pt").to(self.model.device)
+            
+            # Generate response
+            generated_ids = self.model.generate(
+                **model_inputs,
+                max_new_tokens=kwargs.get('max_new_tokens', 512),
+                do_sample=kwargs.get('do_sample', True),
+                temperature=kwargs.get('temperature', 0.7),
+                top_p=kwargs.get('top_p', 0.9),
+                pad_token_id=self.tokenizer.eos_token_id
+            )
+            
+            # Extract only the generated part (remove input tokens)
+            generated_ids = [
+                output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
+            ]
+            
+            # Decode response
+            response = self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
+            return response
+            
+        except Exception as e:
+            error_msg = f"Error processing prompt: {str(e)}"
+            print(error_msg)
+            return error_msg
+    
+    def _print_gpu_memory(self, stage: str) -> None:
+        """Print GPU memory usage for debugging."""
+        if torch.cuda.is_available():
+            print(f"Memory {stage}:")
+            for i in range(torch.cuda.device_count()):
+                allocated = torch.cuda.memory_allocated(i) / 1024**3
+                reserved = torch.cuda.memory_reserved(i) / 1024**3
+                print(f"GPU {i}: Allocated {allocated:.2f} GB, Reserved {reserved:.2f} GB") 

backend/utils/__init__.py

@@ -0,0 +1,29 @@
+from .image_processing import (
+    build_transform, 
+    find_closest_aspect_ratio, 
+    dynamic_preprocess, 
+    load_image
+)
+from .data_processing import (
+    extract_file_dict,
+    validate_data,
+    extract_binary_output
+)
+from .metrics import (
+    create_confusion_matrix_plot,
+    create_accuracy_table,
+    save_dataframe_to_csv
+)
+
+__all__ = [
+    'build_transform',
+    'find_closest_aspect_ratio', 
+    'dynamic_preprocess',
+    'load_image',
+    'extract_file_dict',
+    'validate_data',
+    'extract_binary_output',
+    'create_confusion_matrix_plot',
+    'create_accuracy_table',
+    'save_dataframe_to_csv'
+] 

backend/utils/data_processing.py

@@ -0,0 +1,100 @@
+import pandas as pd
+import os
+from pathlib import Path
+from typing import Dict, List, Tuple, Union, Any
+
+def extract_file_dict(folder_path: List[Path]) -> Dict[str, Path]:
+    """
+    Extract file dictionary from folder path.
+    
+    Args:
+        folder_path: List of Path objects from Gradio file upload
+        
+    Returns:
+        Dictionary mapping filename to full path
+    """
+    file_dict = {}
+    for file in folder_path:
+        filepath = file
+        filename = filepath.name.split("/")[-1]
+        file_dict[filename] = filepath
+    return file_dict
+
+
+def validate_data(file_dict: Dict[str, Path]) -> Tuple[Union[bool, str], str]:
+    """
+    Validate the uploaded data structure.
+    
+    Args:
+        file_dict: Dictionary of filename to path mappings
+        
+    Returns:
+        Tuple of (validation_result, message)
+        validation_result can be:
+        - True: Valid data with CSV
+        - False: Invalid data
+        - "no_csv": Valid but no CSV file
+        - "multiple_csv": Valid but multiple CSV files
+    """
+    # Find CSV file
+    csv_files = [fname for fname in file_dict if fname.lower().endswith('.csv')]
+    
+    # Find image files
+    image_exts = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff']
+    image_files = [fname for fname in file_dict if any(fname.lower().endswith(ext) for ext in image_exts)]
+    
+    if not image_files:
+        return False, "No image files found in the folder or subfolders"
+    
+    # If no CSV or multiple CSVs, we'll proceed with file-based processing
+    if len(csv_files) == 0:
+        return "no_csv", "No CSV file found. Will extract data from file paths and names."
+    elif len(csv_files) > 1:
+        return "multiple_csv", "Multiple CSV files found. Will extract data from file paths and names."
+    
+    # Check if single CSV has required columns
+    try:
+        df = pd.read_csv(file_dict[csv_files[0]])
+        if 'Ground Truth' not in df.columns:
+            return False, "CSV file does not contain 'Ground Truth' column"
+        if 'Image Name' not in df.columns:
+            return False, "CSV file does not contain 'Image Name' column"
+    except Exception as e:
+        return False, f"Error reading CSV file: {str(e)}"
+    
+    return True, "Data validation successful"
+
+
+def extract_binary_output(
+    model_output: str, 
+    ground_truth: str = "", 
+    all_ground_truths: List[str] = None
+) -> str:
+    """
+    Extract binary output from model response based on unique ground truth keywords.
+    
+    Args:
+        model_output: The model's text response
+        ground_truth: Current item's ground truth (for fallback)
+        all_ground_truths: List of all ground truth values to extract unique keywords
+    
+    Returns:
+        Extracted keyword that best matches the model output
+    """
+    if all_ground_truths is None:
+        all_ground_truths = []
+        
+    # Unique lowercase keywords
+    unique_keywords = sorted({str(gt).strip().lower() for gt in all_ground_truths if gt})
+    
+    # Take only the first line of model output
+    first_line = model_output.split("\n", 1)[0].lower()
+
+    print(f"DEBUG: Unique keywords extracted: {first_line}")
+    print(f"DEBUG: Model output: {model_output[:100]}...")  # First 100 chars
+    
+    for keyword in unique_keywords:
+        if keyword in first_line:
+            return keyword
+    
+    return "Enter the output manually" 

backend/utils/image_processing.py

@@ -0,0 +1,147 @@
+import torch
+import numpy as np
+from PIL import Image
+import torchvision.transforms as T
+from torchvision.transforms.functional import InterpolationMode
+from typing import List, Tuple, Union
+
+# Constants from InternVL preprocessing
+IMAGENET_MEAN = (0.485, 0.456, 0.406)
+IMAGENET_STD = (0.229, 0.224, 0.225)
+
+
+def build_transform(input_size: int = 448) -> T.Compose:
+    """
+    Return torchvision transform matching InternVL pre‑training.
+    
+    Args:
+        input_size: Input image size (default: 448)
+        
+    Returns:
+        Composed torchvision transforms
+    """
+    return T.Compose([
+        T.Lambda(lambda img: img.convert("RGB") if img.mode != "RGB" else img),
+        T.Resize((input_size, input_size), interpolation=InterpolationMode.BICUBIC),
+        T.ToTensor(),
+        T.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD),
+    ])
+
+
+def find_closest_aspect_ratio(
+    aspect_ratio: float, 
+    target_ratios: List[Tuple[int, int]], 
+    width: int, 
+    height: int, 
+    image_size: int
+) -> Tuple[int, int]:
+    """
+    Find the closest aspect ratio from target ratios.
+    
+    Args:
+        aspect_ratio: Current image aspect ratio
+        target_ratios: List of target aspect ratios as (width, height) tuples
+        width: Original image width
+        height: Original image height
+        image_size: Target image size
+        
+    Returns:
+        Best matching aspect ratio as (width, height) tuple
+    """
+    best_ratio_diff = float("inf")
+    best_ratio = (1, 1)
+    area = width * height
+    
+    for ratio in target_ratios:
+        tgt_ar = ratio[0] / ratio[1]
+        diff = abs(aspect_ratio - tgt_ar)
+        
+        if (diff < best_ratio_diff or 
+            (diff == best_ratio_diff and area > 0.5 * image_size * image_size * ratio[0] * ratio[1])):
+            best_ratio_diff = diff
+            best_ratio = ratio
+            
+    return best_ratio
+
+
+def dynamic_preprocess(
+    image: Image.Image, 
+    min_num: int = 1, 
+    max_num: int = 12, 
+    image_size: int = 448, 
+    use_thumbnail: bool = False
+) -> List[Image.Image]:
+    """
+    Split arbitrarily‑sized image into ≤12 tiles sized 448×448 (InternVL spec).
+    
+    Args:
+        image: Input PIL Image
+        min_num: Minimum number of tiles
+        max_num: Maximum number of tiles
+        image_size: Size of each tile (default: 448)
+        use_thumbnail: Whether to add a thumbnail version
+        
+    Returns:
+        List of processed image tiles
+    """
+    ow, oh = image.size
+    aspect_ratio = ow / oh
+    
+    # Generate target ratios
+    target_ratios = sorted(
+        {(i, j) for n in range(min_num, max_num + 1) 
+         for i in range(1, n + 1) 
+         for j in range(1, n + 1) 
+         if min_num <= i * j <= max_num},
+        key=lambda x: x[0] * x[1],
+    )
+    
+    ratio = find_closest_aspect_ratio(aspect_ratio, target_ratios, ow, oh, image_size)
+    tw, th = image_size * ratio[0], image_size * ratio[1]
+    blocks = ratio[0] * ratio[1]
+    
+    resized = image.resize((tw, th))
+    
+    # Create tiles
+    tiles = []
+    for idx in range(blocks):
+        tile = resized.crop((
+            (idx % (tw // image_size)) * image_size,
+            (idx // (tw // image_size)) * image_size,
+            ((idx % (tw // image_size)) + 1) * image_size,
+            ((idx // (tw // image_size)) + 1) * image_size,
+        ))
+        tiles.append(tile)
+    
+    # Add thumbnail if requested and more than one tile
+    if use_thumbnail and blocks != 1:
+        tiles.append(image.resize((image_size, image_size)))
+        
+    return tiles
+
+
+def load_image(
+    path: str, 
+    input_size: int = 448, 
+    max_num: int = 12
+) -> torch.Tensor:
+    """
+    Load and preprocess image for InternVL model.
+    
+    Args:
+        path: Path to the image file
+        input_size: Input image size (default: 448)
+        max_num: Maximum number of tiles (default: 12)
+        
+    Returns:
+        Tensor of shape (N, 3, H, W) ready for InternVL
+    """
+    img = Image.open(path).convert("RGB")
+    transform = build_transform(input_size)
+    tiles = dynamic_preprocess(
+        img, 
+        image_size=input_size, 
+        use_thumbnail=True, 
+        max_num=max_num
+    )
+    return torch.stack([transform(t) for t in tiles]) 

backend/utils/metrics.py

@@ -0,0 +1,160 @@
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+import seaborn as sns
+import tempfile
+from typing import Tuple, Optional
+from sklearn.metrics import confusion_matrix, accuracy_score, precision_score, recall_score, f1_score
+
+
+def create_confusion_matrix_plot(
+    cm: np.ndarray, 
+    accuracy: float, 
+    labels: list = ['No', 'Yes']
+) -> str:
+    """
+    Create a confusion matrix plot and save it to a temporary file.
+    
+    Args:
+        cm: Confusion matrix array
+        accuracy: Accuracy score
+        labels: Labels for the confusion matrix
+        
+    Returns:
+        Path to the saved plot file
+    """
+    plt.figure(figsize=(6, 5))
+    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=labels, yticklabels=labels)
+    plt.title(f'Confusion Matrix (Accuracy: {accuracy:.1%})')
+    plt.ylabel('Ground Truth')
+    plt.xlabel('Model Prediction')
+    
+    temp_file = tempfile.mktemp(suffix='.png')
+    plt.savefig(temp_file, dpi=150, bbox_inches='tight')
+    plt.close()
+    
+    return temp_file
+
+
+def create_accuracy_table(df: pd.DataFrame) -> Tuple[pd.DataFrame, str, pd.DataFrame]:
+    """
+    Create accuracy metrics table and confusion matrix from results dataframe.
+    
+    Args:
+        df: DataFrame with 'Ground Truth' and 'Binary Output' columns
+        
+    Returns:
+        Tuple of (metrics_df, confusion_matrix_plot_path, confusion_matrix_values_df)
+        
+    Raises:
+        ValueError: If insufficient data for binary classification
+    """
+    df_copy = df.copy()
+    
+    # Get unique values from both Ground Truth and Binary Output
+    # Convert to string first, then apply .str operations
+    ground_truth_values = df_copy['Ground Truth'].dropna().astype(str).str.lower().unique()
+    binary_output_values = df_copy['Binary Output'].dropna().astype(str).str.lower().unique()
+    
+    # Combine and get all unique values
+    all_values = set(list(ground_truth_values) + list(binary_output_values))
+    all_values = [v for v in all_values if v.strip()]  # Remove empty strings
+    
+    if len(all_values) < 2:
+        raise ValueError("Need at least 2 different values for binary classification")
+    
+    # Sort values to ensure consistent mapping (alphabetical order)
+    sorted_values = sorted(all_values)
+    
+    # Create mapping: first value (alphabetically) = 0, second = 1
+    # This ensures consistent mapping regardless of order in data
+    value_mapping = {sorted_values[0]: 0}
+    if len(sorted_values) >= 2:
+        value_mapping[sorted_values[1]] = 1
+    
+    # If there are more than 2 values, map the rest to 1 (positive class)
+    for i in range(2, len(sorted_values)):
+        value_mapping[sorted_values[i]] = 1
+    
+    print(f"Detected binary mapping: {value_mapping}")
+    
+    # Apply mapping - convert to string first, then apply .str operations
+    df_copy['Ground Truth Binary'] = df_copy['Ground Truth'].astype(str).str.lower().map(value_mapping)
+    df_copy['Binary Output Binary'] = df_copy['Binary Output'].astype(str).str.lower().map(value_mapping)
+    
+    # Remove rows where either ground truth or binary output is NaN
+    df_copy = df_copy.dropna(subset=['Ground Truth Binary', 'Binary Output Binary'])
+    
+    if len(df_copy) == 0:
+        raise ValueError("No valid data for accuracy calculation after mapping. Check that Ground Truth and Binary Output contain valid binary values.")
+    
+    # Calculate metrics
+    cm = confusion_matrix(df_copy['Ground Truth Binary'], df_copy['Binary Output Binary'])
+    accuracy = accuracy_score(df_copy['Ground Truth Binary'], df_copy['Binary Output Binary'])
+    precision = precision_score(df_copy['Ground Truth Binary'], df_copy['Binary Output Binary'], zero_division=0)
+    recall = recall_score(df_copy['Ground Truth Binary'], df_copy['Binary Output Binary'], zero_division=0)
+    f1 = f1_score(df_copy['Ground Truth Binary'], df_copy['Binary Output Binary'], zero_division=0)
+    
+    # Create metrics dataframe
+    metrics_data = [
+        ["Accuracy", f"{accuracy:.3f}"],
+        ["Precision", f"{precision:.3f}"],
+        ["Recall", f"{recall:.3f}"],
+        ["F1 Score", f"{f1:.3f}"],
+        ["Total Samples", f"{len(df_copy)}"]
+    ]
+    metrics_df = pd.DataFrame(metrics_data, columns=["Metric", "Value"])
+    
+    # Create labels for confusion matrix based on detected values
+    # Find the original case versions of the labels
+    original_labels = []
+    for mapped_val in sorted([k for k, v in value_mapping.items() if v in [0, 1]]):
+        # Find original case version from the data
+        original_case = None
+        for val in df_copy['Ground Truth'].dropna():
+            if str(val).lower() == mapped_val:
+                original_case = str(val)
+                break
+        if original_case is None:
+            for val in df_copy['Binary Output'].dropna():
+                if str(val).lower() == mapped_val:
+                    original_case = str(val)
+                    break
+        original_labels.append(original_case if original_case else mapped_val.title())
+    
+    # Ensure we have exactly 2 labels
+    if len(original_labels) < 2:
+        original_labels = ['Class 0', 'Class 1']
+    
+    cm_plot_path = create_confusion_matrix_plot(cm, accuracy, original_labels)
+    
+    # Confusion matrix values table
+    if cm.shape == (2, 2):
+        tn, fp, fn, tp = cm.ravel()
+        cm_values = pd.DataFrame(
+            [[tn, fp], [fn, tp]],
+            columns=[f"Predicted {original_labels[0]}", f"Predicted {original_labels[1]}"],
+            index=[f"Actual {original_labels[0]}", f"Actual {original_labels[1]}"]
+        )
+    else:
+        cm_values = pd.DataFrame(cm)
+    
+    return metrics_df, cm_plot_path, cm_values
+
+
+def save_dataframe_to_csv(df: pd.DataFrame) -> Optional[str]:
+    """
+    Save dataframe to a temporary CSV file.
+    
+    Args:
+        df: DataFrame to save
+        
+    Returns:
+        Path to saved CSV file or None if failed
+    """
+    if df is None or df.empty:
+        return None
+        
+    temp_file = tempfile.mktemp(suffix='.csv')
+    df.to_csv(temp_file, index=False)
+    return temp_file 

config/models.yaml

@@ -0,0 +1,68 @@
+# Model Configuration for Vision Language Models and Language Models
+# This file contains model configurations for easy integration
+
+models:
+  # InternVL Vision-Language Models
+  InternVL3-8B:
+    name: "InternVL3-8B"
+    model_id: "OpenGVLab/InternVL3-8B"
+    model_type: "internvl"
+    description: "Fastest model, good for quick processing"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "non-quantized(fp16)"
+    
+  InternVL3-14B:
+    name: "InternVL3-14B"
+    model_id: "OpenGVLab/InternVL3-14B"
+    model_type: "internvl"
+    description: "Balanced performance and quality"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "quantized(8bit)"
+    
+  InternVL3-38B:
+    name: "InternVL3-38B"
+    model_id: "OpenGVLab/InternVL3-38B"
+    model_type: "internvl"
+    description: "Highest quality, requires significant GPU memory"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "quantized(8bit)"
+    
+  InternVL3_5-8B:
+    name: "InternVL3_5-8B"
+    model_id: "OpenGVLab/InternVL3_5-8B"
+    model_type: "internvl"
+    description: "Fastest model, good for quick processing"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "non-quantized(fp16)"
+    
+  # Qwen Language Models (Text-only)
+  Qwen2.5-7B-Instruct:
+    name: "Qwen2.5-7B-Instruct"
+    model_id: "Qwen/Qwen2.5-7B-Instruct"
+    model_type: "qwen"
+    description: "Qwen2.5 7B instruction-tuned model for text generation"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "quantized(8bit)"
+    
+  Qwen2.5-14B-Instruct:
+    name: "Qwen2.5-14B-Instruct"
+    model_id: "Qwen/Qwen2.5-14B-Instruct"
+    model_type: "qwen"
+    description: "Qwen2.5 14B instruction-tuned model for better text generation"
+    supported_quantizations:
+      - "non-quantized(fp16)"
+      - "quantized(8bit)"
+    default_quantization: "quantized(8bit)"
+
+# Default model selection
+default_model: "InternVL3-8B" 

debug_files.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""
+File Upload Diagnostic Script
+This script helps debug why some images are not being processed.
+"""
+
+import os
+from pathlib import Path
+from typing import Dict, List
+
+def analyze_uploaded_files(folder_path: str) -> None:
+    """
+    Analyze uploaded files to understand why some images might not be processed.
+    
+    Args:
+        folder_path: Path to the uploaded folder
+    """
+    print("🔍 File Upload Diagnostic Tool")
+    print("=" * 50)
+    
+    if not os.path.exists(folder_path):
+        print(f"❌ Folder not found: {folder_path}")
+        return
+    
+    # Get all files in the folder
+    all_files = []
+    for root, dirs, files in os.walk(folder_path):
+        for file in files:
+            full_path = os.path.join(root, file)
+            all_files.append(Path(full_path))
+    
+    print(f"📁 Total files found: {len(all_files)}")
+    print("\n📋 All files:")
+    for i, file_path in enumerate(all_files, 1):
+        print(f"  {i}. {file_path.name} (ext: {file_path.suffix.lower()})")
+    
+    # Analyze image files
+    image_exts = ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff']
+    print(f"\n🖼️  Looking for image extensions: {image_exts}")
+    
+    image_files = []
+    non_image_files = []
+    
+    for file_path in all_files:
+        if any(file_path.suffix.lower().endswith(ext) for ext in image_exts):
+            image_files.append(file_path)
+        else:
+            non_image_files.append(file_path)
+    
+    print(f"\n✅ Image files detected ({len(image_files)}):")
+    for i, img in enumerate(image_files, 1):
+        print(f"  {i}. {img.name}")
+    
+    print(f"\n📄 Non-image files ({len(non_image_files)}):")
+    for i, file in enumerate(non_image_files, 1):
+        print(f"  {i}. {file.name} (ext: {file.suffix.lower()})")
+    
+    # Check for CSV files
+    csv_files = [f for f in all_files if f.suffix.lower() == '.csv']
+    print(f"\n📊 CSV files found ({len(csv_files)}):")
+    for i, csv in enumerate(csv_files, 1):
+        print(f"  {i}. {csv.name}")
+    
+    # If CSV exists, check its content
+    if csv_files:
+        try:
+            import pandas as pd
+            df = pd.read_csv(csv_files[0])
+            print(f"\n📈 CSV Analysis for '{csv_files[0].name}':")
+            print(f"  - Rows: {len(df)}")
+            print(f"  - Columns: {list(df.columns)}")
+            
+            if 'Image Name' in df.columns:
+                image_names_in_csv = df['Image Name'].tolist()
+                print(f"  - Image names in CSV: {len(image_names_in_csv)}")
+                
+                # Check which images from CSV actually exist as files
+                existing_images = []
+                missing_images = []
+                
+                for img_name in image_names_in_csv:
+                    if any(img.name == img_name for img in image_files):
+                        existing_images.append(img_name)
+                    else:
+                        missing_images.append(img_name)
+                
+                print(f"\n🔗 CSV-to-File Matching:")
+                print(f"  - Images in CSV that exist as files: {len(existing_images)}")
+                print(f"  - Images in CSV that are missing: {len(missing_images)}")
+                
+                if existing_images:
+                    print("  ✅ Matching files:")
+                    for img in existing_images:
+                        print(f"    - {img}")
+                
+                if missing_images:
+                    print("  ❌ Missing files:")
+                    for img in missing_images:
+                        print(f"    - {img}")
+                        
+        except Exception as e:
+            print(f"  ❌ Error reading CSV: {e}")
+    
+    # Summary
+    print(f"\n📊 SUMMARY:")
+    print(f"  - Total files uploaded: {len(all_files)}")
+    print(f"  - Image files detected: {len(image_files)}")
+    print(f"  - CSV files: {len(csv_files)}")
+    
+    if csv_files and 'df' in locals():
+        if 'Image Name' in df.columns:
+            print(f"  - Images that will be processed: {len(existing_images)}")
+        else:
+            print(f"  - CSV exists but no 'Image Name' column - will process all {len(image_files)} images")
+    else:
+        print(f"  - No CSV - will process all {len(image_files)} images")
+
+if __name__ == "__main__":
+    print("Please provide the path to your uploaded folder:")
+    folder_path = input("Folder path: ").strip()
+    analyze_uploaded_files(folder_path) 

frontend/__init__.py

@@ -0,0 +1,3 @@
+from .gradio_app import GradioApp
+
+__all__ = ['GradioApp'] 

frontend/gradio_app.py

@@ -0,0 +1,487 @@
+import gradio as gr
+import pandas as pd
+import os
+from pathlib import Path
+import shutil
+import tempfile
+import uuid
+import spaces
+from typing import Optional
+
+from backend import ConfigManager, ModelManager, InferenceEngine
+from backend.utils.metrics import create_accuracy_table, save_dataframe_to_csv
+
+
+class GradioApp:
+    """Gradio application for InternVL3 prompt engineering."""
+    
+    def __init__(self):
+        """Initialize the Gradio application."""
+        # Initialize backend components
+        self.config_manager = ConfigManager()
+        self.model_manager = ModelManager(self.config_manager)
+        self.inference_engine = InferenceEngine(self.model_manager, self.config_manager)
+        
+        # Try to preload default model
+        try:
+            self.model_manager.preload_default_model()
+            print("✅ Default model preloaded successfully!")
+        except Exception as e:
+            print(f"⚠️ Default model preloading failed: {str(e)}")
+            print("The model will be loaded when first needed.")
+    
+    def get_current_model_status(self) -> str:
+        """Get current model status for display."""
+        return self.model_manager.get_current_model_status()
+    
+    def handle_stop_button(self):
+        """Handle stop button click."""
+        message = self.inference_engine.set_stop_flag()
+        return message, gr.update(visible=True)
+    
+    def on_model_change(self, model_selection: str, quantization_type: str) -> str:
+        """Handle model/quantization dropdown changes."""
+        current_status = self.get_current_model_status()
+        if model_selection and quantization_type:
+            available_models = self.config_manager.get_available_models()
+            target_id = available_models.get(model_selection)
+            current_model_id = None
+            if self.model_manager.current_model:
+                current_model_id = self.model_manager.current_model.model_id
+            
+            if (current_model_id != target_id or 
+                (self.model_manager.current_model and 
+                 self.model_manager.current_model.current_quantization != quantization_type)):
+                return f"🔄 Will load {model_selection} with {quantization_type} when processing starts"
+        return current_status
+    
+    def get_model_choices_with_info(self) -> list[str]:
+        """Get model choices with type information for dropdown."""
+        choices = []
+        for model_name in self.config_manager.get_available_models().keys():
+            model_config = self.config_manager.get_model_config(model_name)
+            model_type = model_config.get('model_type', 'unknown').upper()
+            choices.append(f"{model_name} ({model_type})")
+        return choices
+    
+    def extract_model_name_from_choice(self, choice: str) -> str:
+        """Extract the actual model name from the dropdown choice."""
+        return choice.split(' (')[0] if ' (' in choice else choice
+    
+    def update_image_preview(self, evt: gr.SelectData, df, folder_path):
+        """Update image preview when table row is selected."""
+        if df is None or evt.index[0] >= len(df):
+            return None, ""
+        try:
+            # Use the full dataframe with image paths
+            full_df = getattr(self.inference_engine, 'full_df', None)
+            if full_df is None or evt.index[0] >= len(full_df):
+                return None, ""
+            selected_row = full_df.iloc[evt.index[0]]
+            image_path = selected_row["Image Path"]
+            model_output = selected_row["Model Output"]
+            if not os.path.exists(image_path):
+                return None, model_output
+            file_extension = Path(image_path).suffix
+            temp_filename = f"gradio_preview_{uuid.uuid4().hex}{file_extension}"
+            temp_path = os.path.join(tempfile.gettempdir(), temp_filename)
+            shutil.copy2(image_path, temp_path)
+            return temp_path, model_output
+        except Exception as e:
+            print(f"Error loading image preview: {e}")
+            return None, ""
+    
+    def download_results_csv(self, results_table_data):
+        """Download results as CSV file."""
+        try:
+            print(f"Download function called with data type: {type(results_table_data)}")
+            
+            if results_table_data is None:
+                print("No data to download")
+                return None
+                
+            # Handle different data types from Gradio
+            if hasattr(results_table_data, 'values'):
+                # If it's a pandas DataFrame
+                df = results_table_data
+            elif isinstance(results_table_data, list):
+                # If it's a list of lists or list of dicts
+                if len(results_table_data) == 0:
+                    print("Empty data")
+                    return None
+                df = pd.DataFrame(results_table_data, columns=["S.No", "Image Name", "Ground Truth", "Binary Output", "Model Output"])
+            else:
+                # Try to convert to DataFrame
+                df = pd.DataFrame(results_table_data)
+            
+            print(f"DataFrame shape: {df.shape}")
+            print(f"DataFrame columns: {df.columns.tolist()}")
+            
+            # Create temporary file
+            temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False)
+            df.to_csv(temp_file.name, index=False)
+            temp_file.close()
+            
+            print(f"CSV file created: {temp_file.name}")
+            return temp_file.name
+            
+        except Exception as e:
+            print(f"Error in download_results_csv: {str(e)}")
+            import traceback
+            traceback.print_exc()
+            return None
+    
+    def submit_and_show_metrics(self, df):
+        """Generate and show metrics for results."""
+        if df is None:
+            return df, df, None, None, None, gr.update(visible=False), gr.update(visible=False), ""
+        
+        # Only create metrics if all outputs are valid yes/no responses
+        try:
+            metrics_df, cm_plot_path, cm_values = create_accuracy_table(df)
+            return df, df, metrics_df, cm_plot_path, cm_values, gr.update(visible=True), gr.update(visible=True), "📊 Metrics calculated successfully!"
+        except Exception as e:
+            print(f"Could not create metrics: {str(e)}")
+            return df, df, None, None, None, gr.update(visible=False), gr.update(visible=True), f"⚠️ Could not calculate metrics: {str(e)}"
+    
+    @spaces.GPU
+    def process_input_ui(self, folder_path, prompt, quantization_type, model_selection):
+        """UI wrapper for processing input with progress updates."""
+        if not folder_path or not prompt.strip():
+            return (gr.update(visible=True), gr.update(visible=False), gr.update(visible=False),
+                   "Please upload a folder and enter a prompt.", None, None, None, 
+                   gr.update(visible=False), gr.update(visible=False), 
+                   gr.update(value="⚠️ Please upload a folder and enter a prompt.", visible=True), "", gr.update(visible=False))
+        
+        # Extract actual model name from the dropdown choice
+        actual_model_name = self.extract_model_name_from_choice(model_selection)
+        
+        # Check if model needs to be downloaded and show progress
+        available_models = self.config_manager.get_available_models()
+        model_id = available_models[actual_model_name]
+        
+        # Show processing message and hide stop status
+        yield (gr.update(visible=False), gr.update(visible=False), gr.update(visible=False),
+               None, None, None, None, 
+               gr.update(visible=False), gr.update(visible=False), 
+               gr.update(value="🚀 Initializing processing...", visible=True), prompt, gr.update(visible=False))
+        
+        # Process the input
+        error, show_results, show_image, table, error_message, final_message = self.inference_engine.process_folder_input(
+            folder_path, prompt, quantization_type, actual_model_name, gr.Progress()
+        )
+        
+        # If error is visible, show results section but keep error visible
+        if error["visible"]:
+            yield (gr.update(visible=False), gr.update(visible=True), gr.update(visible=True),
+                   error, None, None, None, 
+                   gr.update(visible=False), gr.update(visible=False), 
+                   gr.update(value=final_message, visible=True), prompt, gr.update(visible=False))
+        else:
+            yield (gr.update(visible=False), gr.update(visible=True), gr.update(visible=True),
+                   None, show_results, show_image, table, 
+                   gr.update(visible=True), gr.update(visible=False), 
+                   gr.update(value=final_message, visible=True), prompt, gr.update(visible=False))
+    
+    def rerun_ui(self, df, new_prompt, quantization_type, model_selection):
+        """UI wrapper for rerun with progress updates."""
+        if df is None or not new_prompt.strip():
+            return (df, None, None, None, 
+                   gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), 
+                   gr.update(visible=False), gr.update(visible=True), "⚠️ Please provide a valid prompt", "")
+        
+        # Extract actual model name from the dropdown choice
+        actual_model_name = self.extract_model_name_from_choice(model_selection)
+        
+        # Hide all sections and show only processing, clear model output display
+        yield (df, None, None, None, 
+               gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), 
+               gr.update(visible=False), gr.update(visible=True), "🚀 Initializing reprocessing...", "Select a row from the table to see model output...")
+        
+        # Process with new prompt
+        updated_df, accuracy_table_data, cm_plot, cm_values, section4_vis, progress_vis, final_message = self.inference_engine.rerun_with_new_prompt(
+            df, new_prompt, quantization_type, actual_model_name, gr.Progress()
+        )
+        
+        # Show prompt editing and results sections again, show Generate Metrics button, hide progress, and clear model output display
+        yield (updated_df, accuracy_table_data, cm_plot, cm_values, 
+               gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), section4_vis, 
+               gr.update(visible=True), gr.update(visible=False), final_message, "Select a row from the table to see updated model output...")
+    
+    def create_interface(self):
+        """Create and return the Gradio interface."""
+        # CSS from original app.py
+        css = """
+        .progress {
+            margin: 15px 0;
+            padding: 20px;
+            border-radius: 12px;
+            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+            border: none;
+            color: white;
+            font-weight: 600;
+            font-size: 16px;
+            text-align: center;
+            box-shadow: 0 4px 15px rgba(102, 126, 234, 0.3);
+            animation: progressPulse 2s ease-in-out infinite alternate;
+        }
+
+        @keyframes progressPulse {
+            0% { 
+                transform: scale(1);
+                box-shadow: 0 4px 15px rgba(102, 126, 234, 0.3);
+            }
+            100% { 
+                transform: scale(1.02);
+                box-shadow: 0 6px 20px rgba(102, 126, 234, 0.4);
+            }
+        }
+
+        .processing {
+            background: linear-gradient(45deg, #f0f9ff, #e3f2fd);
+            border: 2px solid #1976d2;
+            border-radius: 10px;
+            padding: 20px;
+            text-align: center;
+            margin: 10px 0;
+        }
+
+        .gr-button.processing {
+            background-color: #ffa726 !important;
+            color: white !important;
+            pointer-events: none;
+        }
+
+        /* Stop button styling */
+        .stop-button {
+            background: linear-gradient(135deg, #ff4757 0%, #c44569 100%) !important;
+            border: none !important;
+            color: white !important;
+            font-weight: 700 !important;
+            font-size: 16px !important;
+            box-shadow: 0 4px 15px rgba(255, 71, 87, 0.4) !important;
+            transition: all 0.3s ease !important;
+        }
+
+        .stop-button:hover {
+            transform: translateY(-2px) !important;
+            box-shadow: 0 8px 25px rgba(255, 71, 87, 0.6) !important;
+            background: linear-gradient(135deg, #ff3742 0%, #b83754 100%) !important;
+        }
+
+        .stop-status {
+            color: #ff4757;
+            font-weight: 600;
+            background: rgba(255, 71, 87, 0.1);
+            padding: 10px;
+            border-radius: 8px;
+            border-left: 4px solid #ff4757;
+            margin: 10px 0;
+        }
+
+        /* Enhanced button styling */
+        .gr-button {
+            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+            border: none;
+            border-radius: 8px;
+            color: white;
+            font-weight: 600;
+            transition: all 0.3s ease;
+        }
+
+        .gr-button:hover {
+            transform: translateY(-2px);
+            box-shadow: 0 8px 25px rgba(102, 126, 234, 0.4);
+        }
+        """
+        
+        with gr.Blocks(theme="origin", css=css) as demo:
+            gr.Markdown("""
+            <h1 style='text-align:center; color:#1976d2; font-size:2.5em; font-weight:bold; margin-bottom:40px!important;'>PROMPT_PILOT</h1>
+            <p style='text-align:center; color:#666; font-size:1.1em; margin-bottom:30px;'>
+            🤖 AI-powered analysis with different vision models
+            </p>
+            <h2 style='text-align:center; color:#666; font-size:1.1em; margin-bottom:30px;'>
+            Note: Currently Accuracy only works properly in case of binary output. For other cases kindly download the csv and calculate the accuracy separately. 
+            </h2>
+            """, elem_id="main-title")
+            
+            # Model and Quantization selection dropdowns at the top
+            model_choices = self.get_model_choices_with_info()
+            default_choice = f"{self.config_manager.get_default_model()} (INTERNVL)"
+            
+            with gr.Row():
+                model_dropdown = gr.Dropdown(
+                    choices=model_choices,
+                    value=default_choice,
+                    label="🤖 Model Selection",
+                    info="Select model: InternVL (vision+text), Qwen (text-only)",
+                    elem_id="model-dropdown"
+                )
+                quantization_dropdown = gr.Dropdown(
+                    choices=["quantized(8bit)", "non-quantized(fp16)"],
+                    value="non-quantized(fp16)",
+                    label="🔧 Model Quantization",
+                    info="Select quantization type: quantized (8bit) uses less memory, non-quantized (fp16) for better quality",
+                    elem_id="quantization-dropdown"
+                )
+            
+            # Model status indicator
+            with gr.Row():
+                model_status = gr.Markdown(
+                    value=self.get_current_model_status(),
+                    label="Model Status",
+                    elem_classes=["model-status"]
+                )
+            
+            # Stop button row
+            with gr.Row():
+                stop_btn = gr.Button("🛑 STOP PROCESSING", variant="stop", size="lg", elem_classes=["stop-button"])
+                stop_status = gr.Markdown("", elem_classes=["stop-status"], visible=False)
+            
+            with gr.Row(visible=True) as section1_row:
+                with gr.Column():
+                    folder_input = gr.File(
+                        label="Upload Folder",
+                        file_count="directory",
+                        type="filepath"
+                    )
+                with gr.Column():
+                    prompt_input = gr.Textbox(
+                        label="Enter your prompt here",
+                        placeholder="Type your prompt...",
+                        lines=3
+                    )
+                with gr.Column():
+                    submit_btn = gr.Button("Proceed", variant="primary")
+            
+            # Progress indicator for section 1
+            with gr.Row(visible=True) as section1_progress_row:
+                section1_progress_message = gr.Markdown("", elem_classes=["progress"], visible=False)
+            
+            # Section 2: Edit Prompt and Rerun Controls (separate section)
+            with gr.Row(visible=False) as section2_prompt_row:
+                with gr.Column():
+                    with gr.Row():
+                        prompt_input_section2 = gr.Textbox(
+                            label="Edit Prompt",
+                            placeholder="Modify your prompt here...",
+                            lines=2,
+                            scale=4
+                        )
+                        rerun_btn = gr.Button("🔄 Rerun", variant="secondary", size="lg", scale=1)
+            
+            # Section 3: Results Display
+            with gr.Row(visible=False) as section3_results_row:
+                error_message = gr.Textbox(label="Error Message", visible=False)
+                with gr.Column(scale=1):
+                    image_preview = gr.Image(label="Selected Image", height=270, width=480)
+                    model_output_display = gr.Textbox(
+                        label="Model Output for Selected Image",
+                        placeholder="Select a row from the table to see model output...",
+                        interactive=False,
+                        lines=3
+                    )
+                with gr.Column(scale=2):
+                    with gr.Row():
+                        gr.HTML("")  # Empty space to push button to right
+                        download_results_btn = gr.Button("📥 CSV", size="sm", scale=1)
+                        results_csv_output = gr.File(label="", visible=True, scale=1, show_label=False)
+                    results_table = gr.Dataframe(
+                        headers=["S.No", "Image Name", "Ground Truth", "Binary Output", "Model Output"],
+                        label="Results",
+                        interactive=True,  # Make it editable for ground truth input
+                        col_count=(5, "fixed")
+                    )
+            
+            # Generate Metrics button
+            with gr.Row(visible=False) as section3_submit_row:
+                with gr.Column():
+                    submit_results_btn = gr.Button("Generate Metrics", variant="primary", size="lg")
+            
+            # Progress indicator row
+            with gr.Row(visible=False) as progress_row:
+                progress_message = gr.Markdown("", elem_classes=["progress"])
+            
+            # Section 4: Metrics and confusion matrix
+            with gr.Row(visible=False) as section4_metrics_row:
+                with gr.Column(scale=2):
+                    confusion_matrix_plot = gr.Image(
+                        label="Confusion Matrix"
+                    )
+                with gr.Column(scale=2):
+                    accuracy_table = gr.Dataframe(
+                        label="Performance Metrics",
+                        interactive=False
+                    )
+                    confusion_matrix_table = gr.Dataframe(
+                        label="Confusion Matrix Table",
+                        interactive=False
+                    )
+            
+            # State to store folder path
+            folder_path_state = gr.State()
+            folder_input.change(
+                fn=lambda x: x,
+                inputs=[folder_input],
+                outputs=[folder_path_state]
+            )
+            
+            # Event handlers
+            submit_btn.click(
+                fn=self.process_input_ui,
+                inputs=[folder_input, prompt_input, quantization_dropdown, model_dropdown],
+                outputs=[section1_row, section2_prompt_row, section3_results_row, error_message, results_table, image_preview, results_table, section3_submit_row, section4_metrics_row, section1_progress_message, prompt_input_section2, stop_status]
+            )
+            
+            results_table.select(
+                fn=self.update_image_preview,
+                inputs=[results_table, folder_path_state],
+                outputs=[image_preview, model_output_display]
+            )
+            
+            submit_results_btn.click(
+                fn=self.submit_and_show_metrics,
+                inputs=[results_table],
+                outputs=[results_table, results_table, accuracy_table, confusion_matrix_plot, confusion_matrix_table, section4_metrics_row, progress_row, progress_message]
+            )
+            
+            download_results_btn.click(
+                fn=self.download_results_csv,
+                inputs=[results_table],
+                outputs=[results_csv_output]
+            )
+            
+            rerun_btn.click(
+                fn=self.rerun_ui,
+                inputs=[results_table, prompt_input_section2, quantization_dropdown, model_dropdown],
+                outputs=[results_table, accuracy_table, confusion_matrix_plot, confusion_matrix_table, 
+                        section1_row, section2_prompt_row, section3_results_row, section4_metrics_row, section3_submit_row, progress_row, progress_message, model_output_display]
+            )
+            
+            # Model change handler to update status
+            model_dropdown.change(
+                fn=self.on_model_change,
+                inputs=[model_dropdown, quantization_dropdown],
+                outputs=[model_status]
+            )
+            
+            quantization_dropdown.change(
+                fn=self.on_model_change,
+                inputs=[model_dropdown, quantization_dropdown],
+                outputs=[model_status]
+            )
+            
+            # Stop button click handler
+            stop_btn.click(
+                fn=self.handle_stop_button,
+                inputs=[],
+                outputs=[stop_status, stop_status]
+            )
+        
+        return demo
+    
+    def launch(self, **kwargs):
+        """Launch the Gradio application."""
+        demo = self.create_interface()
+        return demo.launch(**kwargs) 

requirements.txt

@@ -0,0 +1,25 @@
+numpy
+Pillow
+Requests
+torch
+torchvision
+decord
+git+https://github.com/huggingface/transformers.git
+accelerate
+einops
+timm
+sentencepiece
+gradio>=4.19.2
+torch>=2.2.0
+torchvision>=0.17.0
+transformers>=4.37.2
+pillow>=10.2.0
+accelerate>=0.27.2
+bitsandbytes>=0.42.0
+pandas>=1.5.0
+matplotlib>=3.5.0
+seaborn>=0.11.0
+scikit-learn>=1.0.0
+pyyaml>=6.0.0
+spaces
+boto3