SDK 参考 · EmbodiFlow 文档

本页由浏览器编辑器做补全时用的同一份 SDK 索引生成，所以始终和你运行的 SDK 一致。所有内容都从 embodiflow 导入：

from embodiflow import Context, Feature, FeatureWriter, Recording, algorithm, param

SDK 自带类型提示（py.typed），你自己的编辑器也能补全；每个类和方法都有文档字符串，浏览器编辑器会在鼠标悬停时显示。下面的说明就是这些文档字符串的中文译文；参数名、类型和代码保持英文。

测试阶段。自定义算法和工作流已经在产品里，可以直接试用，但还没有在真实数据上完成完整验收。可能会遇到不完善的地方，依赖结果之前请先核对。

装饰器

@algorithm.standardize

声明一个标准化算法：把一条 raw 录制转换成标准的 recording。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`description`	`str \| None`	否	`None`
`checks`	`list[Check] \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`
`services`	`list[str] \| None`	否	`None`

@algorithm.check

声明一个质检算法：判断数据，但从不修改数据。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`checks`	`list[Check]`	是
`description`	`str \| None`	否	`None`
`level`	`str \| None`	否	`None`
`reads`	`str \| None`	否	`None`
`inputs`	`dict[str, Input] \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`
`services`	`list[str] \| None`	否	`None`

@algorithm.transform

声明一个变换算法：返回一条替换或裁剪了部分数据流的录制。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`description`	`str \| None`	否	`None`
`inputs`	`dict[str, Input] \| None`	否	`None`
`checks`	`list[Check] \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`
`services`	`list[str] \| None`	否	`None`
`privacy`	`str \| None`	否	`None`

@algorithm.enrich

声明一个增强算法：计算新的信息，但不修改录制。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`features`	`list[Feature]`	是
`description`	`str \| None`	否	`None`
`level`	`str \| None`	否	`None`
`inputs`	`dict[str, Input] \| None`	否	`None`
`checks`	`list[Check] \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`
`services`	`list[str] \| None`	否	`None`
`privacy`	`str \| None`	否	`None`

@algorithm.segment

声明一个切分算法：提出任务片段，即带有任务文字的时间范围。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`description`	`str \| None`	否	`None`
`labels`	`dict[str, str] \| None`	否	`None`
`inputs`	`dict[str, Input] \| None`	否	`None`
`checks`	`list[Check] \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`
`services`	`list[str] \| None`	否	`None`
`segment_source`	`str \| None`	否	`None`

@algorithm.curate

声明一个整理算法：逐个任务片段决定数据集版本是否保留它、它的划分和任务。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`description`	`str \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`

@algorithm.export

声明一个导出算法：把数据集版本写成可下载的文件。

参数	类型	必填	默认值
`name`	`str`	是
`title`	`str`	是
`description`	`str \| None`	否	`None`
`requires`	`dict[str, Any] \| None`	否	`None`
`params`	`type \| None`	否	`None`
`resources`	`dict[str, Any] \| None`	否	`None`
`requirements`	`list[str] \| None`	否	`None`
`retries`	`int \| None`	否	`None`
`cache`	`bool \| None`	否	`None`

函数

param(default, *, minimum=None, maximum=None, enum=None, title=None, description=None, widget=None)

声明 Params 数据类中的一个字段，包括默认值和取值范围。

类

ArtifactWriter

收集导出的文件（embodiflow-artifact/v1）。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。

方法	含义
`format(name: str) -> None`	指定文件的格式（`lerobot-v3`、`csv` 等）。
`path(rel: str) -> Path`	产物中要写入的路径（会自动创建父目录）。
`add(src: str \| Path, rel: str) -> None`	把文件或文件夹 `src` 复制到产物的 `rel` 位置。
`summary(doc: dict[str, Any]) -> None`	描述本次导出（数量、大小等），显示在下载页面上。

Camera

录制中的一个相机。

