test_balanced_detection.py

#!/usr/bin/env python3
"""
Test script for balanced damage detection that checks both sensitivity and specificity.

This script helps evaluate if the detection system properly balances:
- Sensitivity: Detecting actual damage when it exists
- Specificity: Not detecting damage in undamaged areas

Author: AI Assistant
Date: June 2025
"""

import os
import sys
import logging
from pathlib import Path
import cv2
import numpy as np
from PIL import Image
import matplotlib.pyplot as plt
import json

# Add the current directory to Python path
sys.path.append(str(Path(__file__).parent))

from advanced_damage_segmentation import AdvancedDamageSegmentation
from pretrained_damage_detector import PretrainedDamageDetector

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class BalancedDetectionTester:
    """
    Test system for evaluating the balance between sensitivity and specificity
    in damage detection algorithms.
    """
    
    def __init__(self):
        """Initialize the testing system."""
        self.basic_detector = PretrainedDamageDetector()
        self.advanced_detector = AdvancedDamageSegmentation()
        self.test_results = {
            'basic': {'true_positives': 0, 'false_positives': 0, 'true_negatives': 0, 'false_negatives': 0},
            'advanced': {'true_positives': 0, 'false_positives': 0, 'true_negatives': 0, 'false_negatives': 0}
        }
    
    def create_test_images(self):
        """Create synthetic test images to evaluate detection performance."""
        test_dir = Path("test_images")
        test_dir.mkdir(exist_ok=True)
        
        # Create undamaged poster-like image
        undamaged = self._create_clean_poster()
        cv2.imwrite(str(test_dir / "undamaged_poster.jpg"), undamaged)
        
        # Create images with different types of simulated damage
        damaged_tear = self._add_simulated_tear(undamaged.copy())
        cv2.imwrite(str(test_dir / "damaged_tear.jpg"), damaged_tear)
        
        damaged_stain = self._add_simulated_stain(undamaged.copy())
        cv2.imwrite(str(test_dir / "damaged_stain.jpg"), damaged_stain)
        
        damaged_hole = self._add_simulated_hole(undamaged.copy())
        cv2.imwrite(str(test_dir / "damaged_hole.jpg"), damaged_hole)
        
        # Create poster with normal content that shouldn't be detected as damage
        poster_with_text = self._create_poster_with_content()
        cv2.imwrite(str(test_dir / "poster_with_text.jpg"), poster_with_text)
        
        logger.info("✅ Test images created in test_images/ directory")
        
        return {
            'undamaged': [test_dir / "undamaged_poster.jpg", test_dir / "poster_with_text.jpg"],
            'damaged': [test_dir / "damaged_tear.jpg", test_dir / "damaged_stain.jpg", test_dir / "damaged_hole.jpg"]
        }
    
    def _create_clean_poster(self):
        """Create a clean poster-like image."""
        # Create base poster
        height, width = 400, 300
        poster = np.ones((height, width, 3), dtype=np.uint8) * 240  # Light gray background
        
        # Add some subtle texture
        noise = np.random.normal(0, 5, (height, width, 3))
        poster = np.clip(poster + noise, 0, 255).astype(np.uint8)
        
        # Add a border
        cv2.rectangle(poster, (10, 10), (width-10, height-10), (200, 200, 200), 2)
        
        return poster
    
    def _create_poster_with_content(self):
        """Create a poster with text and graphics that shouldn't be detected as damage."""
        poster = self._create_clean_poster()
        height, width = poster.shape[:2]
        
        # Add title text
        cv2.putText(poster, "POSTER TITLE", (30, 50), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 0), 2)
        
        # Add some geometric shapes (normal poster content)
        cv2.rectangle(poster, (50, 80), (150, 120), (100, 150, 200), -1)
        cv2.circle(poster, (200, 150), 30, (200, 100, 100), -1)
        
        # Add more text
        cv2.putText(poster, "Event Date: June 2025", (30, 200), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (50, 50, 50), 1)
        cv2.putText(poster, "Location: Test Venue", (30, 220), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (50, 50, 50), 1)
        
        return poster
    
    def _add_simulated_tear(self, image):
        """Add a simulated tear to the image."""
        height, width = image.shape[:2]
        
        # Create irregular tear shape
        points = []
        start_x = width // 3
        start_y = height // 4
        
        for i in range(10):
            x = start_x + i * 5 + np.random.randint(-3, 4)
            y = start_y + i * 8 + np.random.randint(-4, 5)
            points.append([x, y])
        
        points = np.array(points, dtype=np.int32)
        
        # Make the tear very dark
        cv2.fillPoly(image, [points], (10, 10, 10))
        
        # Add some edge highlighting to simulate torn paper
        cv2.polylines(image, [points], False, (50, 50, 50), 2)
        
        return image
    
    def _add_simulated_stain(self, image):
        """Add a simulated stain to the image."""
        height, width = image.shape[:2]
        
        # Create irregular stain shape
        center = (width // 2, height // 2)
        axes = (40, 25)
        
        # Create mask for stain
        mask = np.zeros((height, width), dtype=np.uint8)
        cv2.ellipse(mask, center, axes, 0, 0, 360, 255, -1)
        
        # Add irregularity to stain
        kernel = np.ones((5, 5), np.uint8)
        mask = cv2.erode(mask, kernel, iterations=1)
        
        # Apply brownish stain color
        stain_color = [60, 80, 120]  # Brownish color
        
        # Blend stain with original image
        for i in range(3):
            image[:, :, i] = np.where(mask > 0, 
                                    image[:, :, i] * 0.4 + stain_color[i] * 0.6,
                                    image[:, :, i])
        
        return image
    
    def _add_simulated_hole(self, image):
        """Add a simulated hole to the image."""
        height, width = image.shape[:2]
        
        center = (width // 4, height // 3)
        radius = 15
        
        # Create hole (very dark region)
        cv2.circle(image, center, radius, (5, 5, 5), -1)
        
        # Add torn edges around hole
        cv2.circle(image, center, radius + 2, (30, 30, 30), 2)
        
        return image
    
    def test_single_image(self, image_path, expected_damaged=True):
        """Test both detectors on a single image."""
        logger.info(f"Testing: {Path(image_path).name}")
        
        # Test basic detector
        basic_result = self.basic_detector.detect_damage(str(image_path))
        basic_detected = basic_result['damage_assessment']['is_damaged']
        
        # Test advanced detector
        advanced_result = self.advanced_detector.segment_damage(str(image_path))
        advanced_detected = advanced_result['damage_percentage'] > 0.5  # Threshold for detection
        
        # Update metrics
        for detector_name, detected in [('basic', basic_detected), ('advanced', advanced_detected)]:
            if expected_damaged and detected:
                self.test_results[detector_name]['true_positives'] += 1
            elif expected_damaged and not detected:
                self.test_results[detector_name]['false_negatives'] += 1
            elif not expected_damaged and detected:
                self.test_results[detector_name]['false_positives'] += 1
            elif not expected_damaged and not detected:
                self.test_results[detector_name]['true_negatives'] += 1
        
        logger.info(f"  Basic detector: {'DAMAGED' if basic_detected else 'UNDAMAGED'}")
        logger.info(f"  Advanced detector: {'DAMAGED' if advanced_detected else 'UNDAMAGED'} ({advanced_result['damage_percentage']:.1f}%)")
        logger.info(f"  Expected: {'DAMAGED' if expected_damaged else 'UNDAMAGED'}")
        
        return {
            'basic': basic_detected,
            'advanced': advanced_detected,
            'basic_confidence': basic_result['damage_assessment']['confidence'],
            'advanced_percentage': advanced_result['damage_percentage']
        }
    
    def run_comprehensive_test(self):
        """Run comprehensive test on all test images."""
        logger.info("🧪 Starting comprehensive damage detection test...")
        
        # Create test images
        test_images = self.create_test_images()
        
        results = []
        
        # Test undamaged images
        logger.info("\\n📋 Testing undamaged images (should NOT detect damage):")
        for image_path in test_images['undamaged']:
            result = self.test_single_image(image_path, expected_damaged=False)
            result['image'] = image_path.name
            result['expected'] = 'undamaged'
            results.append(result)
        
        # Test damaged images  
        logger.info("\\n📋 Testing damaged images (should detect damage):")
        for image_path in test_images['damaged']:
            result = self.test_single_image(image_path, expected_damaged=True)
            result['image'] = image_path.name
            result['expected'] = 'damaged'
            results.append(result)
        
        # Calculate and display metrics
        self._display_performance_metrics()
        
        # Save detailed results
        self._save_test_results(results)
        
        return results
    
    def _calculate_metrics(self, detector_results):
        """Calculate precision, recall, and F1 score."""
        tp = detector_results['true_positives']
        fp = detector_results['false_positives']
        tn = detector_results['true_negatives']
        fn = detector_results['false_negatives']
        
        # Avoid division by zero
        precision = tp / (tp + fp) if (tp + fp) > 0 else 0
        recall = tp / (tp + fn) if (tp + fn) > 0 else 0
        specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
        accuracy = (tp + tn) / (tp + fp + tn + fn) if (tp + fp + tn + fn) > 0 else 0
        f1_score = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
        
        return {
            'precision': precision,
            'recall': recall,
            'specificity': specificity,
            'accuracy': accuracy,
            'f1_score': f1_score,
            'true_positives': tp,
            'false_positives': fp,
            'true_negatives': tn,
            'false_negatives': fn
        }
    
    def _display_performance_metrics(self):
        """Display performance metrics for both detectors."""
        logger.info("\\n" + "="*60)
        logger.info("📊 PERFORMANCE METRICS")
        logger.info("="*60)
        
        for detector_name in ['basic', 'advanced']:
            metrics = self._calculate_metrics(self.test_results[detector_name])
            
            logger.info(f"\\n🔍 {detector_name.upper()} DETECTOR:")
            logger.info(f"  Accuracy:    {metrics['accuracy']:.2f} ({metrics['accuracy']*100:.1f}%)")
            logger.info(f"  Precision:   {metrics['precision']:.2f} (of detected damage, how much was real)")
            logger.info(f"  Recall:      {metrics['recall']:.2f} (of real damage, how much was detected)")
            logger.info(f"  Specificity: {metrics['specificity']:.2f} (of undamaged, how much correctly identified)")
            logger.info(f"  F1 Score:    {metrics['f1_score']:.2f} (overall balance)")
            
            logger.info(f"  ✅ True Positives:  {metrics['true_positives']} (correct damage detection)")
            logger.info(f"  ❌ False Positives: {metrics['false_positives']} (false alarms)")
            logger.info(f"  ✅ True Negatives:  {metrics['true_negatives']} (correct undamaged detection)")
            logger.info(f"  ❌ False Negatives: {metrics['false_negatives']} (missed damage)")
    
    def _save_test_results(self, results):
        """Save detailed test results to file."""
        output_file = "test_results_balanced_detection.json"
        
        # Calculate metrics for saving
        basic_metrics = self._calculate_metrics(self.test_results['basic'])
        advanced_metrics = self._calculate_metrics(self.test_results['advanced'])
        
        full_results = {
            'summary': {
                'basic_detector': basic_metrics,
                'advanced_detector': advanced_metrics
            },
            'individual_results': results,
            'raw_counts': self.test_results
        }
        
        with open(output_file, 'w') as f:
            json.dump(full_results, f, indent=2, default=str)
        
        logger.info(f"\\n💾 Detailed results saved to: {output_file}")
    
    def test_existing_images(self, input_dir="input"):
        """Test the detectors on existing images in the input directory."""
        input_path = Path(input_dir)
        
        if not input_path.exists():
            logger.warning(f"Input directory {input_path} does not exist")
            return
        
        # Find image files
        image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.webp'}
        image_files = [f for f in input_path.iterdir() 
                      if f.is_file() and f.suffix.lower() in image_extensions]
        
        if not image_files:
            logger.warning("No image files found in input directory")
            return
        
        logger.info(f"\\n🖼️  Testing {len(image_files)} existing images...")
        
        for image_file in image_files:
            logger.info(f"\\n📸 Testing: {image_file.name}")
            
            # Basic detector
            basic_result = self.basic_detector.detect_damage(str(image_file))
            basic_detected = basic_result['damage_assessment']['is_damaged']
            basic_confidence = basic_result['damage_assessment']['confidence']
            
            # Advanced detector
            advanced_result = self.advanced_detector.segment_damage(str(image_file))
            advanced_percentage = advanced_result['damage_percentage']
            
            logger.info(f"  Basic:    {'DAMAGED' if basic_detected else 'UNDAMAGED'} (confidence: {basic_confidence:.2f})")
            logger.info(f"  Advanced: {advanced_percentage:.1f}% damaged")
            
            if advanced_result['damage_regions']:
                damage_types = set(r['type'] for r in advanced_result['damage_regions'])
                logger.info(f"  Types:    {', '.join(damage_types)}")


def main():
    """Main function to run the balanced detection test."""
    logger.info("🎯 Balanced Damage Detection Tester")
    logger.info("="*50)
    
    tester = BalancedDetectionTester()
    
    # Run comprehensive test with synthetic images
    logger.info("\\n1️⃣  Creating and testing synthetic images...")
    synthetic_results = tester.run_comprehensive_test()
    
    # Test existing images if available
    logger.info("\\n2️⃣  Testing existing images in input directory...")
    tester.test_existing_images()
    
    logger.info("\\n" + "="*60)
    logger.info("🎉 TESTING COMPLETE!")
    logger.info("="*60)
    logger.info("Check the following for detailed analysis:")
    logger.info("  📄 test_results_balanced_detection.json - Detailed metrics")
    logger.info("  📁 test_images/ - Generated test images")
    logger.info("  📁 output/ - Detection results and visualizations")


if __name__ == "__main__":
    main()