跳转至

⚙️ OrcaGym Euler 架构

📌 前置阅读:如果你是第一次接触 OrcaGym 架构,建议先阅读 架构总览 了解整体分层与 API 边界,再回到本文查看组件设计细节。

1. 为什么需要新架构

1.1 现有体系的问题

OrcaGymLocalEnv + OrcaGymLocal 作为当前主用路径,存在三类结构性缺陷:

缺陷类型 表现 后果
上帝类 OrcaGymLocal 单类承担仿真核心、Studio 集成、模型注册、求解器配置、物体操作等所有职责 难以维护、难以扩展、职责耦合
API 不完备 OrcaGymData 只缓存 5 个字段(qpos/qvel/qacc/qfrc_bias/time),缺 xfrc_applied/cvel/contact 等 你不得不绕道 gym._mjData 直接访问
封装泄漏 _mjModel/_mjData 作为公共属性暴露,self.gym 既是内部组件又作外部库句柄 83 处直接访问,封装形同虚设

1.2 用户代码现状

对 OrcaPlayground 和 OrcaManipulation 两个仓库的分析显示:

  • 17 个直接子类 + 4 个间接子类继承自 OrcaGymLocalEnv
  • 83 处直接访问 gym._mjData / gym._mjModel
  • 典型绕道场景:外力注入(xfrc_applied)、求解器配置(opt.*)、body 属性查询(body_subtreemass/cvel)、等式约束结构访问(eq_data)

你绕道不是因为不懂封装,而是封装没有覆盖你的需求。

1.3 新架构做了什么

OrcaGymEulerEnv + OrcaGymEuler 采用 Facade + 职责内聚分解的设计,替代 OrcaGymLocalEnv + OrcaGymLocal 的上帝类模式:

  • 完备的公共 API:覆盖所有合法的 MuJoCo 操作需求,你不再需要绕道访问内部数据
  • 封装隔离:_mjModel/_mjData 不对外暴露,从机制上引导你和 AI 走正确路径
  • 平滑迁移:约 70% API 零改动,25% 机械替换,5% 设计调整

2. 核心设计原则

2.1 六大原则

原则 含义 对比旧体系
P1 完备性 公共 API 覆盖所有合法的 MuJoCo 操作需求 大量缺口迫使绕道
P2 不暴露引擎内部 _mjModel/_mjData 不作为公共属性暴露 直接暴露
P3 状态一致性 任何写操作后,self.data 保证一致;任何读操作都走 self.data 或显式查询 self.data 与 _mjData 双轨制
P4 力应用可追踪 外力注入通过显式方法,未来 Euler 耦合器可感知 xfrc_applied 直接写,无感知
P5 职责内聚 按职责内聚划分模块,一组方法因同一原因变化、共享同一组数据 上帝类
P6 框架无状态、业务自编排 框架只提供无状态原语(单次原子读写),多步编排流程由你的业务代码自行组合实现并自管状态 框架混杂原语与编排,易误用

P6 补充说明

框架公共 API 都是无状态原语——单次调用完成单一数据读写,不依赖前后调用顺序,不持有跨调用的快照或绑定标记。例如:

  • equality_find_slot_by_body — 按 body 查槽位
  • equality_constraint(slot) — 读单槽位
  • equality_update(slot, **fields) — 写单槽位
  • set_mocap_pos_and_quat — 写 mocap 位姿

如果你的业务需要"绑定/释放/抓取"等多步编排流程,请自行组合这些原语实现,业务状态由你自己管理。这比框架代管更易审查、更不易误用。

2.2 设计模式

模式 应用位置 解决的问题
Facade OrcaGymEulerEnv / OrcaGymEuler 组合多个子组件,提供统一 API,避免上帝类
组合优于继承 Env 持有 Gym,Gym 持有子组件 避免继承链腐化,职责可独立演进
策略模式 OrcaGymEuler._euler 字段(占位) 当前恒为 None,通过 has_euler() / step_with_coupling() 封装切换有无 Euler 的策略(Euler 编排器后续设计)
依赖反转 OrcaStudioBridge 不持有 mjData,通过接收数据参数实现解耦 Studio 集成与仿真核心解耦
只读视图 OrcaGymDataView 提供完整状态读取,禁止写入