属性	含义
`anonymized: bool`
`distortion: list[float] \| None`
`height: int`
`intrinsics: list[list[float]] \| None`
`key: str`
`model: str \| None`
`role: str`
`video_path: Path`	录制目录中该相机的 mp4 文件。
`width: int`

方法	含义
`valid() -> 'np.ndarray'`	每帧一个布尔值：相机有缺帧的位置为 `False`，视频中对应位置是黑色占位帧。
`frames(start: int=0, stop: int \| None=None, step: int=1, gray: bool=False, size: tuple[int, int] \| None=None) -> Iterator[Frame]`	按 `step` 解码 `start..stop` 之间的帧（索引相对于本录制或任务片段窗口）。
`frame(i: int) -> Frame`	解码第 `i` 帧。

Check

算法可以通过 CheckWriter.result 报告的一个质检结果 id。

属性	含义
`description: str`
`id: str`
`title: str`

方法	含义
`to_dict() -> dict[str, Any]`	作为清单输入的声明形式。

CheckWriter

收集质检结果（embodiflow-checks/v1）。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。

方法	含义
`result(id: str, status: str, value: float \| None=None, threshold: float \| None=None, message: str='', evidence: Sequence[str \| Path] \| None=None, at: Mapping[str, Any] \| None=None) -> None`	报告一条结果。
`evidence_path(name: str) -> Path`	`evidence/` 下的路径，用来写入图片或文件；传给 `result(evidence=[...])`。
`metrics(doc: dict[str, Any]) -> None`	附加一个 `embodiflow-quality-metrics/v1` 内容（仅限内置质检）。

Column

录制中的一个表格列：name、dtype、shape、names，以及约定字段 unit、frame、source。

属性	含义
`dtype: str`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`shape: list[int]`
`source: str \| None`
`type: str \| None`
`unit: str \| None`

Context

运行时可以获取的全部信息：参数、事实、标识符、临时空间、日志和进度。

属性	含义
`facts: Facts`
`log: logging.Logger`
`params: P`
`platform_params: dict[str, Any]`	平台为标准化步骤解析出的内容（`rigProfiles`、`teamDataFormat`），否则为空。
`recording_id: str \| None`	本步骤处理的录制（数据集级步骤为 `None`）。
`run_id: str`	工作流运行 id。
`step_id: str`	本步骤在工作流中的 id。
`team_slug: str \| None`	团队的存储标识（用于识别团队功能）；作业未声明时为 `None`。
`tmp: Path`

方法	含义
`progress(done: int, total: int, message: str='') -> None`	向运行视图报告进度（进度条和预计剩余时间）；可以频繁调用。
`not_applicable(code: str, message: str) -> NoReturn`	以「不适用」结束本步骤（退出码 85，运行视图中显示为灰色，不算失败）。
`input_unusable(code: str, message: str) -> NoReturn`	因数据损坏结束本步骤（退出码 87）：步骤失败，录制会被隔离。

DatasetVersion

冻结的数据集版本：id、features（声明的名称和类型）和 members。

属性	含义
`document: dict[str, Any]`
`features: dict[str, dict[str, Any]]`
`id: str`
`members: list[Member]`

方法	含义
`to_pandas() -> 'pd.DataFrame'`	每个成员一行：id、录制、起止与时长、任务、划分、标签、质检状态、会话和操作员。

Episode

录制中的一个任务片段：index、start_s、end_s（录制时间轴）、task、labels、confidence。

属性	含义
`confidence: float \| None`
`duration_s: float`	长度，单位为秒。
`end_s: float`
`index: int`
`labels: dict[str, Any]`
`start_s: float`
`task: str \| None`

EpisodeWriter

收集提出的任务片段（embodiflow-episodes/v1）。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。

方法	含义
`episode(start_s: float, end_s: float, task: str, labels: Mapping[str, Any] \| None=None, confidence: float \| None=None) -> None`	添加一个任务片段：`startS`/`endS` 为秒，`task` 为任务文字，可选 `labels`，`confidence` 取值 0 到 1。

Facts

主输入的事实（契约 2.1），可以用属性访问，例如 ctx.facts.kind、ctx.facts.stereo。

