import gradio as gr import subprocess import sys import os import base64 import tempfile import time import json import urllib.parse import numpy as np from PIL import Image # ── ZeroGPU optional ── try: import spaces HAS_GPU = True except Exception: HAS_GPU = False class spaces: @staticmethod def GPU(duration=60): return lambda fn: fn # ── gradio_client optional (for the TripoSplat alt-model call) ── try: from gradio_client import Client as GradioClient, handle_file as gradio_handle_file HAS_GRADIO_CLIENT = True except Exception: HAS_GRADIO_CLIENT = False # ── plyfile optional (for the auto-clean / noise-pruning step) ── try: from plyfile import PlyData, PlyElement HAS_PLYFILE = True except Exception: HAS_PLYFILE = False # ── Install SHARP ── def install_sharp(): try: r = subprocess.run(["sharp", "--help"], capture_output=True, timeout=10) if r.returncode == 0: return except Exception: pass print("Installing Apple SHARP...") subprocess.check_call([ sys.executable, "-m", "pip", "install", "git+https://github.com/apple/ml-sharp.git", "--quiet" ]) install_sharp() # ── Resize before SHARP ── def resize_image(path, max_size=512): img = Image.open(path).convert("RGB") w, h = img.size if max(w, h) > max_size: r = max_size / max(w, h) img = img.resize((int(w*r), int(h*r)), Image.LANCZOS) tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False, prefix="sharp_in_") img.save(tmp.name, "JPEG", quality=90) return tmp.name # ──────────────────────────────────────────────────────────────── # Core SHARP runner — shared by the GPU and CPU code paths. # # Progress: SHARP's CLI doesn't stream fine-grained step-by-step progress, # so we can't show a "true" per-layer progress bar. What we *can* do # honestly: launch it with Popen (not the blocking .run()) and poll it in # a loop from this same function, calling gr.Progress() each tick. Gradio # streams those calls live to the frontend as the function runs — no # threading needed, and no risk of interfering with ZeroGPU's execution # context. The percentage is paced against historically observed timing # (we logged ~49s for a typical run), so it's a real, live-updating bar — # just not literally tied to internal model layers, which SHARP doesn't # expose to us. # ──────────────────────────────────────────────────────────────── def _sharp_core(image_path, timeout_s, env_overrides=None, progress=None, label="", desc_prefix="", est_total=45.0): t0 = time.time() print(f"[{label}] START image_path={image_path}", flush=True) if image_path is None: return None, "⚠ Please upload a photo first." out_dir = tempfile.mkdtemp(prefix="splat_") resized = None def _p(frac, desc): if progress is not None: try: progress(frac, desc=desc_prefix + desc) except Exception: pass try: _p(0.05, "Resizing photo…") resized = resize_image(image_path, 512) print(f"[{label}] resized -> {resized} (+{time.time()-t0:.1f}s)", flush=True) env = os.environ.copy() if env_overrides: env.update(env_overrides) _p(0.12, "Starting SHARP…") proc = subprocess.Popen( ["sharp", "predict", "-i", resized, "-o", out_dir], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, env=env ) while proc.poll() is None: elapsed = time.time() - t0 if elapsed > timeout_s: proc.kill() proc.wait(timeout=5) raise subprocess.TimeoutExpired(cmd="sharp", timeout=timeout_s) frac = 0.12 + 0.78 * min(elapsed / est_total, 1.0) _p(frac, f"Generating gaussians… {elapsed:.0f}s") time.sleep(0.4) stdout, _ = proc.communicate() print(f"[{label}] sharp exited code={proc.returncode} (+{time.time()-t0:.1f}s)", flush=True) _p(0.93, "Reading result…") ply_files = [f for f in os.listdir(out_dir) if f.endswith(".ply")] if not ply_files: err = (stdout or "No .ply produced.")[-800:] print(f"[{label}] FAILED, no .ply produced:\n{err}", flush=True) return None, f"SHARP failed:\n{err}" ply_path = os.path.join(out_dir, ply_files[0]) size_mb = os.path.getsize(ply_path) / 1024 / 1024 print(f"[{label}] DONE ply={ply_path} size={size_mb:.2f}MB total={time.time()-t0:.1f}s", flush=True) _p(1.0, "Done") return ply_path, "✓ Done — 3D scene loading below ↓" except subprocess.TimeoutExpired: print(f"[{label}] TIMEOUT after {time.time()-t0:.1f}s", flush=True) return None, "⚠ Timed out. Try a smaller/simpler photo, or (if on GPU) increase GPU_DURATION near the top of app.py if you have more quota." except FileNotFoundError: print(f"[{label}] SHARP binary not found on PATH", flush=True) return None, "⚠ SHARP not found yet — wait 1 minute and try again." except Exception as e: print(f"[{label}] EXCEPTION: {e}", flush=True) return None, f"⚠ Error: {str(e)}" finally: if resized: try: os.unlink(resized) except: pass # NOTE on GPU_DURATION: HuggingFace's free-tier ZeroGPU accounts get a # 5-minute (300s) TOTAL daily quota. Requesting a duration at or above that # cap makes the platform reject the call outright ("illegal duration") — # instantly, before your function even runs. Keeping this comfortably # under the cap lets a free/unauthenticated user get a few conversions in # per day instead of burning the whole quota on one call. GPU_DURATION = 90 # CPU has no ZeroGPU quota at all, but is much slower — this is just a # generous soft cap for our own subprocess, not a platform guarantee. CPU_TIMEOUT = 300 @spaces.GPU(duration=GPU_DURATION) def run_sharp_gpu(image_path, progress=None, desc_prefix=""): return _sharp_core(image_path, GPU_DURATION - 15, env_overrides=None, progress=progress, label="run_sharp_gpu", desc_prefix=desc_prefix, est_total=45.0) def run_sharp_cpu(image_path, progress=None, desc_prefix=""): # Forcing CUDA_VISIBLE_DEVICES empty makes SHARP (and torch under it) # fall back to CPU even if a GPU happens to be visible in this # container, regardless of whether SHARP itself exposes a --device flag. return _sharp_core(image_path, CPU_TIMEOUT, env_overrides={"CUDA_VISIBLE_DEVICES": ""}, progress=progress, label="run_sharp_cpu", desc_prefix=desc_prefix, est_total=150.0) # ──────────────────────────────────────────────────────────────── # Alt model: TripoSplat, called via its own live public HF Space # (VAST-AI/TripoSplat) instead of vendored locally. # # Why not run it in this Space directly? TripoSplat's own pipeline needs # several GB of diffusion/VAE/background-removal checkpoints and a # heavier runtime than SHARP's single CLI call — that would blow past # both our GPU_DURATION budget and this Space's disk allowance. Calling # their existing Space offloads the heavy compute to VAST-AI's own # ZeroGPU quota instead of ours. # # Honesty note: the exact positional argument order below is my best # reading of their public UI (image, seed, steps, guidance scale, gaussian # count, format) — I couldn't verify the literal function signature # without live-testing it. If it's wrong, client.view_api()'s output # (logged below) will show the real one, and it's a one-line fix. # ──────────────────────────────────────────────────────────────── _tripo_client = None def get_tripo_client(): global _tripo_client if _tripo_client is None: _tripo_client = GradioClient("VAST-AI/TripoSplat") try: print(f"[tripo] API schema:\n{_tripo_client.view_api(print_info=False)}", flush=True) except Exception as e: print(f"[tripo] view_api() failed (non-fatal): {e}", flush=True) return _tripo_client def _extract_ply_path(result): """gradio_client results can be nested tuples/dicts/strings depending on the target Space's output components — walk them looking for anything that looks like a .ply (or .splat) file path/url.""" def _walk(x): if isinstance(x, str) and x.lower().endswith((".ply", ".splat")): return x if isinstance(x, dict): for k in ("path", "url", "name"): v = x.get(k) if isinstance(v, str) and v.lower().endswith((".ply", ".splat")): return v for v in x.values(): found = _walk(v) if found: return found if isinstance(x, (list, tuple)): for v in x: found = _walk(v) if found: return found return None return _walk(result) def run_tripo(image_path, progress=None, desc_prefix="", num_gaussians=131072): t0 = time.time() print(f"[run_tripo] START image_path={image_path}", flush=True) def _p(frac, desc): if progress is not None: try: progress(frac, desc=desc_prefix + desc) except Exception: pass if not HAS_GRADIO_CLIENT: return None, "⚠ gradio_client isn't installed on this Space yet — add `gradio_client` to requirements.txt and redeploy." try: _p(0.1, "Connecting to TripoSplat's live demo…") client = get_tripo_client() _p(0.25, "Generating on TripoSplat (may take 20–90s, longer if their Space is asleep)…") # Verified against TripoSplat's own app.py source (they use a custom # gradio.Server + @app.api() endpoint, not a plain gr.Blocks demo — # their real signature is: # generate(image, seed=42, steps=20, guidance_scale=3.0, # num_gaussians=262144, output_format="ply") # -> (preprocessed_image, ply_file, download_file, info_string) # Calling with keyword args + their real api_name, not guessed # positional args against an assumed "/predict" name. result = client.predict( image=gradio_handle_file(image_path), seed=42, steps=20, guidance_scale=3.0, num_gaussians=int(num_gaussians), output_format="ply", api_name="/generate" ) print(f"[run_tripo] raw result (+{time.time()-t0:.1f}s): {str(result)[:500]}", flush=True) _p(0.9, "Finalizing…") ply_path = _extract_ply_path(result) if not ply_path or not os.path.exists(ply_path): return None, (f"⚠ TripoSplat responded but I couldn't find a .ply in the result. " f"Raw response logged to the HF Space logs for debugging.") _p(1.0, "Done") return ply_path, "✓ Done (via TripoSplat) — 3D scene loading below ↓" except Exception as e: print(f"[run_tripo] EXCEPTION: {e}", flush=True) return None, (f"⚠ TripoSplat call failed: {str(e)}\n" f"(This calls VAST-AI's public demo directly — it may be asleep, " f"rate-limited, or have changed its API since. Check the HF " f"Space logs for the exact error and the real API schema.)") # ──────────────────────────────────────────────────────────────── # Auto-clean: prune near-invisible / statistically-stray gaussians. # This is a pure-CPU numpy pass over the vertex data — runs outside the # GPU-decorated functions on purpose, so it never eats into ZeroGPU quota. # ──────────────────────────────────────────────────────────────── def clean_ply(ply_path, opacity_thresh=0.04, mad_k=6.0): if not HAS_PLYFILE: return ply_path, "skipped (plyfile not installed)" plydata = PlyData.read(ply_path) if "vertex" not in plydata: return ply_path, "skipped (no vertex element)" v = plydata["vertex"] n0 = len(v.data) if n0 == 0: return ply_path, "skipped (empty)" names = v.data.dtype.names if not all(k in names for k in ("x", "y", "z")): return ply_path, "skipped (unexpected format)" xyz = np.stack([v["x"], v["y"], v["z"]], axis=1).astype(np.float64) if "opacity" in names: op = np.asarray(v["opacity"], dtype=np.float64) # 3DGS often stores opacity as a raw pre-sigmoid logit rather than # a 0-1 probability — detect that and convert if so. if op.min() < -0.01 or op.max() > 1.01: op_prob = 1.0 / (1.0 + np.exp(-op)) else: op_prob = op keep_opacity = op_prob >= opacity_thresh else: keep_opacity = np.ones(n0, dtype=bool) # Robust outlier rejection via median absolute deviation (MAD) — more # resistant to the very outliers we're trying to remove than mean/std. med = np.median(xyz, axis=0) mad = np.median(np.abs(xyz - med), axis=0) + 1e-6 dist = np.abs(xyz - med) / mad keep_pos = np.all(dist <= mad_k, axis=1) keep = keep_opacity & keep_pos n1 = int(keep.sum()) if n1 == 0 or n1 == n0: return ply_path, f"{n0:,} gaussians (no change)" new_data = v.data[keep] new_el = PlyElement.describe(new_data, "vertex") out_path = ply_path[:-4] + "_clean.ply" if ply_path.endswith(".ply") else ply_path + "_clean.ply" PlyData([new_el], text=plydata.text, byte_order=plydata.byte_order).write(out_path) size_before = os.path.getsize(ply_path) / 1024 / 1024 size_after = os.path.getsize(out_path) / 1024 / 1024 pct = (1 - n1 / n0) * 100 return out_path, f"{n0:,} → {n1:,} gaussians ({pct:.0f}% pruned) · {size_before:.1f}MB → {size_after:.1f}MB" # ── Animation HTML (two angles, camera flies between them) ── def generate_animation_html(ply1, ply2): def enc(p): with open(p, "rb") as f: return base64.b64encode(f.read()).decode("utf-8") b1, b2 = enc(ply1), enc(ply2) return f"""
PHOTO → 3D GAUSSIAN SPLAT · MULTI-MODEL · FREE