Application 入口与生命周期
import asyncio
import fastsim
async def main() -> None:
async with fastsim.app("run.yaml") as simulation:
await simulation.start()
snapshot = await simulation.state()
targets = await simulation.control.targets()
print(snapshot, targets)
asyncio.run(main())
上下文负责最终 close,但业务代码仍应显式处理 start、Operation 结果、取消和超时。不要从 Application 中提取后端原生对象并在外部线程操作。
稳定服务与值对象
scene:catalog、state、pose/drag/attach 等受控场景操作。frames:frame catalog、lookup 与 transform。geometry:规划几何能力、catalog、transform 与资源读取。kinematics:按当前可用 solver 执行查询。control:目标发现、joint/track Chunk、Operation 状态和取消。visualization:发布调试几何、metric 与 event。- checkpoint、camera 与 output:按 Run/插件能力出现。
调用前应通过能力与目标发现确认当前部署可用内容。缺少可选能力是明确错误或不可用状态,不应靠捕获任意异常后继续。
API 是异步的,但后端世界仍由单一权威边界串行化。超时限制调用者等待时间,不自动等价于取消物理操作;长操作应保留 Operation 并根据状态决定 cancel 或继续等待。
简单脚本可使用 FastSimRun 的同步上下文与生命周期方法。不要在已有 asyncio loop 中嵌套同步门面;服务与 UI 应保持异步 Application。
所有当前签名由源码生成在Python SDK中。
推荐调用顺序与服务发现
fastsim.app(...) 先编译并持有 Plan,进入异步上下文时打开应用边界。可显式 prepare() 以构造世界但不持续运行,也可直接 start();读取状态和相机等能力会确保应用已准备。结束时先处理业务 Operation,再 stop/close。
import asyncio
import fastsim
async def inspect_run() -> None:
async with fastsim.app("run.yaml", timeout=30.0) as simulation:
prepared = await simulation.prepare()
print(prepared.run_id, prepared.plan_digest, prepared.state)
running = await simulation.start()
state = await simulation.state()
targets = await simulation.control.targets()
print(running.state, state.world.generation, targets)
await simulation.pause()
await simulation.stop()
asyncio.run(inspect_run())
不要把示例中的打印字段当作固定 UI 格式;需要准确类型和签名时查询 SDK 参考。业务代码应基于 typed snapshot 与稳定枚举,而不是解析 repr()。
Snapshot、值对象与原生对象。 snapshot() 返回 Application/Runtime/Plugin 的公开状态,state() 返回已提交世界与 observation。它们是可跨线程阅读的不可变值,不是可写后端句柄。修改 Scene 或提交控制使用相应服务方法,不能修改 snapshot 期望后端跟随变化。
缓存值对象时同时保存 generation/revision。Reset 后可以保留它作历史记录,但不能继续用作当前控制输入。
服务发现与可选能力。 Application 始终提供生命周期和 run identity;Scene、Frame、Geometry、Kinematics、Camera、Visualization、Control 或 Output 是否可用取决于 Plan、Provider 和插件组合。先读取 descriptor/capabilities/targets,再显示 UI 或启动 worker。
可选服务不可用时,应向调用方呈现准确原因和安装/配置方向。不要用 except Exception: pass 把服务缺失、请求错误和 Provider 故障混成“没有数据”。
Operation、并发与结构化退出
提交控制得到 Operation 后,把 operation ID 与当前 run/generation 记录到业务状态。等待结果可设置调用预算,但超时后仍需选择继续查询或 cancel();完成后 release() 清理客户端持有资源。
一个稳健的服务端通常把“提交”和“等待”分成两个请求,使客户端断线不会丢失操作身份。FastSim Server 也采用这种异步 operation 模型。
并发任务的退出结构。 相机订阅、状态流、规划 worker 和控制 Operation 应归属于当前 Application scope。使用 task group 或显式任务集合,在 reset/stop/close 时停止生产、取消等待、关闭订阅并收集失败。不要留下后台线程继续访问已关闭 service。
后端对象只在 owner loop 内由 Provider 操作;asyncio.to_thread 适合纯 CPU 或阻塞外部工作,不适合把 simulation 或后端句柄带到任意线程。
错误证据与接口选择
区分编译错误、Application 生命周期错误、服务不可用、请求无效、stale generation、capacity exhausted、Provider failure 与 Operation 业务终态。日志至少关联 run ID、plan digest、generation、operation ID 和实体/插件 ID。
何时使用 CLI、Python 或 Server。 一次性验证和部署脚本优先 CLI;与本进程算法紧密组合的应用使用 Python;跨进程、浏览器、TUI 或远程 Agent 使用 Server。三者最终应进入同一 Application 合同,不应各自实现仿真逻辑。
接下来可查Python SDK确认签名,或阅读运行 FastSim Server把同一 Application 暴露给远程客户端。