Update processing/fallback.py

processing/fallback.py

@@ -1,3 +1,4 @@
+#!/usr/bin/env python3
 """
 Fallback strategies for BackgroundFX Pro.
 Implements robust fallback mechanisms when primary processing fails.
@@ -12,16 +13,15 @@
 import logging
 import traceback
 
-from ..utils.logger import setup_logger
-from ..utils.device import DeviceManager
-from ..utils.config import ConfigManager
-from ..core.quality import QualityAnalyzer
+# ABSOLUTE IMPORTS for Hugging Face Spaces
+from utils.logger import setup_logger
+from utils.device import DeviceManager
+from utils.config import ConfigManager
+from core.quality import QualityAnalyzer
 
 logger = setup_logger(__name__)
 
-
 class FallbackLevel(Enum):
-    """Fallback hierarchy levels."""
     NONE = 0
     QUALITY_REDUCTION = 1
     METHOD_SWITCH = 2
@@ -29,10 +29,8 @@ class FallbackLevel(Enum):
     MINIMAL_PROCESSING = 4
     PASSTHROUGH = 5
 
-
 @dataclass
 class FallbackConfig:
-    """Configuration for fallback strategies."""
     max_retries: int = 3
     quality_reduction_factor: float = 0.75
     min_quality: float = 0.3
@@ -43,10 +41,7 @@ class FallbackConfig:
     progressive_downscale: bool = True
     min_resolution: Tuple[int, int] = (320, 240)
 
-
 class FallbackStrategy:
-    """Intelligent fallback strategy manager."""
-    
     def __init__(self, config: Optional[FallbackConfig] = None):
         self.config = config or FallbackConfig()
         self.device_manager = DeviceManager()
@@ -54,173 +49,108 @@ def __init__(self, config: Optional[FallbackConfig] = None):
         self.cache = {}
         self.fallback_history = []
         self.current_level = FallbackLevel.NONE
-        
+
     def execute_with_fallback(self, func, *args, **kwargs) -> Dict[str, Any]:
-        """
-        Execute function with automatic fallback on failure.
-        
-        Args:
-            func: Function to execute
-            *args: Function arguments
-            **kwargs: Function keyword arguments
-            
-        Returns:
-            Result dictionary with status and output
-        """
         attempt = 0
         last_error = None
         original_args = args
         original_kwargs = kwargs.copy()
-        
+
         while attempt < self.config.max_retries:
             try:
-                # Log attempt
                 logger.info(f"Attempt {attempt + 1}/{self.config.max_retries} for {func.__name__}")
-                
-                # Try execution
                 result = func(*args, **kwargs)
-                
-                # Success - reset fallback level
                 self.current_level = FallbackLevel.NONE
-                
                 return {
                     'success': True,
                     'result': result,
                     'attempts': attempt + 1,
                     'fallback_level': self.current_level
                 }
-                
             except Exception as e:
                 last_error = e
                 logger.warning(f"Attempt {attempt + 1} failed: {str(e)}")
-                
-                # Apply fallback strategy
-                fallback_result = self._apply_fallback(
-                    func, e, attempt, 
-                    original_args, original_kwargs
-                )
-                
+                fallback_result = self._apply_fallback(func, e, attempt, original_args, original_kwargs)
                 if fallback_result['handled']:
                     args = fallback_result.get('new_args', args)
                     kwargs = fallback_result.get('new_kwargs', kwargs)
                 else:
                     break
-                
                 attempt += 1
-        
-        # All attempts failed - apply final fallback
+
         logger.error(f"All attempts failed for {func.__name__}")
         return self._final_fallback(func, last_error, original_args)
-    
-    def _apply_fallback(self, func, error: Exception, 
-                       attempt: int, original_args: tuple,
-                       original_kwargs: dict) -> Dict[str, Any]:
-        """Apply appropriate fallback strategy based on error type."""
-        
+
+    def _apply_fallback(self, func, error: Exception, attempt: int, original_args: tuple, original_kwargs: dict) -> Dict[str, Any]:
         error_type = type(error).__name__
         self.fallback_history.append({
             'function': func.__name__,
             'error': error_type,
             'attempt': attempt
         })
-        
-        # GPU memory error - switch to CPU
+
         if 'CUDA' in str(error) or 'GPU' in str(error):
             return self._handle_gpu_error(original_kwargs)
-        
-        # Memory error - reduce quality
         elif 'memory' in str(error).lower():
             return self._handle_memory_error(original_args, original_kwargs)
-        
-        # Timeout error - simplify processing
         elif 'timeout' in str(error).lower():
             return self._handle_timeout_error(original_kwargs)
-        
-        # Model loading error - use simpler model
         elif 'model' in str(error).lower():
             return self._handle_model_error(original_kwargs)
-        
-        # Generic error - progressive degradation
         else:
             return self._handle_generic_error(attempt, original_kwargs)
-    
+
     def _handle_gpu_error(self, kwargs: dict) -> Dict[str, Any]:
-        """Handle GPU-related errors."""
         logger.info("GPU error detected, falling back to CPU")
-        
         if self.config.gpu_fallback_to_cpu:
-            # Switch to CPU
             self.device_manager.device = torch.device('cpu')
             kwargs['device'] = 'cpu'
-            
-            # Reduce batch size if present
             if 'batch_size' in kwargs:
                 kwargs['batch_size'] = max(1, kwargs['batch_size'] // 2)
-            
             self.current_level = FallbackLevel.METHOD_SWITCH
-            
             return {
                 'handled': True,
                 'new_kwargs': kwargs
             }
-        
         return {'handled': False}
-    
-    def _handle_memory_error(self, args: tuple, 
-                            kwargs: dict) -> Dict[str, Any]:
-        """Handle memory-related errors."""
+
+    def _handle_memory_error(self, args: tuple, kwargs: dict) -> Dict[str, Any]:
         logger.info("Memory error detected, reducing quality")
-        
-        # Try to find image in args
         image = None
         image_idx = -1
-        
         for i, arg in enumerate(args):
             if isinstance(arg, np.ndarray) and len(arg.shape) == 3:
                 image = arg
                 image_idx = i
                 break
-        
         if image is not None and self.config.progressive_downscale:
-            # Reduce image size
             h, w = image.shape[:2]
             new_h = int(h * self.config.quality_reduction_factor)
             new_w = int(w * self.config.quality_reduction_factor)
-            
-            # Ensure minimum resolution
             new_h = max(new_h, self.config.min_resolution[1])
             new_w = max(new_w, self.config.min_resolution[0])
-            
             if new_h < h or new_w < w:
                 resized = cv2.resize(image, (new_w, new_h))
                 args = list(args)
                 args[image_idx] = resized
-                
                 self.current_level = FallbackLevel.QUALITY_REDUCTION
-                
                 return {
                     'handled': True,
                     'new_args': tuple(args),
                     'new_kwargs': kwargs
                 }
-        
-        # Reduce other memory-intensive parameters
         if 'quality' in kwargs:
             kwargs['quality'] = max(
                 self.config.min_quality,
                 kwargs['quality'] * self.config.quality_reduction_factor
             )
-        
         return {
             'handled': True,
             'new_kwargs': kwargs
         }
-    
+
     def _handle_timeout_error(self, kwargs: dict) -> Dict[str, Any]:
-        """Handle timeout errors by simplifying processing."""
         logger.info("Timeout detected, simplifying processing")
-        
-        # Disable expensive operations
         simplifications = {
             'use_refinement': False,
             'use_temporal': False,
@@ -228,93 +158,65 @@ def _handle_timeout_error(self, kwargs: dict) -> Dict[str, Any]:
             'iterations': 1,
             'num_samples': 1
         }
-        
         for key, value in simplifications.items():
             if key in kwargs:
                 kwargs[key] = value
-        
         self.current_level = FallbackLevel.BASIC_PROCESSING
-        
         return {
             'handled': True,
             'new_kwargs': kwargs
         }
-    
+
     def _handle_model_error(self, kwargs: dict) -> Dict[str, Any]:
-        """Handle model loading errors."""
         logger.info("Model error detected, using simpler model")
-        
-        # Switch to simpler model
         if 'model_type' in kwargs:
             model_hierarchy = ['large', 'base', 'small', 'tiny']
             current = kwargs.get('model_type', 'base')
-            
             if current in model_hierarchy:
                 idx = model_hierarchy.index(current)
                 if idx < len(model_hierarchy) - 1:
                     kwargs['model_type'] = model_hierarchy[idx + 1]
                     self.current_level = FallbackLevel.METHOD_SWITCH
-                    
                     return {
                         'handled': True,
                         'new_kwargs': kwargs
                     }
-        
-        # Disable model-based processing
         kwargs['use_model'] = False
         self.current_level = FallbackLevel.BASIC_PROCESSING
-        
         return {
             'handled': True,
             'new_kwargs': kwargs
         }
-    
-    def _handle_generic_error(self, attempt: int, 
-                            kwargs: dict) -> Dict[str, Any]:
-        """Handle generic errors with progressive degradation."""
+
+    def _handle_generic_error(self, attempt: int, kwargs: dict) -> Dict[str, Any]:
         logger.info(f"Generic error, applying degradation level {attempt + 1}")
-        
-        # Progressive degradation based on attempt
         if attempt == 0:
-            # First attempt - minor quality reduction
             self.current_level = FallbackLevel.QUALITY_REDUCTION
             if 'quality' in kwargs:
                 kwargs['quality'] *= 0.8
-                
         elif attempt == 1:
-            # Second attempt - switch methods
             self.current_level = FallbackLevel.METHOD_SWITCH
             kwargs['method'] = 'basic'
-            
         else:
-            # Final attempt - minimal processing
             self.current_level = FallbackLevel.MINIMAL_PROCESSING
             kwargs['skip_refinement'] = True
             kwargs['fast_mode'] = True
-        
         return {
             'handled': True,
             'new_kwargs': kwargs
         }
-    
-    def _final_fallback(self, func, error: Exception, 
-                       original_args: tuple) -> Dict[str, Any]:
-        """Apply final fallback when all attempts fail."""
+
+    def _final_fallback(self, func, error: Exception, original_args: tuple) -> Dict[str, Any]:
         logger.error(f"Final fallback for {func.__name__}: {str(error)}")
         self.current_level = FallbackLevel.PASSTHROUGH
-        
-        # Try to return something useful
         for arg in original_args:
             if isinstance(arg, np.ndarray):
-                # Return original image/mask
                 return {
                     'success': False,
                     'result': arg,
                     'fallback_level': self.current_level,
                     'error': str(error)
                 }
-        
-        # Return empty result
         return {
             'success': False,
             'result': None,
@@ -322,123 +224,56 @@ def _final_fallback(self, func, error: Exception,
             'error': str(error)
         }
 
-
 class ProcessingFallback:
-    """Specific fallback implementations for processing operations."""
-    
     def __init__(self):
         self.logger = setup_logger(f"{__name__}.ProcessingFallback")
         self.quality_analyzer = QualityAnalyzer()
-        
+
     def basic_segmentation(self, image: np.ndarray) -> np.ndarray:
-        """
-        Basic segmentation using traditional CV methods.
-        Used as fallback when ML models fail.
-        
-        Args:
-            image: Input image
-            
-        Returns:
-            Binary mask
-        """
         try:
-            # Convert to grayscale
             if len(image.shape) == 3:
                 gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
             else:
                 gray = image
-            
-            # Apply GrabCut for basic foreground extraction
             mask = np.zeros(gray.shape[:2], np.uint8)
             bgd_model = np.zeros((1, 65), np.float64)
             fgd_model = np.zeros((1, 65), np.float64)
-            
-            # Initialize rectangle (center 80% of image)
             h, w = gray.shape[:2]
-            rect = (int(w * 0.1), int(h * 0.1), 
-                   int(w * 0.8), int(h * 0.8))
-            
-            # Apply GrabCut
-            cv2.grabCut(image, mask, rect, bgd_model, fgd_model, 
-                       5, cv2.GC_INIT_WITH_RECT)
-            
-            # Extract foreground
+            rect = (int(w * 0.1), int(h * 0.1), int(w * 0.8), int(h * 0.8))
+            cv2.grabCut(image, mask, rect, bgd_model, fgd_model, 5, cv2.GC_INIT_WITH_RECT)
             mask2 = np.where((mask == 2) | (mask == 0), 0, 255).astype('uint8')
-            
             return mask2
-            
         except Exception as e:
             self.logger.error(f"Basic segmentation failed: {e}")
-            # Return center blob as last resort
             return self._center_blob_mask(image.shape[:2])
-    
+
     def _center_blob_mask(self, shape: Tuple[int, int]) -> np.ndarray:
-        """Create a center ellipse mask as ultimate fallback."""
         h, w = shape
         mask = np.zeros((h, w), dtype=np.uint8)
-        
-        # Create center ellipse
         center = (w // 2, h // 2)
         axes = (w // 3, h // 3)
         cv2.ellipse(mask, center, axes, 0, 0, 360, 255, -1)
-        
-        # Smooth edges
         mask = cv2.GaussianBlur(mask, (21, 21), 10)
         _, mask = cv2.threshold(mask, 127, 255, cv2.THRESH_BINARY)
-        
         return mask
-    
-    def basic_matting(self, image: np.ndarray, 
-                     mask: np.ndarray) -> np.ndarray:
-        """
-        Basic matting using morphological operations.
-        
-        Args:
-            image: Input image
-            mask: Binary mask
-            
-        Returns:
-            Alpha matte
-        """
+
+    def basic_matting(self, image: np.ndarray, mask: np.ndarray) -> np.ndarray:
         try:
-            # Ensure uint8
             if mask.dtype != np.uint8:
                 mask = (mask * 255).astype(np.uint8)
-            
-            # Morphological smoothing
             kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
             mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel)
             mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)
-            
-            # Edge softening
             mask = cv2.GaussianBlur(mask, (5, 5), 2)
-            
-            # Normalize to [0, 1]
             alpha = mask.astype(np.float32) / 255.0
-            
             return alpha
-            
         except Exception as e:
             self.logger.error(f"Basic matting failed: {e}")
             return mask.astype(np.float32) / 255.0
-    
-    def color_difference_keying(self, image: np.ndarray,
-                              key_color: Optional[np.ndarray] = None,
-                              threshold: float = 30) -> np.ndarray:
-        """
-        Simple color difference keying for solid backgrounds.
-        
-        Args:
-            image: Input image
-            key_color: Background color to remove
-            threshold: Color difference threshold
-            
-        Returns:
-            Alpha matte
-        """
+
+    def color_difference_keying(self, image: np.ndarray, key_color: Optional[np.ndarray] = None, threshold: float = 30) -> np.ndarray:
         try:
             if key_color is None:
-                # Estimate background color from corners
                 h, w = image.shape[:2]
                 corners = [
                     image[0:10, 0:10],
@@ -447,97 +282,47 @@ def color_difference_keying(self, image: np.ndarray,
                     image[h-10:h, w-10:w]
                 ]
                 key_color = np.mean([np.mean(c, axis=(0, 1)) for c in corners], axis=0)
-            
-            # Calculate color difference
             diff = np.sqrt(np.sum((image - key_color) ** 2, axis=2))
-            
-            # Create mask
             mask = (diff > threshold).astype(np.float32)
-            
-            # Smooth edges
             mask = cv2.GaussianBlur(mask, (5, 5), 2)
-            
             return mask
-            
         except Exception as e:
             self.logger.error(f"Color keying failed: {e}")
             return np.ones(image.shape[:2], dtype=np.float32)
-    
+
     def edge_based_segmentation(self, image: np.ndarray) -> np.ndarray:
-        """
-        Edge-based segmentation as fallback.
-        
-        Args:
-            image: Input image
-            
-        Returns:
-            Binary mask
-        """
         try:
-            # Convert to grayscale
             if len(image.shape) == 3:
                 gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
             else:
                 gray = image
-            
-            # Edge detection
             edges = cv2.Canny(gray, 50, 150)
-            
-            # Close contours
             kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5))
             closed = cv2.morphologyEx(edges, cv2.MORPH_CLOSE, kernel, iterations=2)
-            
-            # Find contours
-            contours, _ = cv2.findContours(
-                closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE
-            )
-            
-            # Create mask from largest contour
+            contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
             mask = np.zeros(gray.shape, dtype=np.uint8)
             if contours:
                 largest = max(contours, key=cv2.contourArea)
                 cv2.drawContours(mask, [largest], -1, 255, -1)
-            
             return mask
-            
         except Exception as e:
             self.logger.error(f"Edge segmentation failed: {e}")
             return self._center_blob_mask(image.shape[:2])
-    
-    def cached_result(self, cache_key: str, 
-                     fallback_func, *args, **kwargs) -> Any:
-        """
-        Try to retrieve cached result or compute with fallback.
-        
-        Args:
-            cache_key: Cache identifier
-            fallback_func: Function to call if not cached
-            *args, **kwargs: Function arguments
-            
-        Returns:
-            Cached or computed result
-        """
-        # Simple in-memory cache implementation
+
+    def cached_result(self, cache_key: str, fallback_func, *args, **kwargs) -> Any:
         if not hasattr(self, '_cache'):
             self._cache = {}
-        
         if cache_key in self._cache:
             self.logger.info(f"Using cached result for {cache_key}")
             return self._cache[cache_key]
-        
         try:
             result = fallback_func(*args, **kwargs)
             self._cache[cache_key] = result
-            
-            # Limit cache size
             if len(self._cache) > 100:
-                # Remove oldest entries
                 keys = list(self._cache.keys())
                 for key in keys[:20]:
                     del self._cache[key]
-            
             return result
-            
         except Exception as e:
             self.logger.error(f"Cached computation failed: {e}")
-            return None
+            return None