Update utils/__init__.py

utils/__init__.py

@@ -1,575 +1,164 @@
 #!/usr/bin/env python3
 """
-BackgroundFX Pro - CSP-Safe Application Entry Point
-Now with: live background preview + sources: Preset / Upload / Gradient / AI Generate
-- Lazy-loaded Diffusers pipeline (VRAM-aware: sd-turbo / sdxl-turbo / sd-2.1 CPU)
-- Preview shows the exact background used
-- Clears stale AI image when switching sources
+utils package (lightweight __init__)
+- Export only light helpers/consts at import time
+- Provide LAZY wrappers for heavy CV functions so legacy imports still work:
+    from utils import segment_person_hq  -> OK (resolved at call time)
 """
 
-import early_env  # <<< must be FIRST
-
-import os, time, math
-from typing import Optional, Dict, Any, Callable, Tuple
-
-# Prefer a writable cache in constrained environments (e.g., HF Spaces)
-os.environ.setdefault("HF_HOME", "/tmp/hf")
-os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
-
-# 1) CSP-safe Gradio env
-os.environ['GRADIO_ALLOW_FLAGGING'] = 'never'
-os.environ['GRADIO_ANALYTICS_ENABLED'] = 'False'
-os.environ['GRADIO_SERVER_NAME'] = '0.0.0.0'
-os.environ['GRADIO_SERVER_PORT'] = '7860'
-
-# 2) Gradio schema patch
-try:
-    import gradio_client.utils as gc_utils
-    _orig_get_type = gc_utils.get_type
-    def _patched_get_type(schema):
-        if not isinstance(schema, dict):
-            if isinstance(schema, bool): return "boolean"
-            if isinstance(schema, str): return "string"
-            if isinstance(schema, (int, float)): return "number"
-            return "string"
-        return _orig_get_type(schema)
-    gc_utils.get_type = _patched_get_type
-except Exception:
-    pass
-
-# 3) Logging early
-from utils.logging_setup import setup_logging, make_logger
-setup_logging(app_name="backgroundfx")
-logger = make_logger("entrypoint")
-logger.info("Entrypoint starting…")
-
-# 4) Imports
-from core.exceptions import ModelLoadingError, VideoProcessingError
-from config.app_config import get_config
-from utils.hardware.device_manager import DeviceManager
-from utils.system.memory_manager import MemoryManager
-from models.loaders.model_loader import ModelLoader
-from processing.video.video_processor import CoreVideoProcessor, ProcessorConfig
-from processing.audio.audio_processor import AudioProcessor
-
-# Background helpers (kept lightweight to avoid cycles)
-from utils import PROFESSIONAL_BACKGROUNDS, validate_video_file, create_professional_background
-# Gradient helper (add to utils; fallback here for preview only if missing)
-try:
-    from utils import create_gradient_background
-except Exception:
-    def create_gradient_background(spec: Dict[str, Any], width: int, height: int):
-        # Lightweight fallback (linear+rotate only)
-        import numpy as np
+from __future__ import annotations
+
+import os
+import logging
+from typing import Dict, Any, Tuple, Optional
+
+import numpy as np  # light; OK at import time
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------
+# Background presets & builders (lightweight)
+# ---------------------------------------------------------------------
+
+PROFESSIONAL_BACKGROUNDS: Dict[str, Dict[str, Any]] = {
+    "office":   {"color": (240, 248, 255), "gradient": True},
+    "studio":   {"color": (32, 32, 32),    "gradient": False},
+    "nature":   {"color": (34, 139, 34),   "gradient": True},
+    "abstract": {"color": (75, 0, 130),    "gradient": True},
+    "white":    {"color": (255, 255, 255), "gradient": False},
+    "black":    {"color": (0, 0, 0),       "gradient": False},
+    # add more if you like
+}
+
+def _solid_bg(color: Tuple[int,int,int], width: int, height: int) -> np.ndarray:
+    return np.full((height, width, 3), tuple(int(x) for x in color), dtype=np.uint8)
+
+def _vertical_gradient(top: Tuple[int,int,int], bottom: Tuple[int,int,int], width: int, height: int) -> np.ndarray:
+    bg = np.zeros((height, width, 3), dtype=np.uint8)
+    for y in range(height):
+        t = y / max(1, height - 1)
+        r = int(top[0] * (1 - t) + bottom[0] * t)
+        g = int(top[1] * (1 - t) + bottom[1] * t)
+        b = int(top[2] * (1 - t) + bottom[2] * t)
+        bg[y, :] = (r, g, b)
+    return bg
+
+def create_professional_background(key_or_cfg: Any, width: int, height: int) -> np.ndarray:
+    """
+    Accepts either:
+      - string key in PROFESSIONAL_BACKGROUNDS
+      - a config dict with {"color": (r,g,b), "gradient": bool}
+    Returns RGB uint8 background (H, W, 3).
+    """
+    if isinstance(key_or_cfg, str):
+        cfg = PROFESSIONAL_BACKGROUNDS.get(key_or_cfg, PROFESSIONAL_BACKGROUNDS["office"])
+    elif isinstance(key_or_cfg, dict):
+        cfg = key_or_cfg
+    else:
+        cfg = PROFESSIONAL_BACKGROUNDS["office"]
+
+    color = tuple(int(x) for x in cfg.get("color", (255, 255, 255)))
+    use_grad = bool(cfg.get("gradient", False))
+
+    if not use_grad:
+        return _solid_bg(color, width, height)
+
+    # simple vertical gradient dark->color
+    dark = (int(color[0]*0.7), int(color[1]*0.7), int(color[2]*0.7))
+    return _vertical_gradient(dark, color, width, height)
+
+def create_gradient_background(spec: Dict[str, Any], width: int, height: int) -> np.ndarray:
+    """
+    spec: {"type": "linear"|"radial", "start": (r,g,b)|"#RRGGBB", "end": (r,g,b)|"#RRGGBB", "angle_deg": float}
+    Returns RGB uint8 background (H, W, 3). (Radial treated as linear fallback unless extended.)
+    """
+    import re
+    import cv2  # import locally to keep top-level light
+
+    def _to_rgb(c):
+        if isinstance(c, (list, tuple)) and len(c) == 3:
+            return tuple(int(x) for x in c)
+        if isinstance(c, str) and re.match(r"^#[0-9a-fA-F]{6}$", c):
+            return tuple(int(c[i:i+2], 16) for i in (1,3,5))
+        return (255, 255, 255)
+
+    start = _to_rgb(spec.get("start", (32, 32, 32)))
+    end   = _to_rgb(spec.get("end", (200, 200, 200)))
+    angle = float(spec.get("angle_deg", 0.0))
+
+    bg = _vertical_gradient(start, end, width, height)
+
+    # rotate by angle
+    center = (width / 2, height / 2)
+    rot = cv2.getRotationMatrix2D(center, angle, 1.0)
+    bg = cv2.warpAffine(bg, rot, (width, height), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REFLECT_101)
+    return bg
+
+# ---------------------------------------------------------------------
+# Video validation (lightweight)
+# ---------------------------------------------------------------------
+def validate_video_file(video_path: str) -> bool:
+    """
+    Fast sanity check: file exists, cv2 can open, first frame is readable.
+    Returns True/False (lightweight for UI).
+    """
+    try:
+        if not video_path or not os.path.exists(video_path):
+            return False
+        import cv2  # local import
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            return False
+        ok, frame = cap.read()
+        cap.release()
+        return bool(ok and frame is not None)
+    except Exception as e:
+        logger.warning("validate_video_file error: %s", e)
+        return False
+
+def validate_video_file_detail(video_path: str) -> Tuple[bool, str]:
+    if not video_path:
+        return False, "No path provided"
+    if not os.path.exists(video_path):
+        return False, "File does not exist"
+    try:
         import cv2
