Framework Internals

框架底层运行原理

本页解释框架运行时到底发生了什么。理解这一页后，写自定义 driver、自定义过渡、复杂 YAML 时会更稳。

1. remote_controller 启动流程

入口：

src/remote_controller/src/main.cpp

启动时做这些事：

解析 --driver 和 --config
  -> load_remote_config(config_path)
  -> 创建 InputMapper
  -> 创建 ROS publisher: /motion_commands
  -> 创建 10ms timer
  -> create_input_driver(driver_type)
  -> input_driver.start()

默认 driver 是 joystick。键盘 launch 会传：

--keyboard

等价于 driver type keyboard。

自定义 driver 用：

--driver crsf

2. RemoteConfig 加载和自检

配置加载在：

src/remote_controller/src/config.cpp

大致流程：

YAML::LoadFile
  -> warn_unknown_root_keys
  -> load_sources
  -> load_curves
  -> load_controls
  -> load_outputs
  -> load_system_commands
  -> load_system_mutexes
  -> load system_reset_motion_after
  -> validate_config

自检会检查：

• source 名是否重复。
• control 名是否重复。
• control 类型是否合法。
• curve 引用是否存在。
• source 引用是否存在。
• expr 是否只用于 bool control。
• expr 是否成环。
• output 字段是否被支持。
• binding 是否引用未知 control。
• system output 是否引用未知 system action。
• 未使用的 source/control/curve 会输出 warning。

这些诊断会在遥控器节点启动时打印。

3. InputMapper 的内部状态

核心文件：

src/remote_controller/src/input_mapper.cpp

内部重要变量：

• signals_：raw source 当前值。
• controls_：control 当前值，包含 analog、pressed、value。
• signal_expiry_：键盘 source 的自动过期时间。
• signal_update_time_：runtime source 最近连接心跳时间。
• timed_out_sources_：已经进入 failsafe 的 source。
• binding_active_：每个 edge/level binding 上一帧是否满足。
• output_slots_：level 输出的 btn_N 当前值。
• edge_pulse_slots_：edge 输出的一帧脉冲。
• height_filtered_：未显式输出 height_des 时的默认高度滤波。

4. 输入进入 InputMapper

joystick 轴：

mapper_.set_axis(event.number, event.value);

内部转成：

js.axis.<number> = event.value / 32767

joystick 按钮：

mapper_.handle_button(event.number, pressed);

内部转成：

js.button.<number> = 1.0 或 0.0

自定义 driver：

mapper_.set_signal("crsf.ch1", value);

set_signal() 会：

写 signals_[source]
  -> 清除该 source 的 expiry
  -> 更新 signal_update_time_
  -> 清除 timed_out 标记
  -> refresh_bindings()

5. keyboard 的特殊逻辑

终端键盘没有 release 事件，所以使用 hold_ms。

按键触发后：

signals_[keyboard.xxx] = 1.0 或 axis 值
signal_expiry_[keyboard.xxx] = now + hold_ms

之后 tick() 发现过期：

signals_[keyboard.xxx] = 0.0

因此 keyboard 是“按一次保持一小段时间”，不是物理按住/松开模型。

6. tick() 做什么

遥控器节点每 10ms 调用：

outputs = mapper_.tick();
mapper_.fill_message(message);

tick() 做两类事：

1. 处理 signal_expiry_，让键盘 source 到期归 0。
2. 处理 runtime timeout，让断联 source 进入 failsafe。

runtime timeout 逻辑：

如果 source 设置 timeout_ms
并且 now - signal_update_time_[source] >= timeout_ms
并且该 source 还没 timed_out
  -> signals_[source] = failsafe
  -> 标记 timed_out
  -> refresh_bindings()

7. failsafe 为什么需要 touch

timeout_ms 应该表示设备断联，不是值不变。

如果设备连接正常，但摇杆保持不动，driver 仍然要调用：

mapper_.touch_runtime_sources_with_prefix("js.");

或：

mapper_.touch_runtime_sources_with_prefix("crsf.");

这会更新对应 prefix 的 runtime source 心跳，不改变数值。

设备断开或长期没有有效帧时不要 touch，让 failsafe 生效。

8. control 计算流程

每次刷新时：

