-
Notifications
You must be signed in to change notification settings - Fork 12
Custom Transition
过渡插件负责状态切换期间如何生成电机目标。每种过渡是一个独立类和独立 Python 文件,由框架自动发现。
相关路径:
bxi_example_py_elf3/transitions/ # 一种过渡一个文件
bxi_example_py_elf3/utils/transition_core.py # 共享类型、协议和插件基础设施
新增过渡不需要修改状态机、注册表、__init__.py 或构建脚本。
请求 source -> target 时:
plan.validate_states(source, target)
target.on_prepare(ctx, source)
plan.create_session(ctx, source, target)
每个控制周期:
session.update(ctx, dt)
完成后:
source.on_exit(ctx)
current = target
target.on_enter(ctx)
过渡期间 source 仍是当前状态。安全 event 可以中断活动过渡:
target.on_prepare_cancel(ctx, source)
Session 创建或执行抛出异常时也会走取消回调,当前状态仍保持为源状态。
零时长 Session 会执行一次 update(ctx, 0.0) 后提交,而不是完全跳过执行。
新建:
bxi_example_py_elf3/transitions/fixed_hold.py
from __future__ import annotations
from collections.abc import Mapping
from typing import TYPE_CHECKING
from bxi_example_py_elf3.utils.transition_core import (
ConfigReader,
MotorFrame,
SingleClassTransition,
)
if TYPE_CHECKING:
from bxi_example_py_elf3.bxi_example_demo import BxiExample
from bxi_example_py_elf3.utils.state_machine import StateBehavior
class FixedHoldTransition(SingleClassTransition):
type_name = "fixed_hold"
def __init__(self, name: str, duration: float):
super().__init__(name, duration)
self._frame: MotorFrame | None = None
@classmethod
def from_config(
cls,
name: str,
raw: Mapping[str, object],
) -> "FixedHoldTransition":
reader = ConfigReader(raw, name)
duration = reader.float("duration", minimum=0.0)
reader.finish()
return cls(name, duration)
def on_start(
self,
ctx: "BxiExample",
from_state: "StateBehavior[BxiExample]",
to_state: "StateBehavior[BxiExample]",
) -> None:
self._frame = MotorFrame.create(ctx.pos_last, ctx.kp_last, ctx.kd_last)
def apply(self, ctx: "BxiExample", dt: float, progress: float) -> None:
frame = self._frame
if frame is None:
raise RuntimeError("fixed hold transition has not started")
ctx.set_motor_target(frame.qpos, frame.kp, frame.kd)启动时自动发现模块,并把 type_name = "fixed_hold" 注册为 YAML 的 type。
声明 profile:
transition_profiles:
my_hold:
type: fixed_hold
duration: 0.1使用 profile:
transitions:
on_event:
normal_event:
to: normal
transition: my_hold也可以直接内联:
transition:
type: fixed_hold
duration: 0.1如果需要基于 profile 覆盖:
transition:
profile: my_hold
duration: 0.3ConfigReader 提供有类型的字段读取:
duration = reader.float("duration", minimum=0.0)
enabled = reader.boolean("enabled", default=True)
curve = reader.literal(
"curve",
("linear", "smoothstep"),
default="smoothstep",
)
steps = reader.mappings("steps")
reader.finish()finish() 会拒绝未读取字段。新增插件应始终调用它,让 YAML 拼写错误在启动时暴露。
多数过渡直接继承 SingleClassTransition:
- 配置编译后得到一个不可运行的 plan 原型。
- 每次切换使用浅复制创建独立 Session。
-
on_start()保存本次切换的运行数据。 -
apply()每周期输出。 -
duration、elapsed、progress已由基类实现。
不要在 from_config() 中读取 ctx 或状态实例;它只负责验证配置。
复杂过渡可以直接继承 TransitionPlugin,自行返回实现 TransitionPlan 和 TransitionSession 的强类型对象。仍然只需添加一个插件文件。
过渡不应要求所有状态实现专用函数。只让真正使用该过渡的状态实现一个可选 Protocol。
当前共享能力位于 transition_core.py:
class EntryFrameProvider(Protocol):
def get_entry_frame(self, ctx: BxiExample) -> MotorFrame:
...
class RunningFrameProvider(Protocol):
def sample_running_frame(
self,
ctx: BxiExample,
dt: float,
*,
advance: bool,
) -> MotorFrame | None:
...状态只在需要时继承并实现:
class SinWaveState(RobotControlState, EntryFrameProvider):
def get_entry_frame(self, ctx: BxiExample) -> MotorFrame:
return self._motor_frame(self.target, ctx.joint_kp, ctx.joint_kd)
def on_update(self, ctx: BxiExample, dt: float) -> None:
ctx.set_motor_target(self.target, ctx.joint_kp, ctx.joint_kd)过渡在图加载时验证:
def validate_states(self, from_state, to_state) -> None:
require_entry_frame_provider(to_state)在 Session 启动时得到已收窄类型:
provider = require_entry_frame_provider(to_state)
self._target = provider.get_entry_frame(ctx)如果能力只属于一个新插件,把 Protocol 和插件放在同一个文件,不要修改 RobotControlState 或公共核心:
from abc import abstractmethod
from typing import Protocol, runtime_checkable
@runtime_checkable
class ContactFrameProvider(Protocol):
@abstractmethod
def get_contact_frame(self, ctx: "BxiExample") -> MotorFrame:
...需要使用它的状态显式实现:
class ContactState(RobotControlState, ContactFrameProvider):
def get_contact_frame(self, ctx: BxiExample) -> MotorFrame:
return self._motor_frame(ctx.target_pos, ctx.joint_kp, ctx.joint_kd)
def on_update(self, ctx: BxiExample, dt: float) -> None:
...插件使用 isinstance(to_state, ContactFrameProvider) 检查并收窄类型。没有使用该插件的状态不需要知道这个函数。
多种插件开始共用同一能力时,再把 Protocol 提升到 transition_core.py。
transitions/instant.py
transitions/hold.py
transitions/entry_gain_ramp.py
transitions/running_blend.py
transitions/sequence.py
它们都通过同一自动发现机制加载,没有“内置过渡”分支。
sequence 可以复用任何已发现插件:
transition_profiles:
safe_entry:
type: sequence
steps:
- type: hold
duration: 0.02
- type: entry_gain_ramp
duration: 0.8
kp_from: zero
kd_from: target每个子 Session 在轮到它时才创建,避免提前获取尚未准备好的运行数据。
插件默认快照包含:
{
"name": "my_hold",
"type": "fixed_hold",
"duration": 0.1
}自定义字段通过 config_snapshot() 暴露:
def config_snapshot(self) -> dict[str, object]:
return {"curve": self._curve}不要把 Session 临时数据放入配置快照。
- 在
bxi_example_py_elf3/transitions/添加一个文件。 - 定义唯一的
type_name。 - 用
ConfigReader读取并严格验证配置。 - 实现
apply();需要运行数据时实现on_start()。 - 需要状态能力时实现
validate_states()和对应 Protocol。 - 在 YAML 中增加 profile 或直接内联使用。
- 启动时确认没有重复 type、未知字段或能力错误。
无需编辑其他注册代码。