fix 10

pipeline.py

@@ -12,6 +12,7 @@
 - Fallbacks: MediaPipe SelfieSegmentation → else OpenCV GrabCut
 - H.264 MP4 output (ffmpeg when available; OpenCV fallback)
 - Audio mux: original audio copied into final output (AAC) if present
+- NEW: Stage-A transparent export (VP9 with alpha or checkerboard preview)
 
 Environment knobs (all optional):
 - THIRD_PARTY_SAM2_DIR, THIRD_PARTY_MATANY_DIR
@@ -25,6 +26,8 @@
 - EDGE_ERODE=1, EDGE_DILATE=2, EDGE_BLUR=1.5
 - LIGHTWRAP_RADIUS=5, LIGHTWRAP_AMOUNT=0.18
 - DESPILL_AMOUNT=0.35
+- RETURN_STAGE_A=0 | 1                (if 1, return Stage-A file instead of final composite)
+- STAGEA_VP9_CRF=28                   (quality for VP9 alpha export)
 """
 
 from __future__ import annotations
@@ -255,6 +258,152 @@ def _video_writer(out_path: Path, fps: int, size: Tuple[int, int]) -> cv2.VideoW
     fourcc = cv2.VideoWriter_fourcc(*"mp4v")
     return cv2.VideoWriter(str(out_path), fourcc, max(1, fps), size)
 
+# --- Stage-A (transparent) builders ----------------------------------------------------
+def _build_stage_a_rgba_vp9_from_fg_alpha(
+    fg_path: Union[str, Path],
+    alpha_path: Union[str, Path],
+    out_webm: Union[str, Path],
+    fps: int,
+    size: Tuple[int, int],
+    src_audio: Optional[Union[str, Path]] = None,
+) -> bool:
+    """Merge FG+ALPHA → RGBA WebM (VP9 with alpha). Optionally mux original audio (Opus)."""
+    if not _probe_ffmpeg():
+        return False
+    w, h = size
+    try:
+        cmd = [
+            _ffmpeg_bin(), "-y",
+            "-i", str(fg_path),                     # 0: FG video
+            "-i", str(alpha_path),                  # 1: ALPHA video (grayscale)
+        ]
+        if src_audio:
+            cmd += ["-i", str(src_audio)]           # 2: original (for audio)
+        fcx = f"[1:v]format=gray,scale={w}:{h},fps={fps}[al];" \
+              f"[0:v]scale={w}:{h},fps={fps}[fg];" \
+              f"[fg][al]alphamerge[outv]"
+        cmd += ["-filter_complex", fcx, "-map", "[outv]"]
+        if src_audio:
+            cmd += ["-map", "2:a:0?", "-c:a", "libopus", "-b:a", "128k"]
+        cmd += [
+            "-c:v", "libvpx-vp9", "-pix_fmt", "yuva420p",
+            "-crf", os.environ.get("STAGEA_VP9_CRF", "28"),
+            "-b:v", "0", "-row-mt", "1",
+            "-shortest",
+            str(out_webm),
+        ]
+        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        return True
+    except Exception as e:
+        logger.warning(f"Stage-A VP9(alpha) build failed: {e}")
+        return False
+
+def _build_stage_a_rgba_vp9_from_mask(
+    video_path: Union[str, Path],
+    mask_png: Union[str, Path],
+    out_webm: Union[str, Path],
+    fps: int,
+    size: Tuple[int, int],
+) -> bool:
+    """Merge original video + static mask → RGBA WebM (VP9 with alpha)."""
+    if not _probe_ffmpeg():
+        return False
+    w, h = size
+    try:
+        cmd = [
+            _ffmpeg_bin(), "-y",
+            "-i", str(video_path),                  # 0: original video
+            "-loop", "1", "-i", str(mask_png),      # 1: static PNG mask (grayscale)
+            "-filter_complex",
+            f"[1:v]format=gray,scale={w}:{h},fps={fps}[al];"
+            f"[0:v]scale={w}:{h},fps={fps}[fg];"
+            f"[fg][al]alphamerge[outv]",
+            "-map", "[outv]",
+            "-c:v", "libvpx-vp9", "-pix_fmt", "yuva420p",
+            "-crf", os.environ.get("STAGEA_VP9_CRF", "28"),
+            "-b:v", "0", "-row-mt", "1",
+            "-shortest",
+            str(out_webm),
+        ]
+        subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        return True
+    except Exception as e:
+        logger.warning(f"Stage-A VP9(alpha) (mask) build failed: {e}")
+        return False
+
+def _checkerboard_bg(w: int, h: int, tile: int = 32) -> np.ndarray:
+    """RGB checkerboard (for preview when no real alpha is possible)."""
+    y, x = np.mgrid[0:h, 0:w]
+    c = ((x // tile) + (y // tile)) % 2
+    a = np.where(c == 0, 200, 150).astype(np.uint8)
+    return np.stack([a, a, a], axis=-1)
+
+def _build_stage_a_checkerboard_from_fg_alpha(
+    fg_path: Union[str, Path],
+    alpha_path: Union[str, Path],
+    out_mp4: Union[str, Path],
+    fps: int,
+    size: Tuple[int, int],
+) -> bool:
+    """Preview: FG+ALPHA over checkerboard → MP4 (no real alpha)."""
+    fg_cap = cv2.VideoCapture(str(fg_path))
+    al_cap = cv2.VideoCapture(str(alpha_path))
+    if not fg_cap.isOpened() or not al_cap.isOpened():
+        return False
+    w, h = size
+    writer = _video_writer(Path(out_mp4), fps, (w, h))
+    bg = _checkerboard_bg(w, h)
+    ok_any = False
+    try:
+        while True:
+            okf, fg = fg_cap.read()
+            oka, al = al_cap.read()
+            if not okf or not oka:
+                break
+            fg = cv2.resize(fg, (w, h))
+            al = cv2.cvtColor(cv2.resize(al, (w, h)), cv2.COLOR_BGR2GRAY)
+            comp = _composite_frame_pro(cv2.cvtColor(fg, cv2.COLOR_BGR2RGB), al, bg)
+            writer.write(cv2.cvtColor(comp, cv2.COLOR_RGB2BGR))
+            ok_any = True
+    finally:
+        fg_cap.release()
+        al_cap.release()
+        writer.release()
+    return ok_any
+
+def _build_stage_a_checkerboard_from_mask(
+    video_path: Union[str, Path],
+    mask_png: Union[str, Path],
+    out_mp4: Union[str, Path],
+    fps: int,
+    size: Tuple[int, int],
+) -> bool:
+    """Preview: original video + static mask over checkerboard → MP4."""
+    cap = cv2.VideoCapture(str(video_path))
+    if not cap.isOpened():
+        return False
+    w, h = size
+    mask = cv2.imread(str(mask_png), cv2.IMREAD_GRAYSCALE)
+    if mask is None:
+        return False
+    mask = cv2.resize(mask, (w, h), interpolation=cv2.INTER_NEAREST)
+    writer = _video_writer(Path(out_mp4), fps, (w, h))
+    bg = _checkerboard_bg(w, h)
+    ok_any = False
+    try:
+        while True:
+            ok, frame = cap.read()
+            if not ok:
+                break
+            frame = cv2.resize(frame, (w, h))
+            comp = _composite_frame_pro(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB), mask, bg)
+            writer.write(cv2.cvtColor(comp, cv2.COLOR_RGB2BGR))
+            ok_any = True
+    finally:
+        cap.release()
+        writer.release()
+    return ok_any
+
 # --------------------------------------------------------------------------------------
 # SAM2 Integration
 # --------------------------------------------------------------------------------------
@@ -292,7 +441,6 @@ def run_sam2_mask(predictor: object,
     if predictor is None:
         return None, False
     try:
-        # --- begin try block body (ensure syntax is correct) ---
         rgb = cv2.cvtColor(first_frame_bgr, cv2.COLOR_BGR2RGB)
         predictor.set_image(rgb)
 
@@ -604,7 +752,7 @@ def process(video_path: Union[str, Path],
             point_y: Optional[float] = None,
             auto_box: bool = False,
             work_dir: Optional[Union[str, Path]] = None) -> Tuple[Optional[str], Dict[str, Any]]:
-    """Orchestrate: SAM2 mask → (optional GrabCut refine) → MatAnyone → composite → mux audio."""
+    """Orchestrate: SAM2 mask → (optional GrabCut refine) → MatAnyone → Stage-A → composite → mux audio."""
     t0 = time.time()
     diagnostics: Dict[str, Any] = {
         "sam2_ok": False,
@@ -673,9 +821,44 @@ def process(video_path: Union[str, Path],
     else:
         ran = False
 
-    output_path = tmp_root / "output.mp4"
+    # --- Build Stage-A (transparent) file for inspection ---
+    stageA_path = None
+    stageA_ok = False
+    if diagnostics["matany_ok"] and fg_path and al_path:
+        stageA_path = tmp_root / "stageA_transparent.webm"
+        if _probe_ffmpeg():
+            stageA_ok = _build_stage_a_rgba_vp9_from_fg_alpha(
+                fg_path, al_path, stageA_path, diagnostics["fps"], (vw, vh), src_audio=video_path
+            )
+        if not stageA_ok:
+            stageA_path = tmp_root / "stageA_checkerboard.mp4"
+            stageA_ok = _build_stage_a_checkerboard_from_fg_alpha(
+                fg_path, al_path, stageA_path, diagnostics["fps"], (vw, vh)
+            )
+    else:
+        stageA_path = tmp_root / "stageA_transparent.webm"
+        if _probe_ffmpeg():
+            stageA_ok = _build_stage_a_rgba_vp9_from_mask(
+                video_path, mask_png, stageA_path, diagnostics["fps"], (vw, vh)
+            )
+        if not stageA_ok:
+            stageA_path = tmp_root / "stageA_checkerboard.mp4"
+            stageA_ok = _build_stage_a_checkerboard_from_mask(
+                video_path, mask_png, stageA_path, diagnostics["fps"], (vw, vh)
+            )
+
+    diagnostics["stageA_path"] = str(stageA_path) if stageA_ok else None
+    diagnostics["stageA_note"] = (
+        "WebM with real alpha (VP9)" if stageA_ok and str(stageA_path).endswith(".webm")
+        else ("MP4 checkerboard preview (no real alpha)" if stageA_ok else "Stage-A build failed")
+    )
 
-    # 3) Composite
+    # Optional: return Stage-A instead of final composite
+    if os.environ.get("RETURN_STAGE_A", "0").strip() == "1" and stageA_ok:
+        return str(stageA_path), diagnostics
+
+    # 3) Composite to final background
+    output_path = tmp_root / "output.mp4"
     if diagnostics["matany_ok"] and fg_path and al_path:
         ok_comp = composite_video(fg_path, al_path, bg_image_path, output_path, diagnostics["fps"], (vw, vh))
         if not ok_comp: