usage_examples.py

#!/usr/bin/env python3
"""
Usage Examples for Pretrained Damage Detector

This script provides practical examples of how to use the pretrained
damage detector in different scenarios and applications.

Author: AI Assistant
Date: June 2025
"""

import os
import sys
from pathlib import Path
import json

# Add the project root to the path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from pretrained_damage_detector import PretrainedDamageDetector


def example_1_basic_usage():
    """Example 1: Basic damage detection on a single image."""
    print("=" * 50)
    print("EXAMPLE 1: Basic Damage Detection")
    print("=" * 50)
    
    # Initialize the detector
    detector = PretrainedDamageDetector(device='auto')
    
    # Analyze an image (replace with your actual image path)
    image_path = "sample_poster.jpg"
    
    if os.path.exists(image_path):
        result = detector.detect_damage(image_path)
        
        print(f"Image: {image_path}")
        print(f"Is Damaged: {result['is_damaged']}")
        print(f"Confidence: {result['confidence']:.3f}")
        print(f"Primary Damage: {result['primary_damage_type']}")
    else:
        print(f"Sample image not found: {image_path}")
        print("Replace with an actual image path to test.")


def example_2_quality_control_system():
    """Example 2: Quality control system for poster printing."""
    print("\n" + "=" * 50)
    print("EXAMPLE 2: Quality Control System")
    print("=" * 50)
    
    class PosterQualityControl:
        def __init__(self):
            self.detector = PretrainedDamageDetector(device='auto')
            self.quality_threshold = 0.3  # Damage confidence threshold
            self.processed_count = 0
            self.rejected_count = 0
        
        def inspect_poster(self, image_path):
            """Inspect a single poster for quality issues."""
            result = self.detector.detect_damage(image_path)
            self.processed_count += 1
            
            # Determine quality status
            is_defective = result['confidence'] > self.quality_threshold
            
            if is_defective:
                self.rejected_count += 1
                status = "REJECTED"
                reason = result['primary_damage_type']
            else:
                status = "APPROVED"
                reason = "No significant damage detected"
            
            return {
                'image_path': image_path,
                'status': status,
                'reason': reason,
                'confidence': result['confidence'],
                'damage_details': result['damage_scores']
            }
        
        def generate_quality_report(self):
            """Generate quality control statistics."""
            approval_rate = ((self.processed_count - self.rejected_count) / 
                           self.processed_count * 100) if self.processed_count > 0 else 0
            
            return {
                'total_processed': self.processed_count,
                'approved': self.processed_count - self.rejected_count,
                'rejected': self.rejected_count,
                'approval_rate': f"{approval_rate:.1f}%"
            }
    
    # Demo usage
    qc_system = PosterQualityControl()
    
    # Simulate processing some posters
    sample_images = ["poster1.jpg", "poster2.jpg", "poster3.jpg"]
    
    print("Processing posters through quality control...")
    for image in sample_images:
        if os.path.exists(image):
            inspection_result = qc_system.inspect_poster(image)
            print(f"Poster: {image} - Status: {inspection_result['status']}")
        else:
            print(f"Sample image not found: {image}")
    
    # Generate report
    report = qc_system.generate_quality_report()
    print(f"\nQuality Control Report:")
    print(f"Total Processed: {report['total_processed']}")
    print(f"Approval Rate: {report['approval_rate']}")


