ROS2►核心

ros2cli 源码详细分析

ros2cli 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/ros2cli
版本：0.18.18（Humble），子包 15 个，构建类型 ament_python，许可证 Apache 2.0。

ros2cli 是 ROS 2 命令行工具框架及 内置 introspection 命令 的 monorepo：单一 ros2 可执行文件通过 setuptools entry point 插件 扩展为 ros2 topic list、ros2 node info 等；图查询类命令默认走 后台 daemon + XML-RPC，数据面命令（echo/pub）使用 DirectNode（rclpy）。

范围说明：本仓库含 15 个包；ros2 bag、ros2 launch、ros2 security 等命令在 其他仓库 注册同一 ros2cli.command 组（见 §12）。ros2cli_common_extensions 仅为依赖聚合 metapackage。

1. 总体认识

1.1 核心职责

能力 实现位置 说明
CLI 框架 ros2cli argparse 分层、entry point 发现、按需加载
插件系统 plugin_system.py / entry_points.py 版本校验、实例化、失败降级
图 introspection NodeStrategy + _ros2_daemon 复用长生命周期 rclpy 节点，加速 list/info
数据面工具 各 ros2topic/ros2param verb 独立 rclpy 节点、订阅/服务客户端
包/可执行文件 ros2pkg / ros2run ament index、subprocess 启动
健康检查 ros2doctor 可插拔 check/report entry point
Shell 补全 argcomplete + *.completer 可选 python3-argcomplete

1.2 在 ROS 2 栈中的位置

用户ros2cli 仓库_ros2_daemon 进程ROS 2 运行时echo/pubros2 topic echo /chatterros2cli/cli.pyCommandExtension\ntopic/node/...VerbExtension\nlist/echo/pub/...NodeStrategyDirectNodeDaemonNode XML-RPCNetworkAwareNodeLocalXMLRPCServer\n:11511+domain_idrclpyrmw / DDS graph
对比项 ros2cli daemon 路径 DirectNode 路径
进程 独立 _ros2_daemon 每次命令内嵌 rclpy
启动成本 首次 spawn，后续 RPC 快 每次 rclpy.init + spin
适用 get_topic_names_and_types 等图 API echo、pub、参数服务
端口 11511 + ROS_DOMAIN_ID 无
禁用 --no-daemon 默认 fallback

1.3 命令行分层模型

1
2
3
ros2                          ← console_scripts: ros2cli.cli:main
  └── <command>               ← entry point 组 ros2cli.command
        └── <verb>            ← entry point 组 ros2<cmd>.verb（可选）

示例：

1
2
3
ros2 topic list -t
# command = topic (TopicCommand)
# verb    = list  (ListVerb)

部分 command 无 verb（扁平结构）：

  • ros2 run pkg exe
  • ros2 doctor -r
  • ros2 daemon status

2. 仓库结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
ros2cli/
├── README.md
├── ros2cli/                    # ★ 框架核心 (~2072 行 Python)
│   └── ros2cli/
│       ├── cli.py              # main() 入口
│       ├── command/            # CommandExtension、add_subparsers_on_demand
│       ├── plugin_system.py
│       ├── entry_points.py
│       ├── node/               # DirectNode、DaemonNode、NodeStrategy
│       ├── daemon/             # _ros2_daemon 实现
│       ├── xmlrpc/             # RPC 客户端/服务端、rclpy 序列化
│       └── verb/daemon/        # ros2 daemon start|stop|status
├── ros2topic/                  # 最大命令包之一
├── ros2node/ ros2service/ ros2param/
├── ros2action/ ros2component/ ros2lifecycle/
├── ros2interface/ ros2doctor/ ros2multicast/
├── ros2pkg/ ros2run/
├── ros2cli_test_interfaces/    # 测试用 msg/srv/action
└── ros2lifecycle_test_fixtures/

3. 子包一览

包 ros2 命令 verbs / 行为
ros2cli daemon, extension_points, extensions start / stop / status
ros2topic topic bw, delay, echo, find, hz, info, list, pub, type
ros2node node info, list
ros2service service call, find, list, type
ros2param param delete, describe, dump, get, list, load, set
ros2action action info, list, send_goal
ros2component component list, load, standalone, types, unload
ros2lifecycle lifecycle get, list, nodes, set
ros2interface interface list, package, packages, proto, show
ros2doctor doctor, wtf check/report + 可扩展 verb
ros2multicast multicast receive, send
ros2pkg pkg create, executables, list, prefix, xml
ros2run run （无 verb）直接启动可执行文件
ros2cli_test_interfaces — 测试接口定义
ros2lifecycle_test_fixtures — lifecycle 测试节点

