2026-09-05

osrf_pycommon 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/osrf/osrf_pycommon
版本：2.1.7（package.xml），构建类型 ament_python，Python ≥3.5，许可证 Apache 2.0。

osrf_pycommon 是 OSRF（Open Source Robotics Foundation）维护的 Python 通用工具库，体量很小（约 47 个文件、4 个子模块），不依赖 ROS 运行时。在 ROS 2 Humble 工作区中，它主要被 launch / launch_ros / launch_testing 依赖，承担子进程异步执行、终端颜色处理和 CLI 扩展模式等基础能力。

1. 总体认识

1.1 设计原则（来自 `docs/index.rst`）

只使用标准库或极少量外部依赖（importlib-metadata）
尽量支持 Linux / macOS / Windows，不支持处优雅降级
纯 Python 3，无 C 扩展

1.2 模块结构

osrf_pycommon/
├── osrf_pycommon/
│   ├── process_utils/     # 子进程执行（同步/异步）★ ROS 2 最常用
│   ├── terminal_color/    # ANSI 颜色与转义序列
│   ├── terminal_utils.py  # 终端尺寸、is_tty
│   └── cli_utils/         # CLI verb 模式、参数解析辅助
├── tests/                 # unittest 单元测试
├── docs/                  # Sphinx 文档
├── setup.py
└── package.xml

2. process_utils — 核心模块

路径：osrf_pycommon/process_utils/
这是 ROS 2 生态中使用最广泛的部分。

2.1 公开 API

from .async_execute_process import async_execute_process
from .async_execute_process import asyncio
from .async_execute_process import AsyncSubprocessProtocol
from .async_execute_process import get_loop

from .impl import execute_process
from .impl import execute_process_split
from .impl import which

__all__ = [
    'async_execute_process',
    'asyncio',
    'AsyncSubprocessProtocol',
    'get_loop',
    'execute_process',
    'execute_process_split',
    'which',
]

函数/类	类型	用途
`execute_process`	同步生成器	逐行 yield 子进程 stdout（合并 stderr）
`execute_process_split`	同步生成器	stdout/stderr 分开 yield
`async_execute_process`	asyncio 协程	异步启动子进程
`AsyncSubprocessProtocol`	Protocol 类	异步 I/O 回调基类
`get_loop`	函数	获取/创建合适的事件循环
`which`	函数	查找可执行文件（`shutil.which` 兼容回退）

2.2 同步执行路径

1
2
3

execute_process / execute_process_split  (impl.py)
    ├── emulate_tty=False → execute_process_nopty.py
    └── emulate_tty=True  → execute_process_pty.py (Unix only)

nopty 实现（execute_process_nopty.py）：

基于 subprocess.Popen + select.select（Unix）或 readline（Windows）
按行缓冲输出，保留换行符
可用 stderr_to_stdout 合并或分离 stderr

pty 实现（execute_process_pty.py）：

用 pty.openpty() 让子进程认为在 TTY 上运行
子进程会输出彩色日志、启用行缓冲（如 Python -u 行为）
注意 pty 数量有限，大量并行可能 OSError: out of pty devices
Windows 无 pty，自动回退 nopty

2.3 异步执行路径

1
2
3

async_execute_process  (async_execute_process_asyncio/impl.py)
    ├── emulate_tty=False → loop.subprocess_exec / subprocess_shell
    └── emulate_tty=True  → pty + connect_read_pipe

AsyncSubprocessProtocol（async_execute_process.py）：

继承 asyncio.SubprocessProtocol
可覆写 on_stdout_received / on_stderr_received / on_process_exited
protocol.complete 是 Future，完成时结果为 return code

get_loop（get_loop_impl.py）：

线程局部单例事件循环
Windows 强制使用 ProactorEventLoop（子进程管道需要）
处理 Python 3.10 的 DeprecationWarning

2.4 which 回退实现

Python 3.3+ 优先用 shutil.which；旧版本使用 _which_backport，支持：

Windows PATHEXT（.exe 等）
相对路径 ./script
目录排除（避免把目录当可执行文件）

3. 在 ROS 2 launch 中的关键作用

3.1 ExecuteLocal — 启动节点进程

launch/actions/execute_local.py 是最大消费者：

1 2	from osrf_pycommon.process_utils import async_execute_process from osrf_pycommon.process_utils import AsyncSubprocessProtocol

ExecuteLocal 用 async_execute_process 启动 ros2 run、节点可执行文件等，并通过自定义 Protocol 将 stdout/stderr 转为 launch 事件（ProcessStdout / ProcessStderr）。

3.2 LaunchService 事件循环

