# Project export: Strata

This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.

## Project metadata

- Hackathon: TreeHacks 2026
- Tagline: We turn real spaces into real-time, semantically labeled 3D datasets that give robots structured spatial intelligence for navigation, planning, and task execution.
- Devpost: https://devpost.com/software/strata-cbq07h
- GitHub: https://github.com/Vort3xed/TreeHacks
- Demo: https://stratadatasets.vercel.app/
- Video: https://www.youtube.com/embed/pQcdsmMBB9o?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Vort3xed (6 commits)

## Devpost submission (written by the team)

### Overview

GITHUB LINK: https://github.com/Vort3xed/TreeHacks

### Inspiration

As manual labor shifts toward robotic automation, we saw an opportunity to give robots live spatial intelligence so they can manage entire factories, warehouses, supply chains, and retail stores. Current solutions rely on LIDAR and raw camera footage, which fail to fully understand spatial geometry. They can’t create fully labeled 3D maps, leaving robots error-prone and unable to scale safely across large spaces. We wanted to create a solution that generates labeled 3D datasets from just minutes of video footage, marking objects, obstacles, and reference points in the environment. This structured labeling gives robots a complete understanding of the space, letting them navigate, plan, and act without trial-and-error. With Strata’s 3D datasets, companies can now build capable robotic hardware that can autonomously complete cognitively demanding tasks. We believe that the future lies in fully autonomous warehouses, factories, and retail stores.

### What it does

Strata is an intelligent spatial mapping platform that turns short video footage of any space into labeled 3D datasets, giving robots full spatial intelligence to navigate, plan, and perform tasks safely. Using 3D reconstruction, video clips are processed in real-time using Gaussian Splatting to generate high-fidelity, navigable 3D maps of any environment. SAM3 (pre-trained model) automatically labels objects and reference points in the 3D space, providing complete situational awareness for robots. The reconstructed and labeled environment is compiled into a digital twin that robots can immediately use to plan paths and perform complex tasks. Robots gain a complete understanding of the space before moving, minimizing errors, and accelerating deployment in warehouses, factories, and retail spaces Additional features Works with just a few minutes of video Supports warehouses, factories, processing plants, and stores of any size Provides predictive planning for logistics Analysis of footage (number of objects)

### How we built it

Our tech stack includes: VITE Webserver, Flask, WebSocket, SocketIO, Three.js, OpenCV, TypeScript, Pi3x, SAM3, Gaussian Splatting, SVD, CUDA, Amazon EC2, OpenAI, Gemini, Vercel. Our pipeline combines modern tools to turn simple video footage into high-fidelity, labeled 3D environments that robots can use immediately: Using 3D reconstruction, short video clips are converted into dense, high-resolution 3D maps, capturing geometry, surfaces, and obstacles (Gaussian Splatting, SVD, Pi3x). Objects, obstacles, and reference points are labeled directly in 3D, giving robots structured spatial intelligence without manual labeling (SAM3). The reconstructed and labeled environment is compiled into a digital twin, ready for robot path planning, navigation, and task execution (Three.js, TypeScript, VITE Webserver, Flask). As the camera streams, the map is constantly updated with low latency (WebSocket and SocketIO) All the above processes (reconstruction and labeling in real time) are computed via the cloud (Amazon EC2, CUDA) AI is able to process natural language input to search for certain objects in the Gaussian splatting, as well as produce a detailed analysis of the area surveyed by the splat (OpenAI, Gemini) Fast deployment of CDN (Vercel)

### Challenges we ran into

Setting up Amazon EC2 with GPUs and configuring all required libraries (CUDA, drivers, model dependencies) so Gaussian Splatting, Pi3x, and SAM3 could run reliably on the server Streaming low-latency video from the camera to the cloud while handling large amounts of data Making the Pi3x reconstruction pipeline fast enough for near real-time use by reducing memory usage and speeding up frame processing Keeping object labels consistent in 3D so SAM3 outputs are aligned correctly with the reconstructed environment Finding the right balance between map quality and processing speed so the system produces detailed 3D maps without long wait times

### Accomplishments we're proud of

