MDP中的events部分

events.py 文件的功能

1. 域随机化（Domain Randomization）核心引擎
这是 Isaac Lab 实现域随机化的主要工具，用于：

物理参数随机化： 摩擦系数、质量、惯性等
执行器参数随机化： 关节刚度、阻尼等
环境条件随机化： 重力、外力干扰等

2. 强化学习训练的关键支撑
（1）提高策略泛化能力
通过随机化训练环境，让智能体学会适应各种条件
避免过拟合到特定的仿真参数
提高在真实世界部署时的成功率
（2）增强训练数据多样性
每个训练回合都有不同的物理条件
模拟现实世界的不确定性
让策略更加鲁棒

3. 仿真到现实（Sim-to-Real）的桥梁
缩小仿真与现实的差距
模拟真实世界干扰

4. 环境状态管理

均匀重置：在指定范围内随机重置
地形感知重置：根据地形特征选择合适的重置位置
渐进式重置：根据训练进度调整重置难度

isaaclab中的events.py

路径：IsaacLab/source/isaaclab/isaaclab/envs/mdp/events.py

randomize_rigid_body_material

randomize_rigid_body_material 类：用于随机化刚体材质的物理属性（静摩擦、动摩擦、恢复系数）

class randomize_rigid_body_material(ManagerTermBase):"""随机化资产所有几何体上的物理材质。此函数创建一组具有随机静摩擦、动摩擦和恢复系数的物理材质。材质数量由 ``num_buckets`` 指定。材质属性通过在给定范围内采样均匀随机值生成。然后这些材质属性被分配给资产的几何体。分配方式是为每个资产实例和形状创建随机整数张量，其中 ``num_instances`` 是生成的资产数量，``max_num_shapes`` 是资产中的最大形状数（所有实体中）。整数值用作索引从材质桶中选择材质属性。如果标志 ``make_consistent`` 设为 ``True``，动摩擦将被设置为小于或等于静摩擦，这遵守物理约束。但对于某些应用场景可能不是必需的，因此默认设为 ``False``。.. attention::此函数使用 CPU 张量分配材质属性，建议仅在环境初始化时使用，否则可能导致显著的性能开销。.. note::PhysX 只允许场景中有 64000 个独特的物理材质。如果超过此限制，模拟将崩溃。因此，我们在初始化时仅采样一次材质，随后将这些材质随机分配给资产的几何体。"""def __init__(self, cfg: EventTermCfg, env: ManagerBasedEnv):"""初始化事件项。Args:cfg: 事件项的配置。env: 环境实例。Raises:ValueError: 如果资产不是 RigidObject 或 Articulation 类型。"""super().__init__(cfg, env)# 提取使用的量（启用类型提示）self.asset_cfg: SceneEntityCfg = cfg.params["asset_cfg"]self.asset: RigidObject | Articulation = env.scene[self.asset_cfg.name]# 检查资产类型是否支持if not isinstance(self.asset, (RigidObject, Articulation)):raise ValueError(f"随机化项 'randomize_rigid_body_material' 不支持资产类型: '{self.asset_cfg.name}'"f" (类型: '{type(self.asset)}')。")# 获取每个实体的形状数量（用于正确索引材质属性）# 注意：这是一个变通方法，因为 Articulation 没有直接提供获取每个实体形状数量的方法。# 我们使用物理模拟视图来获取每个实体的形状数量。if isinstance(self.asset, Articulation) and self.asset_cfg.body_ids != slice(None):self.num_shapes_per_body = []# 遍历所有链接路径for link_path in self.asset.root_physx_view.link_paths[0]:link_physx_view = self.asset._physics_sim_view.create_rigid_body_view(link_path)  # type: ignoreself.num_shapes_per_body.append(link_physx_view.max_shapes)# 确保解析正确num_shapes = sum(self.num_shapes_per_body)expected_shapes = self.asset.root_physx_view.max_shapesif num_shapes != expected_shapes:raise ValueError("随机化项 'randomize_rigid_body_material' 未能正确解析每个实体的形状数量。"f" 预期形状总数: {expected_shapes}, 实际: {num_shapes}。")else:# 这种情况下不需要特殊索引self.num_shapes_per_body = None# 获取摩擦和恢复系数的采样参数static_friction_range = cfg.params.get("static_friction_range", (1.0, 1.0))dynamic_friction_range = cfg.params.get("dynamic_friction_range", (1.0, 1.0))restitution_range = cfg.params.get("restitution_range", (0.0, 0.0))num_buckets = int(cfg.params.get("num_buckets", 1))# 从给定范围采样材质属性# 注意：我们仅在初始化时采样一次材质，之后随机分配给资产的几何体range_list = [static_friction_range, dynamic_friction_range, restitution_range]ranges = torch.tensor(range_list, device="cpu")self.material_buckets = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (num_buckets, 3), device="cpu")# 确保动摩擦始终小于静摩擦（如果需要）make_consistent = cfg.params.get("make_consistent", False)if make_consistent:self.material_buckets[:, 1] = torch.min(self.material_buckets[:, 0], self.material_buckets[:, 1])def __call__(self,env: ManagerBasedEnv,env_ids: torch.Tensor | None,static_friction_range: tuple[float, float],dynamic_friction_range: tuple[float, float],restitution_range: tuple[float, float],num_buckets: int,asset_cfg: SceneEntityCfg,make_consistent: bool = False,):# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device="cpu")else:env_ids = env_ids.cpu()# 随机分配材质ID给几何体total_num_shapes = self.asset.root_physx_view.max_shapesbucket_ids = torch.randint(0, num_buckets, (len(env_ids), total_num_shapes), device="cpu")material_samples = self.material_buckets[bucket_ids]# 从物理模拟中检索材质缓冲区materials = self.asset.root_physx_view.get_material_properties()# 使用新样本更新材质缓冲区if self.num_shapes_per_body is not None:# 按实体分组处理形状for body_id in self.asset_cfg.body_ids:# 获取该实体的形状索引范围start_idx = sum(self.num_shapes_per_body[:body_id])end_idx = start_idx + self.num_shapes_per_body[body_id]# 分配新材质 (material_samples形状: num_env_ids x total_num_shapes x 3)materials[env_ids, start_idx:end_idx] = material_samples[:, start_idx:end_idx]else:# 分配所有材质materials[env_ids] = material_samples[:]# 应用到模拟self.asset.root_physx_view.set_material_properties(materials, env_ids)

