1. 点云数据读取与格式转换
任务说明
本案例实现一个轻量级点云数据读取与格式转换工具，支持PLY、PCD、LAS三种格式的读取，并将点云统一转换为内部表示（XYZ坐标+强度值），随后可保存为PLY或PCD格式。通过该工具，用户可以在不同格式间转换点云数据，便于后续处理流程使用统一的数据结构。
环境依赖
需要安装Python 3.8+，以及以下库：
pip install numpy plyfile python-plyfile laspy完整实现代码
import numpy as np
import struct
from pathlib import Path
class PointCloud:
"""统一内部点云表示：xyz坐标 + 强度intensity"""
def __init__(self, points, intensity=None):
self.points = np.asarray(points, dtype=np.float64)
if intensity is None:
self.intensity = np.zeros(len(self.points), dtype=np.float32)
else:
self.intensity = np.asarray(intensity, dtype=np.float32)
assert len(self.points) == len(self.intensity), "points与intensity长度不一致"
@property
def size(self):
return len(self.points)
def read_ply(filepath):
"""读取PLY格式（仅支持vertex x,y,z,intensity字段的ASCII或binary_little_endian）"""
from plyfile import PlyData
plydata = PlyData.read(filepath)
vertex = plydata['vertex']
xyz = np.column_stack([vertex['x'], vertex['y'], vertex['z']])
if 'intensity' in vertex.data.dtype.names:
intensity = vertex['intensity']
else:
intensity = np.zeros(len(xyz), dtype=np.float32)
return PointCloud(xyz, intensity)
def read_pcd(filepath):
"""读取PCD格式（支持ASCII和binary，字段包含x,y,z,intensity）"""
with open(filepath, 'rb') as f:
header = {}
while True:
line = f.readline().decode('ascii').strip()
if not line or line.startswith('#'):
continue
if line == 'DATA ascii':
header['DATA'] = 'ascii'
break
elif line == 'DATA binary':
header['DATA'] = 'binary'
break
elif ' ' in line:
key, _, val = line.partition(' ')
header[key] = val
fields = header['FIELDS'].split()
sizes = [int(s) for s in header['SIZE'].split()]
types = header['TYPE'].split()
count = [int(c) for c in header['COUNT'].split()]
n_points = int(header['POINTS'])
# 解析字段索引
xyz_idx = [fields.index(f) for f in ['x', 'y', 'z']]
has_intensity = 'intensity' in fields
inten_idx = fields.index('intensity') if has_intensity else None
# 计算每点字节数
point_size = sum(s * c for s, c in zip(sizes, count))
if header['DATA'] == 'ascii':
data = np.loadtxt(f, max_rows=n_points)
xyz = data[:, xyz_idx]
intensity = data[:, inten_idx] if has_intensity else np.zeros(n_points)
else:
raw = f.read(point_size * n_points)
# 仅支持float32/float64的x,y,z和intensity
dtype_map = {'F': {4: np.float32, 8: np.float64}, 'I': {4: np.int32, 8: np.int64}}
offsets = []
offset = 0
for i, (sz, ty, cnt) in enumerate(zip(sizes, types, count)):
if i in xyz_idx or i == inten_idx:
offsets.append((offset, dtype_map[ty][sz], cnt))
offset += sz * cnt
xyz = np.zeros((n_points, 3))
intensity = np.zeros(n_points)
for idx, (off, dt, cnt) in enumerate(offsets):
col_data = np.frombuffer(raw, dtype=dt, count=n_points * cnt, offset=off).reshape(n_points, cnt)
if idx < 3:
xyz[:, idx] = col_data[:, 0]
else:
intensity[:] = col_data[:, 0]
return PointCloud(xyz, intensity)
def read_las(filepath):
"""读取LAS格式（使用laspy库）"""
import laspy
las = laspy.read(filepath)
xyz = np.column_stack([las.x, las.y, las.z])
if hasattr(las, 'intensity') and las.intensity.size > 0:
intensity = las.intensity.astype(np.float32)
else:
intensity = np.zeros(len(xyz), dtype=np.float32)
return PointCloud(xyz, intensity)
def write_ply(pc, filepath):
"""保存为PLY格式（binary_little_endian，含intensity）"""
from plyfile import PlyData, PlyElement
dtype = [('x', 'f8'), ('y', 'f8'), ('z', 'f8'), ('intensity', 'f4')]
arr = np.empty(pc.size, dtype=dtype)
arr['x'] = pc.points[:, 0]
arr['y'] = pc.points[:, 1]
arr['z'] = pc.points[:, 2]
arr['intensity'] = pc.intensity
ply = PlyData([PlyElement.describe(arr, 'vertex')], text=False)
ply.write(filepath)
def write_pcd(pc, filepath):
"""保存为PCD格式（binary，含intensity）"""
with open(filepath, 'wb') as f:
header = f"# .PCD v0.7 - Point Cloud Data file format\nVERSION 0.7\nFIELDS x y z intensity\nSIZE 8 8 8 4\nTYPE F F F F\nCOUNT 1 1 1 1\nWIDTH {pc.size}\nHEIGHT 1\nVIEWPOINT 0 0 0 1 0 0 0\nPOINTS {pc.size}\nDATA binary\n"
f.write(header.encode('ascii'))
# 按字段顺序写入
for i in range(pc.size):
f.write(struct.pack('fff f', *pc.points[i], pc.intensity[i]))
def convert_pointcloud(input_path, output_path):
"""主转换函数：读取任意支持格式，写入目标格式（由扩展名决定）"""
in_path = Path(input_path)
out_path = Path(output_path)
suffix_in = in_path.suffix.lower()
suffix_out = out_path.suffix.lower()
# 读取
if suffix_in == '.ply':
pc = read_ply(in_path)
elif suffix_in == '.pcd':
pc = read_pcd(in_path)
elif suffix_in == '.las':
pc = read_las(in_path)
else:
raise ValueError(f"不支持的输入格式: {suffix_in}")
# 写入
if suffix_out == '.ply':
write_ply(pc, out_path)
elif suffix_out == '.pcd':
write_pcd(pc, out_path)
else:
raise ValueError(f"不支持的输出格式: {suffix_out}")
return pc.size
if __name__ == '__main__':
import sys
if len(sys.argv) != 3:
print("用法: python pointcloud_convert.py <输入文件> <输出文件>")
print("支持: .ply .pcd .las 输入; .ply .pcd 输出")
sys.exit(1)
n = convert_pointcloud(sys.argv[1], sys.argv[2])
print(f"转换完成，共 {n} 个点")运行与验证
运行方式：
# 转换示例（将LAS转为PLY）
python pointcloud_convert.py input.las output.ply
# 转换示例（将PCD转为LAS格式需先转为PLY再转，或直接扩展支持）
python pointcloud_convert.py input.pcd output.ply验证方法：
- 准备测试数据：使用CloudCompare或PCL工具导出一个包含XYZ和强度字段的PLY/PCD/LAS文件，或使用以下代码生成测试PCD文件：
import numpy as np
# 生成100个随机点并保存为PCD
points = np.random.rand(100, 3) * 10
intensity = np.random.rand(100).astype(np.float32)
with open('test.pcd', 'wb') as f:
header = f"# .PCD v0.7\nVERSION 0.7\nFIELDS x y z intensity\nSIZE 8 8 8 4\nTYPE F F F F\nCOUNT 1 1 1 1\nWIDTH 100\nHEIGHT 1\nVIEWPOINT 0 0 0 1 0 0 0\nPOINTS 100\nDATA binary\n"
f.write(header.encode())
for i in range(100):
f.write(np.array([*points[i], intensity[i]], dtype=np.float32).tobytes())- 执行转换命令，观察输出：
python pointcloud_convert.py test.pcd test.ply- 通过条件：程序正常退出且输出“转换完成，共 100 个点”；使用CloudCompare或pypcd打开test.ply，确认点云坐标与强度值与原始test.pcd一致（允许浮点精度误差小于1e-5）。
常见问题
- LAS读取失败：确保laspy已安装且LAS文件版本受支持（LAS 1.0-1.4均可）。
- PLY文件无intensity字段：读取时自动填充强度为0，不影响坐标转换。
- PCD二进制格式兼容性：本实现仅支持每个字段count=1且类型为F（浮点）的PCD文件；若遇到其他类型（如I整数型），需扩展read_pcd中的类型映射。