# 处理器

处理器是机器人、数据集和策略之间的数据转换层。流水线是一系列`ProcessorStep`的链式组合；每个步骤都声明了它如何转换数据以及特征契约。

请参阅 [机器人处理器简介](../introduction_processors) 了解相关概念，[实现你自己的处理器](../implement_your_own_processor) 以编写一个步骤，并在流水线出现异常时参考 [调试你的处理器流水线](../debug_processor_pipeline)。

## ProcessorStep[[lerobot.processor.ProcessorStep]]

#### lerobot.processor.ProcessorStep[[lerobot.processor.ProcessorStep]]

```python
lerobot.processor.ProcessorStep()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L150)

数据处理流水线中单个步骤的抽象基类。

每个步骤必须实现 `__call__` 方法，以在数据转换上执行其变换，并实现 `transform_features` 方法，以描述它如何更改数据特征的形状或类型。

子类可以通过实现 `state_dict` 和 `load_state_dict` 来可选地支持有状态。

#### get_config[[lerobot.processor.ProcessorStep.get_config]]

```python
get_config()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L193)

**返回：**

一个可序列化为 JSON 的配置参数字典。

返回用于序列化的步骤配置。

#### load_state_dict[[lerobot.processor.ProcessorStep.load_state_dict]]

```python
load_state_dict(state: dict[str, torch.Tensor])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L209)

**参数：**

state : 状态张量字典。

从状态字典中加载该步骤的状态。

#### 重置[[lerobot.processor.ProcessorStep.reset]]

```python
reset()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L221)

如果存在，重置处理器步骤的内部状态。

#### save_artifacts[[lerobot.processor.ProcessorStep.save_artifacts]]

```python
save_artifacts(save_directory: Path)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L217)

保存非张量资产，并将构造函数参数映射到相对路径。

#### state_dict[[lerobot.processor.ProcessorStep.state_dict]]

```python
state_dict()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L201)

**返回：**

一个将状态名称映射到张量的字典。

返回步骤的状态（例如，学习到的参数、运行均值）。

#### transform_features[[lerobot.processor.ProcessorStep.transform_features]]

```python
transform_features(features: dict[PipelineFeatureType, dict[str, PolicyFeature]])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L225)

**参数：**

features : 一个字典，用于描述观测、动作等输入特征。

**返回：**

一个字典，用于描述此步骤转换后的输出特征。

定义此步骤如何修改流水线功能的描述。

此方法用于跟踪数据在流经管道过程中形状、数据类型或模态的变化，而无需处理实际数据。

## DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

#### lerobot.processor.DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

```python
lerobot.processor.DataProcessorPipeline(steps: Sequence[ProcessorStep] = <factory>, name: str = 'DataProcessorPipeline', to_transition: Callable[[TInput], EnvTransition] = <factory>, to_output: Callable[[EnvTransition], TOutput] = <factory>, before_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory>, after_step_hooks: list[Callable[[int, EnvTransition], None]] = <factory>)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L273)

**参数：**

steps : 构成流水线的 `ProcessorStep` 对象序列。

name : 管道的描述性名称。

to_transition : 一个将原始输入数据转换为标准化`EnvTransition`格式的函数。

to_output : 一个用于将最终的 `EnvTransition` 转换为所需输出格式的函数。

before_step_hooks : 一个函数列表，将在每个步骤执行之前被调用。

after_step_hooks : 一个函数列表，将在每个步骤执行后被调用。

用于处理数据的顺序流水线，并与 Hugging Face Hub 集成。

此类将多个 `ProcessorStep` 实例串联起来，以形成完整的数据处理工作流。它是通用的，支持自定义的输入和输出类型，这些类型由 `to_transition` 和 `to_output` 转换器处理。

#### from_config[[lerobot.processor.DataProcessorPipeline.from_config]]

