Create core/quality.py

core/quality.py

@@ -0,0 +1,409 @@
+"""
+Quality analysis and metrics for BackgroundFX Pro.
+Provides REAL metrics instead of fake 100% values.
+"""
+
+import numpy as np
+import cv2
+import torch
+from typing import Dict, List, Optional, Tuple, Any
+from dataclasses import dataclass, field
+from collections import deque
+import logging
+from scipy import signal, ndimage
+from skimage import metrics as skmetrics
+import json
+from pathlib import Path
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class QualityMetrics:
+    """Real quality metrics container."""
+    # Edge Quality
+    edge_accuracy: float = 0.0
+    edge_smoothness: float = 0.0
+    edge_completeness: float = 0.0
+    
+    # Temporal Quality
+    temporal_stability: float = 0.0
+    temporal_consistency: float = 0.0
+    flicker_score: float = 0.0
+    
+    # Mask Quality
+    mask_coverage: float = 0.0
+    mask_accuracy: float = 0.0
+    mask_confidence: float = 0.0
+    hole_ratio: float = 0.0
+    
+    # Detail Preservation
+    detail_preservation: float = 0.0
+    hair_detail_score: float = 0.0
+    texture_quality: float = 0.0
+    
+    # Overall Scores
+    overall_quality: float = 0.0
+    processing_confidence: float = 0.0
+    
+    # Detailed breakdown
+    breakdown: Dict[str, float] = field(default_factory=dict)
+    warnings: List[str] = field(default_factory=list)
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary."""
+        return {
+            'edge_accuracy': round(self.edge_accuracy, 3),
+            'edge_smoothness': round(self.edge_smoothness, 3),
+            'edge_completeness': round(self.edge_completeness, 3),
+            'temporal_stability': round(self.temporal_stability, 3),
+            'temporal_consistency': round(self.temporal_consistency, 3),
+            'flicker_score': round(self.flicker_score, 3),
+            'mask_coverage': round(self.mask_coverage, 3),
+            'mask_accuracy': round(self.mask_accuracy, 3),
+            'mask_confidence': round(self.mask_confidence, 3),
+            'hole_ratio': round(self.hole_ratio, 3),
+            'detail_preservation': round(self.detail_preservation, 3),
+            'hair_detail_score': round(self.hair_detail_score, 3),
+            'texture_quality': round(self.texture_quality, 3),
+            'overall_quality': round(self.overall_quality, 3),
+            'processing_confidence': round(self.processing_confidence, 3),
+            'breakdown': self.breakdown,
+            'warnings': self.warnings
+        }
+    
+    def get_summary(self) -> str:
+        """Get human-readable summary."""
+        status = "Excellent" if self.overall_quality > 0.9 else \
+                 "Good" if self.overall_quality > 0.75 else \
+                 "Fair" if self.overall_quality > 0.6 else "Poor"
+        
+        return (f"Quality: {status} ({self.overall_quality:.1%})\n"
+                f"Edge: {self.edge_accuracy:.1%} | "
+                f"Temporal: {self.temporal_stability:.1%} | "
+                f"Detail: {self.detail_preservation:.1%}")
+
+
+@dataclass
+class QualityConfig:
+    """Configuration for quality analysis."""
+    enable_deep_analysis: bool = True
+    temporal_window: int = 5
+    edge_threshold: float = 0.1
+    min_confidence: float = 0.6
+    detect_artifacts: bool = True
+    compute_ssim: bool = True
+    compute_psnr: bool = True
+    save_reports: bool = True
+    report_dir: str = "LOGS/quality_reports"
+    warning_thresholds: Dict[str, float] = field(default_factory=lambda: {
+        'edge_accuracy': 0.7,
+        'temporal_stability': 0.75,
+        'mask_accuracy': 0.8,
+        'detail_preservation': 0.7
+    })
+
+
+class QualityAnalyzer:
+    """Comprehensive quality analysis system."""
+    
+    def __init__(self, config: Optional[QualityConfig] = None):
+        self.config = config or QualityConfig()
+        self.frame_buffer = deque(maxlen=self.config.temporal_window)
+        self.mask_buffer = deque(maxlen=self.config.temporal_window)
+        self.metrics_history = deque(maxlen=100)
+        self.frame_count = 0
+        
+        # Initialize analyzers
+        self.edge_analyzer = EdgeQualityAnalyzer()
+        self.temporal_analyzer = TemporalQualityAnalyzer()
+        self.detail_analyzer = DetailPreservationAnalyzer()
+        self.artifact_detector = ArtifactDetector()
+        
+        # Create report directory
+        if self.config.save_reports:
+            Path(self.config.report_dir).mkdir(parents=True, exist_ok=True)
+    
+    def analyze_frame(self, 
+                     original_frame: np.ndarray,
+                     processed_frame: np.ndarray,
+                     mask: np.ndarray,
+                     alpha: Optional[np.ndarray] = None) -> QualityMetrics:
+        """Analyze frame quality with REAL metrics."""
+        self.frame_count += 1
+        metrics = QualityMetrics()
+        
+        # Add to buffers
+        self.frame_buffer.append(processed_frame)
+        self.mask_buffer.append(mask)
+        
+        # 1. Edge Quality Analysis
+        edge_metrics = self.edge_analyzer.analyze(original_frame, mask, alpha)
+        metrics.edge_accuracy = edge_metrics['accuracy']
+        metrics.edge_smoothness = edge_metrics['smoothness']
+        metrics.edge_completeness = edge_metrics['completeness']
+        
+        # 2. Temporal Quality (if we have history)
+        if len(self.mask_buffer) >= 2:
+            temporal_metrics = self.temporal_analyzer.analyze(
+                self.mask_buffer, self.frame_buffer
+            )
+            metrics.temporal_stability = temporal_metrics['stability']
+            metrics.temporal_consistency = temporal_metrics['consistency']
+            metrics.flicker_score = temporal_metrics['flicker']
+        else:
+            # First frame defaults
+            metrics.temporal_stability = 1.0
+            metrics.temporal_consistency = 1.0
+            metrics.flicker_score = 0.0
+        
+        # 3. Mask Quality Analysis
+        mask_metrics = self._analyze_mask_quality(mask, alpha)
+        metrics.mask_coverage = mask_metrics['coverage']
+        metrics.mask_accuracy = mask_metrics['accuracy']
+        metrics.mask_confidence = mask_metrics['confidence']
+        metrics.hole_ratio = mask_metrics['hole_ratio']
+        
+        # 4. Detail Preservation
+        detail_metrics = self.detail_analyzer.analyze(
+            original_frame, processed_frame, mask
+        )
+        metrics.detail_preservation = detail_metrics['overall']
+        metrics.hair_detail_score = detail_metrics['hair_detail']
+        metrics.texture_quality = detail_metrics['texture']
+        
+        # 5. Artifact Detection
+        if self.config.detect_artifacts:
+            artifacts = self.artifact_detector.detect(processed_frame, mask)
+            if artifacts['found']:
+                for artifact in artifacts['types']:
+                    metrics.warnings.append(f"Artifact detected: {artifact}")
+        
+        # 6. Compute Overall Quality (weighted average)
+        metrics.overall_quality = self._compute_overall_quality(metrics)
+        metrics.processing_confidence = self._compute_confidence(metrics)
+        
+        # 7. Generate warnings based on thresholds
+        self._generate_warnings(metrics)
+        
+        # 8. Store in history
+        self.metrics_history.append(metrics)
+        
+        # 9. Save report if configured
+        if self.config.save_reports and self.frame_count % 30 == 0:
+            self._save_report(metrics)
+        
+        return metrics
+    
+    def _analyze_mask_quality(self, mask: np.ndarray,
+                             alpha: Optional[np.ndarray] = None) -> Dict[str, float]:
+        """Analyze mask quality metrics."""
+        h, w = mask.shape[:2]
+        total_pixels = h * w
+        
+        # Coverage ratio
+        coverage = np.sum(mask > 0.5) / total_pixels
+        
+        # Hole detection
+        mask_binary = (mask > 0.5).astype(np.uint8)
+        
+        # Find contours
+        contours, _ = cv2.findContours(
+            mask_binary, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE
+        )
+        
+        # Find holes (internal contours)
+        hole_area = 0
+        if len(contours) > 0:
+            # Create filled mask
+            filled = np.zeros_like(mask_binary)
+            cv2.drawContours(filled, contours, -1, 1, -1)
+            
+            # Holes are the difference
+            holes = filled - mask_binary
+            hole_area = np.sum(holes) / np.sum(filled) if np.sum(filled) > 0 else 0
+        
+        # Accuracy (based on gradient consistency)
+        gradient_x = cv2.Sobel(mask, cv2.CV_64F, 1, 0, ksize=3)
+        gradient_y = cv2.Sobel(mask, cv2.CV_64F, 0, 1, ksize=3)
+        gradient_mag = np.sqrt(gradient_x**2 + gradient_y**2)
+        
+        # Good masks have smooth gradients
+        gradient_smoothness = 1.0 - np.std(gradient_mag) / (np.mean(gradient_mag) + 1e-6)
+        accuracy = np.clip(gradient_smoothness, 0, 1)
+        
+        # Confidence (alpha vs mask consistency if alpha provided)
+        if alpha is not None:
+            diff = np.abs(alpha - mask)
+            confidence = 1.0 - np.mean(diff)
+        else:
+            # Use mask value distribution as confidence
+            hist, _ = np.histogram(mask.flatten(), bins=10, range=(0, 1))
+            hist = hist / hist.sum()
+            # High confidence = values clustered near 0 or 1
+            confidence = (hist[0] + hist[-1]) / 2.0
+        
+        return {
+            'coverage': coverage,
+            'accuracy': accuracy,
+            'confidence': confidence,
+            'hole_ratio': hole_area
+        }
+    
+    def _compute_overall_quality(self, metrics: QualityMetrics) -> float:
+        """Compute weighted overall quality score."""
+        weights = {
+            'edge': 0.25,
+            'temporal': 0.25,
+            'mask': 0.25,
+            'detail': 0.25
+        }
+        
+        # Component scores
+        edge_score = np.mean([
+            metrics.edge_accuracy,
+            metrics.edge_smoothness,
+            metrics.edge_completeness
+        ])
+        
+        temporal_score = np.mean([
+            metrics.temporal_stability,
+            metrics.temporal_consistency,
+            1.0 - metrics.flicker_score  # Invert flicker
+        ])
+        
+        mask_score = np.mean([
+            metrics.mask_accuracy,
+            metrics.mask_confidence,
+            1.0 - metrics.hole_ratio  # Invert hole ratio
+        ])
+        
+        detail_score = np.mean([
+            metrics.detail_preservation,
+            metrics.hair_detail_score,
+            metrics.texture_quality
+        ])
+        
+        # Weighted average
+        overall = (
+            weights['edge'] * edge_score +
+            weights['temporal'] * temporal_score +
+            weights['mask'] * mask_score +
+            weights['detail'] * detail_score
+        )
+        
+        # Apply penalties for warnings
+        penalty = len(metrics.warnings) * 0.05
+        overall = max(0, overall - penalty)
+        
+        return np.clip(overall, 0, 1)
+    
+    def _compute_confidence(self, metrics: QualityMetrics) -> float:
+        """Compute processing confidence."""
+        # Factors that affect confidence
+        factors = []
+        
+        # High edge accuracy increases confidence
+        factors.append(metrics.edge_accuracy)
+        
+        # Good temporal stability increases confidence
+        factors.append(metrics.temporal_stability)
+        
+        # Low hole ratio increases confidence
+        factors.append(1.0 - metrics.hole_ratio)
+        
+        # Mask confidence directly affects overall confidence
+        factors.append(metrics.mask_confidence)
+        
+        # No warnings increases confidence
+        warning_factor = 1.0 if len(metrics.warnings) == 0 else 0.8
+        factors.append(warning_factor)
+        
+        return np.mean(factors)
+    
+    def _generate_warnings(self, metrics: QualityMetrics):
+        """Generate warnings based on quality thresholds."""
+        for metric_name, threshold in self.config.warning_thresholds.items():
+            if hasattr(metrics, metric_name):
+                value = getattr(metrics, metric_name)
+                if value < threshold:
+                    metrics.warnings.append(
+                        f"Low {metric_name.replace('_', ' ')}: {value:.1%} < {threshold:.1%}"
+                    )
+    
+    def _save_report(self, metrics: QualityMetrics):
+        """Save quality report to file."""
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        report_path = Path(self.config.report_dir) / f"quality_report_{timestamp}.json"
+        
+        report = {
+            'timestamp': timestamp,
+            'frame_count': self.frame_count,
+            'metrics': metrics.to_dict(),
+            'config': {
+                'temporal_window': self.config.temporal_window,
+                'edge_threshold': self.config.edge_threshold,
+                'min_confidence': self.config.min_confidence
+            }
+        }
+        
+        with open(report_path, 'w') as f:
+            json.dump(report, f, indent=2)
+        
+        logger.info(f"Quality report saved to {report_path}")
+    
+    def get_statistics(self) -> Dict[str, Any]:
+        """Get quality statistics over time."""
+        if not self.metrics_history:
+            return {}
+        
+        # Compute statistics
+        all_metrics = list(self.metrics_history)
+        
+        stats = {
+            'average_quality': np.mean([m.overall_quality for m in all_metrics]),
+            'min_quality': np.min([m.overall_quality for m in all_metrics]),
+            'max_quality': np.max([m.overall_quality for m in all_metrics]),
+            'std_quality': np.std([m.overall_quality for m in all_metrics]),
+            'total_warnings': sum(len(m.warnings) for m in all_metrics),
+            'frames_analyzed': len(all_metrics)
+        }
+        
+        return stats
+
+
+class EdgeQualityAnalyzer:
+    """Analyzes edge quality in masks."""
+    
+    def analyze(self, image: np.ndarray, mask: np.ndarray,
+               alpha: Optional[np.ndarray] = None) -> Dict[str, float]:
+        """Analyze edge quality metrics."""
+        # Convert to grayscale if needed
+        if len(image.shape) == 3:
+            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        else:
+            gray = image
+        
+        # Detect edges in image
+        image_edges = cv2.Canny(gray, 50, 150) / 255.0
+        
+        # Detect edges in mask
+        mask_uint8 = (mask * 255).astype(np.uint8)
+        mask_edges = cv2.Canny(mask_uint8, 50, 150) / 255.0
+        
+        # Edge accuracy: how well mask edges align with image edges
+        overlap = np.logical_and(image_edges > 0, mask_edges > 0)
+        accuracy = np.sum(overlap) / (np.sum(mask_edges) + 1e-6)
+        
+        # Edge smoothness: measure edge roughness
+        contours, _ = cv2.findContours(
+            mask_uint8, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
+        )
+        
+        smoothness = 1.0
+        if len(contours) > 0:
+            # Approximate contours and measure approximation quality
+            for contour in contours:
+                perimeter = cv2.arcLength(contour, True)
+                if perimeter