kkK

app.py

@@ -1,146 +1,138 @@
 #!/usr/bin/env python3
 """
 VideoBackgroundReplacer2 - SAM2 + MatAnyone Integration
-================================================
+=======================================================
 - Sets up Gradio UI and launches pipeline
-- Aligned with torch==2.3.1+cu121, MatAnyone v1.0.0, SAM2 commit 3c76f73c1a7e7b4a2e8a0a9a3e5b92f7e6e3f2f5
+- Aligned with torch/cu121 stack; runs on HF Spaces (Docker)
 
-Changes (2025-09-16):
-- Enhanced error handling and model verification
-- Added GPU memory management
-- Improved logging and diagnostics
-- Added model verification on startup
+Changes (2025-09-18):
+- Added precise web-stack probes (FastAPI/Starlette/Pydantic/etc. versions + file paths)
+- Added toggleable "mount mode": run Gradio inside our own FastAPI app
+  and provide a safe /config route shim (uses demo.get_config_file()).
+- Kept your startup diagnostics, GPU logging, and heartbeats
 """
 
 print("=== APP STARTUP: Initializing VideoBackgroundReplacer2 ===")
+
+# ---------------------------------------------------------------------
+# Imports & basic setup
+# ---------------------------------------------------------------------
 import sys
 import os
 import gc
-import torch
+import json
 import logging
 import threading
 import time
 import warnings
 import traceback
+import subprocess
 from pathlib import Path
 from loguru import logger
 
-# Configure logging
+# Logging (loguru to stderr)
 logger.remove()
 logger.add(
     sys.stderr,
-    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
+    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> "
+           "| <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
 )
 
-# Set up warnings
+# Warnings
 warnings.filterwarnings("ignore", category=UserWarning)
 warnings.filterwarnings("ignore", category=FutureWarning)
 warnings.filterwarnings("ignore", module="torchvision.io._video_deprecation_warning")
 
-# Environment setup
-os.environ["OMP_NUM_THREADS"] = "1"
-os.environ["TOKENIZERS_PARALLELISM"] = "false"
+# Environment (lightweight & safe in Spaces)
+os.environ.setdefault("OMP_NUM_THREADS", "1")
+os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
 
-# --- Path Configuration ---
+# Paths
 BASE_DIR = Path(__file__).parent.absolute()
 THIRD_PARTY_DIR = BASE_DIR / "third_party"
 SAM2_DIR = THIRD_PARTY_DIR / "sam2"
 CHECKPOINTS_DIR = BASE_DIR / "checkpoints"
 
-# Add to Python path
-for p in [str(THIRD_PARTY_DIR), str(SAM2_DIR)]:
+# Python path extends
+for p in (str(THIRD_PARTY_DIR), str(SAM2_DIR)):
     if p not in sys.path:
         sys.path.insert(0, p)
 
 logger.info(f"Base directory: {BASE_DIR}")
-logger.info(f"Python path: {sys.path}")
+logger.info(f"Python path[0:5]: {sys.path[:5]}")
+
+# ---------------------------------------------------------------------
+# GPU / Torch diagnostics (non-blocking)
+# ---------------------------------------------------------------------
+try:
+    import torch
+except Exception as e:
+    logger.warning("Torch import failed at startup: %s", e)
+    torch = None
 
-# --- GPU Configuration ---
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DEVICE = "cuda" if (torch and torch.cuda.is_available()) else "cpu"
 if DEVICE == "cuda":
     os.environ["SAM2_DEVICE"] = "cuda"
     os.environ["MATANY_DEVICE"] = "cuda"
-    os.environ["CUDA_LAUNCH_BLOCKING"] = "0"
-    logger.info(f"CUDA available: {torch.cuda.get_device_name(0)}")
+    os.environ.setdefault("CUDA_LAUNCH_BLOCKING", "0")
+    try:
+        logger.info(f"CUDA available: {torch.cuda.get_device_name(0)}")
+    except Exception:
+        logger.info("CUDA device name not available at startup.")
 else:
     os.environ["SAM2_DEVICE"] = "cpu"
     os.environ["MATANY_DEVICE"] = "cpu"
     logger.warning("CUDA not available, falling back to CPU")
 
-# --- Model Verification ---
 def verify_models():