def example_3_batch_archive_processing():
    """Example 3: Processing an archive of historical posters."""
    print("\n" + "=" * 50)
    print("EXAMPLE 3: Archive Processing")
    print("=" * 50)
    
    class ArchiveProcessor:
        def __init__(self):
            self.detector = PretrainedDamageDetector(device='auto')
        
        def process_archive(self, archive_directory):
            """Process all images in an archive directory."""
            if not os.path.exists(archive_directory):
                print(f"Archive directory not found: {archive_directory}")
                return
            
            # Get all image files
            image_extensions = {'.jpg', '.jpeg', '.png', '.bmp', '.tiff'}
            image_files = []
            
            for root, dirs, files in os.walk(archive_directory):
                for file in files:
                    if Path(file).suffix.lower() in image_extensions:
                        image_files.append(os.path.join(root, file))
            
            if not image_files:
                print(f"No images found in {archive_directory}")
                return
            
            print(f"Found {len(image_files)} images in archive")
            
            # Process in batches
            results = self.detector.batch_analyze(image_files)
            
            # Categorize results
            preservation_priority = {
                'critical': [],    # Severely damaged, needs immediate attention
                'moderate': [],    # Some damage, needs preservation
                'good': []         # Good condition
            }
            
            for result in results:
                confidence = result.get('confidence', 0)
                if confidence > 0.7:
                    preservation_priority['critical'].append(result)
                elif confidence > 0.3:
                    preservation_priority['moderate'].append(result)
                else:
                    preservation_priority['good'].append(result)
            
            # Generate preservation report
            print(f"\n📋 PRESERVATION PRIORITY REPORT")
            print(f"Critical (needs immediate attention): {len(preservation_priority['critical'])}")
            print(f"Moderate (schedule preservation): {len(preservation_priority['moderate'])}")
            print(f"Good condition: {len(preservation_priority['good'])}")
            
            # Save detailed report
            report_path = "archive_preservation_report.json"
            with open(report_path, 'w') as f:
                json.dump(preservation_priority, f, indent=2)
            print(f"📄 Detailed report saved to: {report_path}")
            
            return preservation_priority
    
    # Demo usage
    processor = ArchiveProcessor()
    
    # Process test archive (replace with actual directory)
    archive_dir = "historical_posters"
    if os.path.exists(archive_dir):
        processor.process_archive(archive_dir)
    else:
        print(f"Create directory '{archive_dir}' with poster images to test this example")


def example_4_real_time_monitoring():
    """Example 4: Real-time monitoring system for poster displays."""
    print("\n" + "=" * 50)
    print("EXAMPLE 4: Real-time Monitoring System")
    print("=" * 50)
    
    class PosterMonitoringSystem:
        def __init__(self):
            self.detector = PretrainedDamageDetector(device='auto')
            self.monitoring_log = []
            self.alert_threshold = 0.4
        
        def monitor_poster(self, poster_id, image_path):
            """Monitor a specific poster for damage changes."""
            if not os.path.exists(image_path):
                return None
            
            result = self.detector.detect_damage(image_path)
            
            # Create monitoring entry
            monitoring_entry = {
                'poster_id': poster_id,
                'timestamp': Path(image_path).stat().st_mtime,
                'image_path': image_path,
                'damage_detected': result['is_damaged'],
                'confidence': result['confidence'],
                'primary_damage': result['primary_damage_type'],
                'needs_attention': result['confidence'] > self.alert_threshold
            }
            
            self.monitoring_log.append(monitoring_entry)
            
            # Generate alert if needed
            if monitoring_entry['needs_attention']:
                self.generate_alert(monitoring_entry)
            
            return monitoring_entry
        
        def generate_alert(self, entry):
            """Generate maintenance alert for damaged poster."""
            print(f"🚨 MAINTENANCE ALERT 🚨")
            print(f"Poster ID: {entry['poster_id']}")
            print(f"Damage Type: {entry['primary_damage']}")
            print(f"Confidence: {entry['confidence']:.3f}")
            print(f"Recommended Action: Immediate inspection and repair")
            print("-" * 30)
        
        def get_monitoring_summary(self):
            """Get summary of all monitored posters."""
            total_posters = len(self.monitoring_log)
            damaged_posters = sum(1 for entry in self.monitoring_log if entry['damage_detected'])
            critical_posters = sum(1 for entry in self.monitoring_log if entry['needs_attention'])
            
            return {
                'total_monitored': total_posters,
                'damaged_count': damaged_posters,
                'critical_count': critical_posters,
                'damage_rate': f"{(damaged_posters/total_posters*100):.1f}%" if total_posters > 0 else "0%"
            }
    
    # Demo usage
    monitoring_system = PosterMonitoringSystem()
    
    # Simulate monitoring some posters
    poster_locations = [
        ("POSTER_001", "display_area_1.jpg"),
        ("POSTER_002", "display_area_2.jpg"),
        ("POSTER_003", "display_area_3.jpg")
    ]
    
    print("Starting poster monitoring...")
    for poster_id, image_path in poster_locations:
        if os.path.exists(image_path):
            entry = monitoring_system.monitor_poster(poster_id, image_path)
            print(f"Monitored {poster_id}: {'⚠️  Damage detected' if entry['damage_detected'] else '✅ Good condition'}")
        else:
            print(f"Monitoring location not found: {image_path}")
    
    # Get summary
    summary = monitoring_system.get_monitoring_summary()
    print(f"\n📊 MONITORING SUMMARY")
    print(f"Total Monitored: {summary['total_monitored']}")
    print(f"Damage Rate: {summary['damage_rate']}")


