Embodied Tabletop Framework — 用一份配置 + 一组硬件能力契约,快速搭出会自主行动的实体桌游。
把摄像头、小车、机械臂、心率这些便宜的现成硬件,收敛成一套统一的能力契约; 把标定、坐标系、路径规划、多设备同步、故障降级这些底层复杂度,全部收进框架。 你写游戏,只需要声明一份配置 + 选用需要的能力,不必碰机器人控制。
传统桌游的世界是"死"的:敌人走位要玩家替它挪、天灾要照规则书宣布、一轮打完账要自己算。
embodied 让桌面"活"起来——自主小车执行玩家的战略,机械臂扮演环境意志实时改写地形,
摄像头把整张桌子的位姿与玩家状态读进来,一份配置把这些编排成一局游戏。
它不是某一款游戏,而是这类游戏共用的运行时与能力抽象:
- 玩家负责意图,Agent 负责行动。 策略与指令被约束到合法动作空间,单位自主完成移动、采集、规避。
- 物理设备是可组合能力。 小车、机械臂、心率、语音、定位通过六种统一契约接入,互不写死。
- 创作者定义玩法,而不是重写驱动。 地图、单位、卡牌、规则、结算由配置声明;同一套硬件承载不同游戏。
一份配置一款游戏 → 结构校验 → 语义校验 → 一条命令 → 拉起整套服务
框架的边界很清楚:复杂性留在框架里,不是从第一次摄像头标定开始。
- 上面:你写一份 YAML/JSON 配置声明游戏,
embodied帮你做两道校验(结构 + 语义)后拉起运行时。 - 下面:六种能力契约。任何设备只要实现对应
Protocol,register进注册表即可即插即用。 - 中间的标定、坐标换算、A* 绕障、闭环控制、看门狗急停、多设备同步、故障降级,全部在框架内。
三层结构:感知 → 决策 → 执行,基础层(几何 / 消息 / 能力契约)贯穿其间。
flowchart TB
subgraph PERC [感知层 perception]
P1[视觉定位<br/>标记→世界位姿]
P2[对象检测<br/>地标/障碍占用栅格]
P3[无接触心率<br/>rPPG 信号管线]
end
subgraph DEC [决策层 decision]
D1[角色内核 Agent<br/>目标·性格·记忆·权限]
D2[规则引擎<br/>合法动作·脚本规则]
D3[决策策略 Policy<br/>LLM 优先·规则兜底]
end
subgraph EXE [执行层 execution]
E1[A* 路径规划<br/>障碍膨胀]
E2[差速驱动 Mover<br/>UDP 闭环]
E3[机械臂 Manipulator<br/>逆运动学]
E4[安全总管<br/>限速·看门狗·急停]
end
PERC --> DEC --> EXE
Config[游戏配置 config] -.装配.-> Runtime[运行时 runtime]
Runtime -.驱动.-> PERC
Geo[基础层:geometry · messaging · capabilities]:::base
classDef base fill:#1f2933,color:#fff,stroke:#f97316;
需要 Python 3.10+。
git clone <this-repo-url> embodied && cd embodied
pip install -e . # 安装 embodied 及依赖(numpy / pyyaml / websockets / httpx)运行期依赖都是惰性导入:只用几何/规划时无需联网;pyyaml 仅在读 YAML 配置时需要,
websockets 仅在跨进程广播时需要,httpx 仅在 LLM 决策或文生 3D 时需要。
embodied examples/moonfall_min.yaml # 安装后的控制台入口
# 或
python -m embodied examples/moonfall_min.yaml常用参数(见 embodied --help):--max-ticks N 跑多少拍、--rate-hz N 节拍频率、
--serve HOST:PORT 同时起 WebSocket 广播服务端供前端大屏连。
from embodied import Runtime
runtime = Runtime.from_config("examples/moonfall_min.yaml")
runtime.start()
for _ in range(12):
snapshot = runtime.tick() # 一拍:读感知 → 决策 → 下发执行 → 汇总状态
runtime.stop()
for car, pose in snapshot["poses"].items():
print(car, pose)或者用一行的 launch() 跑到结束:
from embodied import launch
runtime = launch("examples/moonfall_min.yaml", max_ticks=60, serve=("0.0.0.0", 8000))examples/quickstart.py 是一份可直接运行的完整示例,输出形如:
载入游戏:Moonfall Minimal v0.1.0
场地 80×60 cm,栅格 2 cm
单位 6 个,地标 5 个,规则 2 条
跑完 12 拍,阶段:play
各车位姿(世界厘米坐标):
r0: x= 23.19 y= 24.51 θ=-0.65 rad 在场内=True
...
无硬件时,运行时用内置的运动学底盘(KinematicChassis)与内存位姿源
(KinematicPoseSource)真实推进,车会从起点朝目标地标移动。
设备与游戏之间只通过这六个 Protocol 交互(embodied.capabilities)。实现契约即可接入,
框架用 isinstance + Protocol 识别一个设备满足哪些能力。
| 契约 | 关键方法 | 语义 | 典型实现 |
|---|---|---|---|
PoseSource |
poll() -> dict[str, Pose] |
提供各单位的世界位姿(厘米+弧度) | 顶拍相机 + 标记定位 |
Mover |
move_to(x, y, speed) · stop() · pose |
自主移动到世界坐标 | 差速小车 |
Manipulator |
pick · place · hover · stop |
抓取 / 放置 / 悬停 | 机械臂 |
BioSignalSource |
read() -> dict[str, float] |
读生理信号(如心率 bpm) | rPPG / 手环 |
VoiceIntentSource |
interpret(utterance, action_space) -> Intent |
自然语言 → 受约束意图 | 语音解析 |
Presenter |
say(text) · emote(kind) |
解说 / 角色化反馈 | 人形机器人 / 屏幕角色 |
设备注册表:
from embodied import DeviceRegistry
reg = DeviceRegistry()
reg.register("Mover", my_car, name="r0") # 未实现契约会当场 TypeError
reg.get("Mover", "r0") # 按名取
reg.all("PoseSource") # 某能力下的全部设备
reg.capabilities_of(my_car) # 该设备满足哪些契约一份配置即一款游戏。顶层小节如下(完整示例见 examples/moonfall_min.yaml):
| 小节 | 作用 |
|---|---|
world |
场地尺寸(width_cm / height_cm)与规划栅格边长(cell_cm) |
units |
可行动单位:role(host/rover/arm)、start 世界坐标、max_speed、capabilities、permissions、goal |
landmarks |
地标 / 障碍,统一建模为世界坐标系下的圆(x/y/radius_cm + kind) |
cards |
卡牌:cost、可选 target(指向地标)、effect 效果声明 |
rules |
脚本规则:when 条件表达式 → then 效果,priority 越大越先评估 |
routes / default_route |
巡逻路线(依序引用地标),无卡牌时的默认序列 |
settlement |
胜负条件(win_when / lose_when)与按地标类型计分的 scoring |
加载时先过结构校验(键 / 类型 / 枚举 / 范围),再过语义校验(引用完整性 /
坐标越界 / id 唯一 / 速度等级),两层都通过才产出强类型 GameConfig:
from embodied import load_game, ConfigError
try:
game = load_game("examples/moonfall_min.yaml")
except ConfigError as e:
print(e) # 一次性列出所有结构 + 语义错误flowchart LR
A[感知<br/>PoseSource.poll<br/>BioSignalSource.read] --> B[决策<br/>Agent.observe<br/>Policy.decide<br/>RuleEngine.evaluate]
B --> C[执行<br/>Mover.move_to<br/>Manipulator.pick<br/>SafetyGovernor 限速]
C --> D[汇总世界状态<br/>DigitalTwin.project]
D --> E[广播 state.world<br/>EventBus / WebSocketHub]
E --> A
每拍:读感知刷新位姿与生理信号 → 各角色 Agent 在合法动作空间内决策(LLM 优先、 规则兜底)→ 经安全总管限速后下发执行 → 数字孪生投影出可序列化快照 → 广播给前端。
每个模块都是真实实现(真算法 / 真协议 / 真数学),可独立使用。
| 模块 | 提供 | 关键实现 |
|---|---|---|
embodied.geometry |
Pose · WorldFrame · Homography · 栅格换算 |
纯 numpy 四点单应(像素→厘米)、角度归一化 |
embodied.messaging |
Message · EventBus · WebSocketHub |
统一封套、进程内同步 pub/sub、真实 WebSocket 广播服务端 |
embodied.capabilities |
六种 Protocol · Intent · DeviceRegistry |
runtime_checkable 契约、按能力注册/查找 |
embodied.perception |
TagPoseSource · ObjectDetector · RppgHeartRate |
标记角点过单应解位姿、圆形地标占用栅格、rPPG(去趋势→带通→FFT→峰值插值) |
embodied.decision |
Agent/HostAgent/RoverAgent/ArmAgent · RuleEngine · Policy |
角色内核(目标/记忆/权限过滤)、条件→效果规则求值、LLM 优先 + 规则兜底 |
embodied.execution |
plan/astar · DifferentialDrive · ArmController · SafetyGovernor |
障碍膨胀 A*、航向误差→轮速 + UDP 闭环、2 连杆逆运动学、看门狗急停状态机 |
embodied.config |
load_game · GameConfig · validate_schema/validate_semantics |
手写结构 + 语义两道校验、强类型 IR |
embodied.creation |
Hyper3DClient · DigitalTwin |
文生 3D 真实 HTTP 客户端(提交→轮询→下载 .glb)、世界状态投影与增量补丁 |
embodied.runtime |
Runtime · launch · main |
从配置装配全套、逐拍主循环、CLI 入口 |
几个可独立调用的例子:
from embodied import estimate_bpm, plan, Homography
bpm = estimate_bpm(green_channel_series, fps=30) # 一段 ROI 绿通道均值 → 心率
path = plan(occupancy_grid, (10, 30), (70, 30), cell_cm=2.0) # 厘米坐标间 A* 规划
H = Homography.from_point_pairs(pixel_corners, world_corners) # 标定像素→世界厘米文生 3D 客户端读环境变量 HYPER3D_API_KEY / HYPER3D_BASE_URL,缺密钥时 submit
直接抛 AuthError(不会静默返回无效结果):
from embodied import Hyper3DClient
client = Hyper3DClient() # 或 Hyper3DClient(api_key=...)
client.text_to_mesh("a lunar rover chess piece", "rover.glb")实现相应契约 → 注册 → 即插即用。以一个走串口的机械臂为例:
from embodied import DeviceRegistry
class SerialArm: # 无需继承,鸭子类型满足 Protocol 即可
def pick(self, x: float, y: float) -> None: ...
def place(self, x: float, y: float) -> None: ...
def hover(self, x: float, y: float) -> None: ...
def stop(self) -> None: ...
reg = DeviceRegistry()
reg.register("Manipulator", SerialArm(), name="arm0")
assert "Manipulator" in reg.capabilities_of(reg.get("Manipulator", "arm0"))自主小车实现 Mover(+可选 PoseSource)、位姿源实现 PoseSource、心率源实现
BioSignalSource,同理。框架其余部分对具体设备一无所知。
全系统只有一套口径,任何模块不得偏离:
| 项 | 定值 |
|---|---|
| 世界坐标 | 80cm × 60cm,x 横向 0..80,y 纵向 0..60,原点左下角 |
| 长度 | 一律厘米(cm) |
| 角度 | 一律弧度(rad) |
| 速度 | 整数等级 0..10(0 = 停) |
| 时间戳 | float 秒 |
| 消息封套 | {topic, source, timestamp, payload} |
python -m unittest discover -s tests # 113 项
# 或(装了 pytest)
pytest覆盖:几何变换与单应精度、消息总线收发与退订、能力契约识别、标记→位姿、占用栅格、 对合成正弦信号的 rPPG 测频、Agent 权限过滤、规则求值、无 key 时 Policy 规则兜底、 配置能过能报错、数字孪生投影与增量补丁、A* 寻路、安全总管看门狗/急停时序等。
embodied/
├── geometry.py 坐标系:Pose / WorldFrame / Homography / 栅格换算
├── messaging.py 消息封套 / 事件总线 / WebSocket 广播
├── capabilities.py 六种能力契约 + 设备注册表
├── perception/ 感知层:pose / objects / vitals
├── decision/ 决策层:agent / roles / rules / policy
├── execution/ 执行层:planner / drive / manipulator / safety
├── config/ 配置驱动:schema / semantics / loader
├── creation/ 创作工具:modeling(文生3D) / twin(数字孪生)
└── runtime.py 一条命令拉起整套服务
examples/
├── moonfall_min.yaml 最小可玩游戏配置
└── quickstart.py 可直接运行的上手示例
tests/ 113 项单元测试