有时候，在用于训练之前，录制本身需要改一改：开头操作员伸手去按键盘的那几秒、一路从来不用的相机、一条帧率不对的数据流。变换算法返回一条新的录制，只替换或裁掉这些部分。你没碰的数据流按引用带过去，不会复制，所以小改动的代价也很小。
变换算法每条录制运行一次。它读取 recording，并通过 out（一个 RecordingEditor）写出 recording：
| 方法 | 作用 |
|---|---|
out.unchanged() | 没有需要做的事时，不复制地把输入原样交出去。 |
out.replace_camera(key, frames=… 或 video=path, anonymized=None) | 替换一路相机。 |
out.drop_camera(key) | 去掉一路相机。 |
out.replace_column(name, values) | 替换一列。 |
out.trim(start_s, end_s) | 只保留一个时间范围。 |
每个变换都会把自己追加到 recording.json 的 lineage 里，并保留之前的所有记录，所以你随时能看到一条录制经历过什么。变换也可以通过 out.checks 写出质检结果。
示例：裁掉首尾
这个变换从开头和结尾各裁掉若干秒。两者都为 0 时，录制原样通过；裁完什么都不剩时，则拒绝。文件是 packages/sdk-python/examples/v1/transform-trim/main.py。
"""Transform: cut seconds off the start and the end of the recording.
In: the canonical recording.
Out: a recording with the same cameras and columns, without the first and last seconds. Lineage records the step.
Why: the first seconds of a teleoperation recording often hold the operator reaching for the keyboard.
With nothing to cut the recording is handed through untouched (zero copy).
"""
from dataclasses import dataclass
from embodiflow import Context, Recording, RecordingEditor, algorithm, param
@dataclass
class Params:
head_s: float = param(0.0, minimum=0.0, title="Seconds to cut at the start")
tail_s: float = param(0.0, minimum=0.0, title="Seconds to cut at the end")
@algorithm.transform(name="trim-ends", title="Trim the ends", params=Params)
def run(recording: Recording, out: RecordingEditor, ctx: Context[Params]) -> None:
head, tail = ctx.params.head_s, ctx.params.tail_s
if head == 0 and tail == 0:
out.unchanged()
return
if head + tail >= recording.duration_s:
ctx.input_unusable("TRIM_EMPTIES_RECORDING", f"cutting {head + tail:.2f} s leaves nothing of {recording.duration_s:.2f} s")
out.trim(head, recording.duration_s - tail)它的测试是 packages/sdk-python/examples/v1/transform-trim/test_main.py：用一份很小的测试数据，通过真实的运行器运行这段代码，并检查写出的值。想从它开始，运行 embodiflow algorithm init my-algorithm --category transform，它会写出 main.py 和 test_main.py；再运行 embodiflow algorithm test my-algorithm（需要装有 pytest 和 SDK 的 Python）。
值得知道
ctx.input_unusable(code, message) 让步骤以失败结束，并隔离这条录制。它用于无法修复的数据，示例里裁完什么都不剩时就是这样处理的。
脱敏是团队变换唯一不能「算数」的改动。把相机标记为已脱敏，属于声明了 privacy: redacts 的变换，而只有内置算法可以这样声明。所以团队变换永远不会让一条录制算作已脱敏。
在处理工作流里，标准化之后每个读取录制的步骤，读到的都是内置脱敏步骤的输出。请把你的变换放在脱敏之后；放在它前面，工作流会被 WORKFLOW_PRIVACY_BARRIER 拒绝。