属性	含义
`action: bool \| None`
`cameras: int \| None`
`duration_s: float \| None`
`episodes: int \| None`
`format: str \| None`
`gripper_width: bool \| None`
`imu: bool \| None`
`kind: str \| None`
`privacy_required: bool \| None`
`recorded_hand_pose: bool \| None`
`stereo: bool \| None`

方法	含义
`get(name: str, default: Any=None) -> Any`	某个事实的值；未知时返回 `default`。
`as_dict() -> dict[str, Any]`	所有已知事实，以普通字典返回。

Feature

enrich 算法声明的一个特征，之后通过 FeatureWriter 写入。

属性	含义
`describe: dict[str, Any] \| None`
`dtype: str \| None`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`per: str`
`shape: list[int] \| None`
`source: str \| None`
`type: str`
`unit: str \| None`

方法	含义
`to_dict() -> dict[str, Any]`	作为清单输入的声明形式：只包含传入过的参数。

FeatureWriter

收集 enrich 算法计算的特征（embodiflow-features/v1）。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。

方法	含义
`per_frame(name: str, values: Any, confidence: str \| None=None) -> None`	每帧写入一行：`values` 的形状为 `(num_frames, *shape)`（标量为一维数组）。
`per_recording(name: str, value: Any) -> None`	为整条录制写入一个值（JSON，最大 1 MiB）。
`per_episode(name: str, value: Any) -> None`	写入正在处理的任务片段的值（仅限任务片段级的 `enrich`）。
`file(name: str, path: str \| Path, describe: Mapping[str, Any] \| None=None) -> None`	为 `per="file"` 的特征附加一个文件（例如深度视频）。

Features

features 输入（包含 features.json、frames.parquet、episodes.parquet 和 files/）。

属性	含义
`document: dict[str, Any]`
`names: list[str]`	所有特征的名称。
`num_frames: int`

方法	含义
`info(name: str) -> dict[str, Any]`	`name` 对应的 `features.json` 条目（类型、per、dtype、shape、unit、frame、source、describe）。
`of_type(semantic_type: str) -> list[str]`	语义类型为 `semantic_type` 的特征名称。
`per_frame(name: str) -> 'np.ndarray'`	逐帧特征的值，形状为 `(num_frames, *shape)`；任务片段窗口之外的行为 NaN/None。
`per_recording(name: str) -> Any`	逐录制特征的值。
`per_episode(name: str) -> dict[int, Any]`	逐任务片段特征的 `{episodeIndex: value}`。
`file(name: str) -> Path`	逐文件特征（例如深度视频）的路径。

Frame

一帧解码后的视频：index（录制或片段窗口内的索引）、time_s 和 image（RGB 的 H x W x 3 uint8；gray 时为 H x W）。

属性	含义
`image: 'np.ndarray'`
`index: int`
`time_s: float`

Input

额外的 features 输入（例如标定、手部姿态）。

属性	含义
`optional: bool`
`requires: list[str]`

方法	含义
`to_dict() -> dict[str, Any]`	作为清单输入的声明形式。

InputUnavailableError

算法请求的输入不存在（端口缺失、文件或下载不存在）。

ManifestDriftError

代码的接口与本版本冻结的清单不一致。

ManifestError

算法接口无效（ALGORITHM_FILE_INVALID 及同类错误码）。

McapMessage

解码后的 ROS 2 消息：topic、schema、log_time_s（从文件第一条消息起的秒数）。

McapReader

通过摘要和查询所需的数据块读取的 MCAP 文件。

属性	含义
`duration_s: float`
`end_ns: int`
`start_ns: int`
`topics: list[McapTopic]`

方法	含义
`messages(topics: Sequence[str] \| None=None, start_s: float \| None=None, end_s: float \| None=None) -> Iterator[McapMessage]`	按日志时间排序的解码消息；`topics` 按名称过滤，`start_s`/`end_s` 限定日志时间。

McapTopic