Running the Pi3x model to generate the Gaussian Splatting in real-time is a much more difficult problem than expected. Since the map needs to be generated in real-time, we developed our own submapping algorithm that records the environment at specific intervals (every 4 seconds, every 5 seconds, etc). Every recorded interval is independently processed on the EC2 instance. The difficult task here is stitching the independent submaps together. To do this, we analyze the entropy of each submap and use an SVD algorithm to figure out which part of the submap should be appended to which part of another submap. We then run another algorithm to perform error correcting and ensure all submaps are appended together correctly. Running the Pi3x model to generate the Gaussian Splatting in real-time is a much more difficult problem than expected. Since the map needs to be generated in real-time, we developed our own submapping algorithm that records the environment at specific intervals (every 4 seconds, every 5 seconds, etc). Every recorded interval is independently processed on the EC2 instance. The difficult task here is stitching the independent submaps together. To do this, we analyze the entropy of each submap and use an SVD algorithm to figure out which part of the submap should be appended to which part of another submap. We then run another algorithm to perform error correcting and ensure all submaps are appended together correctly. Rendering pointclouds on the browser can be computationally very taxing, as some large-scale submaps can contain around 30 to 40 million points. To optimize for this, we use the numpy compression algorithm to reduce the computational load on the browser. Rendering pointclouds on the browser can be computationally very taxing, as some large-scale submaps can contain around 30 to 40 million points. To optimize for this, we use the numpy compression algorithm to reduce the computational load on the browser. While we could’ve used a YOLO model to identify objects in the scene and label them, we took this a step further and developed a SAM3 identification system. When a user queries to find “boxes” in the 3d map, CLIP embeds the camera stream and user query (CLIP embeddings map text and images to the same dimensional space) to reduce the search space of all frames from the camera to look through. SAM3 segments the object we’re looking for. After we’ve segmented the camera frames and identified the objects we’re looking for, we can map those pixels to the point cloud and (with very high accuracy) pinpoint the object in 3D space. While we could’ve used a YOLO model to identify objects in the scene and label them, we took this a step further and developed a SAM3 identification system. When a user queries to find “boxes” in the 3d map, CLIP embeds the camera stream and user query (CLIP embeddings map text and images to the same dimensional space) to reduce the search space of all frames from the camera to look through. SAM3 segments the object we’re looking for. After we’ve segmented the camera frames and identified the objects we’re looking for, we can map those pixels to the point cloud and (with very high accuracy) pinpoint the object in 3D space.

### What we learned

We became a lot more experienced with optimizing our algorithm for GPUs and running models on EC2. We also performed time complexity and amortized analysis on our algorithms to improve the point cloud generation speed.

### What's next

Future Features: Enable multiple robots to operate in the same digital twin simultaneously, optimizing paths, avoiding collisions, and sharing spatial intelligence in real-time Use LLMs to prompt the robots to autonomously perform their regular tasks, removing the need for constant human oversight after initial scanning Use AI to simulate and optimize complex tasks like heavy-lifting, inventory sorting, or assembly workflows before execution.

## README (from the GitHub repository)

# Treehacks 2026


## Detected evidence (automated analysis)

Indexed codebase: 54 recognized source files, 410 KB.
- CSS (language) — detected in the code
- FastAPI (technology) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — detected in the code
- Tailwind CSS (technology) — detected in the code
- TypeScript (language) — detected in the code
- Flask (technology) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (66 of 66)

