pi-fastdds

把 FastDDS 封装成 Python 通信库，提供 publish/subscribe、RPC service、peer discovery 等原语。

特性

• 发布/订阅 — 基于 DDS topic 的单向数据流，支持 device 和 topic 两种地址模式
• RPC 服务 — 请求-响应模式，通过共享 topic 路由
• 远程参数 — 键值存储，复用 RPC 通道
• 节点发现 — 自动心跳，peer 上下线回调，RTT 测量（不依赖跨机时钟同步）
• 多实例隔离 — 每个 Comm 独立 DomainParticipant，不同 domain_id 严格隔离
• 跨平台分发 — FastDDS 静态链入 wheel，目标机无需预装 FastDDS

文档

快速开始

安装

# 从 wheel 安装（无需在目标机编译 FastDDS）
pip install wheelhouse/pi_fastdds-*.whl

最小示例

import time
import pi_fastdds as dds

# 创建两个通信节点
a = dds.Comm(device_name="node_a", device_id=1)
b = dds.Comm(device_name="node_b", device_id=2)

# 发布/订阅
received = []
addr = dds.Address.topic("telemetry")
sub = b.subscribe(addr, lambda msg: received.append(msg.payload))
time.sleep(0.3)

a.publish(addr, b"hello")
time.sleep(0.3)
print(f"收到: {received}")  # [b'hello']

# 清理
sub.shutdown()
a.close()
b.close()

运行内置示例：

# 单机端到端示例（pub/sub + service + peer）
python examples/demo_python.py

# 上层插件 dogfood 示例
python examples/demo_uav_state.py

# 跨机端到端测试（需要两台机器）
# 在 server 机器:
python examples/demo_multihost.py server --id 1 --name node_a --domain 7
# 在 client 机器:
python examples/demo_multihost.py client --id 2 --target-id 1 --domain 7

编译指南

前置依赖

步骤 1：编译 FastDDS 静态库

FastDDS 以源码形式 vendored 在 thirdparty/fastdds/，需要编译为静态库。

cd thirdparty/fastdds
bash build_static.sh

这会编译 foonathan_memory、fastcdr、fastdds 三个库，安装到 thirdparty/fastdds/install/。

关键编译选项（已在脚本中设置）：

• -DCMAKE_POSITION_INDEPENDENT_CODE=ON — 必须，否则链接 .so 时会报 recompile with -fPIC
• -DBUILD_SHARED_LIBS=OFF — 静态库
• -DCMAKE_BUILD_TYPE=Release — 默认 Release

编译产物：

thirdparty/fastdds/install/
├── include/          # 头文件
└── lib/
    ├── libfastdds.a
    ├── libfastcdr.a
    └── libfoonathan_memory-0.7.4.a

步骤 2：编译 Python wheel

# 安装构建工具
pip install scikit-build-core auditwheel patchelf ninja

# 运行构建脚本
bash scripts/build_wheel.sh

脚本会：

1. 用 pip wheel 编译初始 wheel（linux_x86_64 标签）
2. 用 auditwheel repair 修复为 manylinux 标签，打包系统库（libssl/libcrypto/libtinyxml2）

产物：

dist/           # 初始 wheel（linux_x86_64）
wheelhouse/     # 修复后的 manylinux wheel（可分发）

指定 Python 版本

默认使用 python3.10。指定其他版本：

# Python 3.8
PYTHON=python3.8 bash scripts/build_wheel.sh

Python 3.8 会使用 thirdparty/nanobind-py38/（v2.9.x），Python 3.10+ 使用 thirdparty/nanobind/（最新版）。

批量编译（conda）

用 conda 自动创建不同 Python 版本的独立环境，批量编译 wheel：

# 全部编译（3.9 / 3.10 / 3.11 / 3.12）
bash scripts/build_wheels_conda.sh

# 指定版本
bash scripts/build_wheels_conda.sh 3.9 3.11

效果：

首次运行会创建 conda 环境，后续复用已存在的环境，只重新编译 wheel。

glibc 说明：本机编译 + auditwheel 产出 manylinux_2_31 标签的 wheel， 要求目标系统 glibc ≥ 2.31（Ubuntu 20.04+、Debian 11+）。 如需支持更旧的 glibc（如 CentOS 7，glibc 2.17），需在对应 manylinux Docker 镜像中 同时重建 FastDDS 静态库和 Python wheel，参考 scripts/build_wheels_batch.sh（Docker 方案）。

测试

# 安装 pytest
pip install pytest

# 运行完整测试套件
python -m pytest tests/ -v

# 运行特定测试
python -m pytest tests/test_pubsub.py -v
python -m pytest tests/test_service.py -v
python -m pytest tests/test_peer.py -v

测试覆盖：

• test_address.py — Address 工厂与校验
• test_pubsub.py — 发布/订阅基本功能、队列、回调串行化、关闭语义
• test_service.py — RPC 服务调用、超时、异常处理
• test_peer.py — 节点发现、上下线回调、心跳配置
• test_param.py — 远程参数读写
• test_exceptions.py — 异常体系

下载源码

git clone git@gitee.com:pi-lab/pi-fastdds.git

目录结构

pi-fastdds/
├── cxx/                          # C++ 核心库
│   ├── pi_fastdds/
│   │   ├── include/pi_fastdds/   # 公共头文件
│   │   └── src/                  # 实现 + Python 绑定
│   └── dds_comm/fastdds/         # IDL 生成代码
├── python/pi_fastdds/            # Python 包
├── examples/                     # 示例程序
├── tests/                        # pytest 测试
├── thirdparty/                   # vendored 依赖
│   ├── fastdds/                  # FastDDS 源码 + 构建脚本
│   ├── nanobind/                 # nanobind（Python 3.10+）
│   └── nanobind-py38/            # nanobind v2.9.x（Python 3.8）
├── wheelhouse/                   # 构建好的 wheel
├── scripts/build_wheel.sh        # wheel 构建脚本
├── docs/                         # 文档
└── pyproject.toml                # Python 包配置

许可证

MIT