LingBot-Map 教程：GPU 感知推理与点云导出

In this tutorial, we implement an end-to-end streaming 3D reconstruction pipeline with LingBot-Map. We begin by configuring the input source, reconstruction settings, checkpoint selection, and output controls, then probe the available GPU and automatically tune frame limits, camera iterations, scale frames, and KV-cache parameters according to the detected VRAM. We install the repository and its dependencies, download the pretrained checkpoint, preprocess image or video frames, and construct the GCTStream model with streaming attention and long-range trajectory memory. We then perform mixed-precision inference, decode the predicted camera poses and intrinsic parameters, convert depth maps into world-coordinate point clouds, validate the recovered geometry, visualize the reconstructed scene and camera trajectory, and export the results as PLY, NPZ, or GLB artifacts. Copy CodeCopiedUse a different BrowserCFG = { "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" 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, check=True): p = subprocess.run(cmd, shell=True, capture_output=True, text=True) if p.returncode != 0 and check: print(p.stdout[-3000:]); print(p.stderr[-3000:]) raise RuntimeError(f"command failed: {cmd}") return p def probe_gpu(): try: q = sh("nvidia-smi --query-gpu=name,memory.total --format=csv,noheader", check=False) if q.returncode != 0 or not q.stdout.strip(): return None, 0.0 name, mem = [x.strip() for x in q.stdout.strip().split("\n")[0].split(",")] return name, float(mem.split()[0]) / 1024.0 except Exception: return None, 0.0 GPU_NAME, VRAM_GB = probe_gpu() print("=" * 78) print(f"GPU : {GPU_NAME or 'NONE — enable Runtime > Change runtime type > GPU'}") print(f"VRAM : {VRAM_GB:.1f} GB") try: import psutil print(f"System RAM : {psutil.virtual_memory().total/2**30:.1f} GB") except Exception: pass 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: raise SystemExit("This model needs 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 = "large (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'] if you have headroom — reconstruction quality " "improves a lot with more views.\n") We define the configuration parameters that control the input source, model behavior, inference mode, point-cloud generation, and export settings. We inspect the available GPU, system memory, and disk space before automatically adjusting frame limits, scale frames, camera iterations, and KV-cache size according to the detected VRAM. We also create reusable shell and GPU-probing functions that prepare the Colab environment for the remaining reconstruction workflow. Copy CodeCopiedUse a different Browserif 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 install -q einops safetensors huggingface_hub") sh(f"pip install -q -e {REPO} --no-deps", check=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 install the required dependencies without replacing Colab’s existing NumPy and OpenCV packages. We register the repository in the Python environment and verify the installed library versions to ensure that the runtime is ready for execution. We then download the selected pretrained LingBot-Map checkpoint from Hugging Face and store its local path for model initialization. Copy CodeCopiedUse a different Browserimport torch import matplotlib.pyplot as plt from lingbot_map.models.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.device("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, round(src_fps / fps)) paths, idx = [], 0 while True: ok, frame = cap.read() if not ok: break if idx % interval == 0: p = os.path.join(out_dir, f"{len(paths):06d}.jpg") cv2.imwrite(p, frame) paths.append(p) idx += 1 cap.release() print(f" decoded {idx} frames @ {src_fps:.1f} fps -> kept {len(paths)} (every {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}/example/{CFG['scene']}" if not os.path.isdir(folder): raise FileNotFoundError(folder) paths = sorted(sum([glob.glob(os.path.join(folder, f"*{e}")) for e in (".jpg", ".jpeg", ".png", ".JPG", ".PNG")], [])) if not paths: raise FileNotFoundError(f"no images in {folder}") return paths, folder print("\n[5] Loading frames") all_paths, SRC_FOLDER = gather_paths() print(f" source: {SRC_FOLDER} ({len(all_paths)} frames available)") 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} -> using {len(paths)} frames " f"(covering {(len(paths)-1)*stride+1}/{len(all_paths)} of the sequence)") t0 = time.time() images = load_and_preprocess_images( paths, mode="crop", image_size=CFG["image_size"], patch_size=CFG["patch_size"], ) S, _, H, W = images.shape n_tok = (H // CFG["patch_size"]) * (W // CFG["patch_size"]) print(f" tensor {tuple(images.shape)} in [0,1] | {W}x{H} | ~{n_tok} patch tokens/frame" f" | {time.time()-t0:.1f}s") k = min(6, S) fig, ax = plt.subplots(1, k, figsize=(3 * k, 2.4)) for i, a in enumerat

原文超出正文长度上限，此处截断——上游还有内容，完整版见上方「原文 ↗」。