from __future__ import annotations import os import shutil import time import uuid from pathlib import Path os.environ.setdefault("GRADIO_SSR_MODE", "false") import spaces # noqa: F401 - must patch torch before model modules are imported from fastapi.responses import HTMLResponse from gradio import Server from gradio.data_classes import FileData from urllib.parse import quote, urlparse from src.demo.hf_runtime import ( InfiniSplatRuntime, ViewerTemplate, export_browser_viewer, export_filtered_gaussian_ply, export_standalone_viewer, prepare_viewer_template, ) OUTPUT_ROOT = Path(os.environ.get("GRADIO_TEMP_DIR", "/tmp/gradio")) / "infinisplat" OUTPUT_ROOT.mkdir(parents=True, exist_ok=True) GPU_DURATION_SECONDS = 6 runtime = InfiniSplatRuntime.load() viewer_template = prepare_viewer_template(OUTPUT_ROOT) app = Server( title="InfiniSplat", description="Implicit Gaussian decoding for large-baseline monocular view synthesis.", ) def _log(stage: str, **metrics) -> None: import json print(f"INFINISPLAT_TIMING {json.dumps({'stage': stage, **metrics}, sort_keys=True)}", flush=True) def _file_url(path: Path) -> str: """Return the public Gradio file URL for a server-side path.""" return f"/gradio_api/file={quote(str(path.resolve()))}" def _resolve_file_path(value: FileData | str) -> Path: """Resolve a FileData input (or URL/path string) back to a local file path.""" if isinstance(value, str): path_str = value elif isinstance(value, dict): path_str = value.get("path") or value.get("url") or "" else: path_str = str(value) if not path_str: raise ValueError(f"Cannot resolve file path from input: {value!r}") # Strip Gradio's /gradio_api/file= URL prefix to get the real path if path_str.startswith("/gradio_api/file="): decoded = urlparse(path_str).path[len("/gradio_api/file="):] return Path(decoded) # Already a server-local path if path_str.startswith("/"): return Path(path_str) return Path(path_str) @app.api(name="reconstruct", queue=True, concurrency_limit=1, concurrency_id="gpu") @spaces.GPU(duration=GPU_DURATION_SECONDS) def reconstruct(image_path: FileData) -> dict: """Run GPU reconstruction and return a public URL for the artifact.""" started = time.perf_counter() request_dir = OUTPUT_ROOT / uuid.uuid4().hex request_dir.mkdir(parents=True, exist_ok=True) artifact = runtime.infer_to_artifact( image_path=_resolve_file_path(image_path), artifact_path=request_dir / "gaussians.pt", ) _log( "gpu_reconstruct", request=request_dir.name, seconds=round(time.perf_counter() - started, 3), bytes=artifact.stat().st_size, ) return {"url": _file_url(artifact), "size": artifact.stat().st_size} @app.api(name="export_ply", queue=True, concurrency_limit=2) def export_ply(artifact_url: str) -> dict: """Filter one PLY artifact and return its public URL.""" started = time.perf_counter() internal = _resolve_file_path(artifact_url) scene_ply = export_filtered_gaussian_ply( artifact_path=internal, output_dir=internal.parent, ) internal.unlink(missing_ok=True) _log( "ply_export", request=scene_ply.parent.name, seconds=round(time.perf_counter() - started, 3), bytes=scene_ply.stat().st_size, ) return {"url": _file_url(scene_ply), "size": scene_ply.stat().st_size} @app.api(name="export_viewer", queue=True, concurrency_limit=2) def export_viewer(scene_ply_url: str) -> dict: """Build the browser viewer and return its iframe-ready HTML URL.""" started = time.perf_counter() scene_ply_path = _resolve_file_path(scene_ply_url) exported = export_browser_viewer( scene_ply=scene_ply_path, viewer_template=viewer_template, ) _log( "browser_viewer", request=exported.viewer_html.parent.name, seconds=round(time.perf_counter() - started, 3), sog_bytes=exported.scene_sog.stat().st_size, viewer_html_bytes=exported.viewer_html.stat().st_size, ) return {"url": _file_url(exported.viewer_html), "size": exported.viewer_html.stat().st_size} @app.api(name="export_html", queue=True, concurrency_limit=2) def export_html(viewer_html_url: str) -> dict: """Bundle a standalone HTML viewer and return its public URL.""" started = time.perf_counter() viewer_html_path = _resolve_file_path(viewer_html_url) standalone = export_standalone_viewer( viewer_html=viewer_html_path, viewer_template=viewer_template, ) _log( "standalone_html", request=standalone.parent.name, seconds=round(time.perf_counter() - started, 3), bytes=standalone.stat().st_size, ) return {"url": _file_url(standalone), "size": standalone.stat().st_size} @app.api(name="viewer_html", queue=False) def viewer_html() -> dict: """Serve the preloaded viewer template HTML for fast first paint.""" return {"url": _file_url(viewer_template.viewer_html)} INDEX_HTML = r""" InfiniSplat — Studio
InfiniSplatv1.0
SCENE untitled.gsplat
GitHub Project

Input

no source
FILE
—
SIZE
—
FORMAT
—

Viewport

idle
STAGE 0 / 4
—
No scene loaded Upload an image to begin reconstruction
""" @app.get("/", response_class=HTMLResponse) async def homepage() -> str: return INDEX_HTML @app.get("/__examples/{name}") async def serve_example(name: str): """Serve curated example images from the bundled examples directory.""" from fastapi.responses import FileResponse examples_dir = Path(__file__).resolve().parent / "examples" / "data" / "rgb_demo" candidate = examples_dir / name if not candidate.is_file() or not str(candidate.resolve()).startswith(str(examples_dir.resolve())): from fastapi import HTTPException raise HTTPException(status_code=404, detail="Example not found") return FileResponse(candidate) def _cleanup() -> None: """Remove expired per-request directories on shutdown.""" if not OUTPUT_ROOT.is_dir(): return cutoff = time.time() - 3600 for request_dir in OUTPUT_ROOT.iterdir(): if request_dir.is_symlink() or not request_dir.is_dir(): continue try: if uuid.UUID(hex=request_dir.name).hex != request_dir.name: continue except ValueError: continue if request_dir.lstat().st_mtime > cutoff: continue shutil.rmtree(request_dir, ignore_errors=True) import atexit as _atexit _atexit.register(_cleanup) if __name__ == "__main__": app.launch( server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860")), allowed_paths=[str(OUTPUT_ROOT)], max_file_size="20mb", show_error=True, ssr_mode=False, footer_links=[], )