```python
from_config(config: dict[str, Any], state_dict: dict[str, dict[str, torch.Tensor]] | None = None, overrides: dict[str, Any] | None = None, to_transition: Callable[[TInput], EnvTransition] | None = None, to_output: Callable[[EnvTransition], TOutput] | None = None)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L792)

**参数：**

config : 一个配置字典，其结构与保存的处理器 JSON 相同。

state_dict : 可选的按无后缀状态键分组的内存中的流水线状态。

overrides : 可选的构造函数覆盖，以注册表名称或类名称作为键。

to_transition : 可选的转换器，用于将输入数据转换为`EnvTransition`。

to_output : 可选的从 `EnvTransition` 到输出数据的转换器。

**返回：**

由配置和可选状态构建的处理器流水线。

从内存中的配置和可选的状态张量构建流水线。

#### from_pretrained[[lerobot.processor.DataProcessorPipeline.from_pretrained]]

```python
from_pretrained(pretrained_model_name_or_path: str | Path, config_filename: str, force_download: bool = False, resume_download: bool | None = None, proxies: dict[str, str] | None = None, token: str | bool | None = None, cache_dir: str | Path | None = None, local_files_only: bool = False, revision: str | None = None, overrides: dict[str, Any] | None = None, to_transition: Callable[[TInput], EnvTransition] | None = None, to_output: Callable[[EnvTransition], TOutput] | None = None, **kwargs)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L647)

**参数：**

pretrained_model_name_or_path : Hugging Face Hub 上仓库的标识符、本地目录的路径或单个配置文件的路径。

config_filename : 流水线 JSON 配置文件的名称。当存在多个配置（例如，preprocessor 与 postprocessor）时，始终需要指定此名称以避免歧义。

force_download : 是否强制（重新）下载文件。

resume_download : 为保持向后兼容性而保留，但会被忽略。下载会自动恢复。

proxies : 为保持向后兼容性而保留，但会被忽略。请改用 HTTP_PROXY 和 HTTPS_PROXY。

token : 用于私有 Hub 仓库的 HTTP 承载者授权令牌。

cache_dir : 用于存储下载文件的特定缓存文件夹的路径。

local_files_only : 如果为 True，则避免从 Hub 下载文件。

revision : 要使用的特定模型版本（例如，分支名称、标签名称或提交 ID）。

overrides : 一个用于覆盖特定步骤配置的字典。键应与步骤的类名或注册表名称匹配。

to_transition : 用于将输入数据转换为 `EnvTransition` 的自定义函数。

to_output : 一个自定义函数，用于将最终的 `EnvTransition` 转换为输出格式。

- ****kwargs**：其他参数（未使用）。

**返回：**

使用指定配置和状态加载的 `DataProcessorPipeline` 实例。

**引发：** ``FileNotFoundError`` 或 ``ValueError`` 或 ``ImportError`` 或 ``KeyError`` 或 ``ProcessorMigrationError``

- ``FileNotFoundError`` -- 如果找不到配置文件。
- ``ValueError`` -- 如果配置存在歧义或实例化失败。
- ``ImportError`` -- 如果某个步骤的类无法导入。
- ``KeyError`` -- 如果覆盖键与流水线中的任何步骤都不匹配。
- ``ProcessorMigrationError`` -- 如果模型需要迁移到处理器格式。

从本地目录、单个文件或 Hugging Face Hub 仓库加载 pipeline。

此方法实现了一个简化的加载流程，并具备智能迁移检测功能：

**简化加载策略**：
1. **配置加载** (_load_config):
   - **目录**：从目录中加载指定的 config_filename
   - **单个文件**：直接加载文件（忽略 config_filename）
   - **Hub 仓库**：从 Hub 下载指定的 config_filename

2. **配置验证** (_validate_loaded_config):
   - 格式验证：确保 config 是有效的处理器格式
   - 迁移检测：引导用户迁移旧版 LeRobot 模型
   - 清除错误：提供可操作的错误信息

3. **步骤构建** (_build_steps_with_overrides):
   - 类解析：注册表查找或动态导入
   - 覆盖合并：用户参数会覆盖已保存的配置
   - 状态加载：为有状态步骤加载 .safetensors 文件

4. **覆盖验证** (_validate_overrides_used):
   - 确保所有用户覆盖项均已应用（捕获拼写错误）
   - 提供包含可用键的有用错误消息

