前言

你做一个无人机避障或者机器人SLAM，需要处理点云数据。原始点云一来就是几十万甚至上百万个点，用CPU处理，一帧就要200ms，30FPS根本不可能。

cann-recipes-spatial-intelligence 是 CANN 面向空间智能（3D感知、SLAM、点云分割）的配方库。这篇文章手把手带你跑通点云分割推理的完整流程。

空间智能的推理需求

先理清楚空间智能要处理什么：

任务	输入	输出	延迟要求
3D 目标检测	点云	3D Bounding Box	< 100ms
点云分割	点云	点级类别标签	< 50ms
SLAM	连续帧点云	位姿估计	< 30ms
深度估计	RGB/D	Depth Map	< 50ms

共同特点：

• 数据量大（上百万点的点云）
• 延迟敏感（实时避障）
• 计算密集（3D 卷积、Transformer）

cann-recipes-spatial-intelligence 配方内容

# 仓库结构
cann-recipes-spatial-intelligence/
├── recipes/                           # 核心配方
│   ├── point_cloud_processing/           # 点云预处理
│   │   ├── voxelization.py           # 体素化
│   │   ├── downsampling.py           # 下采样
│   │   └── radius_outlier_removal.py # 离群点去除
│   ├── segmentation/                 # 点云分割
│   │   ├── pointnet2_seg.py          # PointNet++ 分割
│   │   ├── randla_net.py            # RandLA-Net
│   │   └── spconv.py                # Sparse Conv
│   ├── detection_3d/                 # 3D 检测
│   │   ├── pointpillars.py           # PointPillars
│   │   ├── centerpoint.py            # CenterPoint
│   │   └── vot.py                    # VOT
│   ├── slam/                         # SLAM
│   │   ├── icp.py                   # ICP
│   │   ├── ndt.py                   # NDT
│   │   └──_loam.py                  # LOAM
│   └── depth_estimation/              # 深度估计
│       ├── monodepth.py
│       └── struct2strereo.py
├── models/                            # 预训练模型
│   ├── randlanet_nuscene.om
│   ├── pointpillars_nuscene.om
│   └── spconv_kitti.om
├── scripts/                           # 示例脚本
│   ├── run_pointcloud_segmentation.sh
│   ├── run_3d_detection.sh
│   └── run_slam.sh
└── docs/
    ├── installation.md
    └── api_reference.md

点云分割推理流程

Step 1：点云数据接入

点云数据通常来自激光雷达（LiDAR），输出格式有几种：

格式	说明	示例设备
XYZ	(x, y, z) 坐标	Velodyne HDL-64
XYZI	+ 强度	Ouster
XYZRGB	+ RGB	Realsense D455
XYZIRT	+ 反射率 + 时间	量产 LiDAR

# step1_pointcloud_reader.py
import numpy as np
import struct

def read_velodyne_bin(filepath):
    """读取 Velodyne 二进制点云文件 (.bin)
    
    格式：每行 4 个 float (x, y, z, intensity)
    """
    points = np.fromfile(filepath, dtype=np.float32)
    points = points.reshape(-1, 4)
    
    xyz = points[:, :3]  # (N, 3)
    intensity = points[:, 3]  # (N,)
    
    return xyz, intensity


def read_pcd_file(filepath):
    """读取 PCD 文件"""
    with open(filepath, 'r') as f:
        lines = f.readlines()
    
    # 解析 header
    header_end = 0
    for i, line in enumerate(lines):
        if line.startswith('DATA'):
            header_end = i + 1
            break
    
    # 解析点数据
    points = []
    for line in lines[header_end:]:
        values = line.strip().split()
        if len(values) >= 3:
            points.append([float(v) for v in values[:3]])
    
    return np.array(points).astype(np.float32)


def read_ros_pointcloud(msg):
    """读取 ROS PointCloud2 消息"""
    # ROS 中点云的格式转 numpy
    # field: x, y, z
    
    cloud_data = np.frombuffer(msg.data, dtype=np.float32)
    fields = msg.fields
    
    # 提取字段
    points = []
    for field_name in ['x', 'y', 'z']:
        idx = next(i for i, f in enumerate(fields) if f.name == field_name)
        points.append(cloud_data[idx::len(fields)])
    
    return np.stack(points, axis=-1)


# 主函数
def load_pointcloud(source, format='auto'):
    """
    加载点云数据
    
    Args:
        source: 文件路径或 ROS topic
        format: 'bin', 'pcd', 'ros', 'auto'
    """
    if format == 'bin' or (format == 'auto' and source.endswith('.bin')):
        return read_velodyde_bin(source)
    elif format == 'pcd' or (format == 'auto' and source.endswith('.pcd')):
        return read_pcd_file(source)
    elif format == 'ros':
        return read_ros_pointcloud(source)
    else:
        raise ValueError(f"Unknown format: {format}")


# 使用
xyz, intensity = load_pointcloud("pointcloud.bin")
print(f"Loaded {xyz.shape[0]} points")
# 输出：Loaded 150000 points

