init

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# Make port 7860 available to the world outside the container
+EXPOSE 7860
+
+# Command to run when the container starts
+CMD ["python", "app.py"]

README copy.md

@@ -0,0 +1,34 @@
+---
+title: Viser Gradio Embed
+emoji: 🚀
+colorFrom: blue
+colorTo: pink
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# Viser + Gradio
+
+Demo for integrating [viser](https://github.com/nerfstudio-project/viser) 3D
+visualizations into a [Gradio](https://www.gradio.app/) application.
+
+- Uses Gradio's session management to create isolated 3D visualization contexts.
+- Exposes both Gradio and Viser over the same port.
+
+## Deploying on HuggingFace Spaces
+
+**[ [Live example](https://huggingface.co/spaces/brentyi/viser-gradio-embed) ]**
+
+This repository should work out-of-the-box with HF Spaces via Docker.
+
+- Unlike a vanilla Gradio Space, this is unfortunately not supported by [ZeroGPU](https://huggingface.co/docs/hub/en/spaces-zerogpu).
+
+## Local Demo
+
+```bash
+pip install -r requirements.txt
+python app.py
+```
+
+https://github.com/user-attachments/assets/b94a117a-b9e5-4854-805a-8666941c7816

app.py

@@ -0,0 +1,71 @@
+import random
+
+import fastapi
+import gradio as gr
+import uvicorn
+
+from viser_proxy_manager import ViserProxyManager
+
+
+def main() -> None:
+    app = fastapi.FastAPI()
+    viser_manager = ViserProxyManager(app)
+
+    # Create a Gradio interface with title, iframe, and buttons
+    with gr.Blocks(title="Viser Viewer") as demo:
+        # Add a title and description
+        gr.Markdown("# 🌐 Viser Interactive Viewer")
+
+        # Add the iframe with a border
+        add_sphere_btn = gr.Button("Add Random Sphere")
+        iframe_html = gr.HTML("")
+
+        @demo.load(outputs=[iframe_html])
+        def start_server(request: gr.Request):
+            assert request.session_hash is not None
+            viser_manager.start_server(request.session_hash)
+
+            # Use the request's base URL if available
+            host = request.headers["host"]
+
+            # Determine protocol (use HTTPS for HuggingFace Spaces or other secure environments)
+            protocol = (
+                "https"
+                if request.headers.get("x-forwarded-proto") == "https"
+                else "http"
+            )
+
+            return f"""
+            <div style="border: 2px solid #ccc; padding: 10px;">
+                <iframe src="{protocol}://{host}/viser/{request.session_hash}/" width="100%" height="500px" frameborder="0"></iframe>
+            </div>
+            """
+
+        @add_sphere_btn.click
+        def add_random_sphere(request: gr.Request):
+            assert request.session_hash is not None
+            server = viser_manager.get_server(request.session_hash)
+
+            # Add icosphere with random properties
+            server.scene.add_icosphere(
+                name=f"sphere_{random.randint(1, 10000)}",
+                position=(
+                    random.uniform(-1, 1),
+                    random.uniform(-1, 1),
+                    random.uniform(-1, 1),
+                ),
+                radius=random.uniform(0.05, 0.2),
+                color=(random.random(), random.random(), random.random()),
+            )
+
+        @demo.unload
+        def stop(request: gr.Request):
+            assert request.session_hash is not None
+            viser_manager.stop_server(request.session_hash)
+
+    gr.mount_gradio_app(app, demo, "/")
+    uvicorn.run(app, host="0.0.0.0", port=7860)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,6 @@
+viser>=0.2.23
+gradio==5.23.1
+fastapi==0.115.11
+uvicorn==0.34.0
+httpx==0.27.2
+websockets==15.0.1

viser_proxy_manager.py

@@ -0,0 +1,206 @@
+import asyncio
+
+import httpx
+import viser
+import websockets
+from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
+from fastapi.responses import Response
+
+
+class ViserProxyManager:
+    """Manages Viser server instances for Gradio applications.
+
+    This class handles the creation, retrieval, and cleanup of Viser server instances,
+    as well as proxying HTTP and WebSocket requests to the appropriate Viser server.
+
+    Args:
+        app: The FastAPI application to which the proxy routes will be added.
+        min_local_port: Minimum local port number to use for Viser servers. Defaults to 8000.
+            These ports are used only for internal communication and don't need to be publicly exposed.
+        max_local_port: Maximum local port number to use for Viser servers. Defaults to 9000.
+            These ports are used only for internal communication and don't need to be publicly exposed.
+    """
+
+    def __init__(
+        self,
+        app: FastAPI,
+        min_local_port: int = 8000,
+        max_local_port: int = 9000,
+    ) -> None:
+        self._min_port = min_local_port
+        self._max_port = max_local_port
+        self._server_from_session_hash: dict[str, viser.ViserServer] = {}
+        self._last_port = self._min_port - 1  # Track last port tried
+
+        @app.get("/viser/{server_id}/{proxy_path:path}")
+        async def proxy(request: Request, server_id: str, proxy_path: str):
+            """Proxy HTTP requests to the appropriate Viser server."""
+            # Get the local port for this server ID
+            server = self._server_from_session_hash.get(server_id)
+            if server is None:
+                return Response(content="Server not found", status_code=404)
+
+            # Build target URL
+            if proxy_path:
+                path_suffix = f"/{proxy_path}"
+            else:
+                path_suffix = "/"
+
+            target_url = f"http://127.0.0.1:{server.get_port()}{path_suffix}"
+            if request.url.query:
+                target_url += f"?{request.url.query}"
+
+            # Forward request
+            async with httpx.AsyncClient() as client:
+                # Forward the original headers, but remove any problematic ones
+                headers = dict(request.headers)
+                headers.pop("host", None)  # Remove host header to avoid conflicts
+                headers["accept-encoding"] = "identity"  # Disable compression
+
+                proxied_req = client.build_request(
+                    method=request.method,
+                    url=target_url,
+                    headers=headers,
+                    content=await request.body(),
+                )
+                proxied_resp = await client.send(proxied_req, stream=True)
+
+                # Get response headers
+                response_headers = dict(proxied_resp.headers)
+
+                # Check if this is an HTML response
+                content = await proxied_resp.aread()
+                return Response(
+                    content=content,
+                    status_code=proxied_resp.status_code,
+                    headers=response_headers,
+                )
+
+        # WebSocket Proxy
+        @app.websocket("/viser/{server_id}")
+        async def websocket_proxy(websocket: WebSocket, server_id: str):
+            """Proxy WebSocket connections to the appropriate Viser server."""
+            await websocket.accept()
+
+            server = self._server_from_session_hash.get(server_id)
+            if server is None:
+                await websocket.close(code=1008, reason="Not Found")
+                return
+
+            # Determine target WebSocket URL
+            target_ws_url = f"ws://127.0.0.1:{server.get_port()}"
+
+            if not target_ws_url:
+                await websocket.close(code=1008, reason="Not Found")
+                return
+
+            try:
+                # Connect to the target WebSocket
+                async with websockets.connect(target_ws_url) as ws_target:
+                    # Create tasks for bidirectional communication
+                    async def forward_to_target():
+                        """Forward messages from the client to the target WebSocket."""
+                        try:
+                            while True:
+                                data = await websocket.receive_bytes()
+                                await ws_target.send(data, text=False)
+                        except WebSocketDisconnect:
+                            try:
+                                await ws_target.close()
+                            except RuntimeError:
+                                pass
+
+                    async def forward_from_target():
+                        """Forward messages from the target WebSocket to the client."""
+                        try:
+                            while True:
+                                data = await ws_target.recv(decode=False)
+                                await websocket.send_bytes(data)
+                        except websockets.exceptions.ConnectionClosed:
+                            try:
+                                await websocket.close()
+                            except RuntimeError:
+                                pass
+
+                    # Run both forwarding tasks concurrently
+                    forward_task = asyncio.create_task(forward_to_target())
+                    backward_task = asyncio.create_task(forward_from_target())
+
+                    # Wait for either task to complete (which means a connection was closed)
+                    done, pending = await asyncio.wait(
+                        [forward_task, backward_task],
+                        return_when=asyncio.FIRST_COMPLETED,
+                    )
+
+                    # Cancel the remaining task
+                    for task in pending:
+                        task.cancel()
+
+            except Exception as e:
+                print(f"WebSocket proxy error: {e}")
+                await websocket.close(code=1011, reason=str(e))
+
+    def start_server(self, server_id: str) -> viser.ViserServer:
+        """Start a new Viser server and associate it with the given server ID.
+
+        Finds an available port within the configured min_local_port and max_local_port range.
+        These ports are used only for internal communication and don't need to be publicly exposed.
+
+        Args:
+            server_id: The unique identifier to associate with the new server.
+
+        Returns:
+            The newly created Viser server instance.
+
+        Raises:
+            RuntimeError: If no free ports are available in the configured range.
+        """
+        import socket
+
+        # Start searching from the last port + 1 (with wraparound)
+        port_range_size = self._max_port - self._min_port + 1
+        start_port = (
+            (self._last_port + 1 - self._min_port) % port_range_size
+        ) + self._min_port
+
+        # Try each port once
+        for offset in range(port_range_size):
+            port = (
+                (start_port - self._min_port + offset) % port_range_size
+            ) + self._min_port
+            try:
+                # Check if port is available by attempting to bind to it
+                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                    s.bind(("127.0.0.1", port))
+                    # Port is available, create server with this port
+                    server = viser.ViserServer(port=port)
+                    self._server_from_session_hash[server_id] = server
+                    self._last_port = port
+                    return server
+            except OSError:
+                # Port is in use, try the next one
+                continue
+
+        # If we get here, no ports were available
+        raise RuntimeError(
+            f"No available local ports in range {self._min_port}-{self._max_port}"
+        )
+
+    def get_server(self, server_id: str) -> viser.ViserServer:
+        """Retrieve a Viser server instance by its ID.
+
+        Args:
+            server_id: The unique identifier of the server to retrieve.
+
+        Returns:
+            The Viser server instance associated with the given ID.
+        """
+        return self._server_from_session_hash[server_id]
+
+    def stop_server(self, server_id: str) -> None:
+        """Stop a Viser server and remove it from the manager.
+
+        Args:
+            server_id: The unique identifier of the server to stop.
+        """
+        self._server_from_session_hash.pop(server_id).stop()