**迁移检测**：
- **智能检测**：分析 JSON 文件以检测旧版 LeRobot 模型
- **精准定位**：避免在其他 HuggingFace 模型上出现误报
- **清晰的指导**：提供要运行的确切迁移命令
- **错误模式**：始终引发 ProcessorMigrationError，以便用户明确执行操作

**加载示例**：
```python
# Directory loading
pipeline = DataProcessorPipeline.from_pretrained("/models/my_model", config_filename="processor.json")

# Single file loading
pipeline = DataProcessorPipeline.from_pretrained(
    "/models/my_model/processor.json", config_filename="processor.json"
)

# Hub loading
pipeline = DataProcessorPipeline.from_pretrained("user/repo", config_filename="processor.json")

# Multiple configs (preprocessor/postprocessor)
preprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_preprocessor.json"
)
postprocessor = DataProcessorPipeline.from_pretrained(
    "model", config_filename="policy_postprocessor.json"
)
```

**覆盖系统**：
- **关键匹配**：使用注册表名称或类名称作为覆盖键
- **配置合并**：用户覆盖项优先于已保存的配置
- **验证**：确保所有覆盖键与实际步骤匹配（捕获拼写错误）
- **示例**：overrides={"NormalizeStep": {"device": "cuda"}}

#### get_config[[lerobot.processor.DataProcessorPipeline.get_config]]

```python
get_config()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L451)

**返回：**

一个字典，其内容与 `save_pretrained()` 写入的 JSON 相同。

返回可 JSON 序列化的流水线配置。

#### load_state_dict[[lerobot.processor.DataProcessorPipeline.load_state_dict]]

```python
load_state_dict(state_dict: dict[str, dict[str, torch.Tensor]])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L515)

**参数：**

state_dict : 一个字典，将无后缀的状态键映射到步骤状态字典。

**引发：** ``KeyError``

- ``KeyError`` -- 如果加载时发现有缺失的预期状态或意外的额外状态。

将流水线状态张量加载到现有步骤中。

#### process_action[[lerobot.processor.DataProcessorPipeline.process_action]]

```python
process_action(action: PolicyAction | RobotAction | EnvAction)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1786)

**参数：**

action : 操作数据。

**返回：**

处理后的操作。

仅通过流水线处理转换中的动作部分。

#### process_complementary_data[[lerobot.processor.DataProcessorPipeline.process_complementary_data]]

```python
process_complementary_data(complementary_data: dict[str, Any])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1853)

**参数：**

complementary_data : 互补数据字典。

**返回：**

处理后的互补数据字典。

仅通过管道处理转换的互补数据部分。

#### process_done[[lerobot.processor.DataProcessorPipeline.process_done]]

```python
process_done(done: bool | torch.Tensor)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1814)

**参数：**

done : 完成标志。

**返回：**

已处理的完成标志。

仅通过流水线处理 transition 的 done 标志。

#### process_info[[lerobot.processor.DataProcessorPipeline.process_info]]

```python
process_info(info: dict[str, Any])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1840)

**参数：**

info : 信息字典。

**返回：**

处理后的信息字典。

仅通过流水线处理转换的信息字典。

#### process_observation[[lerobot.processor.DataProcessorPipeline.process_observation]]

```python
process_observation(observation: RobotObservation)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1773)

**参数：**

observation : 观测字典。

**返回：**

处理后的观测字典。

仅通过流水线处理转移中的观察部分。

#### process_reward[[lerobot.processor.DataProcessorPipeline.process_reward]]

```python
process_reward(reward: float | torch.Tensor)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1801)

**参数：**

reward : 奖励值。

**返回：**

处理后的奖励。

仅通过流水线处理转移中的奖励部分。

#### process_truncated[[lerobot.processor.DataProcessorPipeline.process_truncated]]

```python
process_truncated(truncated: bool | torch.Tensor)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1827)

**参数：**

truncated : 截断标志。

**返回：**

处理后的截断标志。

仅处理通过管道时的截断标志。

#### register_after_step_hook[[lerobot.processor.DataProcessorPipeline.register_after_step_hook]]

```python
register_after_step_hook(fn: Callable[[int, EnvTransition], None])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1697)

**参数：**