4. 框架包 ros2cli

4.1 入口 cli.py

1
2
3
4
5
6
7
8
9
def main(*, script_name='ros2', argv=None, ...):
    parser = argparse.ArgumentParser(...)
    add_subparsers_on_demand(parser, 'ros2', '_command', 'ros2cli.command', ...)
    args = parser.parse_args(args=argv)
    extension = getattr(args, '_command', None)
    if extension is None:
        parser.print_help()
        return 0
    return extension.main(parser=parser, args=args)

特性：

  • 默认 行缓冲 stdout（reconfigure(line_buffering=True) 或 patch print(flush=True)）
  • 捕获 KeyboardInterrupt → 返回 SIGINT
  • 捕获 ExternalShutdownException → SIGTERM
  • 可选 argcomplete 自动补全

4.2 插件发现：entry_points.py

函数 作用
get_entry_points(group_name) 读取 setuptools entry point 组
load_entry_points(group_name) entry_point.load() 得到类
get_all_entry_points() 扫描 ros2cli.extension_point 注册的所有组

扩展点注册：第三方包在 setup.py 中声明：

1
2
3
'ros2cli.extension_point': [
    'ros2mytool.verb = ros2mytool.verb:VerbExtension',
],

只有注册在 ros2cli.extension_point 的组才会被 ros2 extension_points 列出。

4.3 按需加载：add_subparsers_on_demand

优化启动速度的关键设计：

  1. 先为每个 command 创建 空 subparser（无参数）
  2. parse_known_args 判断用户选了哪个 command
  3. 仅实例化被选中的 CommandExtension 并调用其 add_arguments
  4. 未选 command 时加载全部 extension 仅为了生成 help 描述

支持 argv= 参数（测试用）与 argcomplete 的 COMP_LINE 解析。

4.4 CommandExtension / VerbExtension

1
2
3
4
class CommandExtension:
    EXTENSION_POINT_VERSION = '0.1'
    def add_arguments(self, parser, cli_name, *, argv=None): ...
    def main(self, *, parser, args): raise NotImplementedError()

构造时 satisfies_version(PLUGIN_SYSTEM_VERSION, '^0.1') — caret 语义 major/minor 兼容检查。

plugin_system.instantiate_extensions：

  • 单例缓存 extension 实例（按 class）
  • PluginException → warning 并跳过
  • 其他异常 → error 并跳过

4.5 内建调试命令

命令 作用
ros2 extensions 列出已加载的 command 扩展
ros2 extension_points 列出所有已注册的 extension point 组

5. 节点策略：DirectNode 与 Daemon

5.1 DirectNode

1
2
3
4
rclpy.init(args=argv)
self.node = rclpy.create_node(NODE_NAME_PREFIX + suffix, ...)
# 默认 spin_time=0.5s 等待图 discovery
rclpy.spin_once(...) until timeout
  • 节点名前缀 _ros2cli_（ros2cli.node.NODE_NAME_PREFIX）
  • 通过 __getattr__ 转发 node.get_* 图 API
  • 补全 action 相关 API（rclpy.action.*）

退出 context 时 destroy_node + rclpy.shutdown()。

5.2 _ros2_daemon 与 XML-RPC

独立进程（console_scripts: _ros2_daemon = ros2cli.daemon:main）：

配置 值
地址 127.0.0.1
端口 11511 + ROS_DOMAIN_ID
路径 /ros2cli/
空闲超时 默认 2 小时 → 自退出

serve() 内 NetworkAwareNode 注册 RPC 方法（与 rclpy graph API 一一对应）：

  • get_node_names_and_namespaces_with_enclaves
  • get_topic_names_and_types / get_service_names_and_types
  • get_publishers_info_by_topic / get_subscriptions_info_by_topic
  • count_publishers / count_subscribers
  • action 相关 get_action_*

DaemonNode 用 ServerProxy 调用；NodeStrategy.__getattr__ 优先走 daemon 方法表。

Spawn 流程（spawn_daemon）：

  • 先绑定 XML-RPC socket（防 TOCTOU）
  • daemonize() 子进程执行 _ros2_daemon
  • 父进程 wait_for 直到 RPC 可用

5.3 NodeStrategy

1
2
with NodeStrategy(args) as node:
    node.get_topic_names_and_types()

