Transformer使用指导

YAML写法

机器人配置YAML里，transforms写在一条mapping rule内部：

device_to_r2c:
  mappings:
    # 腕部相机：ndarray → JPEG bytes（Observations.images.* 必须是 bytes）
    - target: "images.color.wrist"
      source: "wrist"
      transforms:
        - ndarray_to_jpeg: {quality: 85}
    # 末端位姿：numpy → list
    - target_key: "joint_states.position"
      source_path: "eef_pose"
      transforms: array_to_list
r2c_to_device:
  mappings:
    # 模型输出 7D → 前 6 维给 eef_pose，第 7 维给夹爪
    - target: "eef_pose"
      source: "joint_states.position"
      transforms:
        - slice: {start: 0, end: 6}

完整可运行样例详见在CloudRobo平台下载的R2C SDK包中的config/robot_so101_lerobot_config.yaml。

程序化写法

from r2c_sdk.core.config_mapper import ConfigDrivenMapper
mapper = ConfigDrivenMapper.from_rule_mappings([
    {"source": "eef_pose", "target": "joint_states.position",
     "transforms": ["array_to_list"]},
])
mapper.map({"eef_pose": [0.1, 0.2, 0.3, 0.0, 0.0, 0.0]})

查看当前有哪些transformer

from r2c_sdk.core.transformers import TransformerRegistry
print(TransformerRegistry.available_transformers())   # 内置 + 第三方，按字母序
print(TransformerRegistry.lookup("ndarray_to_jpeg"))  # 拿到实例，或 None

transform与transforms

单个条目的四种写法

transforms: to_float                       # 写法 1
transforms: "crop:bottom_left"             # 写法 2（注意整个条目要加引号）
transforms: [{"ndarray_to_jpeg": {"quality": 85}}]   # 写法 3
transforms: [{"subtract": 1.0}]            # 写法 4

写法2的config永远是字符串，其适配 crop（区域名）、decode_image（bgr/rgb/gray）这类字符串参数；但对ndarray_to_jpeg这种要dict的会失败："ndarray_to_jpeg:85" → Invalid config for transformer 'ndarray_to_jpeg': ... got str。

不支持name/config双键写法

下面这种写法（在部分旧文档里出现过）会直接报错：

transforms:
  - name: my_scale          # ✗ 非法
    config:
      factor: 0.001

正确写法是将名字作为唯一key，config作为它的值：

transforms:
  - my_scale: {factor: 0.001}     # ✓

流水线：链式组合

列表即流水线，前一个的输出是后一个的输入：

transforms:
  - "crop:center"                  # ndarray 裁剪
  - {resize: [224, 224]}           # 缩放
  - {bgr_to_rgb: null}             # 通道序（无参数用 null 占位）
  - {ndarray_to_jpeg: {quality: 90}}

链中任何一步抛出异常，整条rule失败，异常向上抛给调用方。

程序化注入overrides

build_transformer_registry(overrides=...) 接受实例（注意与entry_point注册类不同），可用于测试或临时替换：

from r2c_sdk.core.config_mapper import ConfigDrivenMapper
from r2c_sdk.core.transformers import build_transformer_registry
registry = build_transformer_registry(overrides={"my_double": MyDouble()})
mapper = ConfigDrivenMapper.from_rule_mappings(rules, transformers=registry)

context参数

transform(value, config, context) 的第三个参数是整个源 payload，供transformer按名称反查索引（subtract / euler_modulo 的names模式用它）。

mapper.map({"pose": [0.1, 7.0], "pose_names": ["roll", "yaw"]})

类型与结构

序列/数值

• subtract的三种模式

  transforms: [{"subtract": 1.0}]                       # 标量：所有元素 -1.0
  transforms: [{"subtract": [0.0, 0.0, 1.0]}]           # 等长列表：逐元素相减
  transforms:                                           # 定向模式：只改指定位置
    - subtract: {values: [1.0], indices: [2]}
  transforms:                                           # 定向模式（按名，需 payload 带名字表）
    - subtract: {values: [1.0], names: [yaw], names_path: pose_names}

  indices 与 names 互斥；names 模式要求 context 中存在 names_path 指向的名字列表，否则抛错。

• crop的三种模式

  transforms: "crop:bottom_left"                       # 预设区域名
  transforms: [{"crop": {"region": "center"}}]         # 映射形式
  transforms: [{"crop": {"box": [100, 300, 50, 400]}}] # 像素框 [y1, y2, x1, x2]

  9个预设：top_left、top_right、bottom_left、bottom_right、left_half、right_half、top_half、bottom_half、center。box越界或y1 >= y2会报错。

图像编解码

WebP体积通常比同画质JPEG小30~40%，编码耗时高约10~15倍，适合带宽敏感场景。

ROS消息

transforms:
  - select_joints_by_name:
      names: ["joint_1", "joint_3", "joint_5"]

ros_image_to_* 支持 rgb8、bgr8、rgba8、bgra8、mono8、8uc1、mono16、16uc1编码。

ROS 消息及机器人夹爪中涉及ROS消息类型的transformer（ros_*、list_to_ros_*、scalar_to_ros_float64、scalar_to_step_motor_gripper*）在调用时才import对应的ROS消息包，因此必须在装有对应ROS运行时的机器上运行；纯Python环境调用会抛ImportError。

机器人夹爪