loop = osrf_pycommon.process_utils.get_loop()
run_async_task = loop.create_task(self.run_async(
    shutdown_when_idle=shutdown_when_idle
))
while True:
    try:
        return loop.run_until_complete(run_async_task)

整个 ros2 launch 的异步调度建立在 get_loop() 返回的事件循环上。

3.3 查找可执行文件

launch/substitutions/find_executable.py → which
launch_ros/substitutions/executable_in_package.py → which

用于解析 launch 文件中 $(find-pkg-prefix) 等替换后的可执行路径。

3.4 launch_testing 颜色剥离

launch_testing/tools/output.py
launch_testing/asserts/assert_output.py

使用 remove_ansi_escape_sequences 在断言输出时去掉 ANSI 转义，避免颜色码导致测试失败。

4. terminal_color — 终端颜色

路径：osrf_pycommon/terminal_color/

4.1 子模块

文件	职责
`impl.py`	`ansi()`、`format_color()`、`print_color()`、颜色字典
`ansi_re.py`	正则匹配/剥离 ANSI 转义序列
`windows.py`	Win32 API 彩色输出（借鉴 colorama 思路，不 hook stdout）

4.2 两种着色方式

1. 直接 ANSI 码

1 2	from osrf_pycommon.terminal_color import ansi print(ansi('red') + "error" + ansi('reset'))

2. @{} 标记替换

1 2	from osrf_pycommon.terminal_color import format_color print(format_color("This is @{bf}blue@{reset}."))

支持 @{redf} / @{rf} / @{r} 等多种简写；背景色必须带 b 后缀（如 @{rb}）。

4.3 平台行为

Linux/macOS：正常输出 ANSI 转义
Windows：ansi() 返回空字符串；需用 print_color() 或 print_ansi_color_win32() 才能显示颜色
可调用 disable_ansi_color_substitution_globally() 全局关闭颜色

4.4 ANSI 处理工具

def remove_ansi_escape_sequences(string):
    """
    Removes any ansi escape sequences found in the given string and returns it.
    """
    global _ansi_re
    return _ansi_re.sub('', string)