逻辑：

  1. 若未 --no-daemon 且 daemon 已运行 → DaemonNode
  2. 若 daemon 未连接 → fallback DirectNode
  3. 若 daemon 未运行 → spawn_daemon 后 DirectNode（首次）

add_strategy_node_arguments 添加：

  • --spin-time — DirectNode discovery 等待
  • --no-daemon — 禁用 daemon
  • --use-sim-time 等

5.4 NetworkAwareNode

Daemon 内使用：监听网卡变化（netifaces），若地址集变化则 重建 DirectNode，避免网络切换后 graph 数据陈旧。

6. 典型命令包模式（以 ros2topic 为例）

6.1 三层文件组织

1
2
3
4
5
6
7
ros2topic/
├── command/topic.py      # TopicCommand
├── verb/
│   ├── __init__.py       # VerbExtension 基类
│   ├── list.py           # ListVerb
│   └── echo.py           # EchoVerb
└── api/__init__.py       # 共享 helper、argcomplete

6.2 TopicCommand

1
2
3
4
5
6
7
8
class TopicCommand(CommandExtension):
    def add_arguments(self, parser, cli_name):
        parser.add_argument('--include-hidden-topics', ...)
        add_subparsers_on_demand(parser, cli_name, '_verb', 'ros2topic.verb')

    def main(self, *, parser, args):
        extension = getattr(args, '_verb', None)
        return extension.main(args=args)

6.3 entry_points（setup.py）

1
2
3
4
5
6
7
'ros2cli.command': ['topic = ros2topic.command.topic:TopicCommand'],
'ros2cli.extension_point': ['ros2topic.verb = ros2topic.verb:VerbExtension'],
'ros2topic.verb': [
    'list = ros2topic.verb.list:ListVerb',
    'echo = ros2topic.verb.echo:EchoVerb',
    ...
],

6.4 两类 verb 实现

A. 仅图查询（走 NodeStrategy）

list.py：

1
2
with NodeStrategy(args) as node:
    topic_names_and_types = get_topic_names_and_types(node=node, ...)

B. 数据面（自建 rclpy 节点）

echo.py：

  • NodeStrategy 仅用于解析 topic 类型（get_msg_class）
  • 随后创建订阅、spin 打印 YAML/CSV

api/__init__.py 提供：

  • TopicNameCompleter / TopicTypeCompleter — argcomplete
  • qos_profile_from_short_keys — --qos-profile sensor_data 等
  • get_msg_class — 阻塞等待 publisher 出现

7. 其他命令包要点

7.1 ros2run

  • 无 verb；RunCommand.main 直接执行
  • get_executable_path() 查 ament index 安装路径
  • subprocess.Popen + 转发 KeyboardInterrupt
  • 依赖 ros2pkg.api.get_executable_paths

7.2 ros2pkg

  • create — 从 .em 模板生成 ament_cmake / ament_python / cargo 包
  • executables / list / prefix / xml — 包元数据 introspection
  • package_name_completer 供其他命令复用

7.3 ros2param

  • 图查询 + 参数服务客户端（AsyncParametersClient）
  • dump/load — YAML 参数文件

7.4 ros2doctor

额外 entry point 组（非 verb 分层）：

1
2
'ros2doctor.checks': ['PlatformCheck = ...', 'NetworkCheck = ...', ...],
'ros2doctor.report': ['PlatformReport = ...', 'RMWReport = ...', ...],

ros2 doctor 运行 checks；-r 输出 reports。ros2 wtf 为别名 command。

7.5 ros2component

  • 与 composition 包交互：加载/卸载 component 到 container
  • 依赖 launch / component_manager 服务

7.6 ros2interface

  • 纯 Python introspection：rosidl_runtime_py / ament index
  • 不 需要 ROS graph（通常不用 daemon）

8. XML-RPC 序列化层

ros2cli/xmlrpc/：

模块 职责
local_server.py 单线程 XML-RPC server
client.py ServerProxy 包装
marshal/rclpy.py 将 rclpy 返回的复杂类型转为可 RPC 传输结构
marshal/generic.py 通用类型

Daemon 暴露的是 纯函数调用，不是 DDS 代理；每次 RPC 在 daemon 进程内执行 rclpy API。

9. Shell 补全

  • 依赖 python3-argcomplete（ros2cli exec_depend）
  • cli.py 调用 autocomplete(parser, ...)
  • 各 verb 为 argument 设置 .completer = TopicNameCompleter(...) 等
  • 安装 share/ros2cli/environment/completion/ros2-argcomplete.bash

