Create scripts/benchmark.py

scripts/benchmark.py

@@ -0,0 +1,432 @@
+#!/usr/bin/env python3
+"""
+Benchmark script for BackgroundFX Pro.
+Tests performance across different configurations and hardware.
+"""
+
+import time
+import psutil
+import torch
+import cv2
+import numpy as np
+from pathlib import Path
+import json
+import argparse
+from typing import Dict, List, Any
+import statistics
+from datetime import datetime
+
+# Add parent directory to path
+import sys
+sys.path.append(str(Path(__file__).parent.parent))
+
+from api import ProcessingPipeline, PipelineConfig
+from models import ModelRegistry, ModelLoader
+
+
+class Benchmarker:
+    """Performance benchmarking tool."""
+    
+    def __init__(self, output_file: str = None):
+        """Initialize benchmarker."""
+        self.results = {
+            'timestamp': datetime.now().isoformat(),
+            'system_info': self._get_system_info(),
+            'benchmarks': []
+        }
+        self.output_file = output_file or f"benchmark_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+    
+    def _get_system_info(self) -> Dict[str, Any]:
+        """Collect system information."""
+        info = {
+            'cpu': {
+                'count': psutil.cpu_count(),
+                'frequency': psutil.cpu_freq().current if psutil.cpu_freq() else 0,
+                'model': self._get_cpu_model()
+            },
+            'memory': {
+                'total_gb': psutil.virtual_memory().total / (1024**3),
+                'available_gb': psutil.virtual_memory().available / (1024**3)
+            },
+            'gpu': self._get_gpu_info(),
+            'python_version': sys.version,
+            'torch_version': torch.__version__,
+            'cuda_available': torch.cuda.is_available()
+        }
+        return info
+    
+    def _get_cpu_model(self) -> str:
+        """Get CPU model name."""
+        try:
+            import platform
+            return platform.processor()
+        except:
+            return "Unknown"
+    
+    def _get_gpu_info(self) -> Dict[str, Any]:
+        """Get GPU information."""
+        if torch.cuda.is_available():
+            return {
+                'name': torch.cuda.get_device_name(0),
+                'memory_gb': torch.cuda.get_device_properties(0).total_memory / (1024**3),
+                'compute_capability': torch.cuda.get_device_capability(0)
+            }
+        return {'available': False}
+    
+    def benchmark_image_processing(self, 
+                                  sizes: List[tuple] = None,
+                                  qualities: List[str] = None,
+                                  num_iterations: int = 5) -> Dict[str, Any]:
+        """Benchmark image processing performance."""
+        print("\n=== Image Processing Benchmark ===")
+        
+        sizes = sizes or [(512, 512), (1024, 1024), (1920, 1080)]
+        qualities = qualities or ['low', 'medium', 'high']
+        
+        results = {
+            'test': 'image_processing',
+            'iterations': num_iterations,
+            'results': []
+        }
+        
+        for size in sizes:
+            for quality in qualities:
+                print(f"Testing {size[0]}x{size[1]} @ {quality} quality...")
+                
+                # Create test image
+                image = np.random.randint(0, 255, (*size, 3), dtype=np.uint8)
+                
+                # Configure pipeline
+                config = PipelineConfig(
+                    quality_preset=quality,
+                    use_gpu=torch.cuda.is_available(),
+                    enable_cache=False
+                )
+                
+                try:
+                    pipeline = ProcessingPipeline(config)
+                    
+                    # Warmup
+                    pipeline.process_image(image, None)
+                    
+                    # Benchmark
+                    times = []
+                    memory_usage = []
+                    
+                    for _ in range(num_iterations):
+                        start_mem = psutil.Process().memory_info().rss / (1024**2)
+                        start_time = time.time()
+                        
+                        result = pipeline.process_image(image, None)
+                        
+                        elapsed = time.time() - start_time
+                        end_mem = psutil.Process().memory_info().rss / (1024**2)
+                        
+                        times.append(elapsed)
+                        memory_usage.append(end_mem - start_mem)
+                    
+                    # Calculate statistics
+                    result_data = {
+                        'size': f"{size[0]}x{size[1]}",
+                        'quality': quality,
+                        'avg_time': statistics.mean(times),
+                        'std_time': statistics.stdev(times) if len(times) > 1 else 0,
+                        'min_time': min(times),
+                        'max_time': max(times),
+                        'fps': 1.0 / statistics.mean(times),
+                        'avg_memory_mb': statistics.mean(memory_usage)
+                    }
+                    
+                    results['results'].append(result_data)
+                    print(f"  Average: {result_data['avg_time']:.3f}s ({result_data['fps']:.1f} FPS)")
+                    
+                except Exception as e:
+                    print(f"  Failed: {str(e)}")
+                    results['results'].append({
+                        'size': f"{size[0]}x{size[1]}",
+                        'quality': quality,
+                        'error': str(e)
+                    })
+        
+        self.results['benchmarks'].append(results)
+        return results
+    
+    def benchmark_model_loading(self) -> Dict[str, Any]:
+        """Benchmark model loading times."""
+        print("\n=== Model Loading Benchmark ===")
+        
+        results = {
+            'test': 'model_loading',
+            'results': []
+        }
+        
+        registry = ModelRegistry()
+        loader = ModelLoader(registry, device='cuda' if torch.cuda.is_available() else 'cpu')
+        
+        # Test loading different models
+        models_to_test = ['rmbg-1.4', 'u2netp', 'modnet']
+        
+        for model_id in models_to_test:
+            print(f"Loading {model_id}...")
+            
+            # Clear cache
+            loader.unload_all()
+            
+            # Measure loading time
+            start_time = time.time()
+            start_mem = psutil.Process().memory_info().rss / (1024**2)
+            
+            try:
+                loaded = loader.load_model(model_id)
+                
+                elapsed = time.time() - start_time
+                end_mem = psutil.Process().memory_info().rss / (1024**2)
+                
+                if loaded:
+                    result_data = {
+                        'model': model_id,
+                        'load_time': elapsed,
+                        'memory_usage_mb': end_mem - start_mem,
+                        'device': loaded.device
+                    }
+                    print(f"  Loaded in {elapsed:.2f}s, Memory: {end_mem - start_mem:.1f}MB")
+                else:
+                    result_data = {
+                        'model': model_id,
+                        'error': 'Failed to load'
+                    }
+                    print(f"  Failed to load")
+                    
+            except Exception as e:
+                result_data = {
+                    'model': model_id,
+                    'error': str(e)
+                }
+                print(f"  Error: {str(e)}")
+            
+            results['results'].append(result_data)
+        
+        self.results['benchmarks'].append(results)
+        return results
+    
+    def benchmark_video_processing(self, 
+                                  duration: int = 5,
+                                  fps: int = 30,
+                                  size: tuple = (1280, 720)) -> Dict[str, Any]:
+        """Benchmark video processing performance."""
+        print("\n=== Video Processing Benchmark ===")
+        
+        results = {
+            'test': 'video_processing',
+            'video_specs': {
+                'duration': duration,
+                'fps': fps,
+                'size': f"{size[0]}x{size[1]}",
+                'total_frames': duration * fps
+            },
+            'results': []
+        }
+        
+        # Create test video
+        import tempfile
+        video_path = Path(tempfile.mkdtemp()) / "test_video.mp4"
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out = cv2.VideoWriter(str(video_path), fourcc, fps, size)
+        
+        print(f"Creating test video: {duration}s @ {fps}fps, {size[0]}x{size[1]}")
+        for i in range(duration * fps):
+            frame = np.random.randint(0, 255, (*size[::-1], 3), dtype=np.uint8)
+            # Add moving rectangle for motion
+            x = int((i / (duration * fps)) * size[0])
+            cv2.rectangle(frame, (x, 100), (x + 100, 200), (0, 255, 0), -1)
+            out.write(frame)
+        out.release()
+        
+        # Test different quality settings
+        for quality in ['low', 'medium', 'high']:
+            print(f"Processing at {quality} quality...")
+            
+            from api import VideoProcessorAPI
+            processor = VideoProcessorAPI()
+            
+            start_time = time.time()
+            start_mem = psutil.Process().memory_info().rss / (1024**2)
+            
+            try:
+                output_path = video_path.parent / f"output_{quality}.mp4"
+                stats = processor.process_video(
+                    str(video_path),
+                    str(output_path),
+                    background=None
+                )
+                
+                elapsed = time.time() - start_time
+                end_mem = psutil.Process().memory_info().rss / (1024**2)
+                
+                result_data = {
+                    'quality': quality,
+                    'total_time': elapsed,
+                    'frames_processed': stats.frames_processed,
+                    'processing_fps': stats.processing_fps,
+                    'time_per_frame': elapsed / stats.frames_processed if stats.frames_processed > 0 else 0,
+                    'memory_usage_mb': end_mem - start_mem
+                }
+                
+                print(f"  Processed in {elapsed:.2f}s @ {stats.processing_fps:.1f} FPS")
+                
+            except Exception as e:
+                result_data = {
+                    'quality': quality,
+                    'error': str(e)
+                }
+                print(f"  Failed: {str(e)}")
+            
+            results['results'].append(result_data)
+        
+        # Cleanup
+        video_path.unlink(missing_ok=True)
+        
+        self.results['benchmarks'].append(results)
+        return results
+    
+    def benchmark_batch_processing(self, 
+                                  batch_sizes: List[int] = None,
+                                  num_workers_list: List[int] = None) -> Dict[str, Any]:
+        """Benchmark batch processing performance."""
+        print("\n=== Batch Processing Benchmark ===")
+        
+        batch_sizes = batch_sizes or [1, 5, 10, 20]
+        num_workers_list = num_workers_list or [1, 2, 4, 8]
+        
+        results = {
+            'test': 'batch_processing',
+            'results': []
+        }
+        
+        # Create test images
+        test_images = []
+        for i in range(max(batch_sizes)):
+            img = np.random.randint(0, 255, (512, 512, 3), dtype=np.uint8)
+            test_images.append(img)
+        
+        for batch_size in batch_sizes:
+            for num_workers in num_workers_list:
+                print(f"Testing batch_size={batch_size}, workers={num_workers}...")
+                
+                config = PipelineConfig(
+                    batch_size=batch_size,
+                    num_workers=num_workers,
+                    use_gpu=torch.cuda.is_available(),
+                    enable_cache=False
+                )
+                
+                try:
+                    pipeline = ProcessingPipeline(config)
+                    
+                    start_time = time.time()
+                    results_batch = pipeline.process_batch(test_images[:batch_size])
+                    elapsed = time.time() - start_time
+                    
+                    successful = sum(1 for r in results_batch if r.success)
+                    
+                    result_data = {
+                        'batch_size': batch_size,
+                        'num_workers': num_workers,
+                        'total_time': elapsed,
+                        'time_per_image': elapsed / batch_size,
+                        'throughput': batch_size / elapsed,
+                        'successful': successful
+                    }
+                    
+                    print(f"  {elapsed:.2f}s total, {result_data['throughput']:.1f} images/sec")
+                    
+                except Exception as e:
+                    result_data = {
+                        'batch_size': batch_size,
+                        'num_workers': num_workers,
+                        'error': str(e)
+                    }
+                    print(f"  Failed: {str(e)}")
+                
+                results['results'].append(result_data)
+        
+        self.results['benchmarks'].append(results)
+        return results
+    
+    def save_results(self):
+        """Save benchmark results to file."""
+        with open(self.output_file, 'w') as f:
+            json.dump(self.results, f, indent=2)
+        print(f"\nResults saved to: {self.output_file}")
+    
+    def print_summary(self):
+        """Print benchmark summary."""
+        print("\n" + "="*50)
+        print("BENCHMARK SUMMARY")
+        print("="*50)
+        
+        for benchmark in self.results['benchmarks']:
+            print(f"\n{benchmark['test'].upper()}:")
+            
+            if 'results' in benchmark:
+                for result in benchmark['results']:
+                    if 'error' not in result:
+                        if benchmark['test'] == 'image_processing':
+                            print(f"  {result['size']} @ {result['quality']}: {result['fps']:.1f} FPS")
+                        elif benchmark['test'] == 'model_loading':
+                            print(f"  {result['model']}: {result['load_time']:.2f}s")
+                        elif benchmark['test'] == 'video_processing':
+                            print(f"  {result['quality']}: {result['processing_fps']:.1f} FPS")
+                        elif benchmark['test'] == 'batch_processing':
+                            print(f"  Batch {result['batch_size']} x {result['num_workers']} workers: {result['throughput']:.1f} img/s")
+
+
+def main():
+    """Main benchmark function."""
+    parser = argparse.ArgumentParser(description='BackgroundFX Pro Performance Benchmark')
+    parser.add_argument('--tests', nargs='+', 
+                       choices=['image', 'model', 'video', 'batch', 'all'],
+                       default=['all'],
+                       help='Tests to run')
+    parser.add_argument('--output', '-o', help='Output file for results')
+    parser.add_argument('--iterations', '-i', type=int, default=5,
+                       help='Number of iterations for each test')
+    
+    args = parser.parse_args()
+    
+    benchmarker = Benchmarker(args.output)
+    
+    tests_to_run = args.tests
+    if 'all' in tests_to_run:
+        tests_to_run = ['image', 'model', 'video', 'batch']
+    
+    print("BackgroundFX Pro Performance Benchmark")
+    print("="*50)
+    print("System Information:")
+    print(f"  CPU: {benchmarker.results['system_info']['cpu']['model']}")
+    print(f"  Memory: {benchmarker.results['system_info']['memory']['total_gb']:.1f}GB")
+    if benchmarker.results['system_info']['cuda_available']:
+        print(f"  GPU: {benchmarker.results['system_info']['gpu']['name']}")
+    else:
+        print("  GPU: Not available")
+    
+    # Run selected benchmarks
+    if 'image' in tests_to_run:
+        benchmarker.benchmark_image_processing(num_iterations=args.iterations)
+    
+    if 'model' in tests_to_run:
+        benchmarker.benchmark_model_loading()
+    
+    if 'video' in tests_to_run:
+        benchmarker.benchmark_video_processing()
+    
+    if 'batch' in tests_to_run:
+        benchmarker.benchmark_batch_processing()
+    
+    # Save and display results
+    benchmarker.save_results()
+    benchmarker.print_summary()
+
+
+if __name__ == "__main__":
+    main()