early_env.py

#!/usr/bin/env python3
"""
Video Background Replacer (GPU-Optimized)

- MatAnyone (primary), SAM2 (mask seeding), rembg (fallback)
- K-Governor guards torch.topk/kthvalue (no __wrapped__ assumption)
- Adaptive MatAnyone loader (from_pretrained | constructor network/model | repo-id)
- Optional repo pinning via MATANYONE_COMMIT / SAM2_COMMIT
- First-run warmup → READY ✅ before first request
- Robust Gradio input coercion (path | dict | file-like | PIL | NumPy)
- Alpha probing & (optional) stitching alpha_*.png sequences to a video
- Short-clip stabilizer (pre-roll) with correct trim
- Concurrency lock for MatAnyone core
"""

# =========================
# EARLY env & imports
# =========================
import os, sys, re, time, gc, shutil, subprocess, tempfile, threading, traceback, inspect, glob
from pathlib import Path

# ---- Thread/env sanitization (must run BEFORE numpy/torch/cv2) ----
def _safe_int_env(var: str, default: int = 2, cap: int = 8) -> int:
    v = os.environ.get(var, "").strip()
    if not v or not re.fullmatch(r"\d+", v):
        os.environ[var] = str(default); return default
    iv = max(1, min(int(v), cap))
    os.environ[var] = str(iv); return iv

_safe_int_env("OMP_NUM_THREADS", 2, 8)
_safe_int_env("MKL_NUM_THREADS", 2, 8)

# General runtime defaults
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True,max_split_size_mb:512")
os.environ.setdefault("CUDA_MODULE_LOADING", "LAZY")
os.environ.setdefault("PYTHONUNBUFFERED", "1")
os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")

# MatAnyone prefs
os.environ.setdefault("MATANYONE_MAX_EDGE", "1024")
os.environ.setdefault("MATANYONE_TARGET_PIXELS", "1000000")
os.environ.setdefault("MATANYONE_WINDOWED", "1")
os.environ.setdefault("MATANYONE_WINDOW", "16")
os.environ.setdefault("MAX_MODEL_SIZE", "1920")

# CUDA + cuDNN
os.environ.setdefault("CUDA_LAUNCH_BLOCKING", "0")
os.environ.setdefault("TORCH_CUDNN_V8_API_ENABLED", "1")
os.environ.setdefault("CUDNN_BENCHMARK", "1")

# HF cache
os.environ.setdefault("HF_HOME", "./checkpoints/hf")
os.environ.setdefault("TRANSFORMERS_CACHE", "./checkpoints/hf")
os.environ.setdefault("HF_DATASETS_CACHE", "./checkpoints/hf")
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS", "1")
os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1")

# Gradio
os.environ.setdefault("GRADIO_SERVER_NAME", "0.0.0.0")
os.environ.setdefault("GRADIO_SERVER_PORT", "7860")

# Features
os.environ.setdefault("USE_MATANYONE", "true")
os.environ.setdefault("USE_SAM2", "true")
os.environ.setdefault("SELF_CHECK_MODE", "false")

# Stabilizer defaults
os.environ.setdefault("MATANYONE_STABILIZE", "true")
os.environ.setdefault("MATANYONE_PREROLL_FRAMES", "12")

# Optional strict re-sanitization later
os.environ.setdefault("STRICT_ENV_GUARD", "1")

# =========================
# Std imports (safe now)
# =========================
import cv2
import numpy as np
from PIL import Image
import gradio as gr
from moviepy.editor import VideoFileClip, ImageSequenceClip, concatenate_videoclips

print("=" * 50)
print("Application Startup at", os.popen('date').read().strip())
print("=" * 50)
print("Environment Configuration:")
print(f"Python: {sys.version}")
print(f"Working directory: {os.getcwd()}")
print(f"CUDA_MODULE_LOADING: {os.getenv('CUDA_MODULE_LOADING')}")
print(f"OMP_NUM_THREADS: {os.getenv('OMP_NUM_THREADS')}")
print("=" * 50)

# =========================
# Third-party repos & optional pinning
# =========================
BASE_DIR = Path(__file__).resolve().parent
TP_DIR = BASE_DIR / "third_party"
CHECKPOINTS_DIR = BASE_DIR / "checkpoints"
TP_DIR.mkdir(exist_ok=True); CHECKPOINTS_DIR.mkdir(exist_ok=True)

def _git_clone_if_missing(url: str, path: Path, name: str):
    if path.exists():
        return
    print(f"Cloning {name}…")
    try:
        subprocess.run(["git", "clone", "--depth", "1", url, str(path)], check=True, timeout=300)
        print(f"{name} cloned successfully")
    except Exception as e:
        print(f"Failed to clone {name}: {e}")

_git_clone_if_missing("https://github.com/facebookresearch/segment-anything-2.git", TP_DIR/"sam2", "SAM2")
_git_clone_if_missing("https://github.com/pq-yang/MatAnyone.git", TP_DIR/"matanyone", "MatAnyone")

def _checkout(repo_dir: Path, commit: str):
    if not commit:
        print(f"{repo_dir.name} not pinned (env is empty) — using current HEAD.")
        return
    try:
        subprocess.run(["git", "-C", str(repo_dir), "fetch", "--depth", "1", "origin", commit], check=True)
        subprocess.run(["git", "-C", str(repo_dir), "checkout", "--detach", commit], check=True)
        print(f"Locked {repo_dir.name} to {commit}")
    except Exception as e:
        print(f"Warning: failed to lock {repo_dir.name} to {commit}: {e}")

MATANYONE_COMMIT = os.getenv("MATANYONE_COMMIT", "").strip()
SAM2_COMMIT      = os.getenv("SAM2_COMMIT", "").strip()
_checkout(TP_DIR / "matanyone", MATANYONE_COMMIT)
_checkout(TP_DIR / "sam2", SAM2_COMMIT)

# Ensure vendored paths are importable
for p in [TP_DIR / "sam2", TP_DIR / "matanyone"]:
    if p.exists() and str(p) not in sys.path:
        sys.path.insert(0, str(p)); print(f"Added to path: {p}")

