Update app.py

app.py

@@ -4,12 +4,15 @@
 GPU-optimized with real verification & self-tests (no mocks).
 
 Key guarantees:
-- MatAnyone init uses InferenceCore("PeiqingYang/MatAnyone") and runs a real clip test.
+- MatAnyone init tries HF repo-id first, then local checkpoint fallback; errors are surfaced.
 - SAM2 marked Verified only after a successful micro-inference.
 - rembg verified by actually removing BG on a tiny image.
 - FFmpeg/MoviePy tested by encoding/decoding a tiny clip.
 - CUDA verified by real CUDA ops; GPU stats shown from PyTorch (NVML if present).
 - "Run Self-Test" button & "--self-test" CLI flag for CI / manual checks.
+
+NOTE: Ensure requirements include at least:
+  huggingface-hub, requests, hydra-core, omegaconf, einops, timm, opencv-python-headless, moviepy, rembg
 """
 
 import os
@@ -55,9 +58,9 @@
 os.environ["GRADIO_SERVER_PORT"] = "7860"
 
 # Feature flags
-os.environ["USE_MATANYONE"] = "true"
-os.environ["USE_SAM2"] = "true"
-os.environ["SELF_CHECK_MODE"] = "false"
+os.environ["USE_MATANYONE"] = os.getenv("USE_MATANYONE", "true")
+os.environ["USE_SAM2"] = os.getenv("USE_SAM2", "true")
+os.environ["SELF_CHECK_MODE"] = os.getenv("SELF_CHECK_MODE", "false")
 
 # =========================
 # Imports
@@ -315,18 +318,20 @@ def build_professional_bg(w, h, preset: str) -> np.ndarray:
     return make_solid(w, h, (240, 240, 240))
 
 # =========================
-# MatAnyone wrapper (fixed)
+# MatAnyone wrapper (robust)
 # =========================
 class OptimizedMatAnyoneProcessor:
     """