randomize_rigid_body_mass

随机化刚体质量

def randomize_rigid_body_mass(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,mass_distribution_params: tuple[float, float],operation: Literal["add", "scale", "abs"],distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",recompute_inertia: bool = True,
):"""通过加法、缩放或设置随机值来随机化实体的质量。此函数允许随机化资产实体的质量。它从给定分布参数采样随机值，然后根据操作类型将这些值添加到、缩放到或设置为物理模拟中的质量。如果 ``recompute_inertia`` 标志设为 ``True``，函数在设置质量后重新计算实体的惯性张量。这在质量发生显著变化时非常有用，因为惯性张量依赖于质量。它假设实体是均匀密度物体，若非如此，惯性张量可能不准确。.. tip::此函数使用 CPU 张量分配实体质量，建议仅在环境初始化时使用。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device="cpu")else:env_ids = env_ids.cpu()# 解析实体索引if asset_cfg.body_ids == slice(None):body_ids = torch.arange(asset.num_bodies, dtype=torch.int, device="cpu")else:body_ids = torch.tensor(asset_cfg.body_ids, dtype=torch.int, device="cpu")# 获取实体的当前质量 (num_assets, num_bodies)masses = asset.root_physx_view.get_masses()# 在默认值上应用随机化# 确保多次调用该函数时，随机化始终基于默认值而不是先前随机化的值masses[env_ids[:, None], body_ids] = asset.data.default_mass[env_ids[:, None], body_ids].clone()# 从给定范围采样# 注意：我们原地修改所有环境的质量，但设置器确保只修改指定环境的实体质量masses = _randomize_prop_by_op(masses, mass_distribution_params, env_ids, body_ids, operation=operation, distribution=distribution)# 将质量设置到物理模拟中asset.root_physx_view.set_masses(masses, env_ids)# 如果需要则重新计算惯性张量if recompute_inertia:# 计算新质量与初始质量的比率ratios = masses[env_ids[:, None], body_ids] / asset.data.default_mass[env_ids[:, None], body_ids]# 按比率缩放惯性张量inertias = asset.root_physx_view.get_inertias()if isinstance(asset, Articulation):# 惯性形状: (num_envs, num_bodies, 9) 用于多刚体inertias[env_ids[:, None], body_ids] = (asset.data.default_inertia[env_ids[:, None], body_ids] * ratios[..., None])else:# 惯性形状: (num_envs, 9) 用于刚体inertias[env_ids] = asset.data.default_inertia[env_ids] * ratios# 将惯性张量设置到物理模拟中asset.root_physx_view.set_inertias(inertias, env_ids)

randomize_rigid_body_collider_offsets

随机化碰撞体参数

def randomize_rigid_body_collider_offsets(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,rest_offset_distribution_params: tuple[float, float] | None = None,contact_offset_distribution_params: tuple[float, float] | None = None,distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""通过添加、缩放或设置随机值来随机化刚体碰撞体参数。此函数允许随机化资产的碰撞体参数，如静止偏移和接触偏移，这些影响碰撞检测的物理引擎属性。函数从给定分布参数采样随机值并将操作应用于碰撞体属性，然后设置到物理模拟中。如果未提供某个属性的分布参数，则不修改该属性。当前分布参数作为绝对值应用。.. tip::此函数使用 CPU 张量分配碰撞属性，建议仅在环境初始化时使用。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device="cpu")# 从给定范围采样碰撞属性并设置到物理模拟# -- 静止偏移if rest_offset_distribution_params is not None:rest_offset = asset.root_physx_view.get_rest_offsets().clone()rest_offset = _randomize_prop_by_op(rest_offset,rest_offset_distribution_params,None,slice(None),  # 应用于所有形状operation="abs",distribution=distribution,)asset.root_physx_view.set_rest_offsets(rest_offset, env_ids.cpu())# -- 接触偏移if contact_offset_distribution_params is not None:contact_offset = asset.root_physx_view.get_contact_offsets().clone()contact_offset = _randomize_prop_by_op(contact_offset,contact_offset_distribution_params,None,slice(None),  # 应用于所有形状operation="abs",distribution=distribution,)asset.root_physx_view.set_contact_offsets(contact_offset, env_ids.cpu())

randomize_physics_scene_gravity

随机化场景重力

def randomize_physics_scene_gravity(env: ManagerBasedEnv,env_ids: torch.Tensor | None,gravity_distribution_params: tuple[list[float], list[float]],operation: Literal["add", "scale", "abs"],distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""通过添加、缩放或设置随机值来随机化物理场景重力。此函数允许随机化物理场景的重力。它从给定分布参数采样随机值，并根据操作类型将这些值添加到、缩放到或设置到物理模拟中。分布参数是两个元素的列表，分别表示重力向量的 x, y, z 分量的分布上下界。每个分量独立采样随机值。.. attention::此函数为所有环境应用相同的重力。.. tip::此函数使用 CPU 张量分配重力。"""# 获取当前重力gravity = torch.tensor(env.sim.cfg.gravity, device="cpu").unsqueeze(0)dist_param_0 = torch.tensor(gravity_distribution_params[0], device="cpu")dist_param_1 = torch.tensor(gravity_distribution_params[1], device="cpu")gravity = _randomize_prop_by_op(gravity,(dist_param_0, dist_param_1),None,  # 应用于所有环境slice(None),  # 所有分量operation=operation,distribution=distribution,)# 将重力张量解压为列表gravity = gravity[0].tolist()# 将重力设置到物理模拟中physics_sim_view: physx.SimulationView = sim_utils.SimulationContext.instance().physics_sim_viewphysics_sim_view.set_gravity(carb.Float3(*gravity))