```
.gitignore
demo_gradio.py
example_mm.py
example_vo.py
example.py
LICENSE
pi3/models/dinov2/__init__.py
pi3/models/dinov2/hub/__init__.py
pi3/models/dinov2/hub/backbones.py
pi3/models/dinov2/hub/utils.py
pi3/models/dinov2/layers/__init__.py
pi3/models/dinov2/layers/attention.py
pi3/models/dinov2/layers/block.py
pi3/models/dinov2/layers/dino_head.py
pi3/models/dinov2/layers/drop_path.py
pi3/models/dinov2/layers/layer_scale.py
pi3/models/dinov2/layers/mlp.py
pi3/models/dinov2/layers/patch_embed.py
pi3/models/dinov2/layers/swiglu_ffn.py
pi3/models/dinov2/models/__init__.py
pi3/models/dinov2/models/vision_transformer.py
pi3/models/dinov2/utils/__init__.py
pi3/models/dinov2/utils/cluster.py
pi3/models/dinov2/utils/config.py
pi3/models/dinov2/utils/dtype.py
pi3/models/dinov2/utils/param_groups.py
pi3/models/dinov2/utils/utils.py
pi3/models/layers/attention.py
pi3/models/layers/block.py
pi3/models/layers/camera_head.py
pi3/models/layers/conv_head.py
pi3/models/layers/pos_embed.py
pi3/models/layers/prope.py
pi3/models/layers/transformer_head.py
pi3/models/pi3.py
pi3/models/pi3x.py
pi3/pipe/pi3x_vo.py
pi3/utils/basic.py
pi3/utils/debug.py
pi3/utils/geometry.py
pyproject.toml
README.md
realtime/.env
realtime/object_labeler.py
realtime/pipeline.py
realtime/README.md
realtime/requirements.txt
realtime/run.sh
realtime/server.py
realtime/webserver/index.html
realtime/webserver/package.json
realtime/webserver/postcss.config.js
realtime/webserver/sender.html
realtime/webserver/server.cert
realtime/webserver/server.key
realtime/webserver/src/index.ts
realtime/webserver/src/main.css
realtime/webserver/src/sender.ts
realtime/webserver/src/viewer.ts
realtime/webserver/tailwind.config.js
realtime/webserver/tsconfig.json
realtime/webserver/tsconfig.node.json
realtime/webserver/viewer.html
realtime/webserver/vite.config.ts
requirements_demo.txt
requirements.txt
```

### Dependencies

- pyproject.toml: safetensors
- realtime/requirements.txt: fastapi, python-multipart, uvicorn[standard], websockets
- realtime/webserver/package.json: @types/node@^25.2.3, @types/three@^0.160.0, autoprefixer@^10.4.20, postcss@^8.4.49, socket.io-client@^4.5.4, tailwindcss@^3.4.17, three@^0.160.0, typescript@^5.3.3, vite@^5.0.11
- requirements.txt: huggingface_hub, numpy@==1.26.4, opencv-python, pillow, plyfile, safetensors, torch@==2.5.1, torchvision@==0.20.1

### Recent commits (newest first)

- updates
- update
- simple modifications + gemini
- optimizations, run on 10fps
- chat gpt wrapper
- object labeling
- rotation matrix and fix representation errrors
- Initial clean commit

## Key source files (fetched from GitHub, selected and truncated for size)

### requirements.txt

```
torch==2.5.1
torchvision==0.20.1
numpy==1.26.4
pillow
opencv-python
plyfile
huggingface_hub
safetensors
```

### pyproject.toml

```
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "pi3"
version = "0.1"
description = "pi3"
readme = "README.md"
requires-python = ">=3.10.0"
dependencies = [
    "safetensors",
]

# Setuptools configuration
[tool.setuptools]
# Disable automatic package discovery to avoid conflicts
packages = ["pi3"]

# Include package data
[tool.setuptools.package-data]
"pi3" = ["**/*"]

```

### realtime/requirements.txt

```
fastapi
uvicorn[standard]
python-multipart
websockets

```

### realtime/webserver/package.json

```
{
  "name": "pi3x-realtime",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "dev": "vite",
    "build": "tsc && vite build",
    "preview": "vite preview"
  },
  "dependencies": {
    "socket.io-client": "^4.5.4",
    "three": "^0.160.0"
  },
  "devDependencies": {
    "@types/node": "^25.2.3",
    "@types/three": "^0.160.0",
    "autoprefixer": "^10.4.20",
    "postcss": "^8.4.49",
    "tailwindcss": "^3.4.17",
    "typescript": "^5.3.3",
    "vite": "^5.0.11"
  }
}

```

### realtime/server.py