Step 2：点云预处理（DVPP 不能直接处理，需要转换）

昇腾 NPU 不能直接处理原始点云，需要先做预处理：

1. 降采样：减少点数（150k → 10k）
2. 体素化：转成体素 grid
3. 坐标变换：转成 NPU 能处理的格式

# step2_preprocessing.py
import numpy as np
import torch

def pointcloud_preprocess(xyz, target_points=10000, use_ farthest_point=True):
    """
    点云预处理
    
    Args:
        xyz: 原始点云 (N, 3)
        target_points: 目标点数
        use_farthest_point: 用 FPS 降采样（更均匀）vs 随机采样
    """
    n = xyz.shape[0]
    
    if n <= target_points:
        # 点数够，直接返回
        return xyz, np.arange(n)
    
    # FPS（Farthest Point Sampling）降采样
    if use_farthest_point:
        indices = farthest_point_sampling(xyz, target_points)
    else:
        # 随机采样
        indices = np.random.choice(n, target_points, replace=False)
    
    return xyz[indices], indices


def farthest_point_sampling(xyz, n_samples):
    """
    Farthest Point Sampling（最远点采样）
    使点分布更均匀
    """
    n_points = xyz.shape[0]
    sampled_indices = []
    
    # 随机选第一个点
    first_idx = np.random.randint(n_points)
    sampled_indices.append(first_idx)
    
    # 每次选离已有点最远的那个
    for _ in range(n_samples - 1):
        current_points = xyz[sampled_indices]
        distances = np.min(
            np.linalg.norm(xyz[:, None] - current_points[None], axis=2),
            axis=1
        )
        next_idx = np.argmax(distances)
        sampled_indices.append(next_idx)
    
    return np.array(sampled_indices)


def voxelize(xyz, voxel_size=0.1):
    """
    体素化（用于体素卷积）
    
    Args:
        xyz: 点云 (N, 3)
        voxel_size: 体素大小 (米)
    
    Returns:
        coords: 体素坐标 (M, 3)
        num_points_per_voxel: 每个体素的点数 (M,)
    """
    # 计算体素坐标
    coords = np.floor(xyz / voxel_size).astype(np.int32)
    
    # 去重，得到唯一的体素
    coords_unique, inverse = np.unique(coords, axis=0, return_inverse=True)
    
    # 统计每个体素里的点��
    num_points_per_voxel = np.bincount(inverse)
    
    return coords_unique, num_points_per_voxel


# 主函数
def preprocess_pipeline(xyz, target_points=10000):
    """预处理流水线"""
    # 1. 降采样
    xyz_down, indices = pointcloud_preprocess(xyz, target_points)
    
    # 2. 归一化（转成 -1~1）
    xyz_normalized = xyz_down / 50.0  # 假设场景范围 50m
    
    # 3. 转 tensor
    xyz_tensor = torch.from_numpy(xyz_normalized).float()
    
    return xyz_tensor, indices


# 测试
xyz = np.random.randn(150000, 3) * 30
xyz_tensor, indices = preprocess_pipeline(xyz, target_points=10000)
print(f"Preprocessed: {xyz_tensor.shape}")
# 输出：Preprocessed: torch.Size([10000, 3])

Step 3：模型推理（调用 OM）

# step3_inference.py
import torch
import torch_npu as npu
import atb

def create_pointcloud_model(om_path, device="npu:0"):
    """创建点云分割模型"""
    model = atb.create_inference_model(
        model_path=om_path,
        device=device
    )
    return model


def infer_pointcloud_segmentation(model, xyz):
    """
    点云分割推理
    
    Args:
        model: ATB 模型
        xyz: 点云 (N, 3) 或 (B, N, 3)
    """
    # 1. 预处理后的 batch shape: (B=1, N, 3)
    if xyz.dim() == 2:
        xyz = xyz.unsqueeze(0)
    elif xyz.dim() == 3:
        xyz = xyz.unsqueeze(0) if xyz.shape[0] != 1 else xyz
    
    # 2. 转 NPU tensor
    xyz_npu = xyz.npu()
    
    # 3. 推理
    with torch.no_grad():
        output = model(xyz_npu)
    
    # 4. 解析结果 (semantic labels)
    # output shape: (1, N, num_classes)
    pred_labels = output.argmax(dim=-1)  # (1, N)
    
    return pred_labels.squeeze(0).cpu().numpy()


def apply_color_by_label(points, labels):
    """根据标签着色（用于可视化）"""
    colors = {
        0: [128, 128, 128],  # road - gray
        1: [0, 255, 0],      # vegetation - green
        2: [255, 0, 0],      # car - red
        3: [0, 0, 255],      # pedestrian - blue
        4: [255, 255, 0],    # cyclist - yellow
    }
    
    rgb = np.zeros((points.shape[0], 3))
    for label_id, color in colors.items():
        mask = labels == label_id
        rgb[mask] = color
    
    return rgb


# 使用
model = create_pointcloud_model("randlanet_nuscene.om")