# =========================
# K-Governor (with bypass; robust for PyTorch 2.2)
# =========================
if os.getenv("SAFE_TOPK_BYPASS", "0") not in ("1","true","TRUE"):
    import re as _re
    def _write_safe_ops_file(pkg_root: Path):
        utils_dir = pkg_root / "matanyone" / "utils"
        if not utils_dir.exists(): utils_dir = pkg_root / "utils"
        utils_dir.mkdir(parents=True, exist_ok=True)
        (utils_dir / "safe_ops.py").write_text(
            """
import os
import torch

_VERBOSE = bool(int(os.environ.get("SAFE_TOPK_VERBOSE", "1")))

# Robust for builds where topk/kthvalue are builtins without attributes.
_ORIG_TOPK = getattr(torch.topk, "__wrapped__", torch.topk)
_ORIG_KTH  = getattr(torch.kthvalue, "__wrapped__", torch.kthvalue)

def _log(msg):
    if _VERBOSE:
        print(f"[K-Governor] {msg}")

def safe_topk(x, k, dim=None, largest=True, sorted=True):
    if not isinstance(k, int):
        k = int(k)
    if dim is None:
        dim = -1
    n = x.size(dim)
    k_eff = max(1, min(k, int(n)))
    if k_eff != k:
        _log(f"torch.topk: clamp k {k} -> {k_eff} for dim={dim} shape={tuple(x.shape)}")
    values, indices = _ORIG_TOPK(x, k_eff, dim=dim, largest=largest, sorted=sorted)
    if k_eff < k:
        pad = k - k_eff
        pad_shape = list(values.shape); pad_shape[dim] = pad
        pad_vals = values.new_full(pad_shape, float('-inf'))
        pad_idx  = indices.new_zeros(pad_shape, dtype=indices.dtype)
        values = torch.cat([values, pad_vals], dim=dim)
        indices = torch.cat([indices, pad_idx],  dim=dim)
    return values, indices

def safe_kthvalue(x, k, dim=None, keepdim=False):
    if not isinstance(k, int):
        k = int(k)
    if dim is None:
        dim = -1
    n = x.size(dim)
    k_eff = max(1, min(k, int(n)))
    if k_eff != k:
        _log(f"torch.kthvalue: clamp k {k} -> {k_eff} for dim={dim} shape={tuple(x.shape)}")
    return _ORIG_KTH(x, k_eff, dim=dim, keepdim=keepdim)
""".lstrip(), encoding="utf-8")

    def _patch_matanyone_sources(repo_dir: Path) -> int:
        root = repo_dir / "matanyone"
        if not root.exists(): root = repo_dir
        changed = 0
        header_import = "from matanyone.utils.safe_ops import safe_topk, safe_kthvalue\n"
        pt = _re.compile(r"\btorch\.topk\s*\(")
        pm = _re.compile(r"(\b[\w\.]+)\.topk\s*\(")
        kt = _re.compile(r"\btorch\.kthvalue\s*\(")
        km = _re.compile(r"(\b[\w\.]+)\.kthvalue\s*\(")
        for py in root.rglob("*.py"):
            try:
                txt = py.read_text(encoding="utf-8"); orig = txt
                if "safe_topk" not in txt and py.name != "__init__.py":
                    lines = txt.splitlines(keepends=True)
                    insert_at = 0
                    for i, L in enumerate(lines[:80]):
                        if L.startswith(("import ","from ")): insert_at = i+1
                    lines.insert(insert_at, header_import)
                    txt = "".join(lines)
                txt = pt.sub("safe_topk(", txt)
                txt = kt.sub("safe_kthvalue(", txt)
                def _mt(m): return f"safe_topk({m.group(1)}, "
                def _mk(m): return f"safe_kthvalue({m.group(1)}, "
                txt = pm.sub(_mt, txt); txt = km.sub(_mk, txt)
                if txt != orig:
                    py.write_text(txt, encoding="utf-8"); changed += 1
            except Exception as e:
                print(f"[K-Governor] Patch warning on {py}: {e}")
        return changed

    try:
        MATANY_REPO_DIR = TP_DIR / "matanyone"
        _write_safe_ops_file(MATANY_REPO_DIR)
        patched_files = _patch_matanyone_sources(MATANY_REPO_DIR)
        print(f"[K-Governor] Patched MatAnyone sources: {patched_files} files updated.")
    except Exception as e:
        print(f"[K-Governor] Patch failed: {e}")
else:
    print("[K-Governor] BYPASSED via SAFE_TOPK_BYPASS")

# =========================
# Torch & device
# =========================
TORCH_AVAILABLE = False; CUDA_AVAILABLE = False; GPU_NAME = "N/A"; DEVICE = "cpu"
try:
    import torch
    TORCH_AVAILABLE = True
    CUDA_AVAILABLE = torch.cuda.is_available()
    if CUDA_AVAILABLE:
        torch.backends.cudnn.enabled = True
        torch.backends.cudnn.benchmark = True
        torch.backends.cudnn.deterministic = False
        GPU_NAME = torch.cuda.get_device_name(0); DEVICE = "cuda"
        gpu_memory = torch.cuda.get_device_properties(0).total_memory / 1024**3
        print(f"GPU: {GPU_NAME}")
        print(f"VRAM: {gpu_memory:.1f} GB")
        print(f"CUDA Capability: {torch.cuda.get_device_capability(0)}")
        try: torch.cuda.set_per_process_memory_fraction(0.9)
        except Exception: pass
    print(f"Torch version: {torch.__version__}")
    print(f"CUDA available: {CUDA_AVAILABLE}")
    print(f"Device: {DEVICE}")
except Exception as e:
    print(f"Torch not available: {e}")

# =========================
# Light GPU monitor
# =========================
class GPUMonitor:
    def __init__(self):
        self.monitoring = False
        self.stats = {"gpu_util": 0, "memory_used": 0, "memory_total": 0}
    def start_monitoring(self):
        if not CUDA_AVAILABLE: return
        self.monitoring = True
        threading.Thread(target=self._monitor_loop, daemon=True).start()
    def stop_monitoring(self): self.monitoring = False
    def _monitor_loop(self):
        while self.monitoring:
            try:
                if CUDA_AVAILABLE:
                    mem_used = torch.cuda.memory_allocated(0) / 1024**3
                    mem_total = torch.cuda.get_device_properties(0).total_memory / 1024**3
                    self.stats.update({
                        "memory_used": mem_used, "memory_total": mem_total,
                        "memory_percent": (mem_used/mem_total)*100 if mem_total else 0
                    })
                    try:
                        import pynvml
                        pynvml.nvmlInit()
                        h = pynvml.nvmlDeviceGetHandleByIndex(0)
                        util = pynvml.nvmlDeviceGetUtilizationRates(h)
                        self.stats["gpu_util"] = util.gpu
                    except Exception:
                        pass
            except Exception as e:
                print(f"GPU monitoring error: {e}")
            time.sleep(1)
    def get_stats(self): return self.stats.copy()