```python
"""
Pi3X Real-Time Streaming Server
FastAPI + WebSocket server for incremental 3D reconstruction.
"""

import asyncio
import base64
import json
import queue
import ssl
import struct
import threading
import time
import os
import sys
import argparse
import tempfile

import cv2
import numpy as np
import torch
import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel

from dotenv import load_dotenv
load_dotenv(os.path.join(os.path.dirname(__file__), '.env'))

from pipeline import IncrementalPi3
from object_labeler import ObjectLabeler, FrameData

# ─────────────────────────────────────────
# FastAPI App
# ─────────────────────────────────────────
app = FastAPI(title="Pi3X Real-Time Reconstruction")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

# ─────────────────────────────────────────
# Global State
# ─────────────────────────────────────────
pipeline: IncrementalPi3 = None
object_labeler: ObjectLabeler = None
frame_queue: queue.Queue = queue.Queue(maxsize=300)
connected_viewers: set[WebSocket] = set()
processing_lock = threading.Lock()
is_processing = False
video_feeder_stop = threading.Event()

# ─── Global counters ───
frames_received_total = 0
frames_dropped_total = 0
chunks_sent_total = 0        # chunks whose point‑cloud was broadcast to viewers
processing_alive = True      # False if the processing thread dies
_counter_lock = threading.Lock()

def frames_received_total_inc():
    global frames_received_total
    with _counter_lock:
        frames_received_total += 1

def frames_dropped_total_inc():
    global frames_dropped_total
    with _counter_lock:
        frames_dropped_total += 1


# ─────────────────────────────────────────
# WebSocket broadcast helper
# ─────────────────────────────────────────
async def broadcast_to_viewers(message: dict):
    """Send JSON message to all connected viewer WebSockets."""
    dead = set()
    for ws in connected_viewers:
        try:
            await ws.send_json(message)
        except Exception:
            dead.add(ws)
    connected_viewers.difference_update(dead)


async def broadcast_binary(data: bytes):
    """Send binary data to all connected viewer WebSockets."""
    dead = set()
    for ws in connected_viewers:
        try:
            await ws.send_bytes(data)
        except Exception:
            dead.add(ws)
    connected_viewers.difference_update(dead)


def encode_point_cloud_binary(points: np.ndarray, colors: np.ndarray, chunk_id: int, total: int, elapsed: float) -> bytes:
    """
    Encode point cloud as binary for efficient transfer.
    Format:
        Header: chunk_id(i32) + total_points(i32) + num_new(i32) + elapsed(f32) = 16 bytes
        Body:   N * (x f32 + y f32 + z f32 + r u8 + g u8 + b u8) = N * 15 bytes
    """
    n = len(points)
    header = struct.pack('<iiif', chunk_id, total, n, elapsed)

    if n == 0:
        return header

    pts = points.astype(np.float32)
    cols = colors.astype(np.uint8)

    # Interleave: for each point, 12 bytes xyz + 3 bytes rgb
    body = bytearray(n * 15)
    for i in range(n):
        offset = i * 15
        struct.pack_into('<fff', body, offset, pts[i, 0], pts[i, 1], pts[i, 2])
        body[offset + 12] = cols[i, 0]
        body[offset + 13] = cols[i, 1]
        body[offset + 14] = cols[i, 2]

    return header + bytes(body)


def encode_point_cloud_binary_fast(points: np.ndarray, colors: np.ndarray, chunk_id: int, total: int, elapsed: float) -> bytes:
    """Fast version: header + flat xyz floats + flat rgb bytes."""
    n = len(points)
    header = struct.pack('<iiif', chunk_id, total, n, elapsed)

    if n == 0:
        return header

    pts = np.ascontiguousarray(points.astype(np.float32))   # (N, 3)
    cols = np.ascontiguousarray(colors.astype(np.uint8))     # (N, 3)

    # Layout: header(16) + positions(N*12 f32) + colors(N*3 u8)
    return header + pts.tobytes() + cols.tobytes()


# ─────────────────────────────────────────
# Processing Loop (runs in background thread)
# ─────────────────────────────────────────
def _broadcast_chunk_result(result: dict, loop: asyncio.AbstractEventLoop):
    """Helper: encode & broadcast a single chunk result to all viewers."""
    if result is None or len(result['points']) == 0:
        if result is not None:
            skip_msg = {
                'type': 'chunk_skipped',
                'chunk_id': result['chunk_id'],
                'total_points': result['total_points'],
                'reason': 'No valid points produced',
            }
            asyncio.run_coroutine_threadsafe(broadcast_to_viewers(skip_msg), loop)
        return

    # Downsample for streaming if needed
    pts = result['points']
    cols = result['colors']
    if len(pts) > 100_000:
        s = len(pts) // 100_000
        pts = pts[::s]
        cols = cols[::s]

    binary_data = encode_point_cloud_binary_fast(
        pts, cols,
        result['chunk_id'],
        result['total_points'],
        result.get('elapsed', 0.0)
    )
    asyncio.run_coroutine_threadsafe(broadcast_binary(binary_data), loop)

    done_msg = {
        'type': 'chunk_done',
        'chunk_id': result['chunk_id'],
        'new_points': len(result['points']),
        'total_points': result['total_points'],
        'elapsed': result.get('elapsed', 0.0),
    }
    asyncio.run_coroutine_threadsafe(broadcast_to_viewers(done_msg), loop)


def processing_loop(loop: asyncio.AbstractEventLoop):
    """Background thread that pulls frames and runs Pi3X inference.

    In parallel mode (num_workers > 1), uses an *accumulation window*:
    after the first chunk becomes ready, keeps pulling frames from the
    queue for a short time so that multiple chunks can be batched together
    and processed on the GPU in parallel.
    """
   
[truncated — 27343 more characters]
```

