LingBot-Map Tutorial: GPU-Aware Inference and Point Cloud Export
In this tutorial, we implement an end-to-end streaming 3D reconstruction pipeline with LingBot-Map. We start by configuring the enter supply, reconstruction settings, checkpoint choice, and output controls, then probe the obtainable GPU and routinely tune body limits, digicam iterations, scale frames, and KV-cache parameters in accordance with the detected VRAM. We set up the repository and its dependencies, obtain the pretrained checkpoint, preprocess picture or video frames, and assemble the GCTStream mannequin with streaming consideration and long-range trajectory reminiscence. We then carry out mixed-precision inference, decode the anticipated digicam poses and intrinsic parameters, convert depth maps into world-coordinate level clouds, validate the recovered geometry, visualize the reconstructed scene and digicam trajectory, and export the outcomes as PLY, NPZ, or GLB artifacts.
CFG = {
"scene": "courthouse",
"image_folder": None,
"video_path": None,
"fps": 10,
"max_frames": None,
"stride": None,
"checkpoint": "lingbot-map.pt",
"image_size": 518,
"patch_size": 14,
"use_sdpa": True,
"mode": "streaming",
"num_scale_frames": None,
"keyframe_interval": None,
"kv_cache_sliding_window": 64,
"camera_num_iterations": None,
"offload_to_cpu": True,
"window_size": 128,
"overlap_keyframes": 8,
"conf_percentile": 55.0,
"pixel_stride": 2,
"max_plot_points": 60000,
"export_ply": True,
"export_glb": False,
"launch_viser": False,
"run_ablation": False,
"seed": 0,
}
WORK = "/content material"
REPO = f"{WORK}/lingbot-map"
OUT = f"{WORK}/lingbot_out"
import os, sys, subprocess, glob, json, time, math, shutil, textwrap
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")
os.makedirs(OUT, exist_ok=True)
def sh(cmd, verify=True):
p = subprocess.run(cmd, shell=True, capture_output=True, textual content=True)
if p.returncode != 0 and verify:
print(p.stdout[-3000:]); print(p.stderr[-3000:])
elevate RuntimeError(f"command failed: {cmd}")
return p
def probe_gpu():
attempt:
q = sh("nvidia-smi --query-gpu=identify,reminiscence.complete --format=csv,noheader", verify=False)
if q.returncode != 0 or not q.stdout.strip():
return None, 0.0
identify, mem = [x.strip() for x in q.stdout.strip().split("n")[0].break up(",")]
return identify, float(mem.break up()[0]) / 1024.0
besides Exception:
return None, 0.0
GPU_NAME, VRAM_GB = probe_gpu()
print("=" * 78)
print(f"GPU : {GPU_NAME or 'NONE — allow Runtime > Change runtime kind > GPU'}")
print(f"VRAM : {VRAM_GB:.1f} GB")
attempt:
import psutil
print(f"System RAM : {psutil.virtual_memory().complete/2**30:.1f} GB")
besides Exception:
move
print(f"Free disk : {shutil.disk_usage(WORK).free/2**30:.1f} GB (checkpoint is 4.6 GB)")
print("=" * 78)
if GPU_NAME is None:
elevate SystemExit("This mannequin wants a CUDA GPU. Switch the Colab runtime to GPU.")
if VRAM_GB < 18:
AUTO = dict(max_frames=48, num_scale_frames=4, camera_num_iterations=2, kvsw=48)
tier = "small (<18 GB)"
elif VRAM_GB < 30:
AUTO = dict(max_frames=96, num_scale_frames=8, camera_num_iterations=4, kvsw=64)
tier = "medium (18-30 GB)"
else:
AUTO = dict(max_frames=240, num_scale_frames=8, camera_num_iterations=4, kvsw=64)
tier = "massive (30+ GB)"
if CFG["max_frames"] is None: CFG["max_frames"] = AUTO["max_frames"]
if CFG["num_scale_frames"] is None: CFG["num_scale_frames"] = AUTO["num_scale_frames"]
if CFG["camera_num_iterations"] is None: CFG["camera_num_iterations"] = AUTO["camera_num_iterations"]
if VRAM_GB < 18: CFG["kv_cache_sliding_window"] = AUTO["kvsw"]
print(f"Auto-tuned for {tier}: max_frames={CFG['max_frames']}, "
f"scale_frames={CFG['num_scale_frames']}, "
f"cam_iters={CFG['camera_num_iterations']}, "
f"kv_window={CFG['kv_cache_sliding_window']}")
print("Raise CFG['max_frames'] in case you have headroom — reconstruction high quality "
"improves loads with extra views.n")
We outline the configuration parameters that management the enter supply, mannequin conduct, inference mode, point-cloud technology, and export settings. We examine the obtainable GPU, system reminiscence, and disk area earlier than routinely adjusting body limits, scale frames, digicam iterations, and KV-cache measurement in accordance with the detected VRAM. We additionally create reusable shell and GPU-probing capabilities that put together the Colab setting for the remaining reconstruction workflow.
if not os.path.isdir(REPO):
print("Cloning repo (shallow) ...")
sh(f"git clone --depth 1 https://github.com/Robbyant/lingbot-map.git {REPO}")
print("Installing deps ...")
sh("pip set up -q einops safetensors huggingface_hub")
sh(f"pip set up -q -e {REPO} --no-deps", verify=False)
if REPO not in sys.path:
sys.path.insert(0, REPO)
import importlib
importlib.invalidate_caches()
import cv2, numpy as np
print(f"numpy {np.__version__} | opencv {cv2.__version__}")
from huggingface_hub import hf_hub_download
print(f"nFetching {CFG['checkpoint']} from robbyant/lingbot-map ...")
t0 = time.time()
CKPT = hf_hub_download(repo_id="robbyant/lingbot-map",
filename=CFG["checkpoint"],
cache_dir=f"{WORK}/hf_cache")
print(f" -> {CKPT} ({os.path.getsize(CKPT)/2**30:.2f} GB, {time.time()-t0:.0f}s)")
We clone the LingBot-Map repository and set up the required dependencies with out changing Colab’s present NumPy and OpenCV packages. We register the repository within the Python setting and confirm the put in library variations to make sure that the runtime is prepared for execution. We then obtain the chosen pretrained LingBot-Map checkpoint from Hugging Face and retailer its native path for mannequin initialization.
import torch
import matplotlib.pyplot as plt
from lingbot_map.fashions.gct_stream import GCTStream
from lingbot_map.utils.load_fn import load_and_preprocess_images
from lingbot_map.utils.pose_enc import pose_encoding_to_extri_intri
from lingbot_map.utils.geometry import (
closed_form_inverse_se3,
closed_form_inverse_se3_general,
depth_to_world_coords_points,
)
torch.manual_seed(CFG["seed"]); np.random.seed(CFG["seed"])
DEVICE = torch.system("cuda")
CC = torch.cuda.get_device_capability()
DTYPE = torch.bfloat16 if CC[0] >= 8 else torch.float16
print(f"torch {torch.__version__} | sm_{CC[0]}{CC[1]} | inference dtype {DTYPE}")
def extract_video_frames(video_path, out_dir, fps=10):
os.makedirs(out_dir, exist_ok=True)
cap = cv2.VideoCapture(video_path)
src_fps = cap.get(cv2.CAP_PROP_FPS) or 30
interval = max(1, spherical(src_fps / fps))
paths, idx = [], 0
whereas True:
okay, body = cap.learn()
if not okay:
break
if idx % interval == 0:
p = os.path.be a part of(out_dir, f"{len(paths):06d}.jpg")
cv2.imwrite(p, body)
paths.append(p)
idx += 1
cap.launch()
print(f" decoded {idx} frames @ {src_fps:.1f} fps -> stored {len(paths)} (each {interval})")
return paths
def gather_paths():
if CFG["video_path"]:
src = f"{OUT}/video_frames"
return extract_video_frames(CFG["video_path"], src, CFG["fps"]), src
folder = CFG["image_folder"] or f"{REPO}/instance/{CFG['scene']}"
if not os.path.isdir(folder):
elevate FileNotFoundError(folder)
paths = sorted(sum([glob.glob(os.path.join(folder, f"*{e}"))
for e in (".jpg", ".jpeg", ".png", ".JPG", ".PNG")], []))
if not paths:
elevate FileNotFoundError(f"no pictures in {folder}")
return paths, folder
print("n[5] Loading frames")
all_paths, SRC_FOLDER = gather_paths()
print(f" supply: {SRC_FOLDER} ({len(all_paths)} frames obtainable)")
stride = CFG["stride"]
if stride is None:
stride = max(1, math.ceil(len(all_paths) / CFG["max_frames"]))
paths = all_paths[::stride][:CFG["max_frames"]]
print(f" stride={stride} -> utilizing {len(paths)} frames "
f"(protecting {(len(paths)-1)*stride+1}/{len(all_paths)} of the sequence)")
t0 = time.time()
pictures = load_and_preprocess_images(
paths, mode="crop",
image_size=CFG["image_size"], patch_size=CFG["patch_size"],
)
S, _, H, W = pictures.form
n_tok = (H // CFG["patch_size"]) * (W // CFG["patch_size"])
print(f" tensor {tuple(pictures.form)} in [0,1] | {W}x{H} | ~{n_tok} patch tokens/body"
f" | {time.time()-t0:.1f}s")
ok = min(6, S)
fig, ax = plt.subplots(1, ok, figsize=(3 * ok, 2.4))
for i, a in enumerate(np.atleast_1d(ax)):
a.imshow(pictures[int(i * (S - 1) / max(k - 1, 1))].permute(1, 2, 0).numpy())
a.set_title(f"body {int(i*(S-1)/max(k-1,1))}", fontsize=9); a.axis("off")
plt.suptitle("Input frames (after crop/resize)"); plt.tight_layout(); plt.present()
We import PyTorch, LingBot-Map parts, geometry utilities, and visualization libraries earlier than deciding on the suitable inference precision for the obtainable GPU. We load frames from a bundled scene, customized picture listing, or video file and uniformly pattern them in accordance with the configured body restrict and stride. We preprocess the chosen pictures into model-ready tensors and preview consultant frames after cropping and resizing.
print("n[6] Building GCTStream")
mannequin = GCTStream(
img_size=CFG["image_size"],
patch_size=CFG["patch_size"],
enable_3d_rope=True,
max_frame_num=1024,
kv_cache_sliding_window=CFG["kv_cache_sliding_window"],
kv_cache_scale_frames=CFG["num_scale_frames"],
kv_cache_cross_frame_special=True,
kv_cache_include_scale_frames=True,
use_sdpa=CFG["use_sdpa"],
camera_num_iterations=CFG["camera_num_iterations"],
)
n_par = sum(p.numel() for p in mannequin.parameters())
print(f" {n_par/1e9:.2f} B params | heads: "
f"digicam={mannequin.camera_head is just not None}, depth={mannequin.depth_head is just not None}, "
f"level={mannequin.point_head is just not None}")
print(" loading state dict ...")
t0 = time.time()
attempt:
ckpt = torch.load(CKPT, map_location="cpu", weights_only=False, mmap=True)
besides Exception:
ckpt = torch.load(CKPT, map_location="cpu", weights_only=False)
sd = ckpt.get("mannequin", ckpt)
lacking, surprising = mannequin.load_state_dict(sd, strict=False)
del ckpt, sd
print(f" loaded in {time.time()-t0:.0f}s | lacking={len(lacking)} surprising={len(surprising)}")
if len(lacking) > 50:
print(" !! plenty of lacking keys — mistaken checkpoint for this mannequin class?")
mannequin = mannequin.to(DEVICE).eval()
if mannequin.aggregator is just not None:
mannequin.aggregator = mannequin.aggregator.to(dtype=DTYPE)
torch.cuda.empty_cache()
print(f" GPU after load: {torch.cuda.memory_allocated()/2**30:.2f} GB allotted, "
f"{torch.cuda.memory_reserved()/2**30:.2f} GB reserved")
kfi = CFG["keyframe_interval"]
if kfi is None:
kfi = math.ceil(S / 320) if (CFG["mode"] == "streaming" and S > 320) else 1
print(f"n[7] {CFG['mode']} inference | frames={S} | keyframe_interval={kfi} | dtype={DTYPE}")
torch.cuda.reset_peak_memory_stats()
out_dev = torch.system("cpu") if CFG["offload_to_cpu"] else None
t0 = time.time()
with torch.no_grad(), torch.amp.autocast("cuda", dtype=DTYPE):
if CFG["mode"] == "streaming":
preds = mannequin.inference_streaming(
pictures,
num_scale_frames=CFG["num_scale_frames"],
keyframe_interval=kfi,
output_device=out_dev,
)
else:
preds = mannequin.inference_windowed(
pictures,
window_size=CFG["window_size"],
overlap_size=16,
overlap_keyframes=CFG["overlap_keyframes"],
num_scale_frames=CFG["num_scale_frames"],
keyframe_interval=kfi,
output_device=out_dev,
)
elapsed = time.time() - t0
print(f" achieved in {elapsed:.1f}s -> {S/elapsed:.2f} FPS")
print(f" GPU peak: {torch.cuda.max_memory_allocated()/2**30:.2f} GB allotted "
f"/ {torch.cuda.max_memory_reserved()/2**30:.2f} GB reserved")
print(f" outputs: " + ", ".be a part of(f"{ok}{tuple(v.form)}" for ok, v in preds.gadgets()
if torch.is_tensor(v)))
attempt:
print(f" kv cache: {mannequin.get_kv_cache_info()}")
besides Exception:
move
mannequin.clean_kv_cache()
torch.cuda.empty_cache()
We assemble the GCTStream mannequin with streaming KV-cache controls, digicam refinement, 3D rotary embeddings, and scaled dot-product consideration. We load the pretrained checkpoint, transfer the mannequin to the GPU, forged the aggregation trunk to the chosen precision, and report its parameter depend and reminiscence utilization. We then run streaming or windowed inference, report processing velocity and peak GPU consumption, and gather the anticipated pictures, depths, confidence maps, and digicam pose encodings.
print("n[8] Decoding digicam poses")
def decode_poses(pose_enc, hw):
extr, intr = pose_encoding_to_extri_intri(pose_enc, hw)
E = torch.zeros((*extr.form[:-2], 4, 4), system=extr.system, dtype=extr.dtype)
E[..., :3, :4] = extr
E[..., 3, 3] = 1.0
E = closed_form_inverse_se3_general(E)
return E[..., :3, :4], intr
def unbatch(t):
return t[0] if (torch.is_tensor(t) and t.form[0] == 1) else t
extrinsic, intrinsic = decode_poses(preds["pose_enc"].float(), (H, W))
extrinsic = unbatch(extrinsic).cpu().numpy()
intrinsic = unbatch(intrinsic).cpu().numpy()
depth = unbatch(preds["depth"]).float().cpu().numpy()
depth_conf = unbatch(preds["depth_conf"]).float().cpu().numpy()
rgb = unbatch(preds["images"]).float().cpu().numpy()
c2w = closed_form_inverse_se3(extrinsic)
cam_centers = c2w[:, :3, 3]
path_len = float(np.linalg.norm(np.diff(cam_centers, axis=0), axis=1).sum())
print(f" focal (px) : {intrinsic[0,0,0]:.1f} principal pt "
f"({intrinsic[0,0,2]:.1f}, {intrinsic[0,1,2]:.1f})")
print(f" depth vary : {np.percentile(depth,1):.3f} .. {np.percentile(depth,99):.3f} "
f"(arbitrary however metric-consistent models)")
print(f" trajectory len : {path_len:.3f} | extent "
f"{np.ptp(cam_centers,axis=0).spherical(3).tolist()}")
print("n[9] Unprojecting depth to a world level cloud")
def conf_threshold(conf, pct):
sub = conf[::max(1, len(conf) // 20)].ravel()
return float(np.percentile(sub, pct))
THR = conf_threshold(depth_conf, CFG["conf_percentile"])
print(f" conf threshold @ {CFG['conf_percentile']}th pct = {THR:.3f}")
ps = CFG["pixel_stride"]
pts_all, col_all, frame_id = [], [], []
for i in vary(S):
wp, _, legitimate = depth_to_world_coords_points(
depth[i].squeeze(-1), extrinsic[i], intrinsic[i]
)
m = legitimate & (depth_conf[i] > THR)
m_sub = np.zeros_like(m); m_sub[::ps, ::ps] = True
m = m & m_sub
if not m.any():
proceed
pts_all.append(wp[m].astype(np.float32))
col_all.append(rgb[i].transpose(1, 2, 0)[m].astype(np.float32))
frame_id.append(np.full(int(m.sum()), i, dtype=np.int32))
factors = np.concatenate(pts_all, 0)
colours = np.clip(np.concatenate(col_all, 0), 0, 1)
frames_of = np.concatenate(frame_id, 0)
del pts_all, col_all, frame_id
print(f" {len(factors):,} factors stored "
f"({100*len(factors)/(S*H*W):.1f}% of all pixels, stride {ps}, conf>{THR:.2f})")
i = S // 2
wp, _, legitimate = depth_to_world_coords_points(depth[i].squeeze(-1), extrinsic[i], intrinsic[i])
m = legitimate & (depth_conf[i] > THR)
ratio = np.linalg.norm(wp[m] - cam_centers[i], axis=1) / np.most(depth[i].squeeze(-1)[m], 1e-6)
print(f" sanity: median ||p - cam|| / depth = {np.median(ratio):.3f} "
f"(anticipated ~1.00-1.15) -> {'OK' if 0.98 < np.median(ratio) < 1.35 else 'SUSPECT'}")
lo, hello = np.percentile(factors, [1, 99], axis=0)
scene_scale = float(np.linalg.norm(hello - lo))
print(f" scene diagonal : {scene_scale:.3f}")
We decode the anticipated pose encodings into digicam extrinsic and intrinsic matrices and calculate the reconstructed digicam trajectory. We convert each confidence-filtered depth map into world-coordinate factors whereas preserving the corresponding RGB values and source-frame identifiers. We additionally carry out a geometrical high quality verify and calculate sturdy scene boundaries to confirm that the reconstructed level cloud follows the anticipated digicam and depth conventions.
print("n[10] Plots")
ok = min(4, S)
idxs = np.linspace(0, S - 1, ok).astype(int)
fig, axes = plt.subplots(3, ok, figsize=(3.1 * ok, 7.2))
axes = np.atleast_2d(axes)
for c, i in enumerate(idxs):
axes[0, c].imshow(rgb[i].transpose(1, 2, 0)); axes[0, c].set_title(f"body {i}", fontsize=9)
d = depth[i].squeeze(-1)
axes[1, c].imshow(d, cmap="turbo", vmin=np.percentile(d, 2), vmax=np.percentile(d, 98))
axes[2, c].imshow(depth_conf[i] > THR, cmap="grey")
for r, lbl in enumerate(["RGB", "depth", f"conf > {THR:.2f}"]):
axes[r, 0].set_ylabel(lbl, fontsize=10)
for a in axes.ravel():
a.set_xticks([]); a.set_yticks([])
plt.tight_layout(); plt.savefig(f"{OUT}/depth_strip.png", dpi=110); plt.present()
plt.determine(figsize=(11, 3))
plt.subplot(1, 2, 1)
plt.hist(depth_conf[::max(1, S // 15)].ravel(), bins=120, colour="#4477aa")
plt.axvline(THR, colour="crimson", ls="--", label=f"threshold {THR:.2f}")
plt.yscale("log"); plt.xlabel("depth confidence"); plt.ylabel("pixels (log)")
plt.legend(); plt.title("Confidence distribution")
plt.subplot(1, 2, 2)
plt.plot([(depth_conf[i] > THR).imply() for i in vary(S)], lw=1.4, colour="#228833")
plt.xlabel("body"); plt.ylabel("fraction stored"); plt.ylim(0, 1)
plt.title("Per-frame confident-pixel ratio")
plt.tight_layout(); plt.present()
fig = plt.determine(figsize=(11, 4.2))
axA = fig.add_subplot(1, 2, 1, projection="3d")
axA.plot(*cam_centers.T, colour="#cc3311", lw=1.8)
axA.scatter(*cam_centers[0], s=45, c="inexperienced", label="begin")
axA.scatter(*cam_centers[-1], s=45, c="black", label="finish")
sub = factors[np.random.choice(len(points), min(4000, len(points)), replace=False)]
axA.scatter(*sub.T, s=0.4, c="#bbbbbb", alpha=0.35)
axA.set_title("Camera trajectory (3D)"); axA.legend(fontsize=8)
axB = fig.add_subplot(1, 2, 2)
axB.plot(cam_centers[:, 0], cam_centers[:, 2], colour="#cc3311", lw=1.8)
axB.scatter(sub[:, 0], sub[:, 2], s=0.4, c="#bbbbbb", alpha=0.35)
axB.set_xlabel("x"); axB.set_ylabel("z"); axB.set_aspect("equal")
axB.set_title("Top-down (x-z)")
plt.tight_layout(); plt.savefig(f"{OUT}/trajectory.png", dpi=110); plt.present()
import plotly.graph_objects as go
n_show = min(CFG["max_plot_points"], len(factors))
sel = np.random.selection(len(factors), n_show, substitute=False)
P, C = factors[sel], (colours[sel] * 255).astype(np.uint8)
inb = np.all((P >= lo) & (P <= hello), axis=1)
P, C = P[inb], C[inb]
fig = go.Figure([
go.Scatter3d(x=P[:, 0], y=P[:, 1], z=P[:, 2], mode="markers",
marker=dict(measurement=1.2, colour=[f"rgb({r},{g},{b})" for r, g, b in C]),
identify="factors", hoverinfo="skip"),
go.Scatter3d(x=cam_centers[:, 0], y=cam_centers[:, 1], z=cam_centers[:, 2],
mode="strains+markers", line=dict(colour="pink", width=4),
marker=dict(measurement=2, colour="pink"), identify="digicam path"),
])
fig.update_layout(peak=680, margin=dict(l=0, r=0, t=28, b=0),
title=f"LingBot-Map reconstruction — {len(P):,} of {len(factors):,} factors",
scene=dict(aspectmode="information",
xaxis=dict(seen=False), yaxis=dict(seen=False),
zaxis=dict(seen=False), bgcolor="rgb(15,15,20)"))
fig.present()
def write_ply(path, xyz, rgb01):
rgb8 = (np.clip(rgb01, 0, 1) * 255).astype(np.uint8)
hdr = (f"plynformat binary_little_endian 1.0nelement vertex {len(xyz)}n"
"property float xnproperty float ynproperty float zn"
"property uchar rednproperty uchar greennproperty uchar bluen"
"end_headern")
dt = np.dtype([("x", "<f4"), ("y", "<f4"), ("z", "<f4"),
("red", "u1"), ("green", "u1"), ("blue", "u1")])
arr = np.empty(len(xyz), dtype=dt)
arr["x"], arr["y"], arr["z"] = xyz[:, 0], xyz[:, 1], xyz[:, 2]
arr["red"], arr["green"], arr["blue"] = rgb8[:, 0], rgb8[:, 1], rgb8[:, 2]
with open(path, "wb") as f:
f.write(hdr.encode()); f.write(arr.tobytes())
if CFG["export_ply"]:
ply = f"{OUT}/{CFG['scene']}_lingbot.ply"
write_ply(ply, factors, colours)
print(f" PLY -> {ply} ({os.path.getsize(ply)/2**20:.1f} MB) "
"— open in MeshLab / CloudExamine / Blender")
np.savez_compressed(f"{OUT}/predictions.npz",
extrinsic=extrinsic, intrinsic=intrinsic,
cam_centers=cam_centers, depth=depth.astype(np.float16),
depth_conf=depth_conf.astype(np.float16))
print(f" NPZ -> {OUT}/predictions.npz (poses + depth, fp16)")
if CFG["export_glb"]:
sh("pip set up -q trimesh", verify=False)
from lingbot_map.vis import predictions_to_glb
wp_full = np.stack([depth_to_world_coords_points(
depth[i].squeeze(-1), extrinsic[i], intrinsic[i])[0] for i in vary(S)])
scene = predictions_to_glb(
{"world_points_from_depth": wp_full, "depth_conf": depth_conf,
"pictures": rgb, "extrinsic": extrinsic, "intrinsic": intrinsic},
conf_thres=CFG["conf_percentile"], prediction_mode="Predicted Depthmap")
scene.export(f"{OUT}/{CFG['scene']}.glb")
print(f" GLB -> {OUT}/{CFG['scene']}.glb")
attempt:
from google.colab import recordsdata
print(" (run `recordsdata.obtain(path)` in a brand new cell to drag a file down)")
besides Exception:
move
if CFG["launch_viser"]:
sh("pip set up -q 'viser>=0.2.23' trimesh", verify=False)
import threading
from lingbot_map.vis import PointCloudViewer
vis_pred = {"pictures": rgb, "depth": depth, "depth_conf": depth_conf,
"extrinsic": extrinsic, "intrinsic": intrinsic}
viewer = PointCloudViewer(pred_dict=vis_pred, port=8080,
vis_threshold=1.5, downsample_factor=10,
point_size=0.00001, use_point_map=False)
threading.Thread(goal=lambda: viewer.run(background_mode=True), daemon=True).begin()
time.sleep(3)
from google.colab.output import serve_kernel_port_as_window
serve_kernel_port_as_window(8080)
print(" viser opened in a brand new tab (permit pop-ups)")
if CFG["run_ablation"]:
print("n[11b] Ablation on the primary 24 frames")
sub_imgs = pictures[:24]
rows = []
for label, kw in [("cam_iters=4, kf=1", dict(keyframe_interval=1)),
("cam_iters=4, kf=2", dict(keyframe_interval=2)),
("cam_iters=4, kf=4", dict(keyframe_interval=4))]:
mannequin.clean_kv_cache(); torch.cuda.empty_cache()
torch.cuda.reset_peak_memory_stats()
t = time.time()
with torch.no_grad(), torch.amp.autocast("cuda", dtype=DTYPE):
p = mannequin.inference_streaming(sub_imgs,
num_scale_frames=CFG["num_scale_frames"],
output_device=torch.system("cpu"), **kw)
dt = time.time() - t
e, _ = decode_poses(p["pose_enc"].float(), (H, W))
cc = closed_form_inverse_se3(unbatch(e).cpu().numpy())[:, :3, 3]
rows.append((label, 24 / dt, torch.cuda.max_memory_allocated() / 2**30,
float(np.linalg.norm(np.diff(cc, axis=0), axis=1).sum())))
del p
print(f" {'setting':<20}{'FPS':>8}{'peak GB':>10}{'traj len':>11}")
for r in rows:
print(f" {r[0]:<20}{r[1]:>8.2f}{r[2]:>10.2f}{r[3]:>11.3f}")
print(" Higher keyframe_interval = much less KV reminiscence and extra velocity; the "
"trajectory size drifting away from the kf=1 row is your high quality value.")
print("n" + "=" * 78)
print(f"DONE. {S} frames -> {len(factors):,} factors at {S/elapsed:.2f} FPS. "
f"Artifacts in {OUT}/")
print("=" * 78)
print(textwrap.dedent("""
Where to go subsequent
----------------
* More frames is the only greatest high quality lever. Raise CFG['max_frames']
till you hit OOM, then again off.
* >320 frames: the KV cache exceeds the 320-view RoPE coaching vary. Set
keyframe_interval (auto-computed right here) moderately than rising the cache.
* >3000 frames: swap CFG['mode'] to 'windowed'. window_size counts KV
slots, not frames — with scale_frames=8 and keyframe_interval=ok, one
window covers 8 + (window_size - 8) * ok precise frames.
* Outdoor scenes: pip set up onnxruntime and use the repo's sky masking
(lingbot_map.vis.apply_sky_segmentation) to drop sky factors, which
in any other case smear into the far discipline.
* FlashInfer (use_sdpa=False) provides paged-KV consideration and roughly 20 FPS at
518x378 on a correct GPU, however JIT-compiles kernels on first name.
* Pose collapse on lengthy runs = state drift. Shorten the run, elevate
keyframe_interval, or transfer to windowed mode.
"""))
We visualize the RGB frames, depth maps, confidence masks, confidence distribution, digicam trajectory, and reconstructed level cloud. We export the reconstruction as PLY and NPZ recordsdata, whereas optionally producing a GLB scene or launching an interactive Viser viewer inside Colab. We additionally help an non-compulsory ablation experiment that compares keyframe configurations by way of velocity, GPU reminiscence utilization, and reconstructed trajectory conduct.
In conclusion, we established a whole and configurable workflow for changing an ordered picture or video sequence right into a spatially constant 3D reconstruction. We used LingBot-Map’s streaming structure to course of frames effectively whereas controlling KV-cache development, GPU reminiscence consumption, keyframe density, and camera-refinement value. We reworked the mannequin’s pose and depth predictions into digicam extrinsics, intrinsics, trajectory coordinates, and confidence-filtered world factors, and we verified the geometry earlier than producing visualizations and reusable reconstruction recordsdata. We additionally retained help for windowed inference, interactive Viser inspection, different export codecs, and ablation experiments, which permits us to increase the pocket book from a Colab demonstration right into a sensible setting for evaluating reconstruction high quality, runtime efficiency, and reminiscence trade-offs throughout completely different GPUs and scene lengths.
Check out the Full Code here. Also, be at liberty to observe us on Twitter and don’t overlook to hitch our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to associate with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and many others.? Connect with us
The put up LingBot-Map Tutorial: GPU-Aware Inference and Point Cloud Export appeared first on MarkTechPost.