3. 架构总览

3.1 整体结构

gym.Env
  └── OrcaGymEulerEnv (新)                    (Facade + 契约执行者,直接继承 gym.Env)
        │   ↑ OrcaGymEnvMixin(名称空间、动作/观测空间生成、reset 编排)
        │
        │   组合(非继承)
        ├── _gym: OrcaGymEuler           (仿真核心 Facade)
        │     ├── _sim: MuJoCoSimCore    # 持有 _mjModel/_mjData(不对外暴露)
        │     ├── _studio: OrcaStudioBridge  # gRPC 集成
        │     ├── _registry: ModelRegistry  # 模型信息
        │     ├── _opt: SimConfig        # 求解器配置(typed)
        │     └── _euler: None  # Euler 耦合占位(当前未实现,后续设计)
        │
        │   公共 API(你面向的接口)
        ├── .data → OrcaGymDataView      # 完整状态视图
        ├── .model → OrcaGymModel        # 模型结构(原样复用)
        ├── .sim_config → SimConfig      # 求解器配置
        ├── .ctrl → np.ndarray           # 控制数组
        │
        ├── 仿真控制
        ├── 状态查询
        ├── 状态设置
        ├── 名称空间
        └── Studio 交互

3.2 与旧体系的对比

维度 OrcaGymLocalEnv + OrcaGymLocal OrcaGymEulerEnv + OrcaGymEuler
类结构 上帝类,单类承担所有职责 Facade + 职责内聚分解
_mjModel/_mjData 公共属性,83 处直接访问 内部组件,多层隔离
OrcaGymData 5 字段缓存,不完整 OrcaGymDataView 完整只读视图
求解器配置 无接口,绕道 opt.* SimConfig typed 配置
外力注入 直接写 xfrc_applied apply_body_force() 显式方法
继承体系 继承自腐化的 OrcaGymBase 链 直接继承 gym.Env + OrcaGymEnvMixin

3.3 与旧体系的共存

原有 OrcaGymBase → Remote / Local / Warp 继承体系不动,现有系统继续运行不受影响。OrcaGymEulerEnv 是独立的新类,与 OrcaGymLocal 长期共存。迁移路径上,OrcaGymLocal 最终将被废弃,届时原有体系直接删除即可。


4. 组件说明

4.1 OrcaGymEulerEnv — 环境 Facade

作为 Gymnasium Env 的实现,组合 OrcaGymEuler 仿真核心,向你暴露统一 API。继承结构:OrcaGymEulerEnv(OrcaGymEnvMixin, gym.Env)。

class OrcaGymEulerEnv(OrcaGymEnvMixin, gym.Env):
    """OrcaGym Euler 双引擎环境。

    使用契约:
        读取状态:   env.data.qpos / env.data.body_xpos(name) / env.query_*()
        写入状态:   env.set_joint_qpos() / env.apply_body_force()
        仿真步进:   env.do_simulation(ctrl, n_frames)
        求解器配置: env.sim_config.timestep = 0.002

    禁止:
        不要访问 env._gym._sim._mjData 或任何内部 MuJoCo 对象。
        env.gym/env.stub/env.channel 不存在,直接继承 gym.Env 不创建这些属性。
        缺少功能时,扩展本类的公共方法。
    """

关键属性:

属性 类型 说明
data OrcaGymDataView 完整状态只读视图,替代 _mjData 读取
model OrcaGymModel 模型结构信息(原样复用)
sim_config SimConfig 求解器参数配置,替代 opt.* 直接访问
ctrl np.ndarray 控制数组
frame_skip int 每次 step() 的物理步进数
dt float 环境时间步长 = timestep × frame_skip(timestep 才是单步物理时间)

4.2 OrcaGymEuler — 仿真核心 Facade

组合仿真子组件,向 OrcaGymEulerEnv 提供仿真操作接口。持有 MuJoCoSimCore、OrcaStudioBridge、ModelRegistry、SimConfig,以及 _euler 占位字段(当前为 None,Euler 耦合编排器后续设计)。不暴露 _mjModel/_mjData,依赖 _ 前缀约定 + ruff SLF001 静态检查,通过 __dir__ 控制 IDE 自动补全只显示公共 API。