MCAP 文件中的一个话题：name、消息 schema 名称、消息数 count 和平均频率 rate（Hz）。

属性	含义
`count: int`
`name: str`
`rate: float`
`schema: str`

Member

数据集版本中的一个任务片段。

属性	含义
`duration_s: float`	任务片段的长度，单位为秒。
`end_s: float`
`episode_id: str`
`features: dict[str, Any]`
`labels: dict[str, Any]`
`operator: str \| None`
`quality: dict[str, Any]`
`recording_id: str`
`root: str`
`session: str \| None`
`split: str \| None`
`start_s: float`
`task: str \| None`

方法	含义
`recording() -> Recording`	任务片段所在录制的窗口；首次使用时会下载其文件。

OutputError

写入器拒绝了算法给出的内容。

RawFile

一个上传的文件：path（相对于上传根目录）、size（字节）和磁盘上的 local_path。

属性	含义
`local_path: Path`
`path: str`
`size: int`

方法	含义
`open(mode: str='rb') -> IO[Any]`	打开文件（只读，默认为二进制）。

RawFiles

一条录制上传的全部文件，使用它们相对上传目录的路径。

属性	含义
`files: list[RawFile]`

方法	含义
`find(pattern: str) -> list[RawFile]`	路径匹配通配符 `pattern` 的文件（`` 匹配单层目录，`*` 跨目录匹配）。
`read_json(path: str) -> Any`	解析 `path` 处的 JSON 文件。
`read_yaml(path: str) -> Any`	解析 `path` 处的 YAML 文件。
`mcap(path: str \| None=None) -> 'McapReader'`	打开 MCAP 文件；上传中只有一个文件时可以省略 `path`。

Recording

标准录制（LeRobot v3 目录树加 meta/embodiflow/recording.json），或它的一段时间窗口。

属性	含义
`calibration: dict[str, Any]`	约定侧车文件中的标定：按原样保存的 `frames`、`stereoPairs` 和 `timeOffsets`。
`cameras: list[Camera]`
`columns: list[Column]`
`conventions: dict[str, Any]`
`duration_s: float`	长度，单位为秒。
`facts: dict[str, Any]`
`fps: float`
`info: dict[str, Any]`
`kind: str \| None`
`meta: dict[str, Any]`
`num_frames: int`
`start_s: float`	本录制（或任务片段窗口）在录制时间轴上的起点，单位为秒。
`stereo_pairs: list[dict[str, str]]`
`task: str \| None`
`total_frames: int`

方法	含义
`window(start_s: float, end_s: float) -> 'Recording'`	本录制中 `start_s` 到 `end_s` 之间的部分（秒，以本录制自己的时间轴计）。
`table(columns: Sequence[str] \| None=None) -> 'pa.Table'`	帧表（本录制或窗口每帧一行），为 `pyarrow.Table`。
`column(name: str) -> 'np.ndarray'`	一列数据，为形状 `(num_frames, *shape)` 的 numpy 数组。

RecordingBuilder

从源格式构建标准录制。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。
`root: Path`	`recording` 端口的输出目录（配合 :meth:`adopt_tree` 使用）。

方法	含义
`adopt_tree() -> None`	声明本次运行已自行把完整的标准目录树写入 `out.root`（内置标准化算法自带目录树写入器）。
`fps(value: int) -> None`	声明录制的帧率：`30`（平台网格；其他帧率会被拒绝）。
`task(text: str) -> None`	设置录制的任务文字（一条录制一个任务）。
`camera(key: str, role: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, intrinsics: Sequence[Sequence[float]] \| None=None, distortion: Sequence[float] \| None=None, model: str \| None=None) -> None`	从 RGB `frames`（形状 `N x H x W x 3` 的 uint8 数组，或逐帧的 `H x W x 3` 迭代器）或 `video` 文件添加一个相机。
`column(name: str, values: Any, type: str='vector', unit: str='none', frame: str='none', source: str='recorded', names: Sequence[str] \| None=None) -> None`	添加一个表格列，每帧一行：`values` 的形状为 `(N,)` 或 `(N, ...)`。
`timestamps(seconds: Any) -> None`	记录每一行的源采集时间（秒，非递减）；与录制一并保存。
`stereo_pair(left: str, right: str) -> None`	把两个已添加的相机声明为一个立体对（键名或简称）。
`dropped(what: str, reason: str) -> None`	声明源数据中未进入录制的内容，以及原因。

