Rename app.py to core/app.py

app.py → core/app.py

@@ -2,32 +2,20 @@
 """
 BackgroundFX Pro - Main Application Entry Point
 Refactored modular architecture - orchestrates specialized components
-
-This file has been refactored from a monolithic 600+ line structure into
-a clean orchestration layer that coordinates specialized modules:
-- config: Application configuration and environment variables
-- device_manager: Hardware detection and optimization
-- memory_manager: Memory and GPU resource management
-- model_loader: AI model loading and validation
-- video_processor: Core video processing pipeline
-- audio_processor: Audio track handling and FFmpeg operations
-- progress_tracker: Progress monitoring and ETA calculations
-- exceptions: Custom exception classes for better error handling
 """
 
 import os
-import sys
 import logging
 import threading
 from pathlib import Path
 from typing import Optional, Tuple, Dict, Any, Callable
 
-# Add parent directory to path for imports
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-# Configure logging
-from utils.logging import setup_logging
-logger = setup_logging(__name__)
+# Configure logging first
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
 
 # Apply Gradio schema patch early (before other imports)
 try:
@@ -50,26 +38,20 @@ def patched_get_type(schema):
 except Exception as e:
     logger.error(f"Gradio patch failed: {e}")
 
-# Import configuration
-from config.app_config import ProcessingConfig
+# Import configuration from new location
+from config.app_config import ProcessingConfig, get_config
 
-# Import core components
+# Import core components from new locations
 from core.exceptions import ModelLoadingError, VideoProcessingError
-
-# Import utilities
 from utils.hardware.device_manager import DeviceManager
 from utils.system.memory_manager import MemoryManager
-from utils.monitoring.progress_tracker import ProgressTracker
-
-# Import models
 from models.loaders.model_loader import ModelLoader
-
-# Import processing components
 from processing.video.video_processor import CoreVideoProcessor
 from processing.audio.audio_processor import AudioProcessor
+from utils.monitoring.progress_tracker import ProgressTracker
 