randomize_actuator_gains

随机化执行器增益

def randomize_actuator_gains(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,stiffness_distribution_params: tuple[float, float] | None = None,damping_distribution_params: tuple[float, float] | None = None,operation: Literal["add", "scale", "abs"] = "abs",distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""通过添加、缩放或设置随机值来随机化多刚体中的执行器增益。此函数允许随机化执行器的刚度和阻尼增益。函数从给定分布参数采样随机值，并将操作应用于关节属性，然后设置到执行器模型中。如果未提供某个属性的分布参数，则不修改该属性。.. tip::对于隐式执行器，此函数使用 CPU 张量将执行器增益设置到模拟中。在这种情况下，建议仅在环境初始化时使用。"""# 提取使用的量（启用类型提示）asset: Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device=asset.device)# 内部随机化函数def randomize(data: torch.Tensor, params: tuple[float, float]) -> torch.Tensor:return _randomize_prop_by_op(data, params, dim_0_ids=None, dim_1_ids=actuator_indices, operation=operation, distribution=distribution)# 遍历执行器并随机化增益for actuator in asset.actuators.values():# 解析要随机化的关节索引if isinstance(asset_cfg.joint_ids, slice):# 取执行器的所有关节actuator_indices = slice(None)if isinstance(actuator.joint_indices, slice):global_indices = slice(None)else:global_indices = torch.tensor(actuator.joint_indices, device=asset.device)elif isinstance(actuator.joint_indices, slice):# 取资产配置中定义的关节global_indices = actuator_indices = torch.tensor(asset_cfg.joint_ids, device=asset.device)else:# 取执行器关节与资产配置关节的交集actuator_joint_indices = torch.tensor(actuator.joint_indices, device=asset.device)asset_joint_ids = torch.tensor(asset_cfg.joint_ids, device=asset.device)# 执行器中需要随机化的关节索引actuator_indices = torch.nonzero(torch.isin(actuator_joint_indices, asset_joint_ids)).view(-1)if len(actuator_indices) == 0:continue  # 没有需要处理的关节则跳过# 将执行器索引映射到全局关节索引global_indices = actuator_joint_indices[actuator_indices]# 随机化刚度if stiffness_distribution_params is not None:stiffness = actuator.stiffness[env_ids].clone()stiffness[:, actuator_indices] = asset.data.default_joint_stiffness[env_ids][:, global_indices].clone()randomize(stiffness, stiffness_distribution_params)actuator.stiffness[env_ids] = stiffness# 如果是隐式执行器，写入模拟if isinstance(actuator, ImplicitActuator):asset.write_joint_stiffness_to_sim(stiffness, joint_ids=actuator.joint_indices, env_ids=env_ids)# 随机化阻尼if damping_distribution_params is not None:damping = actuator.damping[env_ids].clone()damping[:, actuator_indices] = asset.data.default_joint_damping[env_ids][:, global_indices].clone()randomize(damping, damping_distribution_params)actuator.damping[env_ids] = damping# 如果是隐式执行器，写入模拟if isinstance(actuator, ImplicitActuator):asset.write_joint_damping_to_sim(damping, joint_ids=actuator.joint_indices, env_ids=env_ids)

randomize_joint_parameters

随机化关节参数

def randomize_joint_parameters(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,friction_distribution_params: tuple[float, float] | None = None,armature_distribution_params: tuple[float, float] | None = None,lower_limit_distribution_params: tuple[float, float] | None = None,upper_limit_distribution_params: tuple[float, float] | None = None,operation: Literal["add", "scale", "abs"] = "abs",distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""通过添加、缩放或设置随机值来随机化多刚体的关节参数。此函数允许随机化资产的关节参数，这些是影响关节行为的物理引擎属性。函数从给定分布参数采样随机值并将操作应用于关节属性，然后设置到物理模拟中。如果未提供某个属性的分布参数，则不修改该属性。.. tip::此函数使用 CPU 张量分配关节属性，建议仅在环境初始化时使用。"""# 提取使用的量（启用类型提示）asset: Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device=asset.device)# 解析关节索引if asset_cfg.joint_ids == slice(None):joint_ids = slice(None)  # 优化目的else:joint_ids = torch.tensor(asset_cfg.joint_ids, dtype=torch.int, device=asset.device)# 从给定范围采样关节属性并设置到物理模拟# -- 摩擦if friction_distribution_params is not None:friction = asset.data.default_joint_friction.to(asset.device).clone()friction = _randomize_prop_by_op(friction, friction_distribution_params, env_ids, joint_ids, operation=operation, distribution=distribution)[env_ids][:, joint_ids]asset.write_joint_friction_to_sim(friction, joint_ids=joint_ids, env_ids=env_ids)# -- 惯性if armature_distribution_params is not None:armature = asset.data.default_joint_armature.to(asset.device).clone()armature = _randomize_prop_by_op(armature, armature_distribution_params, env_ids, joint_ids, operation=operation, distribution=distribution)[env_ids][:, joint_ids]asset.write_joint_armature_to_sim(armature, joint_ids=joint_ids, env_ids=env_ids)# -- 自由度限制if lower_limit_distribution_params is not None or upper_limit_distribution_params is not None:dof_limits = asset.data.default_joint_limits.to(asset.device).clone()# 下限if lower_limit_distribution_params is not None:lower_limits = dof_limits[..., 0]lower_limits = _randomize_prop_by_op(lower_limits,lower_limit_distribution_params,env_ids,joint_ids,operation=operation,distribution=distribution,)[env_ids][:, joint_ids]dof_limits[env_ids[:, None], joint_ids, 0] = lower_limits# 上限if upper_limit_distribution_params is not None:upper_limits = dof_limits[..., 1]upper_limits = _randomize_prop_by_op(upper_limits,upper_limit_distribution_params,env_ids,joint_ids,operation=operation,distribution=distribution,)[env_ids][:, joint_ids]dof_limits[env_ids[:, None], joint_ids, 1] = upper_limits# 检查限制是否合法if (dof_limits[env_ids[:, None], joint_ids, 0] > dof_limits[env_ids[:, None], joint_ids, 1]).any():raise ValueError("随机化项 'randomize_joint_parameters' 设置的下关节限制大于上关节限制。")# 设置关节限制到模拟asset.write_joint_limits_to_sim(dof_limits[env_ids][:, joint_ids], joint_ids=joint_ids, env_ids=env_ids, warn_limit_violation=False)

randomize_fixed_tendon_parameters

随机化肌腱参数

def randomize_fixed_tendon_parameters(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,stiffness_distribution_params: tuple[float, float] | None = None,damping_distribution_params: tuple[float, float] | None = None,limit_stiffness_distribution_params: tuple[float, float] | None = None,lower_limit_distribution_params: tuple[float, float] | None = None,upper_limit_distribution_params: tuple[float, float] | None = None,rest_length_distribution_params: tuple[float, float] | None = None,offset_distribution_params: tuple[float, float] | None = None,operation: Literal["add", "scale", "abs"] = "abs",distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""通过添加、缩放或设置随机值来随机化固定肌腱参数。此函数允许随机化资产的肌腱参数，这些是影响关节行为的物理引擎属性。函数从给定分布参数采样随机值并将操作应用于肌腱属性，然后设置到物理模拟中。如果未提供某个属性的分布参数，则不修改该属性。"""# 提取使用的量（启用类型提示）asset: Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device=asset.device)# 解析关节索引if asset_cfg.fixed_tendon_ids == slice(None):fixed_tendon_ids = slice(None)  # 优化目的else:fixed_tendon_ids = torch.tensor(asset_cfg.fixed_tendon_ids, dtype=torch.int, device=asset.device)# 从给定范围采样肌腱属性并设置到物理模拟# -- 刚度if stiffness_distribution_params is not None:stiffness = asset.data.default_fixed_tendon_stiffness.clone()stiffness = _randomize_prop_by_op(stiffness,stiffness_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]asset.set_fixed_tendon_stiffness(stiffness, fixed_tendon_ids, env_ids)# -- 阻尼if damping_distribution_params is not None:damping = asset.data.default_fixed_tendon_damping.clone()damping = _randomize_prop_by_op(damping,damping_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]asset.set_fixed_tendon_damping(damping, fixed_tendon_ids, env_ids)# -- 限制刚度if limit_stiffness_distribution_params is not None:limit_stiffness = asset.data.default_fixed_tendon_limit_stiffness.clone()limit_stiffness = _randomize_prop_by_op(limit_stiffness,limit_stiffness_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]asset.set_fixed_tendon_limit_stiffness(limit_stiffness, fixed_tendon_ids, env_ids)# -- 限制if lower_limit_distribution_params is not None or upper_limit_distribution_params is not None:limit = asset.data.default_fixed_tendon_limit.clone()# 下限if lower_limit_distribution_params is not None:lower_limit = limit[..., 0]lower_limit = _randomize_prop_by_op(lower_limit,lower_limit_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]limit[env_ids[:, None], fixed_tendon_ids, 0] = lower_limit# 上限if upper_limit_distribution_params is not None:upper_limit = limit[..., 1]upper_limit = _randomize_prop_by_op(upper_limit,upper_limit_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]limit[env_ids[:, None], fixed_tendon_ids, 1] = upper_limit# 检查限制是否合法if (limit[env_ids[:, None], fixed_tendon_ids, 0] > limit[env_ids[:, None], fixed_tendon_ids, 1]).any():raise ValueError("随机化项 'randomize_fixed_tendon_parameters' 设置的下肌腱限制大于上肌腱限制。")# 设置限制asset.set_fixed_tendon_limit(limit, fixed_tendon_ids, env_ids)# -- 静止长度if rest_length_distribution_params is not None:rest_length = asset.data.default_fixed_tendon_rest_length.clone()rest_length = _randomize_prop_by_op(rest_length,rest_length_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]asset.set_fixed_tendon_rest_length(rest_length, fixed_tendon_ids, env_ids)# -- 偏移if offset_distribution_params is not None:offset = asset.data.default_fixed_tendon_offset.clone()offset = _randomize_prop_by_op(offset,offset_distribution_params,env_ids,fixed_tendon_ids,operation=operation,distribution=distribution,)[env_ids][:, fixed_tendon_ids]asset.set_fixed_tendon_offset(offset, fixed_tendon_ids, env_ids)# 将肌腱属性写入物理模拟asset.write_fixed_tendon_properties_to_sim(fixed_tendon_ids, env_ids)

apply_external_force_torque

施加随机外力

def apply_external_force_torque(env: ManagerBasedEnv,env_ids: torch.Tensor,force_range: tuple[float, float],torque_range: tuple[float, float],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""对实体施加随机外部力和扭矩。此函数创建一组从给定范围采样的随机力和扭矩。力和扭矩的数量等于实体数量乘以环境数量。通过调用 ``asset.set_external_force_and_torque`` 施加到实体上。只有调用 ``asset.write_data_to_sim()`` 时才会实际应用。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device=asset.device)# 解析实体数量num_bodies = len(asset_cfg.body_ids) if isinstance(asset_cfg.body_ids, list) else asset.num_bodies# 采样随机力和扭矩size = (len(env_ids), num_bodies, 3)forces = math_utils.sample_uniform(*force_range, size, asset.device)torques = math_utils.sample_uniform(*torque_range, size, asset.device)# 设置到缓冲区# 注意：只有在调用 `asset.write_data_to_sim()` 时才会应用asset.set_external_force_and_torque(forces, torques, env_ids=env_ids, body_ids=asset_cfg.body_ids)

push_by_setting_velocity

通过设置速度模拟推力

def push_by_setting_velocity(env: ManagerBasedEnv,env_ids: torch.Tensor,velocity_range: dict[str, tuple[float, float]],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""通过将根速度设置为给定范围内的随机值来推动资产。这会产生类似于用随机脉冲推动资产的效果，改变资产的速度。它从给定范围采样根速度并设置到物理模拟中。函数接受每个轴和旋转的速度范围字典。字典的键是：``x``, ``y``, ``z``, ``roll``, ``pitch``, ``yaw``。值是 ``(min, max)`` 形式的元组。如果字典缺少某个键，则将该轴的流速设为零。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 获取当前速度vel_w = asset.data.root_vel_w[env_ids]# 采样随机速度range_list = [velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)vel_w += math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], vel_w.shape, device=asset.device)# 将速度设置到物理模拟asset.write_root_velocity_to_sim(vel_w, env_ids=env_ids)

reset_root_state_uniform

均匀随机重置根状态

def reset_root_state_uniform(env: ManagerBasedEnv,env_ids: torch.Tensor,pose_range: dict[str, tuple[float, float]],velocity_range: dict[str, tuple[float, float]],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""将资产根状态重置为给定范围内的随机位置和速度。此函数随机化资产的根位置和速度：* 从给定范围采样根位置，添加到默认根位置，然后设置到物理模拟。* 从给定范围采样根方向，设置到物理模拟。* 从给定范围采样根速度，设置到物理模拟。函数接受每个轴和旋转的位置和速度范围字典。字典的键是：``x``, ``y``, ``z``, ``roll``, ``pitch``, ``yaw``。值是 ``(min, max)`` 形式的元组。如果字典缺少某个键，则将该轴的位置或速度设为零。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 获取默认根状态root_states = asset.data.default_root_state[env_ids].clone()# 位置和方向range_list = [pose_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 6), device=asset.device)# 计算位置和方向positions = root_states[:, 0:3] + env.scene.env_origins[env_ids] + rand_samples[:, 0:3]orientations_delta = math_utils.quat_from_euler_xyz(rand_samples[:, 3], rand_samples[:, 4], rand_samples[:, 5])orientations = math_utils.quat_mul(root_states[:, 3:7], orientations_delta)# 速度range_list = [velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 6), device=asset.device)velocities = root_states[:, 7:13] + rand_samples# 设置到物理模拟asset.write_root_pose_to_sim(torch.cat([positions, orientations], dim=-1), env_ids=env_ids)asset.write_root_velocity_to_sim(velocities, env_ids=env_ids)

reset_root_state_with_random_orientation

随机方向重置根状态

def reset_root_state_with_random_orientation(env: ManagerBasedEnv,env_ids: torch.Tensor,pose_range: dict[str, tuple[float, float]],velocity_range: dict[str, tuple[float, float]],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""将资产根位置和速度重置为给定范围内的随机值，并从 SO(3) 均匀采样根方向。此函数随机化资产的根位置和速度：* 从给定范围采样根位置，添加到默认根位置，然后设置到物理模拟。* 从 SO(3) 均匀采样根方向，设置到物理模拟。* 从给定范围采样根速度，设置到物理模拟。函数接受以下字典：* :attr:`pose_range` - 每个轴的位置范围字典，键是 ``x``, ``y``, ``z``。方向从 SO(3) 均匀采样。* :attr:`velocity_range` - 每个轴和旋转的速度范围字典，键是：``x``, ``y``, ``z``, ``roll``, ``pitch``, ``yaw``。值是 ``(min, max)`` 形式的元组。如果字典缺少某个键，则将该轴的位置设为零。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 获取默认根状态root_states = asset.data.default_root_state[env_ids].clone()# 位置range_list = [pose_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 3), device=asset.device)positions = root_states[:, 0:3] + env.scene.env_origins[env_ids] + rand_samplesorientations = math_utils.random_orientation(len(env_ids), device=asset.device)  # SO(3)均匀采样# 速度range_list = [velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 6), device=asset.device)velocities = root_states[:, 7:13] + rand_samples# 设置到物理模拟asset.write_root_pose_to_sim(torch.cat([positions, orientations], dim=-1), env_ids=env_ids)asset.write_root_velocity_to_sim(velocities, env_ids=env_ids)

reset_root_state_from_terrain

从地形采样重置根状态

def reset_root_state_from_terrain(env: ManagerBasedEnv,env_ids: torch.Tensor,pose_range: dict[str, tuple[float, float]],velocity_range: dict[str, tuple[float, float]],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""通过从地形的平坦区域采样随机有效位置来重置资产根状态。此函数从地形采样随机有效位置（基于平坦区域），并将资产的根状态设置为该位置。函数还从给定范围采样随机速度并设置到物理模拟中。函数接受以下字典：* :attr:`pose_range` - 每个轴的方向范围字典，键是 ``roll``, ``pitch``, ``yaw``。位置从地形的平坦区域采样。* :attr:`velocity_range` - 每个轴和旋转的速度范围字典，键是：``x``, ``y``, ``z``, ``roll``, ``pitch``, ``yaw``。值是 ``(min, max)`` 形式的元组。如果字典缺少某个键，则将该轴的位置设为零。Note:函数期望地形在键 "init_pos" 下有有效的平坦区域，用于采样机器人的随机位置。Raises:ValueError: 如果地形没有有效的平坦区域。"""# 提取使用的量（启用类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]terrain: TerrainImporter = env.scene.terrain  # 地形导入器# 获取对应于有效位置的所有平坦区域valid_positions: torch.Tensor = terrain.flat_patches.get("init_pos")if valid_positions is None:raise ValueError("事件项 'reset_root_state_from_terrain' 需要 'init_pos' 下的有效平坦区域。"f" 找到: {list(terrain.flat_patches.keys())}")# 采样随机有效位置ids = torch.randint(0, valid_positions.shape[2], size=(len(env_ids),), device=env.device)positions = valid_positions[terrain.terrain_levels[env_ids], terrain.terrain_types[env_ids], ids]positions += asset.data.default_root_state[env_ids, :3]  # 添加默认偏移# 采样随机方向range_list = [pose_range.get(key, (0.0, 0.0)) for key in ["roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 3), device=asset.device)# 转换为四元数orientations = math_utils.quat_from_euler_xyz(rand_samples[:, 0], rand_samples[:, 1], rand_samples[:, 2])# 采样随机速度range_list = [velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z", "roll", "pitch", "yaw"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 6), device=asset.device)velocities = asset.data.default_root_state[env_ids, 7:13] + rand_samples# 设置到物理模拟asset.write_root_pose_to_sim(torch.cat([positions, orientations], dim=-1), env_ids=env_ids)asset.write_root_velocity_to_sim(velocities, env_ids=env_ids)

reset_joints_by_scale

按比例缩放重置关节状态

def reset_joints_by_scale(env: ManagerBasedEnv,env_ids: torch.Tensor,position_range: tuple[float, float],velocity_range: tuple[float, float],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""通过使用给定范围缩放默认位置和速度来重置机器人关节。此函数从给定范围采样随机值，并将默认关节位置和速度缩放这些值，然后设置到物理模拟中。"""# 提取使用的量（启用类型提示）asset: Articulation = env.scene[asset_cfg.name]# 获取默认关节状态joint_pos = asset.data.default_joint_pos[env_ids].clone()joint_vel = asset.data.default_joint_vel[env_ids].clone()# 随机缩放这些值joint_pos *= math_utils.sample_uniform(*position_range, joint_pos.shape, joint_pos.device)joint_vel *= math_utils.sample_uniform(*velocity_range, joint_vel.shape, joint_vel.device)# 将关节位置限制在合法范围内joint_pos_limits = asset.data.soft_joint_pos_limits[env_ids]joint_pos = joint_pos.clamp_(joint_pos_limits[..., 0], joint_pos_limits[..., 1])# 将关节速度限制在合法范围内joint_vel_limits = asset.data.soft_joint_vel_limits[env_ids]joint_vel = joint_vel.clamp_(-joint_vel_limits, joint_vel_limits)# 设置到物理模拟asset.write_joint_state_to_sim(joint_pos, joint_vel, env_ids=env_ids)

reset_joints_by_offset

按偏移量重置关节状态

def reset_joints_by_offset(env: ManagerBasedEnv,env_ids: torch.Tensor,position_range: tuple[float, float],velocity_range: tuple[float, float],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""使用围绕默认位置和速度的偏移量重置机器人关节。此函数从给定范围采样随机值，并将默认关节位置和速度偏移这些值，然后设置到物理模拟中。"""# 提取使用的量（启用类型提示）asset: Articulation = env.scene[asset_cfg.name]# 获取默认关节状态joint_pos = asset.data.default_joint_pos[env_ids].clone()joint_vel = asset.data.default_joint_vel[env_ids].clone()# 随机偏移这些值joint_pos += math_utils.sample_uniform(*position_range, joint_pos.shape, joint_pos.device)joint_vel += math_utils.sample_uniform(*velocity_range, joint_vel.shape, joint_vel.device)# 将关节位置限制在合法范围内joint_pos_limits = asset.data.soft_joint_pos_limits[env_ids]joint_pos = joint_pos.clamp_(joint_pos_limits[..., 0], joint_pos_limits[..., 1])# 将关节速度限制在合法范围内joint_vel_limits = asset.data.soft_joint_vel_limits[env_ids]joint_vel = joint_vel.clamp_(-joint_vel_limits, joint_vel_limits)# 设置到物理模拟asset.write_joint_state_to_sim(joint_pos, joint_vel, env_ids=env_ids)

reset_nodal_state_uniform

def reset_nodal_state_uniform(env: ManagerBasedEnv,env_ids: torch.Tensor,position_range: dict[str, tuple[float, float]],velocity_range: dict[str, tuple[float, float]],asset_cfg: SceneEntityCfg = SceneEntityCfg("robot"),
):"""将资产节点状态重置为给定范围内的随机位置和速度。此函数随机化资产的节点位置和速度：* 从给定范围采样节点位置，添加到默认节点位置，然后设置到物理模拟。* 从给定范围采样节点速度，设置到物理模拟。函数接受每个轴的位置和速度范围字典。字典的键是：``x``, ``y``, ``z``。值是 ``(min, max)`` 形式的元组。如果字典缺少某个键，则将该轴的位置或速度设为零。"""# 提取使用的量（启用类型提示）asset: DeformableObject = env.scene[asset_cfg.name]# 获取默认根状态nodal_state = asset.data.default_nodal_state_w[env_ids].clone()# 位置range_list = [position_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 1, 3), device=asset.device)nodal_state[..., :3] += rand_samples# 速度range_list = [velocity_range.get(key, (0.0, 0.0)) for key in ["x", "y", "z"]]ranges = torch.tensor(range_list, device=asset.device)rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 1, 3), device=asset.device)nodal_state[..., 3:] += rand_samples# 设置到物理模拟asset.write_nodal_state_to_sim(nodal_state, env_ids=env_ids)

reset_scene_to_default

def reset_scene_to_default(env: ManagerBasedEnv, env_ids: torch.Tensor):"""将场景重置为场景配置中指定的默认状态。"""# 刚体重置for rigid_object in env.scene.rigid_objects.values():# 获取默认状态并处理环境原点的偏移default_root_state = rigid_object.data.default_root_state[env_ids].clone()default_root_state[:, 0:3] += env.scene.env_origins[env_ids]# 设置到物理模拟rigid_object.write_root_pose_to_sim(default_root_state[:, :7], env_ids=env_ids)rigid_object.write_root_velocity_to_sim(default_root_state[:, 7:], env_ids=env_ids)# 多刚体重置for articulation_asset in env.scene.articulations.values():# 获取默认状态并处理环境原点的偏移default_root_state = articulation_asset.data.default_root_state[env_ids].clone()default_root_state[:, 0:3] += env.scene.env_origins[env_ids]# 设置到物理模拟articulation_asset.write_root_pose_to_sim(default_root_state[:, :7], env_ids=env_ids)articulation_asset.write_root_velocity_to_sim(default_root_state[:, 7:], env_ids=env_ids)# 获取默认关节位置default_joint_pos = articulation_asset.data.default_joint_pos[env_ids].clone()default_joint_vel = articulation_asset.data.default_joint_vel[env_ids].clone()# 设置到物理模拟articulation_asset.write_joint_state_to_sim(default_joint_pos, default_joint_vel, env_ids=env_ids)# 可变形物体重置for deformable_object in env.scene.deformable_objects.values():# 获取默认节点状态nodal_state = deformable_object.data.default_nodal_state_w[env_ids].clone()deformable_object.write_nodal_state_to_sim(nodal_state, env_ids=env_ids)

_randomize_prop_by_op

对输入数据张量的指定部分（通过索引选择）进行三种操作之一：加法、乘法或绝对赋值
支持三种随机分布：均匀分布、对数均匀分布和高斯分布

# 内部辅助函数 --------------------------------------------------------------def _randomize_prop_by_op(data: torch.Tensor,distribution_parameters: tuple[float | torch.Tensor, float | torch.Tensor],dim_0_ids: torch.Tensor | None,dim_1_ids: torch.Tensor | slice,operation: Literal["add", "scale", "abs"],distribution: Literal["uniform", "log_uniform", "gaussian"],
) -> torch.Tensor:"""基于给定操作和分布执行数据随机化。Args:data: 要随机化的数据张量，形状为 (dim_0, dim_1)。distribution_parameters: 采样分布的参数。dim_0_ids: 要随机化的第一维度的索引。dim_1_ids: 要随机化的第二维度的索引。operation: 对数据执行的操作：'add'（添加）、'scale'（缩放）、'abs'（绝对设置）。distribution: 采样分布：'uniform'（均匀）、'log_uniform'（对数均匀）、'gaussian'（高斯）。Returns:随机化后的数据张量，形状为 (dim_0, dim_1)。Raises:NotImplementedError: 如果操作或分布不支持。"""# 解析形状# -- 维度 0if dim_0_ids is None:n_dim_0 = data.shape[0]dim_0_ids = slice(None)else:n_dim_0 = len(dim_0_ids)if not isinstance(dim_1_ids, slice):dim_0_ids = dim_0_ids[:, None]  # 为广播准备# -- 维度 1if isinstance(dim_1_ids, slice):n_dim_1 = data.shape[1]else:n_dim_1 = len(dim_1_ids)# 解析分布函数if distribution == "uniform":dist_fn = math_utils.sample_uniformelif distribution == "log_uniform":dist_fn = math_utils.sample_log_uniformelif distribution == "gaussian":dist_fn = math_utils.sample_gaussianelse:raise NotImplementedError(f"未知分布: '{distribution}' 用于属性随机化。请使用 'uniform', 'log_uniform' 或 'gaussian'。")# 执行操作if operation == "add":data[dim_0_ids, dim_1_ids] += dist_fn(*distribution_parameters, (n_dim_0, n_dim_1), device=data.device)elif operation == "scale":data[dim_0_ids, dim_1_ids] *= dist_fn(*distribution_parameters, (n_dim_0, n_dim_1), device=data.device)elif operation == "abs":data[dim_0_ids, dim_1_ids] = dist_fn(*distribution_parameters, (n_dim_0, n_dim_1), device=data.device)else:raise NotImplementedError(f"未知操作: '{operation}' 用于属性随机化。请使用 'add', 'scale' 或 'abs'。")return data

robot_lab中的events.py

randomize_rigid_body_inertia

随机化刚体的惯性张量

def randomize_rigid_body_inertia(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,inertia_distribution_params: tuple[float, float],operation: Literal["add", "scale", "abs"],distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""随机化刚体的惯性张量该函数通过添加、缩放或设置随机值来随机化物体的惯性张量。只能随机化惯性张量的对角线分量（xx, yy, zz），这些分量对应物体绕三个主轴的转动惯量。Args:env: 仿真环境对象env_ids: 需要随机化的环境ID，None表示所有环境asset_cfg: 目标资产的配置信息inertia_distribution_params: 分布参数元组（如均匀分布的最小值和最大值）operation: 随机化操作类型- "add": 在原值基础上添加随机值- "scale": 将原值乘以随机因子- "abs": 直接设置为随机值distribution: 随机分布类型，默认为"uniform"- "uniform": 均匀分布- "log_uniform": 对数均匀分布- "gaussian": 高斯分布注意：该函数使用CPU张量来分配物体惯性，建议仅在环境初始化时使用，以避免影响仿真性能。应用场景：- 模拟制造误差导致的惯性参数变化- 增强机器人对负载变化的适应性- 提高控制算法的鲁棒性"""# 提取目标资产（支持类型提示）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 解析环境IDif env_ids is None:env_ids = torch.arange(env.scene.num_envs, device="cpu")else:env_ids = env_ids.cpu()删除线格式# 解析物体索引if asset_cfg.body_ids == slice(None):body_ids = torch.arange(asset.num_bodies, dtype=torch.int, device="cpu")else:body_ids = torch.tensor(asset_cfg.body_ids, dtype=torch.int, device="cpu")# 获取当前的惯性张量（形状：num_assets, num_bodies, 9）# 惯性张量是3x3矩阵，展平为9个元素inertias = asset.root_physx_view.get_inertias()# 将惯性张量重置为默认值，作为随机化的基础inertias[env_ids[:, None], body_ids, :] = asset.data.default_inertia[env_ids[:, None], body_ids, :].clone()# 随机化对角线元素（xx, yy, zz -> 索引 0, 4, 8）# 这些元素代表绕x、y、z轴的主转动惯量for idx in [0, 4, 8]:# 提取并随机化特定的对角线元素randomized_inertias = _randomize_prop_by_op(inertias[:, :, idx],inertia_distribution_params,env_ids,body_ids,operation,distribution,)# 将随机化的值赋回惯性张量inertias[env_ids[:, None], body_ids, idx] = randomized_inertias# 将更新后的惯性张量设置到物理仿真中asset.root_physx_view.set_inertias(inertias, env_ids)

randomize_com_positions

随机化刚体的质心位置

def randomize_com_positions(env: ManagerBasedEnv,env_ids: torch.Tensor | None,asset_cfg: SceneEntityCfg,com_distribution_params: tuple[float, float],operation: Literal["add", "scale", "abs"],distribution: Literal["uniform", "log_uniform", "gaussian"] = "uniform",
):"""随机化刚体的质心位置该函数允许随机化物体在物理仿真中的质心位置。质心位置的变化会影响物体的动力学行为，包括平衡、稳定性和运动响应。Args:env: 仿真环境对象env_ids: 需要随机化的环境ID，None表示所有环境asset_cfg: 目标资产的配置信息com_distribution_params: 分布参数元组（如均匀分布的最小值和最大值）operation: 随机化操作类型- "add": 在原质心位置基础上添加偏移- "scale": 将质心位置乘以随机因子- "abs": 直接设置为随机位置distribution: 随机分布类型，默认为"uniform"注意：该函数用于初始化或离线调整，因为它直接修改物理属性。应用场景：- 模拟负载分布的不确定性- 测试机器人对质心变化的适应能力- 增强平衡控制算法的鲁棒性- 模拟磨损或损坏导致的质心偏移"""# 提取资产（关节机构或刚体对象）asset: RigidObject | Articulation = env.scene[asset_cfg.name]# 解析环境索引if env_ids is None:env_ids = torch.arange(env.scene.num_envs, device="cpu")else:env_ids = env_ids.cpu()# 解析物体索引if asset_cfg.body_ids == slice(None):body_ids = torch.arange(asset.num_bodies, dtype=torch.int, device="cpu")else:body_ids = torch.tensor(asset_cfg.body_ids, dtype=torch.int, device="cpu")# 获取当前的质心偏移（形状：num_assets, num_bodies, 3）# 3个分量分别对应x、y、z方向的偏移com_offsets = asset.root_physx_view.get_coms()# 独立随机化x、y、z三个方向的质心偏移for dim_idx in range(3):  randomized_offset = _randomize_prop_by_op(com_offsets[:, :, dim_idx],com_distribution_params,env_ids,body_ids,operation,distribution,)com_offsets[env_ids[:, None], body_ids, dim_idx] = randomized_offset[env_ids[:, None], body_ids]# 将随机化的质心偏移设置到仿真中asset.root_physx_view.set_coms(com_offsets, env_ids)