refresh_controls()
  -> evaluate_control_recursive()
  -> evaluate_control()

递归是为了支持派生 control：

command.flip:
  type: bool
  expr:
    any:
      - [trigger.left, button.west]

如果 expr 依赖成环，加载配置时会报错。

analog control 计算：

读取 sources
  -> source direction/scale/offset
  -> source curve/deadzone/expo
  -> mix
  -> control invert
  -> control curve/expo
  -> control deadzone
  -> min/max 映射
  -> alpha 低通

bool control 计算：

raw >= threshold +/- hysteresis

enum control 计算：

先尝试用 hysteresis 保持 previous.value
否则按 positions 区间匹配
没有匹配则 default

9. binding 刷新流程

refresh_bindings()
  -> refresh_controls()
  -> output_slots_ 清零
  -> 遍历所有 bindings

level：

条件满足
  -> apply_level_output()
  -> 写 output_slots_[slot]
条件不满足
  -> 不写

因为每次刷新前 output_slots_ 都清零，所以 level 条件不满足时自然回 0。

edge：

条件满足 且 上一帧不满足
  -> apply_edge_pulse_output()
  -> 写 edge_pulse_slots_[slot]

edge 的 binding_active_ 会记录上一帧状态，因此只在上升沿触发。

10. fill_message()

fill_message() 把当前 mapper 状态写到 MotionCommands。

流程：

refresh_controls()
  -> 写 analog outputs
  -> 对 btn_1..btn_10:
       edge_pulse_slots 优先
       否则 output_slots
       写入 MotionCommands.btn_N
       清空 edge_pulse_slots
  -> 如果没有写 height_des:
       height_des 缓慢回到默认站立高度

edge 脉冲只存在一帧，因为写入后立刻清空。

11. publish_on_change

main.cpp 中：

mapper.fill_message(message)
if publish_on_change and message == last_published_payload:
    return
message.header.stamp = now
publish(message)

比较发生在填 header 之前，所以 header 时间戳不会导致每帧都被认为变化。

publish_on_change: true 的语义：

• payload 没变就不发布。
• 减少 /motion_commands 占用。
• 方便其他节点临时发布命令。

接收端应该保持上一条速度命令。机器人不应该因为遥控器没有重复发布相同 payload 而停下。

12. system action 执行流程

当 binding 输出 system.start：

dispatch_outputs()
  -> run_system_action("start")
  -> blocking_system_mutex()
  -> run_commands()
  -> update_system_mutexes()
  -> reset_motion_after_system

system_mutexes 用于避免重复启动：

system_mutexes:
  launch:
    acquire: start
    release: stop

13. BxiExample 启动流程

核心文件：

src/bxi_example_py_elf3/bxi_example_py_elf3/bxi_example_demo.py

启动时：

load_files()
  -> 读取 topic_prefix
  -> 读取 npz_file_dict / onnx_file_dict
  -> 读取 state_machine_config
  -> 读取 state_machine_info 参数

加载模型
init_pub_sub()
build_robot_states()
RobotStateMachine(...)
RemoteEventAdapter(...)
创建 timer

14. BxiExample timer_callback

定时器周期：

dt = 0.02

主要流程：

前置 reset 流程
  -> 读取传感器缓存
  -> 取出 pending_remote_events
  -> state_machine.update(dt, events)
  -> 如果不在 transition:
       state_machine.update_current_state(dt)
  -> 如果状态产生 motor_target:
       send_to_motor(qpos, kp, kd)
  -> publish_state_machine_info_if_due(events)

普通过渡期间不会直接调用当前状态的 on_update()，而是调用 transition hook。dual_running_blend 是例外：它会在 transition hook 内部调用旧状态和新状态的 get_transition_frame()，再混合后写入 motor_target。

15. joy_callback 和 remote event

BxiExample 订阅：

motion_commands

收到消息：

apply_velocity_profile(msg)
events = remote_event_adapter.extract_events(msg)
pending_remote_events.extend(events)

RemoteEventAdapter 对每个 event 记录上一次 slot 值。

完整 event：

back_flip:
  slot: btn_10
  value: 1

触发：

当前 btn_10 == 1
并且 btn_10 对 back_flip 来说发生变化

