这台机器人的点云链路不是“厂商发一个 ROS 话题，前端直接订阅”。厂商的 Fast DDS 数据面和我们通用的 ROS2 客户端不互通，所以项目在传感器侧自己接雷达，在后端把点云打包成浏览器可读的数据，再在前端用 Three.js 画出来。整条链路分成三段，每一段都藏着一个容易踩的坑。

传感器侧：绕开厂商 DDS 自建 rslidar_sdk

M20Pro 前后各有一颗 RSAIRY 雷达。厂商进程会把点云发到 /LIDAR/POINTS，但这个话题只在发现层可见，通用 ROS2 客户端读不到数据。项目没有继续和厂商 DDS 较劲，而是直接收雷达组播。

链路大致是这样：

前雷达组播 224.10.10.201，后雷达组播 224.10.10.202。
在 106 主机上编译标准 rslidar_sdk，用平台自己的 SHM DDS 环境运行。
输出到 /skill/pointcloud（前）和 /skill/pointcloud_back（后）。

这里有两个编译参数特别重要。ENABLE_IMU_DATA_PARSE=ON 是因为 RSAIRY 内置 IMU，不开这个参数，雷达可能一帧点云都不发。ENABLE_TRANSFORM=ON 则是为了把外参应用进点云本身。

如果不开 ENABLE_TRANSFORM，config 里的 x/y/z/roll/pitch/yaw 外参会被忽略，点云仍然在雷达原始帧里。RSAIRY 的原始坐标轴和机体坐标系不同，前端看到的角度就会完全错。开了之后，点云直接落在 base_link，前端不用再做一次坐标变换。

ROS 侧：控制面和数据面分开

sensor.pointcloud 是一个 streaming Skill，它不自己生产点云，只负责订阅 /skill/pointcloud，并回答“这个能力当前有没有在输出”。

点云订阅使用 SensorDataQoS，也就是 BEST_EFFORT。这是关键取舍：点云是高频大消息，一帧前向大约两万多个点。如果走 RELIABLE，订阅端处理不过来时，发送端和接收端都会被重传拖住，反而让新数据更难送达。BEST_EFFORT 允许丢旧帧，保证浏览器里看到的始终是尽量新的点云。

项目里点云的逻辑处理都发生在 106，104 侧只做可视化。前端注释说得很清楚：

1
2
3

NOTE: the pointcloud transform itself is done on 106
(rslidar_sdk ENABLE_TRANSFORM); this is pure 104-side visualization --
no cloud logic here.

所以浏览器里不是“又读点云又算坐标”，而是后端已经把点云转到了机体坐标系。

前端侧：Three.js 怎么画四万个点

前端通过 /api/pointcloud/data 拿点云。API 返回的是二进制 Float32 数组，每四个数一组：x, y, z, intensity。

前端的坐标系和 API 的打包顺序不同：

API 打包成 x=left, y=up, z=fwd。
前端展示使用 base_link，即 x=fwd, y=left, z=up。

所以写 BufferGeometry 时做了显式重映射：

1
2
3

pos[i * 3] = a[i * 4 + 2];       // fwd
pos[i * 3 + 1] = a[i * 4];       // left
pos[i * 3 + 2] = a[i * 4 + 1];   // up

颜色按高度做 jet 映射。它先扫一遍点云找出最小和最大高度，再把每个点的高度归一化后映射成颜色。这样即使场地高度跨度不同，颜色也能稳定地表达“高”和“低”。

渲染本身不复杂，关键是数据量友好的写法：

geo = new THREE.BufferGeometry();
geo.setAttribute('position', new THREE.BufferAttribute(new Float32Array(n * 3), 3));
geo.setAttribute('color', new THREE.BufferAttribute(new Float32Array(n * 3), 3));

pts = new THREE.Points(geo, new THREE.PointsMaterial({
  size: 0.06,
  vertexColors: true,
  sizeAttenuation: true
}));

点云不是 Mesh，而是 THREE.Points。vertexColors 让每个点使用独立颜色，sizeAttenuation 让点的大小随距离变化，远处看起来更密。