class OrcaGymEuler:
    """双引擎编排核心。

    ┌─────────────────────────────────────────────────────────────┐
    │  API 契约:你不应直接访问 _mjData / _mjModel。              │
    │  读取 MuJoCo 状态 → 使用 env.data(OrcaGymDataView)        │
    │  写入外力 → 使用 env.apply_body_force()                     │
    │  配置求解器 → 使用 env.sim_config                           │
    │  缺少功能时 → 扩展 OrcaGymEulerEnv 公共方法                 │
    └─────────────────────────────────────────────────────────────┘
    """

4.3 MuJoCoSimCore — 仿真核心

持有 _mjModel/_mjData,执行 mj_step/mj_forward/set_ctrl 等纯 MuJoCo 操作。_mjModel/_mjData 只存在于此类内部,不对外暴露。

class MuJoCoSimCore:
    def __init__(self):
        self._mjModel: mujoco.MjModel | None = None
        self._mjData: mujoco.MjData | None = None

    def init_simulation(self, model_xml_path: str) -> None: ...
    def step(self, nstep: int) -> None: ...
    def forward(self) -> None: ...
    def set_ctrl(self, ctrl: np.ndarray) -> None: ...
    def sync_to_view(self, view: OrcaGymDataView) -> None: ...
    def apply_body_force(self, body_id: int, force: np.ndarray, torque: np.ndarray) -> None: ...

4.4 OrcaStudioBridge — Studio 集成

处理与 OrcaStudio 的 gRPC 交互,包括渲染、视频保存、物体操作等。依赖反转设计:不持有 _mjData,通过接收数据参数实现解耦;不碰 mj_step,只负责通信和场景同步。

class OrcaStudioBridge:
    def __init__(self, stub=None) -> None: ...
    async def render(self, qpos: np.ndarray, sim_time: float) -> None: ...
    async def load_model_xml(self) -> str: ...
    async def begin_save_video(self, file_path: str, capture_mode) -> None: ...
    async def stop_save_video(self) -> None: ...
    async def get_current_frame(self) -> int: ...
    async def get_body_manipulation_anchored(self) -> tuple: ...
    async def get_body_manipulation_movement(self) -> dict: ...

4.5 ModelRegistry — 模型注册

构建 OrcaGymModel/OrcaGymData,提供 query_all_* 等模型信息查询。OrcaGymModel 是成功抽象,原样复用,并扩展了缺失的模型结构查询。

class ModelRegistry:
    def __init__(self, mj_model=None) -> None: ...   # None 时需 _bind(mj_model) 后再使用
    def build_orca_gym_model(self) -> OrcaGymModel: ...
    def build_orca_gym_data(self): ...   # 抛 NotImplementedError:Euler 体系使用 OrcaGymDataView 替代
    def body_subtree_mass(self, body_name: str) -> float: ...
    def equality_data_width(self) -> int: ...
    def equality_object_ids(self, eq_idx: int) -> tuple[int, int]: ...

4.6 SimConfig — 求解器配置

提供 typed 的 MuJoCo 求解器参数读写接口,替代 _mjModel.opt.* 直接访问。覆盖所有用户可访问的 opt 字段,修改在下次 mj_step 时生效。

class SimConfig:
    """MuJoCo 求解器参数配置。替代直接访问 _mjModel.opt.*。修改在下次 mj_step 时生效。"""

    @property
    def timestep(self) -> float: ...
    @timestep.setter
    def timestep(self, value: float): ...

    @property
    def integrator(self) -> int: ...
    @integrator.setter
    def integrator(self, value: int): ...

    @property
    def iterations(self) -> int: ...
    @iterations.setter
    def iterations(self, value: int): ...

    @property
    def gravity(self) -> np.ndarray: ...
    @gravity.setter
    def gravity(self, value: np.ndarray): ...

    # ... 覆盖 opt 的所有用户可访问字段 ...

    def load_from_dict(self, config: dict) -> None: ...

迁移映射:

旧代码 新代码
gym._mjModel.opt.timestep = 0.002 env.sim_config.timestep = 0.002
gym._mjModel.opt.iterations = 100 env.sim_config.iterations = 100
gym._mjModel.opt.integrator = 0 env.sim_config.integrator = 0
30 行 opt.* 设置 env.sim_config.load_from_dict({...})

