"""3DGS Robotics zero-install demo: photos or a short video -> browser 3DGS .splat. Runs on Hugging Face Spaces (ZeroGPU-aware) and locally: python app.py """ from __future__ import annotations import logging import os import tempfile import threading from pathlib import Path import gradio as gr import pipeline as pl logging.basicConfig(level=logging.INFO) logger = logging.getLogger("gs_mapper_space") GPU_DURATION = int(os.environ.get("GS_MAPPER_GPU_DURATION", "240")) DEFAULT_METHOD = os.environ.get("GS_MAPPER_DEMO_METHOD", "dust3r") MAX_FRAMES = 16 REPO_URL = "https://github.com/rsasaki0109/3dgs-robotics" VIEWER_URL = "https://rsasaki0109.github.io/3dgs-robotics/splat.html" try: # ZeroGPU decorator; falls back to identity off-Spaces. import spaces gpu = spaces.GPU(duration=GPU_DURATION) except Exception: # pragma: no cover - local runs def gpu(fn): return fn def _warmup() -> None: try: if DEFAULT_METHOD == "dust3r": pl.ensure_dust3r() pl.prefetch_checkpoint() except Exception as exc: # network failures retried lazily on first run logger.warning("warmup failed: %s", exc) @gpu def generate(photos, video, num_frames, iterations, progress=gr.Progress()): if not photos and not video: raise gr.Error("Upload at least 2 photos or one short video.") workdir = Path(tempfile.mkdtemp(prefix="gs_mapper_")) staged = workdir / "images" try: if video: count = pl.extract_video_frames(Path(video), staged, max_frames=int(num_frames)) logger.info("extracted %d frames from video", count) else: paths = [Path(f.name if hasattr(f, "name") else f) for f in photos] pl.prepare_images(paths, staged) result = pl.run_pipeline( staged, workdir / "out", method=DEFAULT_METHOD, num_frames=int(num_frames), iterations=int(iterations), progress=lambda fraction, message: progress(fraction, desc=message), ) except gr.Error: raise except Exception as exc: logger.exception("pipeline failed") raise gr.Error(f"Reconstruction failed: {exc}") from exc size_mb = result.splat_path.stat().st_size / 1e6 talk = " You can now **talk to this map** in the tab below." if result.session_dir else "" summary = ( f"{result.num_used_frames} frames -> {size_mb:.1f} MB .splat " f"in {result.elapsed_sec:.0f}s ({result.method}). " f"Tip: drag the downloaded .splat onto the [3DGS Robotics web viewer]({VIEWER_URL}) " f"or self-host it — see the [repo]({REPO_URL}).{talk}" ) splat = str(result.splat_path) session = str(result.session_dir) if result.session_dir else "" return splat, splat, summary, session @gpu def run_query(session_dir, prompt, threshold, progress=gr.Progress()): """Open-vocabulary query against the just-built map (CLIPSeg over keyframes).""" if not session_dir: raise gr.Error("Build a map first — then ask where something is.") if not (prompt or "").strip(): raise gr.Error('Type what to look for, e.g. "car" or "door".') progress(0.1, desc=f'Looking for "{prompt}"…') try: from gs_sim2real.robotics.language_query import QueryParams, query_map, write_query_preview params = QueryParams(score_threshold=float(threshold)) result, points = query_map(Path(session_dir), prompt.strip(), params=params, device="cuda") except ImportError as exc: raise gr.Error(f"Query needs the optional `transformers` package: {exc}") from exc except (ValueError, FileNotFoundError) as exc: raise gr.Error(str(exc)) from exc except Exception as exc: logger.exception("query failed") raise gr.Error(f"Query failed: {exc}") from exc preview = Path(tempfile.mkdtemp(prefix="gs_query_")) / "query.png" write_query_preview(result, points, preview) if not result.hits: return str(preview), f'No hits for "{prompt}" at threshold {threshold:.2f} — try lowering it.' lines = [f'**"{prompt}"** — {len(result.hits)} hit(s):', ""] for rank, hit in enumerate(result.hits[:5], start=1): gx, gy = hit.goal_xy lines.append(f"{rank}. {hit.gaussians} gaussians, score {hit.mean_score:.2f}, goal ≈ ({gx:.2f}, {gy:.2f})") return str(preview), "\n".join(lines) @gpu def run_inventory(session_dir, progress=gr.Progress()): """Open-vocabulary census of the just-built map.""" if not session_dir: raise gr.Error("Build a map first — then take inventory.") progress(0.1, desc="Taking inventory…") try: from gs_sim2real.robotics.inventory import ( DEFAULT_VOCAB, build_inventory, write_inventory_preview, ) report, points, basis = build_inventory(Path(session_dir), vocab=DEFAULT_VOCAB, device="cuda") except ImportError as exc: raise gr.Error(f"Inventory needs the optional `transformers` package: {exc}") from exc except (ValueError, FileNotFoundError) as exc: raise gr.Error(str(exc)) from exc except Exception as exc: logger.exception("inventory failed") raise gr.Error(f"Inventory failed: {exc}") from exc preview = Path(tempfile.mkdtemp(prefix="gs_inv_")) / "inventory.png" write_inventory_preview(report, points, basis, preview) rows = [ f"- **{cat['prompt']}** — {cat['clusters']} cluster(s), {cat['gaussians']} gaussians" for cat in report.get("categories", []) if cat.get("clusters") ] summary = "**Inventory** ({} clusters total):\n\n{}".format( report.get("total_clusters", 0), "\n".join(rows) or "_nothing matched the default vocabulary._", ) return str(preview), summary def build_ui() -> gr.Blocks: with gr.Blocks(title="3DGS Robotics — Photos to 3D Gaussian Splat") as demo: gr.Markdown( "# 3DGS Robotics — Photos to 3D Gaussian Splat\n" "Turn **a handful of photos or a short walkaround video** into a browser-ready " "3D Gaussian Splat. Pose-free (DUSt3R) — no COLMAP, no install.\n\n" f"[GitHub]({REPO_URL}) · [Live scene gallery]({VIEWER_URL}) · " "Best results: 8–16 photos orbiting one subject with ~70% overlap." ) with gr.Row(): with gr.Column(scale=1): with gr.Tab("Photos"): photos = gr.File( label="Photos (2–16 images, jpg/png)", file_count="multiple", file_types=["image"], ) with gr.Tab("Video"): video = gr.Video(label="Short walkaround video (frames are sampled evenly)") num_frames = gr.Slider(4, MAX_FRAMES, value=10, step=1, label="Frames to use") iterations = gr.Slider(500, 4000, value=2000, step=100, label="Training iterations") run_btn = gr.Button("Build my splat", variant="primary") with gr.Column(scale=2): viewer = gr.Model3D(label="Result (drag to orbit)", height=520) splat_file = gr.File(label="Download .splat") status = gr.Markdown() # Holds the queryable session assembled from the last reconstruction. session_state = gr.State("") gr.Markdown("---\n### 🗣️ Talk to your map") gr.Markdown( "Once a map is built, ask **where things are** (open-vocabulary query) or take a " "whole-map **inventory** — the same language tools the robotics stack and MCP server use. " "Coordinates are in the map's reconstruction gauge (camera-height units)." ) with gr.Tab("Find an object"): with gr.Row(): query_prompt = gr.Textbox(label="What to find", placeholder='e.g. "car", "door", "chair"', scale=3) query_threshold = gr.Slider(0.1, 0.9, value=0.4, step=0.05, label="Relevance threshold", scale=2) query_btn = gr.Button("Find it", variant="primary") with gr.Row(): query_preview = gr.Image(label="Top-down hits", height=360) query_text = gr.Markdown() with gr.Tab("Inventory"): inventory_btn = gr.Button("Take inventory", variant="primary") with gr.Row(): inventory_preview = gr.Image(label="What is where", height=360) inventory_text = gr.Markdown() run_btn.click( generate, inputs=[photos, video, num_frames, iterations], outputs=[viewer, splat_file, status, session_state], concurrency_limit=1, ) query_btn.click( run_query, inputs=[session_state, query_prompt, query_threshold], outputs=[query_preview, query_text], concurrency_limit=1, ) inventory_btn.click( run_inventory, inputs=[session_state], outputs=[inventory_preview, inventory_text], concurrency_limit=1, ) examples_dir = Path(__file__).parent / "examples" / "campus" if examples_dir.is_dir(): example_images = sorted(str(p) for p in examples_dir.glob("*.jpg")) if len(example_images) >= 2: gr.Examples(examples=[[example_images]], inputs=[photos], label="Example photo set") return demo threading.Thread(target=_warmup, daemon=True).start() demo = build_ui() if __name__ == "__main__": demo.launch()