RecordingEditor

编辑输入录制：替换或删除相机、替换列、裁剪。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。
`root: Path`	`recording` 端口的输出目录（配合 :meth:`adopt_tree` 使用）。

方法	含义
`adopt_tree() -> None`	声明本次运行已自行把完整的编辑后目录树写入 `out.root`（内置变换自带视频处理路径，例如使用硬件编码器的脱敏）。
`unchanged() -> None`	原样透传输入录制（零拷贝）。
`replace_camera(key: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, anonymized: bool \| None=None) -> None`	用 `frames`（逐帧的 `H x W x 3` uint8 图像迭代器，每输入帧一张）或 `video` 替换相机 `key` 的画面。
`drop_camera(key: str) -> None`	从录制中删除相机 `key`。
`replace_column(name: str, values: Any) -> None`	替换列 `name` 的值；`values` 的形状与现有列相同（每输入帧一行）。
`trim(start_s: float, end_s: float) -> None`	只保留录制中 `start_s` 到 `end_s`（秒）的部分。

SelectionWriter

收集逐任务片段的决定（embodiflow-selection/v1）。

属性	含义
`checks: 'CheckWriter'`	本算法声明的质检结果对应的 `CheckWriter`（`checks=[Check(...)]`）。

方法	含义
`keep(episode_id: str, split: str \| None=None, task: str \| None=None, reason: str \| None=None) -> None`	保留该任务片段，可选指定 `split`（`train`/`val`/`test`），并替换它的 `task` 文字。
`count(name: str, value: int) -> None`	在选择摘要（`summary.counts`）中报告一个命名计数，例如 `count("grouped_by_recording", 3)`。
`drop(episode_id: str, reason: str) -> None`	从新版本中删除该任务片段；`reason` 会显示给团队。

SetupContext

setup(ctx) 能看到的内容：参数、事实、临时目录和日志记录器。

属性	含义
`facts: Facts`
`log: logging.Logger`
`params: Any`
`run_id: str`	工作流运行 id。
`step_id: str`	本步骤在工作流中的 id。
`tmp: Path`

TransientError

当重试可能成功时抛出（网络、GPU 忙碌）：步骤以退出码 88 结束并会被重试。

词表

名称	允许的值
`categories`	`standardize`, `check`, `transform`, `enrich`, `segment`, `curate`, `export`
`levels`	`recording`, `episode`, `dataset`
`statuses`	`pass`, `warn`, `fail`, `error`, `not_evaluated`, `not_applicable`
`per`	`frame`, `recording`, `episode`, `file`
`sources`	`recorded`, `imported`, `estimated`, `computed`, `retargeted`
`dtypes`	`float32`, `float64`, `int8`, `int16`, `int32`, `int64`, `uint8`, `uint16`
`splits`	`train`, `val`, `test`
`semanticTypes`	`scalar`, `vector`, `label.bool`, `label.category`, `label.text`, `spans`, `embedding`, `pose.se3`, `keypoints3d.hand21`, `skeleton.hand21`, `skeleton.body22`, `keypoints2d`, `confidence`, `calibration.stereo`, `calibration.cam_imu`, `depth.video`, `mask.video`, `video.preview`, `ros2.mcap`, `action.joint`, `action.ee_pose`, `json`
`facts`	`kind`, `format`, `privacy_required`, `duration_s`, `cameras`, `stereo`, `imu`, `action`, `recorded_hand_pose`, `gripper_width`, `episodes`
`privacy`	`redacts`, `numbers_only`
`services`	`vlm`
`runParameters`	`raw`, `recording`, `version`, `episode`, `out`, `ctx`, `state`

下一步