4.7 OrcaGymDataView — 完整状态视图

提供 MuJoCo 状态的完整只读视图,替代直接访问 _mjData。覆盖所有你需要读取的字段(不仅是原 OrcaGymData 的 5 个)。通过方法提供 body/site/geom 属性查询,你按名称访问,不需要知道 id。

class OrcaGymDataView:
    """MuJoCo 状态的完整只读视图。

    替代直接访问 _mjData。所有字段在 update_data() 后保证一致。
    你永远不需要访问 _mjData。

    如果需要此视图未提供的字段,请在 OrcaGymDataView 中扩展,
    不要通过 env._gym._sim._mjData 绕道访问。
    """

    # --- 基本状态 ---
    qpos: np.ndarray
    qvel: np.ndarray
    qacc: np.ndarray
    qfrc_bias: np.ndarray
    time: float

    # --- 扩展字段 ---
    xfrc_applied: np.ndarray       # 只读视图(写入用 apply_body_force)
    actuator_force: np.ndarray     # 执行器力
    contact: list                  # 接触列表

    def body_xpos(self, body_name: str) -> np.ndarray: ...
    def body_xquat(self, body_name: str) -> np.ndarray: ...
    def body_xmat(self, body_name: str) -> np.ndarray: ...
    def body_cvel(self, body_name: str) -> np.ndarray: ...
    def body_subtree_mass(self, body_name: str) -> float: ...

    def site_xpos(self, site_name: str) -> np.ndarray: ...
    def site_xmat(self, site_name: str) -> np.ndarray: ...

迁移映射:

旧代码 新 API
gym._mjData.qpos env.data.qpos
gym._mjData.body(id).xpos env.data.body_xpos(name)
gym._mjData.cvel[id] env.data.body_cvel(name)
gym._mjData.xpos[body_id, 2] env.data.body_xpos(name)[2]
gym._mjData.time env.data.time

4.8 Euler 耦合 — 占位(当前未实现)

编排 Euler 非刚体求解器与 MuJoCo 刚体求解器的耦合步进。当前为设计占位,代码中不存在 EulerOrchestrator 类:OrcaGymEuler._euler 字段恒为 None,OrcaGymEulerEnv 表现为纯 MuJoCo 环境。Euler 耦合的开关通过 OrcaGymEuler.has_euler() 查询、步进通过 OrcaGymEuler.step_with_coupling(ctrl, n_frames, dt) 封装(has_euler() 为 False 时等价于纯 MuJoCo 步进)。具体编排器设计后续单独文档论述。

# 当前代码中的实际封装(OrcaGymEuler 公共 API)
def has_euler(self) -> bool:
    """查询是否存在 Euler 耦合编排器。骨架阶段恒返回 False。"""
    return self._euler is not None

def step_with_coupling(self, ctrl: np.ndarray, n_frames: int, dt: float) -> None:
    """带 Euler 耦合的步进。has_euler()=False 时等价于 set_ctrl + step。"""
    # 骨架阶段:set_ctrl + step(无 Euler 耦合)
    self._sim.set_ctrl(ctrl)
    self._sim.step(n_frames)

4.9 OrcaGymEnvMixin — 环境公共方法混入

抽取 OrcaGymLocalEnv/OrcaGymBaseEnv 中与仿真引擎无关的公共方法,供 OrcaGymEulerEnv 使用。当前 OrcaGymLocalEnv 仍直接继承 OrcaGymBaseEnv,未接入 OrcaGymEnvMixin(可选重构,不强制)。

