hhh

ui.py

@@ -1,25 +1,14 @@
 #!/usr/bin/env python3
 """
-BackgroundFX Pro — Standalone UI Application
-Complete file with all necessary functions included
+BackgroundFX Pro — Main UI Application
+Clean, focused main file that coordinates the application
 """
 
 # ==== Runtime hygiene & paths (high in file) ====
 import os
 import sys
-import gc
-import time
-import json
-import uuid
-import shutil
 import logging
-import tempfile
-import threading
-import traceback
-import subprocess
 from pathlib import Path
-from typing import Optional, Callable, Dict, Any
-from datetime import datetime
 
 # Quiet BLAS/OpenMP thread spam (fixes "libgomp: Invalid value..." warning)
 os.environ.setdefault("OMP_NUM_THREADS", "4")
@@ -45,321 +34,11 @@
 
 # Standard imports
 import torch
-import numpy as np
-from PIL import Image
-import cv2
 import gradio as gr
 
-# Configure logging
-logger = logging.getLogger("backgroundfx_pro")
-if not logger.handlers:
-    h = logging.StreamHandler()
-    h.setFormatter(logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s"))
-    logger.addHandler(h)
-logger.setLevel(logging.INFO)
-
-# ===============================================================================
-# UTILITY FUNCTIONS
-# ===============================================================================
-
-def _disk_stats(p: Path) -> str:
-    """Get disk usage statistics for a path"""
-    try:
-        total, used, free = shutil.disk_usage(str(p))
-        to_mb = lambda x: x // (1024 * 1024)
-        return f"disk(total={to_mb(total)}MB, used={to_mb(used)}MB, free={to_mb(free)}MB)"
-    except Exception:
-        return "disk(n/a)"
-
-def startup_probe():
-    """Comprehensive startup probe - validates system readiness"""
-    try:
-        logger.info("🚀 BACKGROUNDFX PRO STARTUP PROBE")
-        logger.info(f"📁 Working directory: {os.getcwd()}")
-        logger.info(f"🐍 Python executable: {sys.executable}")
-        
-        # Write probe (critical - fail fast if can't write)
-        probe_file = TMP_ROOT / "startup_probe.txt"
-        probe_file.write_text("startup_test_ok", encoding="utf-8")
-        assert probe_file.read_text(encoding="utf-8") == "startup_test_ok"
-        logger.info(f"✅ WRITE PROBE OK: {probe_file} | {_disk_stats(APP_ROOT)}")
-        
-        probe_file.unlink(missing_ok=True)
-        
-        # GPU/Torch status
-        try:
-            logger.info(f"🔧 Torch version: {torch.__version__}")
-            if hasattr(torch.version, 'cuda') and torch.version.cuda:
-                logger.info(f"🔧 CUDA version: {torch.version.cuda}")
-            
-            if torch.cuda.is_available():
-                gpu_count = torch.cuda.device_count()
-                gpu_name = torch.cuda.get_device_name(0) if gpu_count > 0 else "Unknown"
-                gpu_memory = torch.cuda.get_device_properties(0).total_memory / 1e9
-                logger.info(f"🔥 GPU Available: {gpu_name} ({gpu_count} devices)")
-                logger.info(f"💾 GPU Memory: {gpu_memory:.1f}GB")
-            else:
-                logger.warning("⚠️  No GPU available - using CPU only")
-        except Exception as e:
-            logger.warning(f"⚠️  Torch check failed: {e}")
-        
-        # Directory verification
-        required_dirs = ['checkpoints', 'models', 'utils']
-        for dir_name in required_dirs:
-            dir_path = APP_ROOT / dir_name
-            if dir_path.exists():
-                logger.info(f"✅ Directory found: {dir_name}")
-            else:
-                logger.warning(f"⚠️  Missing directory: {dir_name}")
-        
-        logger.info("🎯 Startup probe completed successfully - system ready!")
-        
-    except Exception as e:
-        logger.error(f"❌ STARTUP PROBE FAILED: {e}")
-        logger.error(f"📊 Disk stats: {_disk_stats(APP_ROOT)}")
-        raise RuntimeError(f"Startup probe failed - system not ready: {e}") from e
-
-def create_job_directory() -> Path:
-    """Create unique job directory for processing"""
-    job_id = str(uuid.uuid4())[:8]
-    job_dir = JOB_ROOT / f"job_{job_id}_{int(time.time())}"
-    job_dir.mkdir(parents=True, exist_ok=True)
-    logger.info(f"📁 Created job directory: {job_dir}")
-    return job_dir
-
-def clear_gpu_memory():
-    """Aggressive GPU memory cleanup"""
-    try:
-        if torch.cuda.is_available():
-            torch.cuda.empty_cache()
-            torch.cuda.synchronize()
-            gc.collect()
-            logger.info("🧹 GPU memory cleared")
-    except Exception as e:
-        logger.warning(f"⚠️ GPU cleanup warning: {e}")
-
-# ===============================================================================
-# PROGRESS TRACKING
-# ===============================================================================
-
-class ProgressTracker:
-    """Thread-safe progress tracking for video processing"""
-    
-    def __init__(self):
-        self.current_step = ""
-        self.progress = 0.0
-        self.start_time = time.time()
-        self.lock = threading.Lock()
-        
-    def update(self, step: str, progress: float = None):
-        """Update progress safely"""
-        with self.lock:
-            self.current_step = step
-            if progress is not None:
-                self.progress = max(0.0, min(1.0, progress))
-                
-    def get_status(self) -> Dict[str, Any]:
-        """Get current status safely"""
-        with self.lock:
-            elapsed = time.time() - self.start_time
-            return {
-                "step": self.current_step, 
-                "progress": self.progress,
-                "elapsed": elapsed
-            }
-
-# Global progress tracker
-progress_tracker = ProgressTracker()
-
-# ===============================================================================
-# VIDEO PROCESSING PIPELINE
-# ===============================================================================
-
-def process_video_safe(
-    video_path: str,
-    background_image: Optional[Image.Image],
-    background_type: str = "custom",
-    background_prompt: str = "",
-    progress_callback: Optional[Callable] = None
-) -> str:
-    """Safe video processing with comprehensive error handling"""
-    job_dir = None
-    try:
-        logger.info("=" * 60)
-        logger.info("=== STARTING SAFE VIDEO PROCESSING ===")
-        logger.info("=" * 60)
-        
-        # Create job directory
-        job_dir = create_job_directory()
-        
-        # Pre-flight checks
-        logger.info(f"DEBUG: Video path: {video_path}")
-        logger.info(f"DEBUG: Video exists: {Path(video_path).exists()}")
-        logger.info(f"DEBUG: Job directory: {job_dir}")
-        logger.info(f"DEBUG: Background image size: {background_image.size if background_image else 'None'}")
-        
-        # Import two-stage pipeline
-        logger.info("DEBUG: Attempting to import two-stage pipeline...")
-        from two_stage_pipeline import process_two_stage as pipeline_process
-        logger.info("✓ Two-stage pipeline imported successfully")
-        
-        # Enhanced progress callback with stage monitoring
-        current_stage = {"stage": "init"}
-        
-        def safe_progress_callback(step: str, progress: float = None):
-            try:
-                # Stage detection
-                if "Stage 1" in step:
-                    if current_stage["stage"] != "stage1":
-                        current_stage["stage"] = "stage1"
-                        logger.info("🔄 STAGE TRANSITION: Entering Stage 1 (SAM2)")
-                elif "Stage 2" in step:
-                    if current_stage["stage"] != "stage2":
-                        current_stage["stage"] = "stage2"
-                        logger.info("🔄 STAGE TRANSITION: Entering Stage 2 (Composition)")
-                elif "Done" in step:
-                    if current_stage["stage"] != "complete":
-                        current_stage["stage"] = "complete"
-                        logger.info("🔄 STAGE TRANSITION: Pipeline Complete")
-                
-                logger.info(f"PROGRESS [{current_stage['stage'].upper()}]: {step} ({progress})")
-                progress_tracker.update(step, progress)
-                
-                if progress_callback:
-                    if progress is not None:
-                        progress_callback(f"Progress: {progress:.1%} - {step}")
-                    else:
-                        progress_callback(step)
-            except Exception as e:
-                logger.error(f"Progress callback error: {e}")
-        
-        # Validation
-        if background_image is None:
-            raise ValueError("Background image is required")
-            
-        logger.info("DEBUG: Pre-pipeline validation complete")
-        logger.info(f"DEBUG: Job dir contents before: {list(job_dir.iterdir()) if job_dir.exists() else 'does not exist'}")
-        
-        # Call two-stage pipeline
-        result_path = pipeline_process(
-            video_path=video_path,
-            background_image=background_image,
-            workdir=job_dir,
-            progress=safe_progress_callback,
-            use_matany=True
-        )
-        
-        logger.info(f"DEBUG: Pipeline returned: {result_path}")
-        
-        # Post-processing validation
-        if result_path:
-            result_file = Path(result_path)
-            logger.info(f"DEBUG: Result file exists: {result_file.exists()}")
-            if result_file.exists():
-                logger.info(f"DEBUG: Result file size: {result_file.stat().st_size} bytes")
-            logger.info(f"DEBUG: Job dir contents after: {list(job_dir.iterdir())}")
-        
-        if not result_path or not Path(result_path).exists():
-            raise RuntimeError("Pipeline processing failed - no output produced")
-            
-        logger.info("=" * 60)
-        logger.info(f"✅ PROCESSING COMPLETED: {result_path}")
-        logger.info("=" * 60)
-        return result_path
-        
-    except Exception as e:
-        logger.error("=" * 60)
-        logger.error(f"❌ PROCESSING FAILED:")
-        logger.error(f"   Error type: {type(e).__name__}")
-        logger.error(f"   Error message: {e}")
-        logger.error(f"   Job directory exists: {job_dir.exists() if job_dir else 'unknown'}")
-        if job_dir and job_dir.exists():
-            logger.error(f"   Job directory contents: {list(job_dir.iterdir())}")
-        logger.error("   Full traceback:")
-        logger.error(f"{traceback.format_exc()}")
-        logger.error("=" * 60)
-        
-        # Cleanup GPU memory on error
-        clear_gpu_memory()
-        raise
-
-# ===============================================================================
-# GRADIO INTERFACE
-# ===============================================================================
-
-def create_interface():
-    """Create and configure Gradio interface"""
-    
-    def process_video_wrapper(video_file, background_image):
-        """Wrapper function for Gradio processing"""
-        try:
-            if not video_file:
-                return None, "Please upload a video file"
-                
-            if not background_image:
-                return None, "Please upload a background image"
-            
-            logger.info(f"Processing video: {video_file}")
-            logger.info(f"Background image size: {background_image.size}")
-            
-            # Process video
-            result_path = process_video_safe(
-                video_path=video_file,
-                background_image=background_image,
-                background_type="custom",
-                background_prompt="",
-                progress_callback=lambda msg: logger.info(f"UI Progress: {msg}")
-            )
-            
-            return result_path, "✅ Processing completed successfully!"
-            
-        except Exception as e:
-            error_msg = f"❌ Processing failed: {str(e)}"
-            logger.error(error_msg)
-            return None, error_msg
-    
-    # Create Gradio interface
-    with gr.Blocks(title="BackgroundFX Pro", theme=gr.themes.Soft()) as demo:
-        gr.Markdown("# 🎬 BackgroundFX Pro")
-        gr.Markdown("Professional video background replacement using SAM2 + MatAnyone")
-        
-        with gr.Row():
-            with gr.Column():
-                video_input = gr.Video(
-                    label="📹 Upload Video",
-                    format="mp4"
-                )
-                background_input = gr.Image(
-                    label="🖼️ Background Image",
-                    type="pil"
-                )
-                
-                process_btn = gr.Button(
-                    "🚀 Process Video",
-                    variant="primary",
-                    size="lg"
-                )
-            
-            with gr.Column():
-                video_output = gr.Video(
-                    label="✨ Processed Video",
-                    format="mp4"
-                )
-                status_output = gr.Textbox(
-                    label="📊 Status",
-                    value="Ready to process...",
-                    interactive=False
-                )
-        
-        # Process button click
-        process_btn.click(
-            fn=process_video_wrapper,
-            inputs=[video_input, background_input],
-            outputs=[video_output, status_output],
-            show_progress=True
-        )
-    
-    return demo
+# Import our modules
+from ui_core_functionality import startup_probe, logger
+from ui_core_interface import create_interface
 
 # ===============================================================================
 # MAIN APPLICATION
@@ -375,8 +54,8 @@ def main():
         logger.info("🚀 Launching Gradio interface...")
         demo = create_interface()
         
-        # Configure for single-user processing (avoid file conflicts)
-        demo.queue(concurrency_count=1, max_size=2, api_open=False)
+        # Fixed queue configuration for Gradio 5.46.0
+        demo.queue()
         
         demo.launch(
             server_name="0.0.0.0",