页面还叠加了 M20 的 URDF 模型，并根据 /api/robotstate 更新腿部关节。它的作用不是让画面好看，而是给人一个空间参照：狗就在自己的雷达点云中心，前后左右、地面高度都一目了然。

大文件预览：不要一次解析几百 MB

地图分图点云的预览器面对的文件可能很大，所以不能像实时点云那样一次拉完。

这个查看器选择 view 角色对应的降采样点云，而不是原始 archive。原始图可能达到 475MB，直接预览会同时拖死浏览器和存储服务。

它的加载流程是分阶段的：

先下载，边下载边更新百分比。
下载完成后，让浏览器先画一帧“解析中”。
再同步调用 PCDLoader.parse。
如果没有颜色，就按高度分块上色，每处理 40 万个点就让出一次主线程。
最后调用 renderer.compile 并渲染首帧，再关闭遮罩。

这样处理千万级点云时，进度条真的在动，不会出现“遮罩消失后黑屏半天”的观感。

关键代码示例

点云订阅：为什么是 BEST_EFFORT

sensor.pointcloud 本身不生产点云，只观察高频 PointCloud2 是否还在流。这里直接用 qos_profile_sensor_data，也就是 BEST_EFFORT：

class PointCloudSkill(SkillBase):
    """Streaming lidar point cloud. Cloud produced by rslidar_sdk on 106.
    This skill gates the request and observes the topic to report real flow.
    """

    def __init__(self, node, host_node: str,
                 topic: str = "/skill/pointcloud") -> None:
        self._active = False
        self._last_msg_t = 0.0
        # BEST_EFFORT (SensorDataQoS): high-rate large PointCloud2.
        # RELIABLE here fails to deliver to this long-running multi-endpoint
        # node over SHM; best-effort is the correct QoS for sensor streams.
        self._sub = node.create_subscription(
            PointCloud2, topic, self._on_cloud, qos_profile_sensor_data)

    def _on_cloud(self, _msg: PointCloud2) -> None:
        self._last_msg_t = time.monotonic()

    def status(self, handle=None) -> dict:
        producing = (time.monotonic() - self._last_msg_t) < 2.0
        return {"state": "RUNNING" if producing else "IDLE",
                "producing": producing}

如果为了“不丢帧”改成 RELIABLE，这台长期跑多端点的节点会更容易被重传拖住，反而一帧都读不到。点云是传感器流，语义上允许丢旧帧，保新帧更重要。

后端合并前后雷达并打包给浏览器

106 已经把前后两路点云转到机体系，后端再合并，并统一做一次 body→viewer 的轴交换：

_BODY_TO_VIEWER = np.array([[0, 1, 0], [0, 0, 1], [1, 0, 0]], dtype="float32")
_PC_VIEW_MAX = 20000

def pointcloud_packed(self, max_points: int = _PC_VIEW_MAX) -> bytes:
    """Merged front+back -> float32[x,y,z,w] bytes.
    Both arrive already in body frame (ENABLE_TRANSFORM), so we just
    concatenate and apply one clean body->viewer axis swap.
    max_points <= 0 = 不抽样,给原始密度。
    """
    with self._lock:
        mf, mb = self._pc_msg, self._pc_msg_back

    parts = []
    for msg in (mf, mb):
        if msg is None:
            continue
        xyz = self._xyz_from_msg(msg)
        if xyz is None or xyz.shape[0] == 0:
            continue
        parts.append(xyz)

    if not parts:
        return b""

    xyz = np.concatenate(parts, axis=0)
    if _BODY_TO_VIEWER is not None:
        xyz = xyz @ _BODY_TO_VIEWER.T

    if max_points > 0 and xyz.shape[0] > max_points:
        idx = np.linspace(0, xyz.shape[0] - 1, max_points).astype(np.int64)
        xyz = xyz[idx]

    buf = np.zeros((xyz.shape[0], 4), dtype=np.float32)
    buf[:, :3] = xyz.astype(np.float32)
    return buf.tobytes()