scalar_to_step_motor_gripper配置项：open_threshold(80.0)、close_threshold(20.0)、id、speed、mode、angle、state、sub_divide。值落在两个阈值之间时返回None（不产生指令）。

启动即校验（fail fast）

构造ConfigDrivenMapper时（服务启动阶段）会对每条 rule 的每个transform调用validate_config并检查名字是否存在。配置错误在启动期暴露，不会拖到运行期。

运行期才会出现的错误：Unknown mapper transform: 'x'，以及各transformer自己的类型/形状校验异常。

开启日志

import logging
logging.getLogger("r2c_sdk.core.transformers").setLevel(logging.DEBUG)

entry_point的冲突与加载失败都以WARNING级别打在这个logger上。

快速定位某个值经过了什么

临时在rule上加identity收敛范围，或写单元测试直接调mapper：

mapper = ConfigDrivenMapper.from_rule_mappings([...])
print(mapper.map(payload))

加载扩展包的时序图

加载扩展包的时序图如图2所示。

加载转换器的优先级

图3 加载转换器的优先级图示

同名时内置永远胜出，并记录日志：

WARNING r2c_sdk.core.transformers: Transformer 'identity' from entry_point is ignored because a builtin with the same name exists. 

第三方名字请添加厂商/项目前缀，例如my_gripper_stroke_to_fingers。

实操：以examples/ur5e_transformer为示例

该示例把内置的平行夹爪“手指 ↔ 行程”(双向转换）换算在SDK之外重新实现了一遍（并换成项目自己的比例系数），是标准的第三方transformer包。

examples/ur5e_transformer/
├── pyproject.toml                        # entry_point 声明
├── my_parallel_gripper_transformers.py   # transformer 实现
└── README.md                             # 安装说明

第 1 步：实现类

my_parallel_gripper_transformers.py：

from dataclasses import dataclass
from typing import Any, Mapping
from r2c_sdk.core.interfaces import IValueTransformer
@dataclass(frozen=True)
class MyGripperFingersToStrokeTransformer(IValueTransformer):
    """两手指位置 -> 行程。"""
    def transform(self, value: Any, config: Any = None, context: Any = None) -> Any:
        left, right = value
        return (float(left) + float(right)) * 1250.0
@dataclass(frozen=True)
class MyGripperStrokeToFingersTransformer(IValueTransformer):
    """行程 -> 两手指位置（逆变换）。"""
    def transform(self, value: Any, config: Any = None, context: Any = None) -> Any:
        return [float(value) / 2500.0, float(value) / 2500.0]

第 2 步：声明entry_point

pyproject.toml：

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"
[project]
name = "my-parallel-gripper-r2c-transformers"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["hw-r2c-sdk"]
[tool.setuptools]
py-modules = ["my_parallel_gripper_transformers"]
[project.entry-points."r2c_sdk.transformers"]
my_gripper_fingers_to_stroke = "my_parallel_gripper_transformers:MyGripperFingersToStrokeTransformer"
my_gripper_stroke_to_fingers = "my_parallel_gripper_transformers:MyGripperStrokeToFingersTransformer"

要点：

• group名固定为r2c_sdk.transformers。
• 等号左边是用户写在YAML里的名字。
• 等号右边是 模块：类，必须是类，不是实例。
• 值用引号包起来（含 : 的TOML值）。

第 3 步：安装

pip install -e .          # 开发期（示例目录 README 用的是 pip install --no-build-isolation -e .）

若构建后端在当前环境不可用（受限网络 / 内网源），添加 --no-build-isolation 并预装 setuptools，即示例 README 的做法。uv 用户可直接 uv pip install -e .。

第 4 步：在YAML中使用

device_to_r2c:
  mappings:
    - target: "gripper_stroke"
      source: "joint_states.position"          # [left, right]
      transform: my_gripper_fingers_to_stroke
r2c_to_device:
  mappings:
    - target: "gripper_fingers"
      source: "gripper_stroke"
      transform: my_gripper_stroke_to_fingers

第 5 步：验证

from r2c_sdk.core.transformers import TransformerRegistry, build_transformer_registry
print([n for n in TransformerRegistry.available_transformers() if n.startswith("my_")])
reg = build_transformer_registry()
print(reg["my_gripper_fingers_to_stroke"].transform([0.02, 0.04]))  # 75.0
print(reg["my_gripper_stroke_to_fingers"].transform(75.0))          # [0.03, 0.03]

接口契约

class IValueTransformer(ABC):
    @abstractmethod
    def transform(self, value: Any, config: Any = None, context: Any = None) -> Any: ...
    @staticmethod
    def validate_config(config: Any) -> None:
        """在加载期校验配置，非法则抛 ValueError。默认空实现。"""

• 必须实现transform，签名固定三参数（context 可忽略）。
• validate_config可选覆盖；覆盖了就会在启动期被调用，可在那里做早失败检查。
• transform返回None是合法语义：该rule本帧不产生值（如夹爪去抖动、阈值中间态）。
• 推荐 @dataclass(frozen=True) 写无状态实现；需要内部状态（计数器等）时用普通类。

注意事项

打包发布

python -m build                     # 产出 wheel + sdist
pip install dist/xxx-0.1.0-py3-none-any.whl

发布到PyPI或内网源后，用户只需pip install <包名> 即可在YAML里按名引用。运行时不需要--extra-index之外的特殊配置，entry_point元数据随wheel一起安装。