-    """Verify that all required models are available and loadable."""
+    """Verify critical model files exist and are loadable (cheap checks)."""
     results = {"status": "success", "details": {}}
-    
-    # Check SAM2 model
     try:
         sam2_model_path = os.getenv("SAM2_MODEL_PATH", str(CHECKPOINTS_DIR / "sam2_hiera_large.pt"))
         if not os.path.exists(sam2_model_path):
             raise FileNotFoundError(f"SAM2 model not found at {sam2_model_path}")
-        
-        # Try to load a small part of the model to verify it's not corrupted
-        state_dict = torch.load(sam2_model_path, map_location=DEVICE)
-        if not isinstance(state_dict, dict):
-            raise ValueError("Invalid SAM2 model format")
-            
+        # Cheap load test (map to CPU to avoid VRAM use during boot)
+        if torch:
+            sd = torch.load(sam2_model_path, map_location="cpu")
+            if not isinstance(sd, dict):
+                raise ValueError("Invalid SAM2 checkpoint format")
         results["details"]["sam2"] = {
             "status": "success",
             "path": sam2_model_path,
-            "size_mb": os.path.getsize(sam2_model_path) / (1024 * 1024)
+            "size_mb": round(os.path.getsize(sam2_model_path) / (1024 * 1024), 2),
         }
     except Exception as e:
         results["status"] = "error"
         results["details"]["sam2"] = {
             "status": "error",
             "error": str(e),
-            "traceback": traceback.format_exc()
+            "traceback": traceback.format_exc(),
         }
-    
     return results
 
-# --- Startup Diagnostics ---
 def run_startup_diagnostics():
-    """Run comprehensive system and model diagnostics."""
     diag = {
         "system": {
             "python": sys.version,
-            "pytorch": torch.__version__,
-            "cuda_available": torch.cuda.is_available(),
-            "device_count": torch.cuda.device_count() if torch.cuda.is_available() else 0,
-            "cuda_version": torch.version.cuda if hasattr(torch.version, 'cuda') else None,
+            "pytorch": getattr(torch, "__version__", None) if torch else None,
+            "cuda_available": bool(torch and torch.cuda.is_available()),
+            "device_count": (torch.cuda.device_count() if torch and torch.cuda.is_available() else 0),
+            "cuda_version": getattr(getattr(torch, "version", None), "cuda", None) if torch else None,
         },
         "paths": {
             "base_dir": str(BASE_DIR),
             "checkpoints_dir": str(CHECKPOINTS_DIR),
             "sam2_dir": str(SAM2_DIR),
-            "python_path": sys.path
         },
-        "environment": dict(os.environ)
+        "env_subset": {k: v for k, v in os.environ.items() if k in ("HOST", "PORT", "SPACE_ID", "SPACE_AUTHOR_NAME")},
     }
-    
-    # Run model verification
     diag["model_verification"] = verify_models()
-    
     return diag
 
-# Run diagnostics on startup
 startup_diag = run_startup_diagnostics()
 logger.info("Startup diagnostics completed")
 
-# Import Gradio after environment setup
-import gradio as gr
-
-# -----------------------------------------------------------------------------
-# Logging early
-# -----------------------------------------------------------------------------
-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)
-
-# Heartbeat so logs never go silent during startup/imports
+# Noisy heartbeat so logs show life during import time
 def _heartbeat():
     i = 0
     while True:
@@ -150,56 +142,17 @@ def _heartbeat():
 
 threading.Thread(target=_heartbeat, daemon=True).start()
 
-# -----------------------------------------------------------------------------
-# Safe, minimal startup diagnostics (no long CUDA probes)
-# -----------------------------------------------------------------------------
-def _safe_startup_diag():
-    # Torch version
-    try:
-        import torch
-        import importlib
-        t = importlib.import_module("torch")
-        logger.info(
-            "torch imported: %s | torch.version.cuda=%s",
-            getattr(t, "__version__", "?"),
-            getattr(getattr(t, "version", None), "cuda", None),
-        )
-    except Exception as e:
-        logger.warning("Torch not available at startup: %s", e)
-
-    # MatAnyone version
-    try:
-        import importlib.metadata
-        version = importlib.metadata.version("matanyone")
-        logger.info(f"[MATANY] MatAnyone version: {version}")
-    except Exception:
-        logger.info("[MATANY] MatAnyone version unknown")
-
-    # nvidia-smi with short timeout (avoid indefinite block)
-    try:
-        out = subprocess.run(
-            ["nvidia-smi", "-L"], capture_output=True, text=True, timeout=2
-        )
-        if out.returncode == 0:
-            logger.info("nvidia-smi -L:\n%s", out.stdout.strip())
-        else:
-            logger.warning("nvidia-smi -L failed or unavailable (rc=%s).", out.returncode)
-    except subprocess.TimeoutExpired:
-        logger.warning("nvidia-smi -L timed out (skipping).")
-    except Exception as e:
-        logger.warning("nvidia-smi not runnable: %s", e)
-
-# Optional perf tuning; never block startup
+# Optional perf tuning import (non-fatal)
 try:
-    import perf_tuning
+    import perf_tuning  # noqa: F401
     logger.info("perf_tuning imported successfully.")
 except Exception as e:
     logger.info("perf_tuning not available: %s", e)
 
-# MatAnyone API detection probe (non-instantiating)
+# MatAnyone non-instantiating probe
 try:
     import inspect
-    from matanyone.inference import inference_core as ic
+    from matanyone.inference import inference_core as ic  # type: ignore
     sigs = {}
     for name in ("InferenceCore",):
         obj = getattr(ic, name, None)
@@ -209,47 +162,149 @@ def _safe_startup_diag():
 except Exception as e:
     logger.info(f"[MATANY] probe skipped: {e}")
 
-# Continue with app startup
-_safe_startup_diag()
+# ---------------------------------------------------------------------
+# Gradio import and web-stack probes
+# ---------------------------------------------------------------------
+import gradio as gr
 
-# -----------------------------------------------------------------------------
-# Post-launch CUDA diag in background (so it never blocks binding the port)
-# -----------------------------------------------------------------------------
-def _post_launch_diag():
+# Standard logger for some libs that use stdlib logging
+py_logger = logging.getLogger("backgroundfx_pro")
+if not py_logger.handlers:
+    h = logging.StreamHandler()
+    h.setFormatter(logging.Formatter("[%(asctime)s] %(levelname)s: %(message)s"))
+    py_logger.addHandler(h)
+py_logger.setLevel(logging.INFO)
+
+def _log_web_stack_versions_and_paths():
+    import inspect
     try:
-        import torch
+        import fastapi, starlette, pydantic, httpx, anyio
         try:
-            avail = torch.cuda.is_available()
-        except Exception as e:
-            logger.warning("torch.cuda.is_available() failed: %s", e)
-            avail = False
-        logger.info("CUDA available: %s", avail)
+            import pydantic_core
+            pc_ver = pydantic_core.__version__
+        except Exception:
+            pc_ver = "unknown"
+        logger.info(
+            "[WEB-STACK] fastapi=%s | starlette=%s | pydantic=%s | pydantic-core=%s | httpx=%s | anyio=%s",
+            getattr(fastapi, "__version__", "?"),
+            getattr(starlette, "__version__", "?"),
+            getattr(pydantic, "__version__", "?"),
+            pc_ver,
+            getattr(httpx, "__version__", "?"),
+            getattr(anyio, "__version__", "?"),
+        )
+    except Exception as e:
+        logger.warning("[WEB-STACK] version probe failed: %s", e)
+
+    try:
+        import gradio
+        import gradio.routes as gr_routes
+        import gradio.queueing as gr_queueing
+        logger.info("[PATH] gradio.__file__     = %s", getattr(gradio, "__file__", "?"))
+        logger.info("[PATH] gradio.routes       = %s", inspect.getfile(gr_routes))
+        logger.info("[PATH] gradio.queueing     = %s", inspect.getfile(gr_queueing))
+        import starlette.exceptions as st_exc
+        logger.info("[PATH] starlette.exceptions= %s", inspect.getfile(st_exc))
+    except Exception as e:
+        logger.warning("[PATH] probe failed: %s", e)
+
+def _post_launch_diag():
+    try:
+        if not torch:
+            return
+        avail = torch.cuda.is_available()
+        logger.info("CUDA available (post-launch): %s", avail)
         if avail:
-            try:
-                idx = torch.cuda.current_device()
-                name = torch.cuda.get_device_name(idx)
-                cap = torch.cuda.get_device_capability(idx)
-                logger.info("CUDA device %d: %s (cc %d.%d)", idx, name, cap[0], cap[1])
-            except Exception as e:
-                logger.warning("CUDA device query failed: %s", e)
+            idx = torch.cuda.current_device()
+            name = torch.cuda.get_device_name(idx)
+            cap = torch.cuda.get_device_capability(idx)
+            logger.info("CUDA device %d: %s (cc %d.%d)", idx, name, cap[0], cap[1])
     except Exception as e:
-        logger.warning("Post-launch torch diag failed: %s", e)
+        logger.warning("Post-launch CUDA diag failed: %s", e)
 
-# -----------------------------------------------------------------------------
-# Build UI (in separate module) and launch
-# -----------------------------------------------------------------------------
+# ---------------------------------------------------------------------
+# UI factory (your existing UI builder)
+# ---------------------------------------------------------------------
 def build_ui() -> gr.Blocks:
-    # Import here so any heavy imports inside ui.py (it shouldn’t) would show up after logs are configured
-    from ui import create_interface
+    from ui import create_interface  # your module
     return create_interface()
 
+# ---------------------------------------------------------------------
+# Optional: custom FastAPI mount mode
+#   Why: if Gradio’s internal FastAPI route schema generation glitches
+#   under certain dependency combos, we can run our own FastAPI app,
+#   mount Gradio, and provide a known-good /config shim.
+# ---------------------------------------------------------------------
+def build_fastapi_with_gradio(demo: gr.Blocks):
+    """
+    Returns a FastAPI app with Gradio mounted at root.
+    Also exposes JSON health and a config shim using demo.get_config_file().
+    """
+    from fastapi import FastAPI
+    from fastapi.responses import JSONResponse
+
+    app = FastAPI(title="VideoBackgroundReplacer2")
+
+    # Simple health
+    @app.get("/healthz")
+    def _healthz():
+        return {"ok": True, "ts": time.time()}
+
+    # Config shim — this bypasses Gradio's internal /config route
+    # and returns the same structure the frontend expects.
+    @app.get("/config")
+    def _config():
+        try:
+            cfg = demo.get_config_file()  # Gradio builds the JSON-able config dict
+            return JSONResponse(content=cfg)
+        except Exception as e:
+            # If something fails, return explicit JSON (so the frontend won't choke on HTML)
+            return JSONResponse(
+                status_code=500,
+                content={"error": "config_generation_failed", "detail": str(e)},
+            )
+
+    # Mount Gradio UI at root; static assets & index served by Gradio
+    app = gr.mount_gradio_app(app, demo, path="/")
+    return app
+
+# ---------------------------------------------------------------------
+# Entrypoint
+# ---------------------------------------------------------------------
 if __name__ == "__main__":
     host = os.environ.get("HOST", "0.0.0.0")
     port = int(os.environ.get("PORT", "7860"))
-    logger.info("Launching Gradio on %s:%s …", host, port)
+    mount_mode = os.environ.get("GRADIO_MOUNT_MODE", "0") == "1"
+
+    logger.info("Launching on %s:%s (mount_mode=%s)…", host, port, mount_mode)
+    _log_web_stack_versions_and_paths()
 
     demo = build_ui()
-    demo.queue(max_size=16, api_open=False)  # Disable public API for security
+    # Good defaults for Spaces
+    demo.queue(max_size=16, api_open=False)
 
     threading.Thread(target=_post_launch_diag, daemon=True).start()
-    demo.launch(server_name=host, server_port=port, show_error=True)
+
+    if mount_mode:
+        # Our own FastAPI + /config shim
+        try:
+            from uvicorn import run as uvicorn_run
+        except Exception:
+            logger.error("uvicorn is not installed; mount mode cannot start.")
+            raise
+
+        app = build_fastapi_with_gradio(demo)
+        # NOTE: In Docker Spaces, this process is PID1; we call uvicorn.run programmatically.
+        uvicorn_run(app=app, host=host, port=port, log_level="info")
+    else:
+        # Standard Gradio server (uses internal FastAPI app & routes)
+        demo.launch(
+            server_name=host,
+            server_port=port,
+            share=False,
+            show_api=False,   # keep off in Spaces
+            show_error=True,
+            quiet=False,
+            debug=True,
+            max_threads=1,
+        )