### realtime/webserver/src/index.ts

```typescript
import './main.css';

```

### example_vo.py

```python
import torch
import argparse
import numpy as np
import os
from pi3.utils.basic import load_multimodal_data, write_ply
from pi3.models.pi3x import Pi3X
from pi3.pipe.pi3x_vo import Pi3XVO

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description="Run inference with the Pi3 model.")
    
    parser.add_argument("--data_path", type=str, default='examples/skating.mp4',
                        help="Path to the input image directory or a video file.")
    
    parser.add_argument("--save_path", type=str, default='examples/result.ply',
                        help="Path to save the output .ply file.")
    parser.add_argument("--interval", type=int, default=-1,
                        help="Interval to sample image. Default: 1 for images dir, 10 for video")
    parser.add_argument("--ckpt", type=str, default=None,
                        help="Path to the model checkpoint file. Default: None")
    parser.add_argument("--device", type=str, default='cuda',
                        help="Device to run inference on ('cuda' or 'cpu'). Default: 'cuda'")
                        
    args = parser.parse_args()
    if args.interval < 0:
        args.interval = 10 if args.data_path.endswith('.mp4') else 1
    print(f'Sampling interval: {args.interval}')

    # 1. Prepare model
    print(f"Loading model...")
    device = torch.device(args.device)
    if args.ckpt is not None:
        model = Pi3X().to(device).eval()
        if args.ckpt.endswith('.safetensors'):
            from safetensors.torch import load_file
            weight = load_file(args.ckpt)
        else:
            weight = torch.load(args.ckpt, map_location=device, weights_only=False)
        
        model.load_state_dict(weight, strict=False)
    else:
        model = Pi3X.from_pretrained("yyfz233/Pi3X").to(device).eval()
        # or download checkpoints from `https://huggingface.co/yyfz233/Pi3X/resolve/main/model.safetensors`, and `--ckpt ckpts/model.safetensors`

    pipe = Pi3XVO(model)

    # 2. Prepare input data
    # Load images (Required)
    imgs, _ = load_multimodal_data(args.data_path, conditions=None, interval=args.interval, device=device) 

    # 3. Infer
    print("Running model inference...")
    dtype = torch.bfloat16 if torch.cuda.get_device_capability()[0] >= 8 else torch.float16
    
    with torch.no_grad():
        res = pipe(
            imgs=imgs, 
            dtype=dtype,
        )

    # 4. process mask
    masks = res['conf'][0] > 0.05

    # 5. Save points
    print(f"Saving point cloud to: {args.save_path}")
    if os.path.dirname(args.save_path):
        os.makedirs(os.path.dirname(args.save_path), exist_ok=True)
        
    write_ply(res['points'][0][masks].cpu(), imgs[0].permute(0, 2, 3, 1)[masks], args.save_path)
    print("Done.")

```

### example.py

```python
import torch
import argparse
from pi3.utils.basic import load_images_as_tensor, write_ply
from pi3.utils.geometry import depth_edge
from pi3.models.pi3 import Pi3