class OrcaGymEnvMixin:
    """OrcaGym 环境公共方法 Mixin。

    提供名称空间解析、动作/观测空间生成、reset 编排等方法。
    不定义 __init__,不持有状态,子类自行初始化 _agent_names 等字段。
    """

    # --- 名称空间解析(自动添加 agent 前缀)---
    def body(self, name: str, agent_id: int = None) -> str: ...
    def joint(self, name: str, agent_id: int = None) -> str: ...
    def actuator(self, name: str, agent_id: int = None) -> str: ...
    def site(self, name: str, agent_id: int = None) -> str: ...
    def mocap(self, name: str, agent_id: int = None) -> str: ...
    def sensor(self, name: str, agent_id: int = None) -> str: ...

    # --- 空间生成 ---
    def generate_action_space(self, bounds: np.ndarray) -> Space: ...
    def generate_observation_space(self, obs: Union[Dict, np.ndarray]) -> Space: ...

    # --- reset 编排 ---
    def reset(self, *, seed: Optional[int] = None, options: Optional[dict] = None): ...
    def set_seed_value(self, seed: int = None) -> list: ...
    def _get_reset_info(self) -> Dict[str, float]: ...

    # --- 辅助 ---
    def _name_with_agent0(self, name: str) -> str: ...
    def _name_with_agent(self, agent_id: int, name: str) -> str: ...
    @property
    def agent_num(self) -> int: ...

使用方式:

# Euler 体系(当前实际使用)
class OrcaGymEulerEnv(OrcaGymEnvMixin, gym.Env):
    def __init__(self, ...):
        self._agent_names = agent_names
        self._gym = OrcaGymEuler(...)
        # Mixin 方法可直接使用
        ...

# Local 体系(当前实际继承 OrcaGymBaseEnv,未接入 Mixin)
class OrcaGymLocalEnv(OrcaGymBaseEnv):
    ...

5. API 使用契约

5.1 契约层级

层级 含义 违反后果
L1 公共 API __dir__ 暴露的方法和属性,你应该使用 正常工作
L2 内部组件 _gym/_sim/_studio 等,你不应访问 ruff SLF001 报警
L3 引擎内部 _mjModel/_mjData,你绝不应访问 ruff SLF001 报警

5.2 状态读取

规则:所有状态读取通过 env.data(OrcaGymDataView)或 env.query_*() 方法。env.data 在 do_simulation() 返回后、mj_forward() 返回后保证一致。env.data 是只读视图,写入操作必须通过显式方法。

# ✅ 正确
qpos = env.data.qpos
body_pos = env.data.body_xpos("link1")

# ❌ 错误
qpos = env._gym._sim._mjData.qpos  # ruff SLF001 报警

5.3 状态写入

规则:所有状态写入通过显式方法,不直接操作 MuJoCo 数据结构。外力注入通过 apply_body_force(),不直接写 xfrc_applied。写入操作后若需立即读取一致状态,须调用 mj_forward()。

# ✅ 正确
env.set_joint_qpos({"joint1": np.array([0.5])})
env.apply_body_force("link1", force, torque)
env.mj_forward()

# ❌ 错误
env._gym._sim._mjData.xfrc_applied[body_id, :3] = force  # ruff SLF001 报警

5.4 仿真步进

方法 职责 Euler 耦合 适用场景
do_simulation(ctrl, n) 标准步进 有(未来) 大多数 Env 的 step()
mj_step(n) 纯 MuJoCo 步进 无 需要精细控制时序的高级用户
mj_forward() 前向计算 无 状态设置后更新派生量

两种使用模式:

# 模式 A(推荐,含 Euler 耦合)
env.do_simulation(ctrl, self.frame_skip)
# do_simulation 返回后 env.data 已自动同步(内部调用 sync_to_view)

# 模式 B(纯 MuJoCo,无耦合)
for _ in range(self.frame_skip):
    env.set_ctrl(torques)
    env.mj_step(1)
# 循环结束后需调用 env.mj_forward() 刷新派生量,再通过 env.data 读取

模式 B 当前与 OrcaGymLocalEnv 行为一致。若未来需要 Euler 耦合,模式 B 用户必须改用模式 A。

5.5 求解器配置

所有 opt.* 参数通过 env.sim_config 读写,配置修改在下次 mj_step 时生效。

# ✅ 正确
env.sim_config.timestep = 0.002
env.sim_config.iterations = 100
env.sim_config.load_from_dict({"integrator": 0, "iterations": 100})

# ❌ 错误
env._gym._sim._mjModel.opt.timestep = 0.002  # ruff SLF001 报警

5.6 名称空间