def example_5_api_integration():
    """Example 5: Integration with web API or microservice."""
    print("\n" + "=" * 50)
    print("EXAMPLE 5: API Integration Example")
    print("=" * 50)
    
    class DamageDetectionAPI:
        def __init__(self):
            self.detector = PretrainedDamageDetector(device='auto')
        
        def analyze_image_endpoint(self, image_data):
            """Simulate API endpoint for image analysis."""
            # In a real API, you'd handle image upload/decoding here
            # For demo, we'll use a file path
            
            try:
                result = self.detector.detect_damage(image_data)
                
                # Format response for API
                api_response = {
                    'status': 'success',
                    'data': {
                        'is_damaged': result['is_damaged'],
                        'confidence': round(result['confidence'], 3),
                        'damage_type': result['primary_damage_type'],
                        'damage_scores': {k: round(v, 3) for k, v in result['damage_scores'].items()},
                        'description': result['detailed_analysis']['blip_caption']
                    },
                    'metadata': {
                        'model_version': result['metadata']['model_version'],
                        'processing_device': result['metadata']['processing_device']
                    }
                }
                
                return api_response
                
            except Exception as e:
                return {
                    'status': 'error',
                    'message': str(e)
                }
        
        def health_check(self):
            """Health check endpoint for API."""
            return {
                'status': 'healthy',
                'models_loaded': True,
                'device': self.detector.device
            }
    
    # Demo usage
    api = DamageDetectionAPI()
    
    print("API Health Check:")
    health = api.health_check()
    print(f"Status: {health['status']}")
    print(f"Device: {health['device']}")
    
    # Simulate API call
    sample_image = "test_poster.jpg"
    if os.path.exists(sample_image):
        response = api.analyze_image_endpoint(sample_image)
        print(f"\nAPI Response:")
        print(json.dumps(response, indent=2))
    else:
        print(f"\nSample API response format:")
        sample_response = {
            'status': 'success',
            'data': {
                'is_damaged': True,
                'confidence': 0.756,
                'damage_type': 'tear',
                'damage_scores': {'tear': 0.756, 'stain': 0.234},
                'description': 'A poster with visible damage'
            }
        }
        print(json.dumps(sample_response, indent=2))


def main():
    """Run all usage examples."""
    print("🎯 PRETRAINED DAMAGE DETECTOR - USAGE EXAMPLES")
    print("=" * 60)
    
    try:
        # Run examples
        example_1_basic_usage()
        example_2_quality_control_system()
        example_3_batch_archive_processing()
        example_4_real_time_monitoring()
        example_5_api_integration()
        
        print("\n" + "=" * 60)
        print("✅ ALL EXAMPLES COMPLETED!")
        print("=" * 60)
        print("\n💡 Next Steps:")
        print("1. Add your own poster images to test the detector")
        print("2. Modify the examples for your specific use case")
        print("3. Integrate the detector into your existing workflow")
        print("4. Adjust confidence thresholds based on your requirements")
        
    except Exception as e:
        print(f"\n❌ Error running examples: {e}")
        print("\nPlease ensure:")
        print("1. All dependencies are installed: pip install -r requirements.txt")
        print("2. You have internet connection for model downloads")
        print("3. Sufficient disk space for pretrained models (~2GB)")


if __name__ == "__main__":
    main()