Upload code/faz3_v6/meshai_api/pipeline.py with huggingface_hub

code/faz3_v6/meshai_api/pipeline.py

@@ -0,0 +1,296 @@
+"""
+Faz3 generate pipeline — TAM yol.
+
+image|prompt -> Microsoft TRELLIS -> Hunyuan3D-Paint maps -> meshai_bridge -> result.glb
+Hafif hybrid: sadece MESHAI_ALLOW_LEGACY_HYBRID=1
+Stub: sadece MESHAI_ALLOW_TRELLIS_STUB=1
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+from pathlib import Path
+from typing import Any
+
+from meshai_api.hunyuan import (
+    ensure_stub_hunyuan_maps,
+    hunyuan_ready,
+    release_hunyuan_cuda,
+    resolve_hunyuan_maps_dir,
+    run_hunyuan_paint,
+    write_hunyuan_maps_from_feats,
+)
+from meshai_api.trellis import (
+    allow_trellis_stub,
+    release_trellis_cuda,
+    run_trellis_geometry,
+    trellis_ready,
+)
+from meshai_api.weights import resolve_faz2_ckpt, runtime_status
+
+PIPELINE_ID = "faz3-full-v1"
+
+
+def _env_int(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        return int(raw)
+    except ValueError:
+        return default
+
+
+def _truthy(name: str) -> bool:
+    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes"}
+
+
+def _write_trellis_stub(
+    *,
+    job_dir: Path,
+    prompt: str,
+    image_path: Path | None,
+    log: Any,
+) -> Path:
+    out = job_dir / "trellis.glb"
+    import numpy as np
+    import trimesh
+
+    mesh = trimesh.creation.box(extents=(1.0, 1.0, 1.0))
+    seed = abs(hash(prompt or (image_path.name if image_path else "x"))) % (2**32)
+    rng = np.random.default_rng(seed)
+    color = (
+        int(rng.integers(80, 220)),
+        int(rng.integers(80, 220)),
+        int(rng.integers(80, 220)),
+        255,
+    )
+    mesh.visual.vertex_colors = color
+    mesh.export(out)
+    log(f"[faz3] trellis STUB GLB (explicit allow): {out}")
+    return out
+
+
+def _resolve_final_glb(bridge_result: dict[str, Any], bridged_dir: Path) -> Path:
+    final = Path(bridge_result.get("final_glb") or "")
+    if final.is_file():
+        return final
+    for name in ("final_textured.glb", "normalized.glb", "final.glb"):
+        candidate = bridged_dir / name
+        if candidate.is_file():
+            return candidate
+    raise FileNotFoundError(f"bridge final GLB yok: {bridged_dir}")
+
+
+def _collect_previews(bridged_dir: Path, job_dir: Path, *, max_n: int = 4) -> list[str]:
+    render_dir = bridged_dir / "renders"
+    if not render_dir.is_dir():
+        return []
+    exts = {".png", ".jpg", ".jpeg", ".webp"}
+    frames = sorted(
+        p for p in render_dir.iterdir() if p.is_file() and p.suffix.lower() in exts
+    )[:max_n]
+    if not frames:
+        return []
+    preview_dir = job_dir / "previews"
+    preview_dir.mkdir(parents=True, exist_ok=True)
+    rels: list[str] = []
+    for i, src in enumerate(frames):
+        dest = preview_dir / f"view_{i:02d}{src.suffix.lower()}"
+        shutil.copy2(src, dest)
+        rels.append(f"previews/{dest.name}")
+    return rels
+
+
+def run_bridge_on_trellis(
+    *,
+    trellis_glb: Path,
+    out_dir: Path,
+    hunyuan_maps_dir: Path | None = None,
+    log: Any = print,
+) -> tuple[Path, dict[str, Any]]:
+    from meshai_bridge.bridge_pipeline import run_bridge_runtime
+    from meshai_bridge.constants import BRIDGE_VERSION
+
+    views = _env_int("MESHAI_BRIDGE_VIEWS", 4)
+    render_size = _env_int("MESHAI_BRIDGE_RENDER_SIZE", 512)
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    maps = hunyuan_maps_dir if hunyuan_maps_dir and hunyuan_maps_dir.exists() else None
+    result = run_bridge_runtime(
+        trellis_glb,
+        out_dir,
+        hunyuan_maps_dir=maps,
+        views=views,
+        render_size=render_size,
+        unwrap_uv=True,
+        textured_renders=not _truthy("MESHAI_BRIDGE_GRAY_RENDERS"),
+        voxel_fn=None,
+        latent_fn=None,
+        final_glb_name="final_textured.glb",
+    )
+    result.setdefault("bridge_version", BRIDGE_VERSION)
+    result["hunyuan_maps_dir"] = str(maps) if maps else None
+    final = _resolve_final_glb(result, out_dir)
+    log(f"[faz3] bridge final: {final}")
+    return final, result
+
+
+def run_generate_job(
+    *,
+    job_dir: Path,
+    prompt: str = "",
+    image_path: Path | None = None,
+    log: Any = print,
+) -> Path:
+    job_dir.mkdir(parents=True, exist_ok=True)
+    faz2 = resolve_faz2_ckpt(download=_truthy("MESHAI_FAZ2_AUTO_DOWNLOAD"), log=log)
+    meta: dict[str, Any] = {
+        "pipeline": PIPELINE_ID,
+        "prompt": prompt,
+        "image": str(image_path) if image_path else None,
+        "trellis": None,
+        "hunyuan_source": None,
+        "previews": [],
+        "trellis_health": trellis_ready(),
+        "hunyuan": hunyuan_ready(job_dir),
+        "runtime": runtime_status(),
+        "faz2_ckpt": str(faz2) if faz2 else None,
+        "faz2_bind": "pending_peft" if faz2 else "missing",
+        "note": "Faz2 ckpt cached; TRELLIS full PEFT bind henuz yok. Paint icin VRAM handoff aktif.",
+    }
+
+    # Onceki job Paint modeli tutuyorsa TRELLIS'e yer ac
+    release_hunyuan_cuda(log=log)
+
+    trellis = job_dir / "trellis.glb"
+    trellis_meta: dict[str, Any] = {}
+    if not trellis.is_file():
+        try:
+            trellis, trellis_meta = run_trellis_geometry(
+                job_dir=job_dir,
+                prompt=prompt,
+                image_path=image_path,
+                log=log,
+            )
+            meta["trellis"] = trellis_meta.get("backend") or "trellis_full"
+            meta["trellis_meta"] = {
+                k: v for k, v in trellis_meta.items() if k != "feats"
+            }
+        except Exception as exc:
+            if allow_trellis_stub():
+                log(f"[faz3] TRELLIS full fail, stub allow: {exc}")
+                trellis = _write_trellis_stub(
+                    job_dir=job_dir, prompt=prompt, image_path=image_path, log=log
+                )
+                meta["trellis"] = "stub"
+                meta["trellis_error"] = str(exc)
+            else:
+                meta["trellis_error"] = str(exc)
+                (job_dir / "FAILED.txt").write_text(
+                    f"trellis failed: {exc}", encoding="utf-8"
+                )
+                (job_dir / "meta.json").write_text(
+                    json.dumps(meta, indent=2, default=str), encoding="utf-8"
+                )
+                raise
+
+    # TRELLIS VRAM'i birak; Paint ayni A100'e sigsin
+    release_trellis_cuda(log=log)
+
+    # Hunyuan full paint
+    hy_info: dict[str, Any] | None = None
+    try:
+        hy_info = run_hunyuan_paint(
+            job_dir=job_dir,
+            trellis_glb=trellis,
+            image_path=image_path,
+            prompt=prompt,
+            log=log,
+        )
+        meta["hunyuan_source"] = "hunyuan_paint"
+        meta["hunyuan_meta"] = hy_info
+    except Exception as exc:
+        meta["hunyuan_paint_error"] = str(exc)
+        log(f"[faz3] Hunyuan paint fail: {exc}")
+        # Legacy latent maps only if hybrid feats present
+        feats = trellis_meta.get("feats") if isinstance(trellis_meta, dict) else None
+        if isinstance(feats, dict) and _truthy("MESHAI_ALLOW_LEGACY_HYBRID"):
+            write_hunyuan_maps_from_feats(
+                job_dir,
+                tex_latent=feats.get("tex_latent"),
+                hunyuan_feat=feats.get("hunyuan_feat"),
+                bridge_out=feats.get("bridge_out"),
+                log=log,
+            )
+            meta["hunyuan_source"] = "legacy_latent"
+        stub = ensure_stub_hunyuan_maps(job_dir)
+        if stub and meta.get("hunyuan_source") != "legacy_latent":
+            meta["hunyuan_source"] = "stub"
+            log(f"[faz3] hunyuan stub maps: {stub}")
+
+    maps_dir = resolve_hunyuan_maps_dir(job_dir)
+    meta["hunyuan"] = hunyuan_ready(job_dir)
+
+    # Prefer Hunyuan's own textured GLB when present (correct UV bake).
+    # Fallback: TRELLIS mesh + extracted maps via bridge packager.
+    bridge_input = trellis
+    textured = None
+    if hy_info and hy_info.get("textured_glb"):
+        candidate = Path(hy_info["textured_glb"])
+        if candidate.is_file():
+            textured = candidate
+            bridge_input = textured
+            meta["bridge_input"] = "hunyuan_textured"
+            # Maps already baked into textured GLB — avoid re-packing onto TRELLIS UVs
+            maps_dir = None
+    elif maps_dir is not None:
+        meta["bridge_input"] = "trellis_plus_maps"
+
+    bridged_dir = job_dir / "bridged"
+    try:
+        final_glb, bridge_result = run_bridge_on_trellis(
+            trellis_glb=bridge_input,
+            out_dir=bridged_dir,
+            hunyuan_maps_dir=maps_dir,
+            log=log,
+        )
+        meta["bridge"] = {
+            "version": bridge_result.get("bridge_version"),
+            "render_mode": bridge_result.get("render_mode"),
+            "final_glb": str(final_glb),
+            "normalized_glb": bridge_result.get("normalized_glb"),
+            "hunyuan_maps_dir": bridge_result.get("hunyuan_maps_dir"),
+            "hunyuan_stage": None,
+        }
+        report_path = bridged_dir / "bridge_report.json"
+        if report_path.is_file():
+            try:
+                report = json.loads(report_path.read_text(encoding="utf-8"))
+                meta["bridge"]["hunyuan_stage"] = report.get("hunyuan_stage")
+                meta["bridge"]["errors"] = report.get("bridge_errors") or report.get("errors")
+                meta["bridge"]["pbr"] = report.get("pbr")
+            except Exception:
+                pass
+        meta["previews"] = _collect_previews(bridged_dir, job_dir)
+    except Exception as exc:
+        meta["bridge_error"] = str(exc)
+        (job_dir / "meta.json").write_text(
+            json.dumps(meta, indent=2, default=str), encoding="utf-8"
+        )
+        log(f"[faz3] bridge failed, falling back to input glb: {exc}")
+        final_glb = bridge_input
+
+    out_glb = job_dir / "result.glb"
+    if Path(final_glb).resolve() != out_glb.resolve():
+        shutil.copy2(final_glb, out_glb)
+    else:
+        out_glb = Path(final_glb)
+
+    meta["artifact_glb"] = str(out_glb)
+    (job_dir / "meta.json").write_text(
+        json.dumps(meta, indent=2, default=str), encoding="utf-8"
+    )
+    log(f"[faz3] result GLB: {out_glb}")
+    return out_glb