-    Thin wrapper around MatAnyone's InferenceCore.
-    Uses HF repo-id constructor InferenceCore("PeiqingYang/MatAnyone").
+    Wrapper around MatAnyone's InferenceCore.
+    Tries HF repo-id first; if that fails (or the local class expects a `network`),
+    falls back to downloading `pretrained_models/matanyone.pth` and building the model.
     """
     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  # for diagnostics
 
     def initialize(self) -> bool:
         if not MATANYONE_IMPORTED:
@@ -334,18 +339,72 @@ def initialize(self) -> bool:
             return False
         if self.initialized and self.processor is not None:
             return True
+
+        self.last_error = None
+
+        # 1) Preferred path: HF repo-id (documented usage)
         try:
-            print(f"Initializing MatAnyone on {self.device}…")
-            self.processor = MatAnyInferenceCore("PeiqingYang/MatAnyone")  # auto-downloads
+            print(f"Initializing MatAnyone (HF repo-id) on {self.device}…")
+            self.processor = MatAnyInferenceCore("PeiqingYang/MatAnyone")  # per README
             self.verified = hasattr(self.processor, "process_video")
             if self.device == "cuda":
                 torch.cuda.empty_cache()
-                _ = torch.rand(1, device="cuda") * 0.0  # warm CUDA context
+                _ = torch.rand(1, device="cuda") * 0.0  # warmup
             self.initialized = True
-            print("✅ MatAnyone initialized (HF repo-id).")
+            print("✅ MatAnyone initialized via HF repo-id.")
             return True
+        except TypeError as e:
+            # Classic sign of older API: __init__ requires a `network` object
+            if "network" in str(e).lower():
+                print("MatAnyone InferenceCore expects a `network` (older API); will build network locally.")
+            else:
+                print(f"HF init TypeError: {e}")
+            self.last_error = f"HF init TypeError: {e}"
         except Exception as e:
-            print(f"MatAnyone initialization failed: {e}")
+            self.last_error = f"HF init failed: {type(e).__name__}: {e}"
+            print(self.last_error)
+
+        # 2) Fallback: local checkpoint → network object → InferenceCore(network)
+        try:
+            print("Falling back to local checkpoint init for MatAnyone…")
+            from pathlib import Path
+            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}")
+
+            # Build the actual network then wrap with InferenceCore
+            network = get_matanyone_model(
+                str(ckpt_path),
+                device=("cuda" if CUDA_AVAILABLE else "cpu")
+            )
+            self.processor = MatAnyInferenceCore(network)
+            self.verified = hasattr(self.processor, "process_video")
+
+            if self.device == "cuda":
+                torch.cuda.empty_cache()
+                _ = torch.rand(1, device="cuda") * 0.0
+
+            self.initialized = True
+            print("✅ MatAnyone initialized via local checkpoint.")
+            return True
+
+        except Exception as e:
+            self.last_error = f"Local init failed: {type(e).__name__}: {e}"
+            print(f"MatAnyone initialization failed: {self.last_error}")
             import traceback; traceback.print_exc()
             return False
 
@@ -374,7 +433,7 @@ def create_mask_optimized(self, video_path: str, output_path: str) -> str:
             except Exception as e:
                 print(f"SAM2 mask creation failed; fallback rectangle. Error: {e}")
 
-        # Fallback: centered box (ensures pipeline continuity)
+        # 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)
@@ -575,8 +634,9 @@ def self_test_sam2() -> (bool, str):
 
 def self_test_matanyone() -> (bool, str):
     try:
-        if not matanyone_processor.initialize():
-            return False, "MatAnyone init failed"
+        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"
         # Create a tiny real video + mask, then run process_video
@@ -588,8 +648,15 @@ def self_test_matanyone() -> (bool, str):
                 x = 8 + t*4
                 cv2.rectangle(frame, (x, 20), (x+12, 44), (200, 200, 200), -1)
                 frames.append(frame)
+
+            # Write temp video and a visible copy you can inspect
             vid_path = os.path.join(td, "tiny_input.mp4")
-            ImageSequenceClip(frames, fps=8).write_videofile(vid_path, audio=False, logger=None)
+            clip = ImageSequenceClip(frames, fps=8)
+            clip.write_videofile(vid_path, audio=False, logger=None)
+            visible_test_clip = CHECKPOINTS_DIR / "selftest_clip.mp4"
+            clip.write_videofile(str(visible_test_clip), audio=False, logger=None)
+            clip.close()
+            print(f"📹 Self-test clip saved to {visible_test_clip}")
 
             # Simple central mask (seed)
             mask = np.zeros((64, 64), dtype=np.uint8)
@@ -703,6 +770,10 @@ def gradio_interface_optimized(video_file, bg_image, use_matanyone=True, bg_pres
 def gradio_run_self_test():
     return run_self_test()
 
+def show_matanyone_diag():
+    ok = matanyone_processor.initialized and matanyone_processor.verified
+    return "READY ✅" if ok else (matanyone_processor.last_error or "Not initialized yet")
+
 # =========================
 # UI
 # =========================
@@ -756,6 +827,10 @@ def gradio_run_self_test():
             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")
@@ -773,11 +848,18 @@ def gradio_run_self_test():
         outputs=[selftest_out],
     )
 
+    mat_diag_btn.click(
+        fn=show_matanyone_diag,
+        inputs=[],
+        outputs=[mat_diag_out],
+    )
+
     gr.Markdown("---")
     gr.Markdown("""
     **Notes**
     - SAM2 shows ✅ only after a real micro-inference passes.
     - MatAnyone shows ✅ only if initialization succeeded and `process_video` exists; the self-test also runs a tiny real video through it.
+    - The self-test saves a visible clip to `checkpoints/selftest_clip.mp4` so you can inspect what was used.
     - FFmpeg/MoviePy, CUDA, and rembg are validated by actually running them.
     """)