-# Import legacy utilities (during migration)
-from utils.legacy import (
+# Import existing utilities (temporary during migration)
+from utilities import (
     segment_person_hq,
     refine_mask_hq,
     replace_background_hq,
@@ -80,21 +62,20 @@ def patched_get_type(schema):
 
 # Import two-stage processor if available
 try:
-    from processing.advanced.two_stage_processor import TwoStageProcessor, CHROMA_PRESETS
+    from two_stage_processor import TwoStageProcessor, CHROMA_PRESETS
     TWO_STAGE_AVAILABLE = True
 except ImportError:
     TWO_STAGE_AVAILABLE = False
     CHROMA_PRESETS = {'standard': {}}
-    logger.warning("Two-stage processor not available")
 
-class BackgroundFXProcessor:
+class VideoProcessor:
     """
     Main video processing orchestrator - coordinates all specialized components
     """
     
-    def __init__(self, config: Optional[ProcessingConfig] = None):
+    def __init__(self):
         """Initialize the video processor with all required components"""
-        self.config = config or ProcessingConfig()
+        self.config = get_config()  # Use singleton config
         self.device_manager = DeviceManager()
         self.memory_manager = MemoryManager(self.device_manager.get_optimal_device())
         self.model_loader = ModelLoader(self.device_manager, self.memory_manager)
@@ -110,21 +91,19 @@ def __init__(self, config: Optional[ProcessingConfig] = None):
         self.loading_lock = threading.Lock()
         self.cancel_event = threading.Event()
         
-        logger.info(f"BackgroundFX Pro initialized on device: {self.device_manager.get_optimal_device()}")
-        logger.info(f"Configuration: {self.config.to_dict()}")
+        logger.info(f"VideoProcessor initialized on device: {self.device_manager.get_optimal_device()}")
     
     def load_models(self, progress_callback: Optional[Callable] = None) -> str:
         """Load and validate all AI models"""
         with self.loading_lock:
             if self.models_loaded:
-                return "✓ Models already loaded and validated"
+                return "Models already loaded and validated"
             
             try:
                 self.cancel_event.clear()
-                self.progress_tracker.start("model_loading")
                 
                 if progress_callback:
-                    progress_callback(0.0, f"🚀 Starting model loading on {self.device_manager.get_optimal_device()}")
+                    progress_callback(0.0, f"Starting model loading on {self.device_manager.get_optimal_device()}")
                 
                 # Load models using the specialized loader
                 sam2_predictor, matanyone_model = self.model_loader.load_all_models(
@@ -133,8 +112,7 @@ def load_models(self, progress_callback: Optional[Callable] = None) -> str:
                 )
                 
                 if self.cancel_event.is_set():
-                    self.progress_tracker.cancel("model_loading")
-                    return "⚠️ Model loading cancelled"
+                    return "Model loading cancelled"
                 
                 # Initialize core processor with loaded models
                 self.core_processor = CoreVideoProcessor(
@@ -147,32 +125,24 @@ def load_models(self, progress_callback: Optional[Callable] = None) -> str:
                 # Initialize two-stage processor if available
                 if TWO_STAGE_AVAILABLE and sam2_predictor and matanyone_model:
                     try:
-                        self.two_stage_processor = TwoStageProcessor(
-                            sam2_predictor, 
-                            matanyone_model,
-                            memory_manager=self.memory_manager
-                        )
-                        logger.info("✓ Two-stage processor initialized")
+                        self.two_stage_processor = TwoStageProcessor(sam2_predictor, matanyone_model)
+                        logger.info("Two-stage processor initialized")
                     except Exception as e:
                         logger.warning(f"Two-stage processor init failed: {e}")
                 
                 self.models_loaded = True
-                self.progress_tracker.complete("model_loading")
-                
                 message = self.model_loader.get_load_summary()
                 logger.info(message)
-                return f"✓ {message}"
+                return message
                 
             except ModelLoadingError as e:
                 self.models_loaded = False
-                self.progress_tracker.fail("model_loading", str(e))
-                error_msg = f"❌ Model loading failed: {str(e)}"
+                error_msg = f"Model loading failed: {str(e)}"
                 logger.error(error_msg)
                 return error_msg
             except Exception as e:
                 self.models_loaded = False
-                self.progress_tracker.fail("model_loading", str(e))
-                error_msg = f"❌ Unexpected error during model loading: {str(e)}"
+                error_msg = f"Unexpected error during model loading: {str(e)}"
                 logger.error(error_msg)
                 return error_msg
     
@@ -185,48 +155,40 @@ def process_video(
         use_two_stage: bool = False,
         chroma_preset: str = "standard",
         preview_mask: bool = False,
-        preview_greenscreen: bool = False,
-        **kwargs
+        preview_greenscreen: bool = False
     ) -> Tuple[Optional[str], str]:
         """Process video with the specified parameters"""
         
         if not self.models_loaded or not self.core_processor:
-            return None, "❌ Models not loaded. Please load models first."
+            return None, "Models not loaded. Please load models first."
         
         if self.cancel_event.is_set():
-            return None, "⚠️ Processing cancelled"
+            return None, "Processing cancelled"
         
         # Validate input file
         is_valid, validation_msg = validate_video_file(video_path)
         if not is_valid:
-            return None, f"❌ Invalid video: {validation_msg}"
+            return None, f"Invalid video: {validation_msg}"
         
         try:
-            self.progress_tracker.start("video_processing")
-            
             # Route to appropriate processing method
             if use_two_stage and TWO_STAGE_AVAILABLE and self.two_stage_processor:
-                result = self._process_two_stage(
+                return self._process_two_stage(
                     video_path, background_choice, custom_background_path,
                     progress_callback, chroma_preset
                 )
             else:
-                result = self._process_single_stage(
+                return self._process_single_stage(
                     video_path, background_choice, custom_background_path,
                     progress_callback, preview_mask, preview_greenscreen
                 )
-            
-            self.progress_tracker.complete("video_processing")
-            return result
                 
         except VideoProcessingError as e:
-            self.progress_tracker.fail("video_processing", str(e))
             logger.error(f"Video processing failed: {e}")
-            return None, f"❌ Processing failed: {str(e)}"
+            return None, f"Processing failed: {str(e)}"
         except Exception as e:
-            self.progress_tracker.fail("video_processing", str(e))
             logger.error(f"Unexpected error during video processing: {e}")
-            return None, f"❌ Unexpected error: {str(e)}"
+            return None, f"Unexpected error: {str(e)}"
     
     def _process_single_stage(
         self,
@@ -263,10 +225,10 @@ def _process_single_stage(
             final_video_path = processed_video_path
         
         success_msg = (
-            f"✓ {process_message}\n"
-            f"📹 Background: {background_choice}\n"
-            f"⚡ Mode: Single-stage\n"
-            f"🖥️ Device: {self.device_manager.get_optimal_device()}"
+            f"{process_message}\n"
+            f"Background: {background_choice}\n"
+            f"Mode: Single-stage\n"
+            f"Device: {self.device_manager.get_optimal_device()}"
         )
         
         return final_video_path, success_msg
@@ -293,7 +255,7 @@ def _process_two_stage(
             background_choice, custom_background_path, frame_width, frame_height
         )
         if background is None:
-            return None, "❌ Failed to prepare background"
+            return None, "Failed to prepare background"
         
         # Process with two-stage pipeline
         import time
@@ -316,11 +278,11 @@ def _process_two_stage(
             return None, message
         
         success_msg = (
-            f"✓ Two-stage success!\n"
-            f"📹 Background: {background_choice}\n"
-            f"🎨 Preset: {chroma_preset}\n"
-            f"🌟 Quality: Cinema-grade\n"
-            f"🖥️ Device: {self.device_manager.get_optimal_device()}"
+            f"Two-stage success!\n"
+            f"Background: {background_choice}\n"
+            f"Preset: {chroma_preset}\n"
+            f"Quality: Cinema-grade\n"
+            f"Device: {self.device_manager.get_optimal_device()}"
         )
         
         return result, success_msg
@@ -328,14 +290,11 @@ def _process_two_stage(
     def get_status(self) -> Dict[str, Any]:
         """Get comprehensive status of all components"""
         base_status = {
-            'app': 'BackgroundFX Pro',
-            'version': '2.0.0',
             'models_loaded': self.models_loaded,
             'two_stage_available': TWO_STAGE_AVAILABLE and self.two_stage_processor is not None,
             'device': str(self.device_manager.get_optimal_device()),
             'memory_usage': self.memory_manager.get_memory_usage(),
-            'config': self.config.to_dict(),
-            'progress': self.progress_tracker.get_all_progress()
+            'config': self.config.to_dict()
         }
         
         # Add model-specific status if available
@@ -346,12 +305,15 @@ def get_status(self) -> Dict[str, Any]:
         if self.core_processor:
             base_status.update(self.core_processor.get_status())
         
+        # Add progress tracking if available
+        if self.progress_tracker:
+            base_status['progress'] = self.progress_tracker.get_all_progress()
+        
         return base_status
     
     def cancel_processing(self):
         """Cancel any ongoing processing"""
         self.cancel_event.set()
-        self.progress_tracker.cancel_all()
         logger.info("Processing cancellation requested")
     
     def cleanup_resources(self):
@@ -359,20 +321,12 @@ def cleanup_resources(self):
         self.memory_manager.cleanup_aggressive()
         if self.model_loader:
             self.model_loader.cleanup()
-        self.progress_tracker.cleanup()
         logger.info("Resources cleaned up")
 
 # Global processor instance for application
-processor = BackgroundFXProcessor()
-
-# ============================================================================
-# BACKWARD COMPATIBILITY LAYER
-# These functions maintain compatibility with existing UI and scripts
-# ============================================================================
-
-# Alias for backward compatibility
-VideoProcessor = BackgroundFXProcessor
+processor = VideoProcessor()
 
+# Backward compatibility functions for existing UI
 def load_models_with_validation(progress_callback: Optional[Callable] = None) -> str:
     """Load models with validation - backward compatibility wrapper"""
     return processor.load_models(progress_callback)
@@ -408,25 +362,22 @@ def get_cache_status() -> Dict[str, Any]:
 def main():
     """Main application entry point"""
     try:
-        logger.info("=" * 60)
-        logger.info("Starting BackgroundFX Pro")
-        logger.info("=" * 60)
+        logger.info("Starting Video Background Replacement application")
         logger.info(f"Device: {processor.device_manager.get_optimal_device()}")
         logger.info(f"Two-stage available: {TWO_STAGE_AVAILABLE}")
         logger.info("Modular architecture loaded successfully")
         
         # Import and create UI
-        from web.ui_components import create_interface
+        from ui_components import create_interface
         demo = create_interface()
         
         # Launch application
-        demo.queue(max_size=10).launch(
+        demo.queue().launch(
             server_name="0.0.0.0",
             server_port=7860,
             share=True,
             show_error=True,
-            debug=False,
-            favicon_path="web/static/favicon.ico"
+            debug=False
         )
         
     except Exception as e: