Return merged point cloud for all frames in the scene. Binary format: int32(N) | float32[N*3](xyz) | uint8[N*3](rgb)
(
scene_id: str,
z_far: float = Query(10.0),
downsample: int = Query(4),
max_pts: int = Query(800000),
depth_mask: bool = Query(True),
conf_threshold: float = Query(0.0),
)
| 1631 | |
| 1632 | @app.get("/api/scene/{scene_id}/pcd") |
| 1633 | def get_scene_pcd( |
| 1634 | scene_id: str, |
| 1635 | z_far: float = Query(10.0), |
| 1636 | downsample: int = Query(4), |
| 1637 | max_pts: int = Query(800000), |
| 1638 | depth_mask: bool = Query(True), |
| 1639 | conf_threshold: float = Query(0.0), |
| 1640 | ): |
| 1641 | """Return merged point cloud for all frames in the scene. |
| 1642 | |
| 1643 | Binary format: int32(N) | float32[N*3](xyz) | uint8[N*3](rgb) |
| 1644 | """ |
| 1645 | scene = _find_scene(scene_id) |
| 1646 | if scene is None: |
| 1647 | return Response(status_code=404) |
| 1648 | |
| 1649 | data = _load_scene_data_raw(scene, z_far, use_depth_mask=depth_mask, conf_threshold=conf_threshold) |
| 1650 | |
| 1651 | all_pts = [] |
| 1652 | all_rgb = [] |
| 1653 | |
| 1654 | for i in range(len(data["depths"])): |
| 1655 | depth = data["depths"][i] |
| 1656 | pose = data["extrinsics"][i] |
| 1657 | img = data["images"][i] |
| 1658 | K = data["intrinsics"][i] if "intrinsics" in data else data["K"] |
| 1659 | |
| 1660 | pts, vs, us = _unproject_frame(depth, K, pose, downsample, max_pts) |
| 1661 | if len(pts) == 0: |
| 1662 | continue |
| 1663 | |
| 1664 | # Sample colors |
| 1665 | vs_c = np.clip(vs, 0, img.shape[0] - 1).astype(int) |
| 1666 | us_c = np.clip(us, 0, img.shape[1] - 1).astype(int) |
| 1667 | colors = img[vs_c, us_c] |
| 1668 | |
| 1669 | all_pts.append(pts) |
| 1670 | all_rgb.append(colors) |
| 1671 | |
| 1672 | if not all_pts: |
| 1673 | buf = io.BytesIO() |
| 1674 | buf.write(struct.pack("<i", 0)) |
| 1675 | return Response(content=buf.getvalue(), media_type="application/octet-stream") |
| 1676 | |
| 1677 | pts_all = np.concatenate(all_pts, axis=0) |
| 1678 | rgb_all = np.concatenate(all_rgb, axis=0) |
| 1679 | |
| 1680 | # Limit total points |
| 1681 | if len(pts_all) > max_pts: |
| 1682 | idx = np.random.choice(len(pts_all), max_pts, replace=False) |
| 1683 | pts_all = pts_all[idx] |
| 1684 | rgb_all = rgb_all[idx] |
| 1685 | |
| 1686 | buf = io.BytesIO() |
| 1687 | buf.write(struct.pack("<i", len(pts_all))) |
| 1688 | buf.write(pts_all.astype(np.float32).tobytes()) |
| 1689 | buf.write(rgb_all.astype(np.uint8).tobytes()) |
| 1690 | return Response(content=buf.getvalue(), media_type="application/octet-stream") |
nothing calls this directly
no test coverage detected