10. 测试体系

包 测试方式
ros2cli pytest：test_strategy.py（NodeStrategy）、test_daemon.py
ros2topic 等 test_cli.py — subprocess 调用真实 ros2
公共 ament lint（flake8/pep257/copyright/xmllint）

CLI 测试常用 launch_testing + fixture nodes（见 ros2topic/test/fixtures/）。

11. 扩展开发指南

11.1 添加新 command（新包）

  1. 创建 ament_python 包，install_requires=['ros2cli']
  2. 实现 XxxCommand(CommandExtension)
  3. setup.py：
1
2
3
entry_points={
    'ros2cli.command': ['mytool = ros2mytool.command.mytool:MytoolCommand'],
}
  1. colcon build 后 ros2 mytool --help

11.2 添加 verb（已有 command）

1
2
3
4
entry_points={
    'ros2cli.extension_point': ['ros2topic.verb = ros2topic.verb:VerbExtension'],
    'ros2topic.verb': ['myverb = ros2topic.verb.myverb:MyVerb'],
}

11.3 使用图 API 的最佳实践

  • 只读 introspection → with NodeStrategy(args) as node:
  • 需要 subscription/publisher/service → 独立 rclpy.create_node，勿长期占用 daemon
  • CI 并行 → 固定 ROS_DOMAIN_ID 或 --no-daemon 避免 daemon 端口冲突

12. 本仓库外的 CLI 扩展

以下命令 不在 ros2cli 目录，但使用同一插件机制：

仓库 / 包 命令
rosbag2/ros2bag bag
launch_ros/ros2launch launch
ros2_tracing/ros2trace trace
ros2_testing/ros2test test
sros2/sros2 security
其他第三方 自定义 ros2cli.command

Humble 桌面安装常通过 ros2cli_common_extensions metapackage 一次性依赖本仓库全部 command + ros2launch + sros2 等。

13. 依赖关系

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
ros2cli (框架)
├── rclpy
├── python3-importlib-metadata / pkg-resources
├── python3-argcomplete
└── python3-netifaces  (daemon NetworkAwareNode)

ros2topic / ros2service / ...
├── ros2cli
├── rclpy
└── rosidl_runtime_py

ros2run / ros2pkg
├── ros2cli
└── ament_index_python (通过 ros2pkg.api)

ros2doctor
├── ros2cli
└── 各 check 模块 (platform, network, rmw, ...)

14. 调试建议

  1. 命令未出现：ros2 extension_points / 检查包是否 overlay source + setup.py entry_points
  2. daemon 问题：ros2 daemon status；--no-daemon 对比；检查 11511+domain_id 端口占用
  3. 空 topic list：增大 --spin-time；检查 ROS_DOMAIN_ID 与节点一致
  4. echo 无输出：QoS 不匹配（试 --qos-profile）；topic 名/remapping
  5. 插件加载失败：查看 stderr warning（Failed to load entry point）
  6. 慢启动：首次 spawn daemon 正常；后续应走 RPC

15. 推荐阅读顺序

  1. ros2cli/ros2cli/cli.py — 总入口
  2. ros2cli/command/__init__.py — add_subparsers_on_demand
  3. ros2cli/node/strategy.py + daemon/__init__.py — daemon 架构
  4. ros2topic/command/topic.py + verb/list.py — command/verb 模式
  5. ros2topic/verb/echo.py — 数据面 + QoS
  6. ros2run/command/run.py — 无 verb 命令
  7. ros2doctor/command/doctor.py — 多 entry point 组
  8. ros2pkg/verb/create.py — 模板生成
  9. 仓库 README.md — 官方扩展示例链接

16. 小结

ros2cli 仓库 = 可扩展 CLI 框架 + ROS 2 标准 introspection 工具集：

  • 对上：统一 ros2 用户体验、argcomplete、daemon 加速图查询；
  • 对中：三层 entry point（command → verb → 可选 checks/reports）与按需加载；
  • 对下：rclpy graph API、ament index、subprocess 启动节点。

理解 NodeStrategy / DirectNode / Daemon 三者关系是阅读各 verb 源码的钥匙：列表类命令几乎总是 NodeStrategy；echo/pub/param set 等在图查询之外另建 rclpy 实体。扩展新工具只需新 Python 包 + setuptools entry point，无需修改框架源码。

文章互动

阅读 --

留言

0 条留言

正在加载留言…