gpu_monitor = GPUMonitor(); gpu_monitor.start_monitoring()

# =========================
# SAM2 (verified micro-inference)
# =========================
SAM2_IMPORTED = False; SAM2_AVAILABLE = False; SAM2_PREDICTOR = None
if TORCH_AVAILABLE and os.getenv("USE_SAM2","true").lower()=="true":
    try:
        print("Setting up SAM2…")
        from hydra import initialize_config_dir, compose
        from hydra.core.global_hydra import GlobalHydra
        from sam2.build_sam import build_sam2
        from sam2.sam2_image_predictor import SAM2ImagePredictor
        SAM2_IMPORTED = True
        ckpt = Path("./checkpoints/sam2.1_hiera_tiny.pt")
        ckpt.parent.mkdir(parents=True, exist_ok=True)
        if not ckpt.exists():
            print("Downloading SAM2.1 checkpoint…")
            import requests
            url = "https://dl.fbaipublicfiles.com/segment_anything_2/092824/sam2.1_hiera_tiny.pt"
            r = requests.get(url, stream=True, timeout=60); r.raise_for_status()
            with open(ckpt, "wb") as f:
                for ch in r.iter_content(chunk_size=8192):
                    if ch: f.write(ch)
            print(f"SAM2 checkpoint downloaded to {ckpt}")
        if GlobalHydra().is_initialized():
            GlobalHydra.instance().clear()
        config_dir = str(TP_DIR / "sam2" / "sam2" / "configs")
        config_file = "sam2.1/sam2.1_hiera_t.yaml"
        initialize_config_dir(config_dir=config_dir, version_base=None)
        _ = compose(config_name=config_file)
        model = build_sam2(config_file, str(ckpt), device="cuda" if CUDA_AVAILABLE else "cpu")
        if CUDA_AVAILABLE and hasattr(torch, "compile"):
            try: model = torch.compile(model, mode="max-autotune")
            except Exception as _e: print(f"torch.compile not used: {_e}")
        SAM2_PREDICTOR = SAM2ImagePredictor(model)
        try:
            dummy = np.zeros((64,64,3), dtype=np.uint8)
            SAM2_PREDICTOR.set_image(dummy)
            pts = np.array([[32,32]], dtype=np.int32); lbs = np.array([1], dtype=np.int32)
            _m,_s,_l = SAM2_PREDICTOR.predict(point_coords=pts, point_labels=lbs, multimask_output=True)
            SAM2_AVAILABLE = True; print("✅ SAM2 verified via micro-inference.")
        except Exception as ver_e:
            SAM2_AVAILABLE = False; SAM2_PREDICTOR = None
            print(f"SAM2 verification failed: {ver_e}")
    except Exception as e:
        print(f"SAM2 setup failed: {e}")

# =========================
# MatAnyone import (canonical first, fallback)
# =========================
MATANYONE_IMPORTED = False; MatAnyInferenceCore = None
try:
    from matanyone.inference.inference_core import InferenceCore as MatAnyInferenceCore
    MATANYONE_IMPORTED = True
    print("MatAnyone import OK: matanyone.inference.inference_core.InferenceCore")
except Exception as e1:
    try:
        from matanyone import InferenceCore as MatAnyInferenceCore
        MATANYONE_IMPORTED = True
        print("MatAnyone import OK: matanyone.InferenceCore")
    except Exception as e2:
        print(f"MatAnyone not importable: {e2 or e1}")

# =========================
# rembg fallback
# =========================
REMBG_AVAILABLE = False
try:
    from rembg import remove
    REMBG_AVAILABLE = True; print("rembg import OK (fallback ready).")
except Exception as e:
    print(f"rembg not available: {e}")

# =========================
# Background helpers
# =========================
def make_solid(w, h, rgb): return np.full((h, w, 3), rgb, dtype=np.uint8)
def make_vertical_gradient(w, h, top_rgb, bottom_rgb):
    top = np.array(top_rgb, dtype=np.float32); bot = np.array(bottom_rgb, dtype=np.float32)
    t = np.linspace(0,1,h,dtype=np.float32)[:,None]
    grad = (1-t)*top + t*bot; grad = np.clip(grad,0,255).astype(np.uint8)
    return np.repeat(grad[None,...], w, axis=0).transpose(1,0,2)
def build_professional_bg(w, h, preset: str) -> np.ndarray:
    p = (preset or "").lower()
    if p == "office (soft gray)": return make_vertical_gradient(w,h,(245,246,248),(220,223,228))
    if p == "studio (charcoal)":  return make_vertical_gradient(w,h,(32,32,36),(64,64,70))
    if p == "nature (green tint)":return make_vertical_gradient(w,h,(180,220,190),(100,160,120))
    if p == "brand blue":         return make_solid(w,h,(18,112,214))
    return make_solid(w,h,(240,240,240))