所有名称通过 env.joint()/env.body()/env.site()/env.actuator()/env.sensor() 解析,自动添加 agent 前缀。这部分 API 与 OrcaGymLocalEnv 完全一致,零改动迁移。

joint_name = env.joint("joint1")  # → "agent_name/joint1"
body_name = env.body("object")

5.7 完整公共 API 清单

类别 API
状态读取 data(OrcaGymDataView), model(OrcaGymModel), ctrl, frame_skip, dt, realtime_step
仿真控制 do_simulation(ctrl, n), mj_step(n), mj_forward()
状态查询 query_joint_qpos/qvel/qacc/offsets/lengths(), query_site_pos_and_mat(), query_site_pos_and_quat_B(), query_site_xvalp_xvalr(), query_site_xvalp_xvalr_B(), query_actuator_torques(), query_sensor_data(), query_contact_simple(), get_body_xpos_xmat_xquat()
状态设置 set_joint_qpos/qvel(), set_mocap_pos_and_quat(), set_geom_friction(), apply_body_force(), clear_body_force(), clear_all_forces()
等式约束原语(无状态,L1) equality_find_slot_by_body(body_name), equality_constraint(slot), equality_update(slot, **fields, forward=True)
求解器配置 sim_config(SimConfig)
名称空间 joint(), body(), site(), actuator(), sensor()
Studio 交互 render(), begin_save_video(), stop_save_video(), get_current_frame(), get_frame_png()
生命周期 initialize_simulation(), initialize_grpc(), pause_simulation(), close()

Studio UI 抓取为内部 API:原 anchor_actor() / release_body_anchored() / do_body_manipulation() 按 P6 原则改为 _ 前缀内部方法,由 render() 内部驱动,不进入公共 API。程序化体操作应使用等式约束无状态原语自行实现。


6. 封装与隔离

6.1 机制总览

本架构通过多层引导让"正确方式"成为阻力最小的路径。OrcaGymEulerEnv 直接继承 gym.Env,不创建 gym/stub/channel 属性,Python 原生即拒绝访问;其余内部对象依赖 _ 前缀约定 + ruff SLF001 静态检查 + AGENTS.md AI 行为约束:

机制 实现 效果
Python 原生属性不存在 OrcaGymEulerEnv 直接继承 gym.Env,__init__ 中赋值的内部组件均带下划线前缀(如 _gym/_stub/_channel/_studio_bridge 等),不创建无前缀的 gym/stub/channel 属性 env.gym/env.stub/env.channel 抛 AttributeError
ruff SLF001 静态检查 配置 ruff check --select SLF001,扫描外部访问 _ 前缀属性的代码 提交前/CI 阶段检测穿墙访问
AGENTS.md AI 约束 每个自研仓库根目录配置 AGENTS.md,明文禁止 AI 使用 _ 前缀属性 从输入端约束 AI 代码生成行为
__dir__ 控制 Env/Gym/DataView 实现 __dir__,只暴露公共 API IDE 自动补全引导正确路径
DataView 兜底 OrcaGymDataView.__getattr__ 缺字段时引导扩展 缺功能时引导扩展而非绕道
类型标注 公共方法返回 typed 对象,不返回 mujoco.MjData AI 代码生成走正确路径
docstring 契约 类文档明确列出正确用法和禁止事项 阅读 API 即知契约
路径深度 _mjData 在 env._gym._sim._mjData 三层之下 天然屏障

6.2 隔离效果对比

场景 触发机制 你/AI 看到的
AI 生成 env._mjData.qpos ruff SLF001 提交前报警:用 env.data.qpos
AI 生成 env._gym._mjData ruff SLF001 提交前报警:用 env.data
AI 生成 env._mjModel.opt.iterations ruff SLF001 提交前报警:用 env.sim_config.iterations
AI 在 IDE 中补全 env. __dir__ 控制 只看到公共 API
AI 阅读 class docstring 类型标注 + 契约 知道正确用法和禁止事项
AI 未执行 ruff 直接提交 AGENTS.md 约束 + CI 门禁 CI 拒绝合并

6.3 与旧体系的隔离强度对比

系统 绕道路径 层数 内部组件是否可见 静态检查
OrcaGymLocalEnv env.gym._mjData 2 gym 是公共属性 无
OrcaGymEulerEnv env._gym._sim._mjData 3 __dir__ 不列出 ruff SLF001 报警