if __name__ == '__main__':
    # --- Argument Parsing ---
    parser = argparse.ArgumentParser(description="Run inference with the Pi3 model.")
    
    parser.add_argument("--data_path", type=str, default='examples/skating.mp4',
                        help="Path to the input image directory or a video file.")
    parser.add_argument("--save_path", type=str, default='examples/result.ply',
                        help="Path to save the output .ply file.")
    parser.add_argument("--interval", type=int, default=-1,
                        help="Interval to sample image. Default: 1 for images dir, 10 for video")
    parser.add_argument("--ckpt", type=str, default=None,
                        help="Path to the model checkpoint file. Default: None")
    parser.add_argument("--device", type=str, default='cuda',
                        help="Device to run inference on ('cuda' or 'cpu'). Default: 'cuda'")
                        
    args = parser.parse_args()
    if args.interval < 0:
        args.interval = 10 if args.data_path.endswith('.mp4') else 1
    print(f'Sampling interval: {args.interval}')

    # 1. Prepare model
    print(f"Loading model...")
    device = torch.device(args.device)
    if args.ckpt is not None:
        model = Pi3().to(device).eval()
        if args.ckpt.endswith('.safetensors'):
            from safetensors.torch import load_file
            weight = load_file(args.ckpt)
        else:
            weight = torch.load(args.ckpt, map_location=device, weights_only=False)
        
        model.load_state_dict(weight)
    else:
        model = Pi3.from_pretrained("yyfz233/Pi3").to(device).eval()
        # or download checkpoints from `https://huggingface.co/yyfz233/Pi3/resolve/main/model.safetensors`, and `--ckpt ckpts/model.safetensors`

    # 2. Prepare input data
    # The load_images_as_tensor function will print the loading path
    imgs = load_images_as_tensor(args.data_path, interval=args.interval).to(device) # (N, 3, H, W)

    # 3. Infer
    print("Running model inference...")
    dtype = torch.bfloat16 if torch.cuda.get_device_capability()[0] >= 8 else torch.float16
    with torch.no_grad():
        with torch.amp.autocast('cuda', dtype=dtype):
            res = model(imgs[None]) # Add batch dimension

    # 4. process mask
    masks = torch.sigmoid(res['conf'][..., 0]) > 0.1
    non_edge = ~depth_edge(res['local_points'][..., 2], rtol=0.03)
    masks = torch.logical_and(masks, non_edge)[0]

    # 5. Save points
    print(f"Saving point cloud to: {args.save_path}")
    write_ply(res['points'][0][masks].cpu(), imgs.permute(0, 2, 3, 1)[masks], args.save_path)
    print("Done.")
```

### example_mm.py

```python
import torch
import argparse
import numpy as np
import os
from pi3.utils.basic import load_multimodal_data, write_ply
from pi3.utils.geometry import depth_edge
from pi3.models.pi3x import Pi3X

if __name__ == '__main__':
    # --- Argument Parsing ---
    parser = argparse.ArgumentParser(description="Run inference with the Pi3 model.")
    
    parser.add_argument("--data_path", type=str, default='examples/skating.mp4',
                        help="Path to the input image directory or a video file.")
    
    # parser.add_argument("--conditions_path", type=str, default='examples/room/condition.npz',
    parser.add_argument("--conditions_path", type=str, default=None,
                        help="Optional path to a .npz file containing 'poses', 'depths', 'intrinsics'.")

    parser.add_argument("--save_path", type=str, default='examples/result.ply',
                        help="Path to save the output .ply file.")
    parser.add_argument("--interval", type=int, default=-1,
                        help="Interval to sample image. Default: 1 for images dir, 10 for video")
    parser.add_argument("--ckpt", type=str, default=None,
                        help="Path to the model checkpoint file. Default: None")
    parser.add_argument("--device", type=str, default='cuda',
                        help="Device to run inference on ('cuda' or 'cpu'). Default: 'cuda'")
                        
    args = parser.parse_args()
    if args.interval < 0:
        args.interval = 10 if args.data_path.endswith('.mp4') else 1
    print(f'Sampling interval: {args.interval}')

    # 1. Prepare model
    print(f"Loading model...")
    device = torch.device(args.device)
    if args.ckpt is not None:
        model = Pi3X().to(device).eval()
        if args.ckpt.endswith('.safetensors'):
            from safetensors.torch import load_file
            weight = load_file(args.ckpt)
        else:
            weight = torch.load(args.ckpt, map_location=device, weights_only=False)
        
        model.load_state_dict(weight, strict=False)
    else:
        model = Pi3X.from_pretrained("yyfz233/Pi3X").to(device).eval()
        # or download checkpoints from `https://huggingface.co/yyfz233/Pi3X/resolve/main/model.safetensors`, and `--ckpt ckpts/model.safetensors`

    # 2. Prepare input data

    # Load optional conditions from .npz
    poses = None
    depths = None
    intrinsics = None

    if args.conditions_path is not None and os.path.exists(args.conditions_path):
        print(f"Loading conditions from {args.conditions_path}...")
        data_npz = np.load(args.conditions_path, allow_pickle=True)

        poses = data_npz['poses']             # Expected (N, 4, 4) OpenCV camera-to-world
        depths = data_npz['depths']           # Expected (N, H, W)
        intrinsics = data_npz['intrinsics']   # Expected (N, 3, 3)

    conditions = dict(
        intrinsics=intrinsics,
        poses=poses,
        depths=depths
    )

    # Load images (Required)
    imgs, conditions = load_multimodal_data(args.data_path, conditions, interval=args.interval, device=device) 

    """
    Args:
        imgs (torch.Tensor): Input RGB images valued in [0, 1].
            Shape: (B, N, 3, H, W).
        intrinsics (torch.Tensor, optional): Camera intrinsic matrices.
            Shape: (B, N, 3, 3).
            Values are in pixel coordinates (not normalized).
        rays (torch.Tensor, optional): Pre-computed ray directions (unit vectors).
            Shape: (B, N, H, W, 3).
            Can replace `intrinsics` as a geometric condition.
        poses (torch.Tensor, optional): Camera-to-World matrices.
            Shape: (B, N, 4, 4).
            Coordinate system: OpenCV convention (Right-Down-Forward).
        depths (torch.Tensor, optional): Ground truth or prior depth maps.
            Shape: (B, N, H, W).
            Invalid values (e.g., sky or missing data) should be set to 0.
        mask_add_depth (torch.Tensor, optional): Mask for depth condition.
            Shape: (B, N, N).
        mask_add_ray (torch.Tensor, optional): Mask for ray/intrinsic condition.
            Shape: (B, N, N).
        mask_add_pose (torch.Tensor, optional): Mask for pose condition.
            Shape: (B, N, N).
            Note: Requires at least two frames to be True to establish a meaningful
            coordinate system (absolute pose for a single frame provides no relative constraint).
    """

    # 3. Infer
    print("Running model inference...")
    dtype = torch.bfloat16 if torch.cuda.get_device_capability()[0] >= 8 else torch.float16
    
    with torch.no_grad():
        with torch.amp.autocast('cuda', dtype=dtype):
            res = model(
                imgs=imgs, 
                **conditions
            )

    # 4. process mask
    masks = torch.sigmoid(res['conf'][..., 0]) > 0.1
    non_edge = ~depth_edge(res['local_points'][..., 2], rtol=0.03)
    masks = torch.logical_and(masks, non_edge)[0]

    # 5. Save points
    print(f"Saving point cloud to: {args.save_path}")
    if os.path.dirname(args.save_path):
        os.makedirs(os.path.dirname(args.save_path), exist_ok=True)
        
    write_ply(res['points'][0][masks].cpu(), imgs[0].permute(0, 2, 3, 1)[masks], args.save_path)
    print("Done.")

```

### realtime/run.sh

```shell
#!/bin/bash
set -e

SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
cd "$SCRIPT_DIR"

# Activate the Pi3 venv
source ~/pi3venv/bin/activate

# Generate self-signed SSL certs if missing
if [ ! -f webserver/server.cert ] || [ ! -f webserver/server.key ]; then
    echo "Generating self-signed SSL certificates..."
    openssl req -x509 -newkey rsa:2048 -keyout webserver/server.key -out webserver/server.cert \
        -days 365 -nodes -subj "/CN=localhost"
fi

# Install Python deps if needed
pip install -q fastapi uvicorn[standard] python-multipart websockets 2>/dev/null

# Install frontend deps and build if needed
if [ ! -d webserver/node_modules ]; then
    echo "Installing frontend dependencies..."
    cd webserver
    npm install
    cd ..
fi

# Build frontend
echo "Building frontend..."
cd webserver
npx vite build 2>/dev/null || echo "Frontend build skipped (dev mode available)"
cd ..

echo ""
echo "Starting Pi3X Real-Time Server..."
echo ""

# Pass through all arguments  
python server.py "$@"

```

[46 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]