#!/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"""

🖥️ System Status (verified)

GPU: {gpu_status}
Device: {DEVICE}
MatAnyone module: {matany_status}
MatAnyone ready: {"✅ Yes" if getattr(matanyone_processor, "verified", False) else "❌ No"}
SAM2: {sam2_status}
rembg: {rembg_status}
PyTorch: {torch_status}
""" 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)