正则 \033\[\d{1,2}[m] 匹配常见 SGR 序列；另有 split_by_ansi_escape_sequence 用于分段处理。

5. terminal_utils — 终端工具

单文件 terminal_utils.py，仅 3 个公开符号：

函数	说明
`get_terminal_dimensions()`	返回 `(width, height)`；Unix 用 `tput cols/lines`，Windows 用 `GetConsoleScreenBufferInfo`
`is_tty(stream)`	判断 stream 是否为 TTY
`GetTerminalDimensionsError`	无法获取尺寸时抛出

用途相对独立，ROS 2 核心路径较少直接引用。

6. cli_utils — CLI 扩展模式

路径：osrf_pycommon/cli_utils/

6.1 verb_pattern — 插件式子命令

这是 ROS 2 早期 ros2 xxx verb 风格 CLI 的基础模式（ros2 topic echo 等），基于 setuptools entry_points 动态加载 verb：

函数	作用
`list_verbs(group)`	从 entry_point group 列出所有 verb 名
`load_verb_description(name, group)`	加载 verb 模块（含 `main`、`prepare_arguments`）
`create_subparsers(...)`	为每个 verb 创建 argparse 子解析器
`split_arguments_by_verb(args)`	拆分 `ros2 [全局选项] verb [verb选项]`
`call_prepare_arguments(func, parser, sysargs)`	兼容 1/2 参数版本的 `prepare_arguments`

verb 模块约定结构（见 docs/cli_utils.rst）：

verb = 'myverb'
description = '...'
def prepare_arguments(parser): ...
def main(args): ...
# 可选
def argument_preprocessor(args): ...

注意：当前 Humble 工作区中，ros2cli 不再直接依赖 osrf_pycommon，而是使用自有的 ros2cli 框架；cli_utils 仍可作为构建类似 CLI 的参考库，也被 colcon 等工具的历史版本使用过。

6.2 common — 参数解析辅助

函数	用途
`extract_jobs_flags(arguments)`	从 make 参数字符串中提取 `-j8`、`-l8`、`--jobs=4` 等
`extract_argument_group(args, '--args')`	用 `--args ... --` 分隔符提取参数组；支持 `---` 转义

典型场景：构建工具需要把「传给 make 的并行参数」与「传给目标的参数」分开。

7. 依赖与构建

7.1 package.xml

1	<exec_depend>python3-importlib-metadata</exec_depend>

verb_pattern.list_verbs 用 importlib.metadata.entry_points() 发现插件；Python 3.8+ 内置，旧版通过 importlib-metadata 包提供。

7.2 setup.py 要点

ament_python 包，注册 resource/osrf_pycommon 索引
测试 extra：flake8、pytest
zip_safe=True

8. 测试结构

tests/
├── test_code_format.py          # flake8
└── unit/
    ├── test_process_utils/      # 同步/异步/pty 子进程
    ├── test_terminal_color/
    ├── test_terminal_utils.py
    └── test_cli_utils/

进程测试包含 stdout_stderr_ordering 等 fixture，验证 nopty/pty 模式下输出顺序行为。

9. 在 ROS 2 栈中的位置

用户: ros2 launch my_pkg launch.py
         │
         ▼
    launch.LaunchService.run()
         │ get_loop()
         ▼
    ExecuteLocal.execute()
         │ async_execute_process(AsyncSubprocessProtocol, cmd)
         ▼
    子进程 (talker, listener, ...)

包	依赖 osrf_pycommon	使用内容
launch	`depend`	`async_execute_process`, `get_loop`, `which`
launch_ros	`depend`	`which`
launch_testing	`exec_depend`	`remove_ansi_escape_sequences`
launch_pytest	`exec_depend`	（传递依赖）

与 CycloneDDS、iceoryx 等不同，osrf_pycommon 不参与通信，只服务于 Python 工具链的运行时基础设施。

10. 设计特点小结

特点	说明
轻量	4 模块、无重型依赖
跨平台子进程	Unix select + Windows readline；可选 pty 模拟 TTY
asyncio 集成	为 launch 提供统一的 ProactorEventLoop（Windows）
流式输出	生成器/Protocol 模式，适合实时日志转发
颜色可剥离	测试断言时可去掉 ANSI，避免误匹配
verb 插件模式	entry_points 驱动的可扩展 CLI 框架（历史影响 ros2cli 设计）
向后兼容	`which` 回退、`importlib_metadata` 双版本 API

11. 推荐阅读顺序

launch 集成：launch/actions/execute_local.py — 看 Protocol 如何包装子进程 I/O
异步基础：process_utils/async_execute_process_asyncio/impl.py + get_loop_impl.py
同步/pty：process_utils/impl.py → execute_process_nopty.py / execute_process_pty.py
测试断言：launch_testing/asserts/assert_output.py — 颜色剥离用法
CLI 模式：docs/cli_utils.rst + cli_utils/verb_pattern.py
文档：docs/process_utils.rst、docs/terminal_color.rst

如果你希望，我可以把本文写入 ros2doc/osrf_pycommon/，或继续分析 launch 中 ExecuteLocal 如何基于 AsyncSubprocessProtocol 转发 stdout 的完整调用链。

文章互动

阅读 --

留言

0 条留言

正在加载留言…

fishBlog

专注嵌入式开发

osrf_pycommon 源码详细分析

osrf_pycommon 源码详细分析

1. 总体认识

1.1 设计原则（来自 `docs/index.rst`）

1.2 模块结构

2. process_utils — 核心模块

2.1 公开 API

2.2 同步执行路径

2.3 异步执行路径

2.4 which 回退实现

3. 在 ROS 2 launch 中的关键作用

3.1 ExecuteLocal — 启动节点进程

3.2 LaunchService 事件循环

3.3 查找可执行文件

3.4 launch_testing 颜色剥离

4. terminal_color — 终端颜色

4.1 子模块

4.2 两种着色方式

4.3 平台行为

4.4 ANSI 处理工具

5. terminal_utils — 终端工具

6. cli_utils — CLI 扩展模式

6.1 verb_pattern — 插件式子命令

6.2 common — 参数解析辅助

7. 依赖与构建

7.1 package.xml

7.2 setup.py 要点

8. 测试结构

9. 在 ROS 2 栈中的位置

10. 设计特点小结

11. 推荐阅读顺序

文章互动

留言

osrf_pycommon 源码详细分析

1. 总体认识

1.1 设计原则（来自 docs/index.rst）

1.2 模块结构

2. process_utils — 核心模块

2.1 公开 API

2.2 同步执行路径

2.3 异步执行路径

2.4 which 回退实现

3. 在 ROS 2 launch 中的关键作用

3.1 ExecuteLocal — 启动节点进程

3.2 LaunchService 事件循环

3.3 查找可执行文件

3.4 launch_testing 颜色剥离

4. terminal_color — 终端颜色

4.1 子模块

4.2 两种着色方式

4.3 平台行为

4.4 ANSI 处理工具

5. terminal_utils — 终端工具

6. cli_utils — CLI 扩展模式

6.1 verb_pattern — 插件式子命令

6.2 common — 参数解析辅助

7. 依赖与构建

7.1 package.xml

7.2 setup.py 要点

8. 测试结构

9. 在 ROS 2 栈中的位置

10. 设计特点小结

11. 推荐阅读顺序

文章互动

留言

1.1 设计原则（来自 `docs/index.rst`）