-        def _to_rgb(c):
-            if isinstance(c, (list, tuple)) and len(c) == 3:
-                return tuple(int(x) for x in c)
-            if isinstance(c, str) and c.startswith("#") and len(c) == 7:
-                return tuple(int(c[i:i+2], 16) for i in (1,3,5))
-            return (255, 255, 255)
-        start = _to_rgb(spec.get("start", "#222222"))
-        end   = _to_rgb(spec.get("end", "#888888"))
-        angle = float(spec.get("angle_deg", 0))
-        bg = np.zeros((height, width, 3), np.uint8)
-        for y in range(height):
-            t = y / max(1, height - 1)
-            r = int(start[0] * (1 - t) + end[0] * t)
-            g = int(start[1] * (1 - t) + end[1] * t)
-            b = int(start[2] * (1 - t) + end[2] * t)
-            bg[y, :] = (r, g, b)
-        center = (width / 2, height / 2)
-        rot = cv2.getRotationMatrix2D(center, angle, 1.0)
-        return cv2.warpAffine(bg, rot, (width, height), flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REFLECT_101)
-
-# 5) CSP-safe fallbacks for models
-class CSPSafeSAM2:
-    def set_image(self, image):
-        self.shape = getattr(image, 'shape', (512, 512, 3))
-    def predict(self, point_coords=None, point_labels=None, box=None, multimask_output=True, **kwargs):
-        import numpy as np
-        h, w = self.shape[:2] if hasattr(self, 'shape') else (512, 512)
-        n = 3 if multimask_output else 1
-        return np.ones((n, h, w), dtype=bool), np.array([0.9, 0.8, 0.7][:n]), np.ones((n, h, w), dtype=np.float32)
-
-class CSPSafeMatAnyone:
-    def step(self, image_tensor, mask_tensor=None, objects=None, first_frame_pred=False, **kwargs):
-        import torch
-        if hasattr(image_tensor, "shape"):
-            if len(image_tensor.shape) == 3:
-                _, H, W = image_tensor.shape
-            elif len(image_tensor.shape) == 4:
-                _, _, H, W = image_tensor.shape
-            else:
-                H, W = 256, 256
-        else:
-            H, W = 256, 256
-        return torch.ones((1, 1, H, W))
-    def output_prob_to_mask(self, output_prob):
-        return (output_prob > 0.5).float()
-    def process(self, image, mask, **kwargs):
-        return mask
-
-# ---------- helpers for UI ----------
-import numpy as np
-import cv2
-from PIL import Image
-from typing import Tuple
-
-PREVIEW_W, PREVIEW_H = 640, 360  # 16:9
-
-def _hex_to_rgb(x: str) -> Tuple[int, int, int]:
-    x = (x or "").strip()
-    if x.startswith("#") and len(x) == 7:
-        return tuple(int(x[i:i+2], 16) for i in (1, 3, 5))
-    return (255, 255, 255)
-
-def _np_to_pil(arr: np.ndarray) -> Image.Image:
-    if arr.dtype != np.uint8:
-        arr = arr.clip(0, 255).astype(np.uint8)
-    return Image.fromarray(arr)
-
-def _div8(n: int) -> int:
-    # Ensure sizes are multiples of 8 for SD/VAEs (min 256)
-    n = int(n)
-    if n < 256: n = 256
-    return int(math.floor(n / 8.0) * 8)
-
-# ---------- main app ----------
-class VideoBackgroundApp:
-    def __init__(self):
-        self.config = get_config()
-        self.device_mgr = DeviceManager()
-        self.memory_mgr = MemoryManager(self.device_mgr.get_optimal_device())
-        self.model_loader = ModelLoader(self.device_mgr, self.memory_mgr)
-        self.audio_proc = AudioProcessor()
-        self.models_loaded = False
-        self.core_processor: Optional[CoreVideoProcessor] = None
-        # Text-to-image cache
-        self.t2i_pipe = None
-        self.t2i_model_id = None
-        logger.info("VideoBackgroundApp initialized (device=%s)", self.device_mgr.get_optimal_device())
-
-    def load_models(self, progress_callback: Optional[Callable] = None) -> str:
-        logger.info("Loading models (CSP-safe)…")
-        try:
-            sam2, matanyone = self.model_loader.load_all_models(progress_callback=progress_callback)
-        except Exception as e:
-            logger.warning("Model loading failed (%s) - Using CSP-safe fallbacks", e)
-            sam2, matanyone = None, None
-
-        sam2_model = getattr(sam2, "model", sam2) if sam2 else CSPSafeSAM2()
-        matanyone_model = getattr(matanyone, "model", matanyone) if matanyone else CSPSafeMatAnyone()
-
-        cfg = ProcessorConfig(
-            background_preset="office",
-            write_fps=None,
-            max_model_size=1280,
-            use_nvenc=True,
-            nvenc_codec="h264",
-            nvenc_preset="p5",
-            nvenc_cq=18,
-            nvenc_tune_hq=True,
-            nvenc_pix_fmt="yuv420p",
-        )
-        self.core_processor = CoreVideoProcessor(config=cfg, models=None)
-        self.core_processor.models = type('FakeModelManager', (), {
-            'get_sam2': lambda self_: sam2_model,
-            'get_matanyone': lambda self_: matanyone_model
-        })()
-
-        self.models_loaded = True
-        logger.info("Models ready (SAM2=%s, MatAnyOne=%s)",
-                    type(sam2_model).__name__, type(matanyone_model).__name__)
-        return "Models loaded (CSP-safe; fallbacks in use if actual AI models failed)."
-
-    # ---- PREVIEWS ----
-    def preview_preset(self, preset_key: str) -> Image.Image:
-        key = preset_key if preset_key in PROFESSIONAL_BACKGROUNDS else "office"
-        bg = create_professional_background(key, PREVIEW_W, PREVIEW_H)  # RGB
-        return _np_to_pil(bg)
-
-    def preview_upload(self, file) -> Optional[Image.Image]:
-        if file is None:
-            return None
-        try:
-            img = Image.open(file.name).convert("RGB")
-            img = img.resize((PREVIEW_W, PREVIEW_H), Image.LANCZOS)
-            return img
-        except Exception as e:
-            logger.warning("Upload preview failed: %s", e)
-            return None
-
-    def preview_gradient(self, gtype: str, color1: str, color2: str, angle: int) -> Image.Image:
-        spec = {
-            "type": (gtype or "linear").lower(),  # "linear" or "radial" (linear in fallback)
-            "start": _hex_to_rgb(color1 or "#222222"),
-            "end": _hex_to_rgb(color2 or "#888888"),
-            "angle_deg": float(angle or 0),
-        }
-        bg = create_gradient_background(spec, PREVIEW_W, PREVIEW_H)
-        return _np_to_pil(bg)
-
-    # ---- AI BG: lazy-load + reuse pipe ----
-    def _ensure_t2i(self):
-        """
-        Load a text-to-image pipeline once with memory-efficient settings.
-        Returns (pipe, model_id, msg).
-        """
-        if self.t2i_pipe is not None:
-            return self.t2i_pipe, self.t2i_model_id, "AI generator ready"
-
-        try:
-            import torch
-            from diffusers import AutoPipelineForText2Image, StableDiffusionPipeline
-        except Exception as e:
-            return None, None, f"AI generation unavailable (missing deps): {e}"
-
-        token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
-        device = "cuda" if getattr(__import__("torch"), "cuda", None) and __import__("torch").cuda.is_available() else "cpu"
-
-        # Try to estimate VRAM to pick a model
-        vram_gb = None
-        try:
-            vram_gb = self.device_mgr.get_device_memory_gb()
-        except Exception:
-            pass
-
-        if device == "cuda":
-            if vram_gb and vram_gb >= 12:
-                model_id = os.environ.get("BGFX_T2I_MODEL", "stabilityai/sdxl-turbo")
-            else:
-                model_id = os.environ.get("BGFX_T2I_MODEL", "stabilityai/sd-turbo")
-        else:
-            model_id = os.environ.get("BGFX_T2I_MODEL", "stabilityai/stable-diffusion-2-1")
-
-        logger.info("Loading text-to-image model: %s (device=%s, VRAM=%s GB)", model_id, device, vram_gb)
-        dtype = __import__("torch").float16 if device == "cuda" else __import__("torch").float32
-
-        pipe = None
-        try:
-            # Unified API for turbo/SDXL/SD
-            pipe = AutoPipelineForText2Image.from_pretrained(
-                model_id,
-                torch_dtype=dtype,
-                use_safetensors=True,
-                token=token,
-            )
-        except Exception as e1:
-            try:
-                pipe = StableDiffusionPipeline.from_pretrained(
-                    model_id,
-                    torch_dtype=dtype,
-                    use_safetensors=True,
-                    safety_checker=None,
-                    feature_extractor=None,
-                    use_auth_token=token,
-                )
-            except Exception as e2:
-                return None, None, f"AI model load failed: {e1} / {e2}"
-
-        # Memory/perf knobs
-        try: pipe.set_progress_bar_config(disable=True)
-        except Exception: pass
-        try: pipe.enable_attention_slicing()
-        except Exception: pass
-        try: pipe.enable_vae_slicing()
-        except Exception: pass
-
-        if device == "cuda":
-            try: pipe.enable_xformers_memory_efficient_attention()
-            except Exception: pass
-            pipe = pipe.to(device)
-        else:
-            try: pipe.enable_sequential_cpu_offload()
-            except Exception: pass
-
-        self.t2i_pipe = pipe
-        self.t2i_model_id = model_id
-        return pipe, model_id, f"AI model loaded: {model_id}"
-
-    def ai_generate_background(self, prompt: str, seed: int, width: int, height: int) -> Tuple[Optional[Image.Image], Optional[str], str]:
-        """
-        Generate a background and save to /tmp. Returns (preview_img, path, status).
-        """
-        pipe, model_id, status_msg = self._ensure_t2i()
-        if pipe is None:
-            logger.warning(status_msg)
-            return None, None, status_msg
-
-        # Sizes: multiples of 8, clamped to safe range
-        w = _div8(width or PREVIEW_W)
-        h = _div8(height or PREVIEW_H)
-        w = max(256, min(w, 1536))
-        h = max(256, min(h, 1536))
-
-        # Prompt defaults aimed at "office-like" backgrounds
-        prompt = (prompt or "professional modern office background, neutral colors, soft depth of field, clean, minimal, photorealistic")
-        negative = "text, watermark, logo, people, person, artifact, noisy, blurry"
-
-        try:
-            import torch
-            device = "cuda" if getattr(torch, "cuda", None) and torch.cuda.is_available() else "cpu"
-            try:
-                g = torch.Generator(device=device).manual_seed(int(seed)) if seed is not None else None
-            except Exception:
-                g = None
-
-            steps = 4 if ("turbo" in (self.t2i_model_id or "").lower()) else 25
-            guidance = 1.0 if ("turbo" in (self.t2i_model_id or "").lower()) else 7.0
-
-            with torch.inference_mode():
-                if device == "cuda":
-                    with torch.autocast("cuda"):
-                        out = pipe(
-                            prompt=prompt,
-                            negative_prompt=negative,
-                            height=h,
-                            width=w,
-                            guidance_scale=guidance,
-                            num_inference_steps=steps,
-                            generator=g,
-                        )
-                else:
-                    out = pipe(
-                        prompt=prompt,
-                        negative_prompt=negative,
-                        height=h,
-                        width=w,
-                        guidance_scale=guidance,
-                        num_inference_steps=steps,
-                        generator=g,
-                    )
-            img = out.images[0]
-            tmp_path = f"/tmp/ai_bg_{int(time.time())}.png"
-            img.save(tmp_path)
-
-            return img.resize((PREVIEW_W, PREVIEW_H), Image.LANCZOS), tmp_path, f"{status_msg} • Generated {w}x{h}"
-        except Exception as e:
-            logger.exception("AI generation error")
-            return None, None, f"AI generation failed: {e}"
-
-    # ---- PROCESS VIDEO ----
-    def process_video(
-        self,
-        video: str,
-        bg_source: str,
-        preset_key: str,
-        custom_bg_file,
-        grad_type: str,
-        grad_color1: str,
-        grad_color2: str,
-        grad_angle: int,
-        ai_bg_path: Optional[str],
-    ):
-        if not self.models_loaded:
-            return None, "Models not loaded yet"
-        if not video:
-            return None, "Please upload a video first."
-
-        logger.info(
-            "process_video called (video=%s, source=%s, preset=%s, file=%s, grad=%s, ai=%s)",
-            video,
-            bg_source,
-            preset_key,
-            getattr(custom_bg_file, "name", None) if custom_bg_file else None,
-            {"type": grad_type, "c1": grad_color1, "c2": grad_color2, "angle": grad_angle},
-            ai_bg_path,
-        )
-
-        output_path = f"/tmp/output_{int(time.time())}.mp4"
-
-        # Validate input video
-        ok = validate_video_file(video)
-        if not ok:
-            logger.warning("Invalid/unreadable video: %s", video)
-            return None, "Invalid or unreadable video file"
-
-        # Build bg_config based on source
-        src = (bg_source or "Preset").lower()
-        if src == "upload" and custom_bg_file is not None:
-            bg_cfg: Dict[str, Any] = {"custom_path": custom_bg_file.name}
-        elif src == "gradient":
-            bg_cfg = {
-                "gradient": {
-                    "type": (grad_type or "linear").lower(),
-                    "start": _hex_to_rgb(grad_color1 or "#222222"),
-                    "end": _hex_to_rgb(grad_color2 or "#888888"),
-                    "angle_deg": float(grad_angle or 0),
-                }
-            }
-        elif src == "ai generate" and ai_bg_path:
-            bg_cfg = {"custom_path": ai_bg_path}
-        else:
-            key = preset_key if preset_key in PROFESSIONAL_BACKGROUNDS else "office"
-            bg_cfg = {"background_choice": key}
-
-        try:
-            result = self.core_processor.process_video(
-                input_path=video,
-                output_path=output_path,
-                bg_config=bg_cfg,
-            )
-            logger.info("Core processing done → %s", output_path)
-
-            output_with_audio = self.audio_proc.add_audio_to_video(video, output_path)
-            logger.info("Audio merged → %s", output_with_audio)
-
-            frames = (result.get('frames') if isinstance(result, dict) else None) or "n/a"
-            return output_with_audio, f"Processing complete ({frames} frames, background={bg_source})"
-
-        except Exception as e:
-            logger.exception("Processing failed")
-            return None, f"Processing failed: {e}"
-
-# 7) Gradio UI
-def create_csp_safe_gradio():
-    import gradio as gr
-    app = VideoBackgroundApp()
-
-    with gr.Blocks(
-        title="BackgroundFX Pro - CSP Safe",
-        analytics_enabled=False,
-        css="""
-        .gradio-container { max-width: 1100px; margin: auto; }
-        """
-    ) as demo:
-        gr.Markdown("# 🎬 BackgroundFX Pro (CSP-Safe)")
-        gr.Markdown("Replace your video background with cinema-quality AI matting. Now with live background preview.")
-
-        with gr.Row():
-            with gr.Column(scale=1):
-                video = gr.Video(label="Upload Video")
-                bg_source = gr.Radio(
-                    ["Preset", "Upload", "Gradient", "AI Generate"],
-                    value="Preset",
-                    label="Background Source",
-                    interactive=True,
-                )
-
-                # PRESET
-                preset_choices = list(PROFESSIONAL_BACKGROUNDS.keys())
-                default_preset = "office" if "office" in preset_choices else (preset_choices[0] if preset_choices else "office")
-                preset_key = gr.Dropdown(choices=preset_choices, value=default_preset, label="Preset")
-
-                # UPLOAD
-                custom_bg = gr.File(label="Custom Background (Image)", file_types=["image"], visible=False)
-
-                # GRADIENT
-                grad_type = gr.Dropdown(choices=["Linear", "Radial"], value="Linear", label="Gradient Type", visible=False)
-                grad_color1 = gr.ColorPicker(value="#222222", label="Start Color", visible=False)
-                grad_color2 = gr.ColorPicker(value="#888888", label="End Color", visible=False)
-                grad_angle = gr.Slider(0, 360, value=0, step=1, label="Angle (degrees)", visible=False)
-
-                # AI
-                ai_prompt = gr.Textbox(label="AI Prompt", placeholder="e.g., sunlit modern office, soft bokeh, neutral palette", visible=False)
-                ai_seed = gr.Slider(0, 2**31-1, step=1, value=42, label="Seed", visible=False)
-                ai_size = gr.Dropdown(choices=["640x360","960x540","1280x720"], value="640x360", label="AI Image Size", visible=False)
-                ai_go = gr.Button("✨ Generate Background", visible=False, variant="secondary")
-                ai_status = gr.Markdown(visible=False)
-                ai_bg_path_state = gr.State(value=None)  # store /tmp path
-
-                btn_load = gr.Button("🔄 Load Models", variant="secondary")
-                btn_run = gr.Button("🎬 Process Video", variant="primary")
-
-            with gr.Column(scale=1):
-                status = gr.Textbox(label="Status", lines=4)
-                bg_preview = gr.Image(label="Background Preview", width=PREVIEW_W, height=PREVIEW_H, interactive=False)
-                out_video = gr.Video(label="Processed Video")
-
-        # ---------- UI wiring ----------
-
-        # background source → show/hide controls
-        def on_source_toggle(src):
-            src = (src or "Preset").lower()
-            return (
-                gr.update(visible=(src == "preset")),
-                gr.update(visible=(src == "upload")),
-                gr.update(visible=(src == "gradient")),
-                gr.update(visible=(src == "gradient")),
-                gr.update(visible=(src == "gradient")),
-                gr.update(visible=(src == "gradient")),
-                gr.update(visible=(src == "ai generate")),
-                gr.update(visible=(src == "ai generate")),
-                gr.update(visible=(src == "ai generate")),
-                gr.update(visible=(src == "ai generate")),
-                gr.update(visible=(src == "ai generate")),
-            )
-        bg_source.change(
-            fn=on_source_toggle,
-            inputs=[bg_source],
-            outputs=[preset_key, custom_bg, grad_type, grad_color1, grad_color2, grad_angle, ai_prompt, ai_seed, ai_size, ai_go, ai_status],
-        )
-
-        # ✅ Clear any previous AI image path when switching source (avoids stale AI background)
-        def _clear_ai_state(_):
-            return None
-        bg_source.change(fn=_clear_ai_state, inputs=[bg_source], outputs=[ai_bg_path_state])
-
-        # When source changes, also refresh preview based on visible controls
-        def on_source_preview(src, pkey, gt, c1, c2, ang):
-            src_l = (src or "Preset").lower()
-            if src_l == "preset":
-                return app.preview_preset(pkey)
-            elif src_l == "gradient":
-                return app.preview_gradient(gt, c1, c2, ang)
-            # For upload/AI we keep whatever the component change handler sets (don’t overwrite)
-            return gr.update()  # no-op
-        bg_source.change(
-            fn=on_source_preview,
-            inputs=[bg_source, preset_key, grad_type, grad_color1, grad_color2, grad_angle],
-            outputs=[bg_preview]
-        )
-
-        # live previews
-        preset_key.change(fn=lambda k: app.preview_preset(k), inputs=[preset_key], outputs=[bg_preview])
-        custom_bg.change(fn=lambda f: app.preview_upload(f), inputs=[custom_bg], outputs=[bg_preview])
-        for comp in (grad_type, grad_color1, grad_color2, grad_angle):
-            comp.change(
-                fn=lambda gt, c1, c2, ang: app.preview_gradient(gt, c1, c2, ang),
-                inputs=[grad_type, grad_color1, grad_color2, grad_angle],
-                outputs=[bg_preview],
-            )
-
-        # AI generate
-        def ai_generate(prompt, seed, size):
-            try:
-                w, h = map(int, (size or "640x360").split("x"))
-            except Exception:
-                w, h = PREVIEW_W, PREVIEW_H
-            img, path, msg = app.ai_generate_background(
-                prompt or "professional modern office background, neutral colors, depth of field",
-                int(seed) if seed is not None else 42,
-                w, h
-            )
-            return img, (path or None), msg
-        ai_go.click(fn=ai_generate, inputs=[ai_prompt, ai_seed, ai_size], outputs=[bg_preview, ai_bg_path_state, ai_status])
-
-        # model load / run
-        def safe_load():
-            msg = app.load_models()
-            logger.info("UI: models loaded")
-            return msg, app.preview_preset(preset_key.value if hasattr(preset_key, "value") else "office")
-        btn_load.click(fn=safe_load, outputs=[status, bg_preview])
-
-        def safe_process(vid, src, pkey, file, gtype, c1, c2, ang, ai_path):
-            return app.process_video(vid, src, pkey, file, gtype, c1, c2, ang, ai_path)
-        btn_run.click(
-            fn=safe_process,
-            inputs=[video, bg_source, preset_key, custom_bg, grad_type, grad_color1, grad_color2, grad_angle, ai_bg_path_state],
-            outputs=[out_video, status]
-        )
-
-    return demo
-
-# 8) Launch
-if __name__ == "__main__":
-    logger.info("Launching CSP-safe Gradio interface for Hugging Face Spaces")
-    demo = create_csp_safe_gradio()
-    demo.queue().launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        show_error=True,
-        debug=False,
-        inbrowser=False
-    )
+        cap = cv2.VideoCapture(video_path)
+        if not cap.isOpened():
+            return False, "cv2 could not open file"
+        ok, frame = cap.read()
+        cap.release()
+        if not ok or frame is None:
+            return False, "Could not read first frame"
+        return True, "OK"
+    except Exception as e:
+        return False, f"cv2 error: {e}"
+
+# ---------------------------------------------------------------------
+# LAZY WRAPPERS (avoid importing utils.cv_processing at module import time)
+# ---------------------------------------------------------------------
+def segment_person_hq(*args, **kwargs):
+    from .cv_processing import segment_person_hq as _f
+    return _f(*args, **kwargs)
+
+def refine_mask_hq(*args, **kwargs):
+    from .cv_processing import refine_mask_hq as _f
+    return _f(*args, **kwargs)
+
+def replace_background_hq(*args, **kwargs):
+    from .cv_processing import replace_background_hq as _f
+    return _f(*args, **kwargs)
+
+__all__ = [
+    # backgrounds
+    "PROFESSIONAL_BACKGROUNDS",
+    "create_professional_background",
+    "create_gradient_background",
+    # validation
+    "validate_video_file",
+    "validate_video_file_detail",
+    # lazy CV exports (back-compat)
+    "segment_person_hq",
+    "refine_mask_hq",
+    "replace_background_hq",
+]