7. 步进编排

7.1 do_simulation 内部流程

def do_simulation(self, ctrl: np.ndarray, n_frames: int):
    """标准仿真步进(含 Euler 耦合)。

    契约:
    - 设置控制输入 → 步进 n_frames 次 → 同步状态
    - Euler 耦合通过 step_with_coupling 封装(不写 if self._gym._euler is not None)
    - 步进完成后 self.data 保证一致
    """
    # K8 合规: 不写 if self._gym._euler is not None,通过 step_with_coupling 封装
    self._gym.step_with_coupling(ctrl, n_frames, self.dt)
    self._gym.sync_to_view()

实际实现见 orca_gym/environment/euler/orca_gym_euler_env.py 的 do_simulation。step_with_coupling 在 has_euler()=False(当前骨架阶段)时等价于 set_ctrl + step,后续 Euler 耦合实现时扩展。

7.2 两种使用模式

模式 A(委托式,推荐):

def step(self, action):
    torque = self._compute_torque(action)
    self.do_simulation(torque, self.frame_skip)
    obs = self._get_obs()
    return obs, reward, terminated, truncated, info

模式 B(手动循环式):

def step(self, action):
    for _ in range(self.frame_skip):
        torque = self._compute_torque(action)
        self.set_ctrl(torque)
        self.mj_step(nstep=1)
    # 循环结束后刷新派生量,再读取 env.data
    self.mj_forward()
    obs = self._get_obs()
    return obs, reward, terminated, truncated, info

契约:模式 B 当前与 OrcaGymLocalEnv 行为一致(纯 MuJoCo)。若未来需要 Euler 耦合,模式 B 用户必须改用模式 A。


8. 迁移指南

8.1 迁移代价

API 类别 兼容难度 说明
生命周期与属性 低 model/data/ctrl/frame_skip 等原样提供
仿真步进(模式 A) 低 do_simulation 内部委托,签名一致
仿真步进(模式 B) 中 mj_step(1) 行为需注意无 Euler 耦合
状态查询 低 query_* 方法原样复制
状态设置 低 set_* 方法原样复制 + 新增 apply_body_force
名称空间解析 低 joint()/body()/site() 等原样提供
_mjData/_mjModel 直接访问 低 有正式 API 替代,机械替换
Studio 交互 低 gRPC 逻辑原样复制

总体:约 70% 零改动,25% 机械替换,5% 设计调整。

8.2 83 处直接访问的替代方案

读取类(→ OrcaGymDataView)

旧代码 新 API
gym._mjData.qpos env.data.qpos
gym._mjData.qvel env.data.qvel
gym._mjData.body(id).xpos env.data.body_xpos(name)
gym._mjData.cvel[id] env.data.body_cvel(name)
gym._mjData.xpos[body_id, 2] env.data.body_xpos(name)[2]
gym._mjData.time env.data.time

写入类(→ 显式方法)

旧代码 新 API
gym._mjData.xfrc_applied[id, :3] = f env.apply_body_force(name, f, tau)
gym._mjData.xfrc_applied[id].fill(0) env.clear_body_force(name)
gym._mjData.eq_active[gi] = bool env.equality_update(slot, active=bool)

配置类(→ SimConfig)

旧代码 新 API
gym._mjModel.opt.timestep = 0.002 env.sim_config.timestep = 0.002
gym._mjModel.opt.iterations = 100 env.sim_config.iterations = 100
gym._mjModel.opt.integrator = 0 env.sim_config.integrator = 0
gym._mjModel.opt.gravity = ... env.sim_config.gravity = ...
30 行 opt.* 设置 env.sim_config.load_from_dict({...})

模型结构类(→ env 公共方法 / OrcaGymModel)

旧代码 新 API
gym._mjModel.body_subtreemass[id] env.body_subtree_mass(name) 或 env.data.body_subtree_mass(name)
gym._mjModel.eq_data.shape[1] 当前 env.model 无此公共 API,需通过 env._gym.equality_data_width() 或扩展 env.model 公共方法
gym._mjModel.eq_obj1id[gi] 当前 env.model 无此公共 API,需通过 env._gym.equality_object_ids(idx) 或扩展 env.model 公共方法
gym._mjModel.joint(i).name env.model.joint_id2name(i)
gym._mjModel.njnt len(env.model.get_joint_dict())