启动 reset 阶段会 sync_only，只同步上一帧值，不触发事件。

16. RobotStateMachine 更新流程

核心文件：

src/bxi_example_py_elf3/bxi_example_py_elf3/state_machine.py

每周期：

如果 active transition 存在:
  _update_active_transition(dt)
  return True

处理 events
处理 pending delay
如果开始 active transition:
  _update_active_transition(dt)
  return True

state_elapsed += dt
处理 after rules
如果开始 active transition:
  _update_active_transition(0.0)
  return True

return False

timer_callback 看到返回 False 才会调用：

state_machine.update_current_state(dt)

17. event 规则优先级

_handle_events() 中：

event_set = set(events)
按 YAML 中当前状态 rules 顺序遍历
找到第一个 rule.event 在 event_set 中的规则
  -> action 或 transition
  -> return

因此同一周期多个 event 进入时，当前状态 YAML 里规则顺序决定优先级。

18. pending transition

带 delay 的规则：

recover:
  to: recover
  delay: 0.5

触发后进入 pending：

pending.elapsed += dt
pending.elapsed >= delay
  -> begin transition

pending 期间如果又触发新的 delayed rule，当前实现会覆盖 _pending。

19. active transition

开始切换：

current.on_exit(ctx)
to_state.on_prepare_enter(ctx, current, profile)
active = ActiveTransition(...)

更新切换：

elapsed += dt
progress = elapsed / profile.duration
exit_progress = elapsed / profile.exit_duration
enter_progress = elapsed / profile.enter_duration

from_state.on_exit_transition(...)
to_state.on_enter_transition(...)

if progress >= 1:
  finish

结束切换：

current = to_state
state_elapsed = 0
pending = None
active = None
fired_after_rules.clear()
current.on_transition_commit(ctx, from_state, profile)

默认 on_transition_commit() 会调用 on_enter()。RobotControlState 在 dual_running_blend 这种目标状态已经提前运行的过渡里，会跳过重复的 on_enter()，避免目标动作在过渡结束瞬间重新初始化。

dual_running_blend 的采样流程：

from_frame = from_state.get_transition_frame(role="from")
to_frame = to_state.get_transition_frame(role="to")
alpha = enter_progress
motor = lerp(from_frame, to_frame, alpha)

RobotControlState.get_transition_frame() 默认调用 get_motor_frame(ctx, ctx.dt)。节点不参与采样逻辑，BxiExample 只负责发送最终的 motor_target。

20. action 解析

规则：

toggle_dance_pause:
  action: toggle_dance_pause

执行顺序：

先查 action_handlers
再调用 current.on_action(ctx, action_name)
都没有处理则报错

状态内 action 返回 True 表示已处理。

21. 状态图自检

初始化时：

_run_graph_checks(initial_state)
_export_graph_from_config(initial_state)

自检内容：

• 转移目标状态是否存在。
• profile 是否存在。
• event 是否在 remote_events 声明。
• 从 initial_state 是否可达。
• 状态是否没有任何输出 transition。
• after 自动转移是否成环。

导出：

/tmp/elf3_state_machine.dot
/tmp/elf3_state_machine.mmd

22. 状态信息话题

BxiExample 发布：

<topic_prefix>state_machine_info

内容来自：

state_machine.snapshot(include_graph=True)

包含：

• mode: state / transition / pending
• 当前状态。
• active transition。
• pending transition。
• 状态图。
• 当前事件。
• 当前速度命令。

23. 高上限扩展点

想扩展遥控器表达能力：

• 增加新的 ControlConfig.type。
• 增加新的 condition kind。
• 增加批量 set_signals()。
• 增加更多 MotionCommands 字段适配。

想扩展状态机：

• 增加 guard 条件。
• 增加 transition priority。
• 增加层级状态机。
• 增加状态组和共享转移。

想扩展机器人过渡：

• 增加更多 enter_behavior / exit_behavior。
• 在 data 中定义行为参数。
• 为状态实现专属 transition hook。

扩展时保持分层：

通用状态机能力 -> state_machine.py
机器人电机过渡 -> robot_state_base.py
具体动作行为 -> 状态类
输入协议读取 -> InputDriver
输入业务解释 -> YAML