fn : 一个可调用对象，接受步数索引和当前转换。

注册一个函数，以便在每一步之后调用。

#### register_before_step_hook[[lerobot.processor.DataProcessorPipeline.register_before_step_hook]]

```python
register_before_step_hook(fn: Callable[[int, EnvTransition], None])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1673)

**参数：**

fn : 一个可调用对象，接受步数索引和当前转换。

注册一个函数，以便在每一步之前调用。

#### 重置[[lerobot.processor.DataProcessorPipeline.reset]]

```python
reset()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1721)

重置管道中所有有状态步骤的状态。

#### save_pretrained[[lerobot.processor.DataProcessorPipeline.save_pretrained]]

```python
save_pretrained(save_directory: str | Path | None = None, repo_id: str | None = None, push_to_hub: bool = False, card_kwargs: dict[str, Any] | None = None, config_filename: str | None = None, **push_to_hub_kwargs)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L595)

**参数：**

save_directory : 流水线将保存到的目录。如果为 None，则保存到 HF_LEROBOT_HOME/processors/{sanitized_pipeline_name}。

repo_id : 您在 Hub 上的仓库 ID。仅在 `push_to_hub=true` 时使用。

push_to_hub : 保存后是否将对象推送到 Hugging Face Hub。

card_kwargs : 传递给卡片模板以自定义卡片的额外参数。

config_filename : JSON 配置文件的名称。如果为 None，则根据流水线的 `name` 属性生成名称。

- ****push_to_hub_kwargs** : Additional key word arguments passed along to the push_to_hub method.

将流水线的配置和状态保存到目录中。

此方法会创建一个 JSON 配置文件，用于定义流水线的结构（名称和步骤）。对于每个有状态的步骤，它还会保存一个包含其状态字典的 `.safetensors` 文件。

#### state_dict[[lerobot.processor.DataProcessorPipeline.state_dict]]

```python
state_dict()
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L488)

**返回：**

一个字典，将无后缀的状态键映射到克隆的步骤状态字典。

按状态键分组返回流水线状态张量。

#### step_through[[lerobot.processor.DataProcessorPipeline.step_through]]

```python
step_through(data: TInput)
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L343)

**参数：**

data : 输入数据。

**返回值：**

`EnvTransition` 对象，从初始状态开始，然后在每个处理步骤之后。

逐步处理数据，并在每个阶段输出转换结果。

这是一个生成器方法，可用于调试和检查数据在通过管道过程中的中间状态。

#### transform_features[[lerobot.processor.DataProcessorPipeline.transform_features]]

```python
transform_features(initial_features: dict[PipelineFeatureType, dict[str, PolicyFeature]])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1750)

**参数：**

initial_features : 描述初始特征的字典。

**返回：**

经过所有转换后的最终特征描述。

按顺序应用所有步骤中的特征转换。

此方法通过每个步骤的 `transform_features` 方法传播特征描述字典，使流水线能够在不处理任何实际数据的情况下静态确定输出特征规范。

#### unregister_after_step_hook[[lerobot.processor.DataProcessorPipeline.unregister_after_step_hook]]

```python
unregister_after_step_hook(fn: Callable[[int, EnvTransition], None])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1705)

**参数：**

fn : 之前注册的精确函数对象。

**引发：** ``ValueError``

- ``ValueError`` -- 如果在列表中未找到该钩子。

注销一个 'after_step' 钩子。

#### unregister_before_step_hook[[lerobot.processor.DataProcessorPipeline.unregister_before_step_hook]]

```python
unregister_before_step_hook(fn: Callable[[int, EnvTransition], None])
```

[源代码](https://github.com/huggingface/lerobot/blob/v0.7.0/src/lerobot/processor/pipeline.py#L1681)

**参数：**

fn : 之前注册的精确函数对象。

**引发：** ``ValueError``

- ``ValueError`` -- 如果在列表中未找到该钩子。

注销一个 'before_step' 钩子。

## PolicyProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

#### lerobot.processor.DataProcessorPipeline[[lerobot.processor.DataProcessorPipeline]]

```python
lerobot.processor.DataProcessorPipeline(*args, **kwargs)
```