8.3 迁移分类示例

第一类:零改动(业务逻辑保持)

关节查询、状态设置、名称空间、渲染、步进——API 签名完全一致。

第二类:机械替换(_mjData → 正式 API)

# 迁移前
self.gym._mjData.xfrc_applied[body_id, :3] = force

# 迁移后
self.apply_body_force(body_name, force, torque)

第三类:设计调整(少数情况)

# 迁移前(Local 体系):手动循环步进
for _ in range(self.frame_skip):
    self.set_ctrl(torques)
    self.mj_step(nstep=1)
    self.gym.update_data()   # Local 体系的老 API

# 迁移后(Euler 体系):若需 Euler 耦合,改用 do_simulation
# do_simulation 内部已封装 step_with_coupling + sync_to_view,数据自动同步
self.do_simulation(torques, self.frame_skip)

9. 设计决策

9.1 为何直接继承 gym.Env + OrcaGymEnvMixin

OrcaGymEulerEnv 直接继承 gym.Env,通过 OrcaGymEnvMixin 共享公共方法,不继承 OrcaGymBaseEnv 也不继承 OrcaGymLocalEnv。

理由: - OrcaGymLocalEnv 是上帝类,继承会继承所有职责耦合 - OrcaGymLocal 的 _mjModel/_mjData 暴露设计与 P2 原则冲突 - OrcaGymBaseEnv 的 self.gym/self.model/self.data 赋值与 Euler 体系直接冲突,需要 __setattr__ 屏蔽等补丁机制 - 直接继承 gym.Env 后,env.gym/env.stub/env.channel 天然不存在(Python 原生 AttributeError),无需任何拦截机制 - OrcaGymEnvMixin 抽取名称空间、空间生成、reset 编排等与引擎无关的方法,避免代码重复

9.2 为何放弃 MuJoCoAdapter

不提供 MuJoCoAdapter(受控的 MuJoCo 句柄适配器)。

理由:原需求来自 robosuite 控制器等外部库需要直接操作 MuJoCo 对象,决定不再支持 robosuite 组件后该需求消失。放弃后设计更简洁——没有"逃生舱",所有需求都通过扩展 Env/Gym 的公共方法解决。

9.3 为何 _mjData/_mjModel 放在 MuJoCoSimCore

_mjModel/_mjData 只存在于 MuJoCoSimCore 内部,OrcaGymEuler 和 OrcaGymEulerEnv 不持有引用。

理由: - 增加绕道路径深度(env._gym._sim._mjData 三层) - 职责内聚:MuJoCo 原生操作集中在 MuJoCoSimCore - OrcaGymEuler 作为 Facade 只协调,不直接操作引擎数据

9.4 为何保留 self.gym 概念但重命名为 _gym

OrcaGymEulerEnv 持有 _gym: OrcaGymEuler,但不作为公共属性暴露。

理由: - 保留分层结构(Env → Gym → SimCore)便于职责划分 - _gym 不在 __dir__ 中列出,AI 难以发现 - 你通过 Env 的公共方法间接使用 Gym,不直接接触


10. 总结

本文档的核心要点:

  1. Facade + 职责内聚分解替代上帝类,组件按职责划分为 MuJoCoSimCore/OrcaStudioBridge/ModelRegistry/SimConfig(_euler 字段为占位,当前未实现)
  2. 直接继承 gym.Env + OrcaGymEnvMixin:不继承 OrcaGymBaseEnv,公共方法通过 Mixin 共享
  3. 完备的公共 API 契约覆盖所有合法 MuJoCo 操作需求,消除绕道理由
  4. 多层封装隔离(ruff SLF001 + AGENTS.md + Python 原生属性不存在 + __dir__ + DataView 兜底 + 类型标注 + docstring)引导你和 AI 走正确路径
  5. 步进编排契约明确 do_simulation(含耦合)与 mj_step(纯 MuJoCo)的语义区分
  6. 迁移策略约 70% 零改动、25% 机械替换、5% 设计调整