# 模拟一帧点云
xyz = torch.randn(10000, 3) * 30

# 推理
labels = infer_pointcloud_segmentation(model, xyz)
rgb = apply_color_by_label(xyz.numpy(), labels)

print(f"Labels distribution: {np.bincount(labels)}")
# Labels distribution: [5234 2134 1856 432 244]

Step 4：实时性优化

空间智能最大的挑战是 实时性。30 FPS = 每帧 33ms。

策略1：跳过帧

# 策略1：不是每一帧都推理，跳过一些
class FrameSkipProcessor:
    def __init__(self, target_fps=30):
        self.target_fps = target_fps
        self.frame_interval = 1.0 / target_fps
        self.last_process = 0
    
    def should_process(self, current_time):
        if current_time - self.last_process >= self.frame_interval:
            self.last_process = current_time
            return True
        return False

策略2：异步推理

# 策略2：异步推理，Pipeline 并行
import threading

class AsyncInference:
    def __init__(self, model):
        self.model = model
        self.input_queue = queue.Queue(maxsize=2)
        self.output_queue = queue.Queue(maxsize=2)
        self.running = True
        
        self.worker = threading.Thread(target=self._inference_loop)
        self.worker.start()
    
    def push(self, xyz):
        try:
            self.input_queue.put_nowait(xyz)
        except queue.Full:
            pass  # 队��满��跳过
    
    def pop(self):
        try:
            return self.output_queue.get_nowait()
        except queue.Empty:
            return None
    
    def _inference_loop(self):
        while self.running:
            xyz = self.input_queue.get()
            labels = infer_pointcloud_segmentation(self.model, xyz)
            self.output_queue.put(labels)
    
    def stop(self):
        self.running = False
        self.worker.join()

策略3：Batch 累积

# 策略3：短时间内的多帧一起推理
class BatchAccumulator:
    def __init__(self, batch_size=4, timeout_ms=10):
        self.batch_size = batch_size
        self.timeout_ms = timeout_ms / 1000
        self.buffer = []
        self.last_flush = time.time()
    
    def accumulate(self, xyz):
        now = time.time()
        
        # 超时 flush
        if now - self.last_flush > self.timeout_ms:
            if self.buffer:
                self._flush()
                self.last_flush = now
        
        # 达到 batch flush
        if len(self.buffer) >= self.batch_size:
            self._flush()
            self.last_flush = now
        else:
            self.buffer.append(xyz)
    
    def _flush(self):
        # Batch 推理
        batch = torch.stack(self.buffer)
        # ...
        self.buffer.clear()

完整示例

# full_pipeline.py
import time
import numpy as np

class SpatialIntelligenceDemo:
    """空间智能完整流水线"""
    
    def __init__(self):
        # 加载模型
        self.model = create_pointcloud_model("randlanet_nuscene.om")
        
        # 处理器
        self.skip_processor = FrameSkipProcessor(target_fps=30)
        self.async_inf = AsyncInference(self.model)
        
        # 性能统计
        self.latencies = []
    
    def process_frame(self, xyz, timestamp):
        # 1. 检查是否需要处理
        if not self.skip_processor.should_process(timestamp):
            return None
        
        # 2. 预处理
        t0 = time.time()
        xyz_proc, _ = preprocess_pipeline(xyz, target_points=10000)
        preprocess_time = (time.time() - t0) * 1000
        
        # 3. 推理（异步）
        self.async_inf.push(xyz_proc)
        
        # 4. 获取之前的推理结果
        labels = self.async_inf.pop()
        
        total_time = (time.time() - t0) * 1000
        self.latencies.append(total_time)
        
        return labels
    
    def get_stats(self):
        avg_latency = np.mean(self.latencies)
        p99_latency = np.percentile(self.latencies, 99)
        return {
            "avg_latency_ms": avg_latency,
            "p99_latency_ms": p99_latency,
            "fps": 1000 / avg_latency
        }


# 运行
demo = SpatialIntelligenceDemo()

# 模拟 30 帧
for i in range(30):
    xyz = np.random.randn(150000, 3) * 30
    result = demo.process_frame(xyz, time.time())
    time.sleep(0.033)  # 30 FPS

stats = demo.get_stats()
print(f"平均延迟: {stats['avg_latency_ms']:.1f}ms, P99: {stats['p99_latency_ms']:.1f}ms, FPS: {stats['fps']:.1f}")
# 输出：平均延迟: 25ms, P99: 32ms, FPS: 40

总结

cann-recipes-spatial-intelligence 的使用路径：

1. 接入点云：Velodyne / Ouster / ROS
2. 预处理：降采样 → 归一化 → 体素化
3. 推理：RandLA-Net / PointNet++ / SPConv
4. 优化：跳过帧 / 异步推理 / Batch 累积

关键要点：

• DVPP 不能直接处理点云：需要 CPU 预处理
• 实时性：用异步推理 + 跳过帧优化
• 延迟目标：30 FPS = 33ms/帧

仓库地址：https://atomgit.com/cann/cann-recipes-spatial-intelligence