# =========================
# MatAnyone wrapper (+ lock, adaptive constructor, alpha stitching)
# =========================
class OptimizedMatAnyoneProcessor:
    def __init__(self):
        self.processor = None
        self.device = "cuda" if (TORCH_AVAILABLE and CUDA_AVAILABLE) else "cpu"
        self.initialized = False
        self.verified = False
        self.last_error = None
        self.stabilize = os.getenv("MATANYONE_STABILIZE","true").lower()=="true"
        try: self.preroll_frames = max(0, int(os.getenv("MATANYONE_PREROLL_FRAMES","12")))
        except Exception: self.preroll_frames = 12
        self._lock = threading.Lock()

    # ---- Adaptive core constructor
    def _construct_inference_core(self, network_or_repo):
        # prefer classmethod if available
        try:
            if hasattr(MatAnyInferenceCore, "from_pretrained"):
                return MatAnyInferenceCore.from_pretrained(
                    network_or_repo,
                    device=("cuda" if CUDA_AVAILABLE else "cpu")
                )
        except Exception:
            pass
        # try constructor with introspection
        try:
            sig = inspect.signature(MatAnyInferenceCore)
            if isinstance(network_or_repo, str):
                return MatAnyInferenceCore(network_or_repo)
            if "network" in sig.parameters:
                return MatAnyInferenceCore(network=network_or_repo)
            if "model" in sig.parameters:
                return MatAnyInferenceCore(model=network_or_repo)
            return MatAnyInferenceCore(network_or_repo)
        except Exception as e:
            raise RuntimeError(f"InferenceCore construction failed: {type(e).__name__}: {e}")

    # ---- Normalize return + disk probe + png sequence stitch
    def _stitch_alpha_sequence(self, outdir: str, fps: float) -> str | None:
        # common patterns
        patt_list = ["alpha_%04d.png", "alpha_%03d.png", "alpha_%05d.png", "alpha_*.png"]
        frames = []
        for patt in patt_list:
            frames = sorted(glob.glob(os.path.join(outdir, patt.replace("%0", "*").replace("d",""))))
            if frames:
                break
        if not frames:
            return None
        # read as float [0,1]
        ary = []
        for p in frames:
            im = cv2.imread(p, cv2.IMREAD_GRAYSCALE)
            if im is None: continue
            ary.append((im.astype(np.float32) / 255.0))
        if not ary:
            return None
        clip = ImageSequenceClip([f for f in ary], fps=max(1, int(round(fps or 24))))
        alpha_mp4 = tempfile.NamedTemporaryFile(delete=False, suffix="_alpha_seq.mp4").name
        clip.write_videofile(alpha_mp4, audio=False, logger=None)
        clip.close()
        return alpha_mp4

    def _normalize_ret_and_probe(self, ret, outdir: str, fallback_fps: float = 24.0):
        fg_path = alpha_path = None
        if isinstance(ret, (list, tuple)):
            if len(ret) >= 2: fg_path, alpha_path = ret[0], ret[1]
            elif len(ret) == 1: alpha_path = ret[0]
        elif isinstance(ret, str):
            alpha_path = ret

        def _valid(p: str) -> bool:
            return p and os.path.exists(p) and os.path.getsize(p) > 0

        # probe common video names
        if not _valid(alpha_path):
            for cand in ("alpha.mp4","alpha.mkv","alpha.mov","alpha.webm"):
                p = os.path.join(outdir, cand)
                if _valid(p):
                    alpha_path = p; break

        # try stitching sequences if needed
        if not _valid(alpha_path):
            stitched = self._stitch_alpha_sequence(outdir, fallback_fps)
            if stitched and _valid(stitched):
                alpha_path = stitched

        return fg_path, alpha_path

    def _warmup(self) -> None:
        import numpy as _np, cv2 as _cv2, os as _os
        from moviepy.editor import ImageSequenceClip as _ISC
        with tempfile.TemporaryDirectory() as td:
            frames = []
            for t in range(8):
                fr = _np.zeros((64,64,3), _np.uint8); x = 8 + t*4
                _cv2.rectangle(fr, (x,20), (x+12,44), 200, -1); frames.append(fr)
            vid = _os.path.join(td,"warmup.mp4"); _ISC(frames, fps=10).write_videofile(vid, audio=False, logger=None)
            m = _np.zeros((64,64), _np.uint8); _cv2.rectangle(m,(24,24),(40,40),255,-1)
            mask = _os.path.join(td,"mask.png"); _cv2.imwrite(mask, m)
            outdir = _os.path.join(td,"out"); os.makedirs(outdir, exist_ok=True)
            # ensure method exists
            if not hasattr(self.processor, "process_video"):
                if hasattr(self.processor, "process"):
                    self.processor.process_video = self.processor.process
                else:
                    raise RuntimeError("MatAnyone core lacks process_video/process")

            ret = self.processor.process_video(input_path=vid, mask_path=mask, output_path=outdir, max_size=512)
            _fg, alpha = self._normalize_ret_and_probe(ret, outdir, fallback_fps=10)
            if not alpha or not os.path.exists(alpha) or os.path.getsize(alpha) == 0:
                raise RuntimeError("Warmup: MatAnyone produced no alpha")

    def initialize(self) -> bool:
        with self._lock:
            if not MATANYONE_IMPORTED:
                print("MatAnyone not importable; skipping init."); return False
            if self.initialized and self.processor is not None:
                return True
            self.last_error = None

            # HF path first
            try:
                print(f"Initializing MatAnyone (HF repo-id) on {self.device}…")
                self.processor = self._construct_inference_core("PeiqingYang/MatAnyone")
                if self.device == "cuda":
                    import torch as _t
                    _t.cuda.empty_cache(); _ = _t.rand(1, device="cuda") * 0.0
                # alias method if needed
                if not hasattr(self.processor, "process_video") and hasattr(self.processor, "process"):
                    self.processor.process_video = self.processor.process
                self._warmup()
                self.verified = True; self.initialized = True
                print("✅ MatAnyone initialized & warmed up (HF repo-id).")
                return True
            except Exception as e:
                self.last_error = f"HF init failed: {type(e).__name__}: {e}"
                print(self.last_error)

            # Local ckpt fallback
            try:
                print("Falling back to local checkpoint init for MatAnyone…")
                from hydra.core.global_hydra import GlobalHydra
                if hasattr(GlobalHydra,"instance") and GlobalHydra().is_initialized():
                    GlobalHydra.instance().clear()
                import requests
                from matanyone.utils.get_default_model import get_matanyone_model
                ckpt_dir = Path("./pretrained_models"); ckpt_dir.mkdir(parents=True, exist_ok=True)
                ckpt_path = ckpt_dir / "matanyone.pth"
                if not ckpt_path.exists():
                    url = "https://github.com/pq-yang/MatAnyone/releases/download/v1.0.0/matanyone.pth"
                    print(f"Downloading MatAnyone checkpoint from: {url}")
                    with requests.get(url, stream=True, timeout=180) as r:
                        r.raise_for_status()
                        with open(ckpt_path, "wb") as f:
                            for chunk in r.iter_content(chunk_size=8192):
                                if chunk: f.write(chunk)
                    print(f"Checkpoint saved to {ckpt_path}")
                network = get_matanyone_model(str(ckpt_path), device=("cuda" if CUDA_AVAILABLE else "cpu"))
                self.processor = self._construct_inference_core(network)
                if self.device == "cuda":
                    import torch as _t
                    _t.cuda.empty_cache(); _ = _t.rand(1, device="cuda") * 0.0
                if not hasattr(self.processor, "process_video") and hasattr(self.processor, "process"):
                    self.processor.process_video = self.processor.process
                self._warmup()
                self.verified = True; self.initialized = True
                print("✅ MatAnyone initialized & warmed up (local checkpoint).")
                return True
            except Exception as e:
                self.last_error = f"Local init/warmup failed: {type(e).__name__}: {e}"
                print(f"MatAnyone initialization failed: {self.last_error}")
                traceback.print_exc(); return False

    # ---- Pre-roll & trimming
    @staticmethod
    def _build_preroll_concat(input_path: str, frames: int) -> tuple[str, float, float]:
        clip = VideoFileClip(input_path)
        fps = float(clip.fps or 24.0)
        preroll_frames = max(0, frames)
        if preroll_frames == 0:
            out = input_path; clip.close(); return out, 0.0, fps
        first = clip.get_frame(0)
        pre = ImageSequenceClip([first]*preroll_frames, fps=max(1, int(round(fps))))
        concat = concatenate_videoclips([pre, clip])
        tmp = tempfile.NamedTemporaryFile(delete=False, suffix="_concat.mp4")
        concat.write_videofile(tmp.name, audio=False, logger=None)
        pre.close(); concat.close(); clip.close()
        return tmp.name, preroll_frames / fps, fps

    @staticmethod
    def _trim_head(video_path: str, seconds: float) -> str:
        if seconds <= 0: return video_path
        clip = VideoFileClip(video_path); dur = clip.duration or 0
        start = min(seconds, max(0.0, dur - 0.001))
        trimmed = tempfile.NamedTemporaryFile(delete=False, suffix="_trim.mp4").name
        clip.subclip(start, None).write_videofile(trimmed, audio=False, logger=None)
        clip.close(); return trimmed

    def create_mask_optimized(self, video_path: str, output_path: str) -> str:
        cap = cv2.VideoCapture(video_path); ret, frame = cap.read(); cap.release()
        if not ret: raise ValueError("Could not read first frame from video.")
        if SAM2_AVAILABLE and SAM2_PREDICTOR is not None:
            try:
                print("Creating mask with SAM2 (first frame)…")
                rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                SAM2_PREDICTOR.set_image(rgb)
                h, w = rgb.shape[:2]
                pts = np.array([[w//2, h//2],[w//3, h//3],[2*w//3, 2*h//3]], dtype=np.int32)
                lbs = np.array([1,1,1], dtype=np.int32)
                masks, scores, _ = SAM2_PREDICTOR.predict(point_coords=pts, point_labels=lbs, multimask_output=True)
                best = masks[np.argmax(scores)]
                mask = ((best.astype(np.uint8) > 0).astype(np.uint8)) * 255  # 1ch u8 {0,255}
                cv2.imwrite(output_path, mask)
                print(f"Self-test mask uniques: {np.unique(mask//255)}")
                return output_path
            except Exception as e:
                print(f"SAM2 mask creation failed; fallback rectangle. Error: {e}")
        # Fallback: centered box
        h, w = frame.shape[:2]
        mask = np.zeros((h,w), dtype=np.uint8)
        mx, my = int(w*0.15), int(h*0.10)
        mask[my:h-my, mx:w-mx] = 255
        cv2.imwrite(output_path, mask); return output_path

    def process_video_optimized(self, input_path: str, output_dir: str):
        with self._lock:
            if not self.initialized and not self.initialize():
                return None
            try:
                print("🚀 MatAnyone processing…")
                if CUDA_AVAILABLE:
                    import torch as _t
                    _t.cuda.empty_cache(); gc.collect()

                concat_path = input_path; preroll_sec = 0.0; fps_used = 24.0
                if self.stabilize and self.preroll_frames > 0:
                    concat_path, preroll_sec, fps_used = self._build_preroll_concat(input_path, self.preroll_frames)
                    print(f"[Stabilizer] Pre-rolled {self.preroll_frames} frames ({preroll_sec:.3f}s).")

                mask_path = os.path.join(output_dir, "mask.png")
                self.create_mask_optimized(input_path, mask_path)

                if not hasattr(self.processor, "process_video") and hasattr(self.processor, "process"):
                    self.processor.process_video = self.processor.process

                ret = self.processor.process_video(
                    input_path=concat_path,
                    mask_path=mask_path,
                    output_path=output_dir,
                    max_size=int(os.getenv("MAX_MODEL_SIZE","1920"))
                )
                fg_path, alpha_path = self._normalize_ret_and_probe(ret, output_dir, fallback_fps=fps_used)

                if not alpha_path or not os.path.exists(alpha_path):
                    raise RuntimeError("MatAnyone finished without a valid alpha video on disk.")

                if preroll_sec > 0.0:
                    alpha_path = self._trim_head(alpha_path, preroll_sec)
                    print(f"[Stabilizer] Trimmed {preroll_sec:.3f}s from alpha.")

                if not os.path.exists(alpha_path) or os.path.getsize(alpha_path) == 0:
                    raise RuntimeError("Alpha exists but is empty/zero bytes after trim.")

                return alpha_path

            except Exception as e:
                print(f"❌ MatAnyone processing failed: {e}")
                traceback.print_exc()
                return None

matanyone_processor = OptimizedMatAnyoneProcessor()

# =========================
# rembg helpers
# =========================
REMBG_AVAILABLE = REMBG_AVAILABLE
def process_frame_rembg_optimized(frame_bgr_u8, bg_img_rgb_u8):
    if not REMBG_AVAILABLE:
        return cv2.cvtColor(frame_bgr_u8, cv2.COLOR_BGR2RGB)
    try:
        frame_rgb = cv2.cvtColor(frame_bgr_u8, cv2.COLOR_BGR2RGB)
        pil_im = Image.fromarray(frame_rgb)
        from rembg import remove  # lazy import in case plugin is heavy
        result = remove(pil_im).convert("RGBA")
        result_np = np.array(result)
        if result_np.shape[2] == 4:
            alpha = (result_np[:, :, 3:4].astype(np.float32) / 255.0)
            comp = alpha * result_np[:, :, :3].astype(np.float32) + (1 - alpha) * bg_img_rgb_u8.astype(np.float32)
            return comp.astype(np.uint8)
        return result_np.astype(np.uint8)
    except Exception as e:
        print(f"rembg processing error: {e}")
        return cv2.cvtColor(frame_bgr_u8, cv2.COLOR_BGR2RGB)

# =========================
# Compositing
# =========================
def composite_with_background(original_path, alpha_path, bg_path=None, bg_preset=None):
    print("🎬 Compositing final video…")
    orig_clip = VideoFileClip(original_path)
    alpha_clip = VideoFileClip(alpha_path)
    fps = orig_clip.fps or 24
    w, h = orig_clip.size
    if bg_path:
        bg_img = cv2.imread(bg_path)
        if bg_img is None: raise ValueError(f"Could not read background image: {bg_path}")
        bg_img = cv2.cvtColor(bg_img, cv2.COLOR_BGR2RGB); bg_img = cv2.resize(bg_img, (w, h))
    else:
        bg_img = build_professional_bg(w, h, bg_preset)

    def process_func(get_frame, t):
        frame = get_frame(t)
        a = alpha_clip.get_frame(t)
        if a.ndim == 2: a = a[..., None]
        elif a.shape[2] > 1: a = a[..., :1]
        a = np.clip(a, 0.0, 1.0).astype(np.float32)
        bg_f32 = (bg_img.astype(np.float32) / 255.0)
        comp = a * frame.astype(np.float32) + (1.0 - a) * bg_f32
        return comp.astype(np.float32)

    new_clip = orig_clip.fl(process_func).set_fps(fps)
    output_path = "final_output.mp4"
    new_clip.write_videofile(output_path, audio=False, logger=None)
    alpha_clip.close(); orig_clip.close(); new_clip.close()
    return output_path

# =========================
# rembg whole-video fallback
# =========================
def process_video_rembg_fallback(video_path, bg_image_path=None, bg_preset=None):
    print("🔄 Processing with rembg fallback…")
    cap = cv2.VideoCapture(video_path); ret, frame = cap.read()
    if not ret: cap.release(); raise ValueError("Could not read video")
    h, w, _ = frame.shape; cap.release()
    if bg_image_path:
        bg_img = cv2.imread(bg_image_path)
        if bg_img is None: raise ValueError(f"Could not read background image: {bg_image_path}")
        bg_img = cv2.cvtColor(bg_img, cv2.COLOR_BGR2RGB); bg_img = cv2.resize(bg_img, (w, h))
    else:
        bg_img = build_professional_bg(w, h, bg_preset)
    clip = VideoFileClip(video_path)
    fps = clip.fps or 24
    def process_func(get_frame, t):
        fr = get_frame(t)
        fr_u8 = (fr * 255).astype(np.uint8)
        comp = process_frame_rembg_optimized(cv2.cvtColor(fr_u8, cv2.COLOR_RGB2BGR), bg_img)
        return (comp.astype(np.float32) / 255.0)
    new_clip = clip.fl(process_func).set_fps(fps)
    output_path = "rembg_output.mp4"
    new_clip.write_videofile(output_path, audio=False, logger=None)
    clip.close(); new_clip.close()
    return output_path

# =========================
# Self-test harness
# =========================
def _ok(flag): return "✅" if flag else "❌"
def self_test_cuda():
    try:
        if not TORCH_AVAILABLE: return False, "Torch not importable"
        if not CUDA_AVAILABLE: return False, "CUDA not available"
        import torch as _t
        a = _t.randn((1024,1024), device="cuda"); b = _t.randn((1024,1024), device="cuda")
        c = (a @ b).mean().item(); return True, f"CUDA matmul ok, mean={c:.6f}"
    except Exception as e: return False, f"CUDA op failed: {e}"
def self_test_ffmpeg_moviepy():
    try:
        ff = shutil.which("ffmpeg")
        if not ff: return False, "ffmpeg not found on PATH"
        frames = [(np.zeros((64,64,3), np.uint8) + i).clip(0,255) for i in range(0,200,25)]
        clip = ImageSequenceClip(frames, fps=4)
        with tempfile.TemporaryDirectory() as td:
            vp = os.path.join(td, "tiny.mp4")
            clip.write_videofile(vp, audio=False, logger=None); clip.close()
            clip_r = VideoFileClip(vp); _ = clip_r.get_frame(0.1); clip_r.close()
        return True, "FFmpeg/MoviePy encode/decode ok"
    except Exception as e: return False, f"FFmpeg/MoviePy test failed: {e}"
def self_test_rembg():
    try:
        if not REMBG_AVAILABLE: return False, "rembg not importable"
        from rembg import remove
        img = np.zeros((64,64,3), dtype=np.uint8); img[:,:] = (0,255,0)
        pil = Image.fromarray(img); out = remove(pil)
        ok = isinstance(out, Image.Image) and out.size == (64,64)
        return ok, "rembg ok" if ok else "rembg returned unexpected output"
    except Exception as e: return False, f"rembg failed: {e}"
def self_test_sam2():
    try:
        if not SAM2_IMPORTED: return False, "SAM2 not importable"
        if not SAM2_PREDICTOR: return False, "SAM2 predictor not initialized"
        dummy = np.zeros((64,64,3), dtype=np.uint8)
        SAM2_PREDICTOR.set_image(dummy)
        pts = np.array([[32,32]], dtype=np.int32); lbs = np.array([1], dtype=np.int32)
        masks, scores, _ = SAM2_PREDICTOR.predict(point_coords=pts, point_labels=lbs, multimask_output=True)
        ok = masks is not None and len(masks) > 0
        return ok, "SAM2 micro-inference ok" if ok else "SAM2 predict returned no masks"
    except Exception as e: return False, f"SAM2 micro-inference failed: {e}"
def self_test_matanyone():
    try:
        ok_init = matanyone_processor.initialize()
        if not ok_init: return False, f"MatAnyone init failed: {getattr(matanyone_processor,'last_error','no details')}"
        if not matanyone_processor.verified: return False, "MatAnyone missing process_video API"
        with tempfile.TemporaryDirectory() as td:
            frames = []
            for t in range(8):
                frame = np.zeros((64,64,3), dtype=np.uint8)
                x = 8 + t*4; cv2.rectangle(frame, (x,20),(x+12,44), 200, -1); frames.append(frame)
            vid_path = os.path.join(td,"tiny_input.mp4")
            clip = ImageSequenceClip(frames, fps=8); clip.write_videofile(vid_path, audio=False, logger=None); clip.close()
            mask = np.zeros((64,64), dtype=np.uint8); cv2.rectangle(mask,(24,24),(40,40),255,-1)
            mask_path = os.path.join(td,"mask.png"); cv2.imwrite(mask_path, mask)
            alpha = matanyone_processor.process_video_optimized(vid_path, td)
            if alpha is None or not os.path.exists(alpha): return False, "MatAnyone did not produce alpha video"
            _alpha_clip = VideoFileClip(alpha); _ = _alpha_clip.get_frame(0.1); _alpha_clip.close()
            return True, "MatAnyone process_video ok"
    except Exception as e: return False, f"MatAnyone test failed: {e}"
def run_self_test() -> str:
    lines = []
    lines.append("=== SELF TEST REPORT ===")
    lines.append(f"Python: {sys.version.split()[0]}")
    lines.append(f"Torch: {torch.__version__ if TORCH_AVAILABLE else 'N/A'} | CUDA: {CUDA_AVAILABLE} | Device: {DEVICE} | GPU: {GPU_NAME}")
    lines.append(f"FFmpeg on PATH: {bool(shutil.which('ffmpeg'))}")
    lines.append("")
    tests = [("CUDA", self_test_cuda), ("FFmpeg/MoviePy", self_test_ffmpeg_moviepy),
             ("rembg", self_test_rembg), ("SAM2", self_test_sam2), ("MatAnyone", self_test_matanyone)]
    for name, fn in tests:
        t0 = time.time(); ok, msg = fn(); dt = time.time() - t0
        lines.append(f"{_ok(ok)} {name}: {msg} [{dt:.2f}s]")
    return "\n".join(lines)

# =========================
# Gradio input coercion helpers
# =========================
def _coerce_video_to_path(video_file):
    if video_file is None:
        return None
    if isinstance(video_file, str):
        return video_file
    if isinstance(video_file, dict) and "name" in video_file:
        return video_file["name"]
    return getattr(video_file, "name", None)

def _coerce_bg_to_path(bg_image, temp_dir):
    """Return filesystem path for background image, writing it to temp_dir if needed."""
    if bg_image is None:
        return None
    if isinstance(bg_image, str):
        return bg_image
    if isinstance(bg_image, dict) and "name" in bg_image:
        return bg_image["name"]
    if hasattr(bg_image, "name") and isinstance(bg_image.name, str):
        return bg_image.name
    if isinstance(bg_image, Image.Image):
        p = os.path.join(temp_dir, "bg_uploaded.png")
        bg_image.save(p); return p
    if isinstance(bg_image, np.ndarray):
        p = os.path.join(temp_dir, "bg_uploaded.png")
        arr = bg_image
        if arr.ndim == 3 and arr.shape[2] == 3:
            cv2.imwrite(p, cv2.cvtColor(arr, cv2.COLOR_RGB2BGR))
        else:
            cv2.imwrite(p, arr)
        return p
    return None

# =========================
# Gradio callback
# =========================
def gradio_interface_optimized(video_file, bg_image, use_matanyone=True, bg_preset="Office (Soft Gray)", stabilize=True, preroll_frames=12):
    try:
        if video_file is None:
            return None, None, "Please upload a video."
        print(f"UI types: video={type(video_file)}, bg={type(bg_image)}")

        with tempfile.TemporaryDirectory() as temp_dir:
            video_path = _coerce_video_to_path(video_file)
            if not video_path or not os.path.exists(video_path):
                return None, None, "Could not read the uploaded video path."
            bg_path = _coerce_bg_to_path(bg_image, temp_dir)  # may be None → preset is used

            # reflect UI choices
            matanyone_processor.stabilize = bool(stabilize)
            try:
                matanyone_processor.preroll_frames = max(0, int(preroll_frames))
            except Exception:
                pass

            start_time = time.time()

            if use_matanyone and MATANYONE_IMPORTED:
                if not matanyone_processor.initialized:
                    matanyone_processor.initialize()

                if matanyone_processor.initialized and matanyone_processor.verified:
                    alpha_video_path = matanyone_processor.process_video_optimized(video_path, temp_dir)
                    if alpha_video_path is None:
                        out = process_video_rembg_fallback(video_path, bg_path, bg_preset=bg_preset)
                        method = "rembg (fallback after MatAnyone error)"
                    else:
                        out = composite_with_background(video_path, alpha_video_path, bg_path, bg_preset=bg_preset)
                        method = f"MatAnyone (GPU: {CUDA_AVAILABLE})"
                else:
                    out = process_video_rembg_fallback(video_path, bg_path, bg_preset=bg_preset)
                    method = "rembg (MatAnyone not verified)"
            else:
                out = process_video_rembg_fallback(video_path, bg_path, bg_preset=bg_preset)
                method = "rembg"

            final_gpu = gpu_monitor.get_stats()
            elapsed = time.time() - start_time
            status = (
                f"✅ Processing complete\n"
                f"Method: {method}\n"
                f"Time: {elapsed:.2f}s\n"
                f"Output: {out}\n\n"
                f"GPU Stats:\n"
                f"• Mem: {final_gpu.get('memory_used', 0):.2f}GB / {final_gpu.get('memory_total', 0):.2f}GB"
                f" ({final_gpu.get('memory_percent', 0):.1f}%)\n"
                f"• Util: {final_gpu.get('gpu_util', 0)}%\n"
                f"• CUDA: {CUDA_AVAILABLE}"
            )
            return out, out, status

    except Exception as e:
        traceback.print_exc()
        msg = (
            f"❌ Error: {e}\n"
            f"- MatAnyone imported: {MATANYONE_IMPORTED}\n"
            f"- MatAnyone initialized: {matanyone_processor.initialized}\n"
            f"- MatAnyone verified: {matanyone_processor.verified}\n"
            f"- MatAnyone last_error: {matanyone_processor.last_error}\n"
            f"- SAM2 imported: {SAM2_IMPORTED}\n"
            f"- SAM2 verified: {SAM2_AVAILABLE}\n"
            f"- rembg: {REMBG_AVAILABLE}\n"
            f"- CUDA: {CUDA_AVAILABLE}\n"
            f"(see server logs for traceback)"
        )
        return None, None, msg

def gradio_run_self_test(): return run_self_test()
def show_matanyone_diag():
    try:
        ok = matanyone_processor.initialized and matanyone_processor.verified
        return "READY ✅" if ok else (matanyone_processor.last_error or "Not initialized yet")
    except Exception as e:
        return f"Diag error: {e}"

# =========================
# UI
# =========================
with gr.Blocks(title="Video Background Replacer - GPU Optimized", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🎬 Video Background Replacer (GPU Optimized)")
    gr.Markdown("All green checks are earned by real tests. No guesses.")
    gpu_status = f"✅ {GPU_NAME}" if CUDA_AVAILABLE else "❌ CPU Only"
    matany_status = "✅ Module Imported" if MATANYONE_IMPORTED else "❌ Not Importable"
    sam2_status = "✅ Verified" if SAM2_AVAILABLE else ("⚠️ Imported but unverified" if SAM2_IMPORTED else "❌ Not Ready")
    rembg_status = "✅ Ready" if REMBG_AVAILABLE else "❌ Not Available"
    torch_status = "✅ GPU" if CUDA_AVAILABLE else "❌ CPU"
    status_html = f"""
    <div style='padding: 15px; background: #f8f9fa; border-radius: 8px; margin-bottom: 20px; border-left: 4px solid #6c757d;'>
        <h4 style='margin-top: 0;'>🖥️ System Status (verified)</h4>
        <strong>GPU:</strong> {gpu_status}<br>
        <strong>Device:</strong> {DEVICE}<br>
        <strong>MatAnyone module:</strong> {matany_status}<br>
        <strong>MatAnyone ready:</strong> {"✅ Yes" if getattr(matanyone_processor, "verified", False) else "❌ No"}<br>
        <strong>SAM2:</strong> {sam2_status}<br>
        <strong>rembg:</strong> {rembg_status}<br>
        <strong>PyTorch:</strong> {torch_status}
    </div>
    """
    gr.HTML(status_html)

    with gr.Row():
        with gr.Column():
            video_input = gr.Video(label="📹 Input Video")
            bg_input = gr.Image(label="🖼️ Background Image (optional)", type="filepath")
            bg_preset = gr.Dropdown(
                label="🎨 Background Preset (if no image)",
                choices=["Office (Soft Gray)","Studio (Charcoal)","Nature (Green Tint)","Brand Blue","Plain Light"],
                value="Office (Soft Gray)",
            )
            use_matanyone = gr.Checkbox(label="🚀 Use MatAnyone (GPU accelerated, best quality)",
                                        value=MATANYONE_IMPORTED, interactive=MATANYONE_IMPORTED)
            stabilize = gr.Checkbox(label="🧱 Stabilize short clips (pre-roll first frame)",
                                    value=os.getenv("MATANYONE_STABILIZE","true").lower()=="true")
            preroll_frames = gr.Slider(label="Pre-roll frames", minimum=0, maximum=24, step=1,
                                       value=int(os.getenv("MATANYONE_PREROLL_FRAMES","12")))
            process_btn = gr.Button("🚀 Process Video", variant="primary")
            gr.Markdown("### 🔎 Self-Verification"); selftest_btn = gr.Button("Run Self-Test")
            selftest_out = gr.Textbox(label="Self-Test Report", lines=16)
            gr.Markdown("### 🛠 MatAnyone Diagnostics"); mat_diag_btn = gr.Button("Show MatAnyone Diagnostics")
            mat_diag_out = gr.Textbox(label="MatAnyone Last Error / Status", lines=6)
        with gr.Column():
            output_video = gr.Video(label="✨ Result")
            download_file = gr.File(label="💾 Download")
            status_text = gr.Textbox(label="📊 Status & Performance", lines=8)

    process_btn.click(fn=gradio_interface_optimized,
                      inputs=[video_input, bg_input, use_matanyone, bg_preset, stabilize, preroll_frames],
                      outputs=[output_video, download_file, status_text])
    selftest_btn.click(fn=gradio_run_self_test, inputs=[], outputs=[selftest_out])
    mat_diag_btn.click(fn=show_matanyone_diag, inputs=[], outputs=[mat_diag_out])

    gr.Markdown("---")
    gr.Markdown("""
    **Notes**
    - K-Governor clamps/pads Top-K inside MatAnyone to prevent 'k out of range' crashes.
    - Short-clip stabilizer pre-roll is trimmed out of alpha automatically.
    - SAM2 shows ✅ only after a real micro-inference passes.
    - FFmpeg/MoviePy, CUDA, and rembg are validated by actually running them.
    """)

# =========================
# Proactive warmup at boot (before UI render)
# =========================
try:
    if MATANYONE_IMPORTED and os.getenv("USE_MATANYONE","true").lower()=="true":
        print("Warming up MatAnyone…")
        matanyone_processor.initialize()
        print("MatAnyone warmup complete.")
except Exception as e:
    print(f"MatAnyone warmup failed (non-fatal): {e}")
    traceback.print_exc()

# =========================
# Late re-sanitization for external .env overrides
# =========================
def _re_sanitize_threads():
    for v in ("OMP_NUM_THREADS", "MKL_NUM_THREADS"):
        val = os.environ.get(v, "")
        if not str(val).isdigit():
            os.environ[v] = "2"
            print(f"{v} had invalid value; reset to 2")

if os.getenv("STRICT_ENV_GUARD","1") in ("1","true","TRUE"):
    _re_sanitize_threads()

# =========================
# Entrypoint / CLI self-test
# =========================
if __name__ == "__main__":
    if "--self-test" in sys.argv:
        report = run_self_test(); print(report)
        exit_code = 0
        for line in report.splitlines():
            if line.startswith("❌"): exit_code = 2; break
        sys.exit(exit_code)
    print("\n" + "="*50)
    print("🚀 Starting GPU-optimized Gradio app…")
    print("URL: http://0.0.0.0:7860")
    print(f"GPU Monitoring: {'Active' if CUDA_AVAILABLE else 'Disabled'}")
    print("="*50 + "\n")
    demo.launch(server_name="0.0.0.0", server_port=7860)