6.isaac sim4.2 教程-Core API-多机器人，多任务

本教程将两种不同类型的机器人集成到同一个仿真中。它详细说明了如何构建程序逻辑以在子任务之间切换。通过本教程，你将获得构建更复杂的机器人交互仿真经验。

1. 创建一个场景

首先，按照以下步骤将 Jetbot、Franka Panda 和 Cube 添加到场景中，并通过子任务来简化代码：

# Adding Multiple Robots
# 创建场景
from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):def __init__(self,name):super().__init__(name=name, offset=None)self._jetbot_goal_position = np.array([130, 30, 0]) # 初始化时，定义了 Jetbot 的目标位置# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]))returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path="/World/Fancy_Robot",name="fancy_robot",wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。self._franka.set_world_pose(position=np.array([1.0, 0, 0]))self._franka.set_default_state(position=np.array([1.0, 0, 0]))return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representationdef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()returndef setup_scene(self):world = self.get_world()world.add_task(RobotsPlaying(name="awesome_task"))  # 添加自定义的任务return

2. 驱动Jetbot 把cube推给Franka.

from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
from omni.isaac.wheeled_robots.controllers.wheel_base_pose_controller import WheelBasePoseController
from omni.isaac.wheeled_robots.controllers.differential_controller import DifferentialController
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):def __init__(self,name):super().__init__(name=name, offset=None)self._jetbot_goal_position = np.array([1.3, 0.3, 0]) # 初始化时，定义了 Jetbot 的目标位置# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]))returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path="/World/Fancy_Robot",name="fancy_robot",wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。self._franka.set_world_pose(position=np.array([1.0, 0, 0]))self._franka.set_default_state(position=np.array([1.0, 0, 0]))return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representationdef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()returndef setup_scene(self):world = self.get_world()world.add_task(RobotsPlaying(name="awesome_task"))  # 添加自定义的任务returnasync def setup_post_load(self):self._world = self.get_world()task_params = self._world.get_task("awesome_task").get_params()self._jetbot = self._world.scene.get_object(task_params["jetbot_name"]["value"])self._cube_name = task_params["cube_name"]["value"]# 这里使用了 WheelBasePoseController 来控制 Jetbot 的运动self._jetbot_controller = WheelBasePoseController(name="cool_controller",open_loop_wheel_controller=DifferentialController(name="simple_control",wheel_radius=0.03,wheel_base=0.1125))self._world.add_physics_callback("sim_step", callback_fn=self.physics_step)await self._world.play_async()returnasync def setup_post_reset(self):self._jetbot_controller.reset()await self._world.play_async()returndef physics_step(self, step_size):current_observations = self._world.get_observations()self._jetbot.apply_action(self._jetbot_controller.forward(start_position=current_observations[self._jetbot.name]["position"],start_orientation=current_observations[self._jetbot.name]["orientation"],goal_position=current_observations[self._jetbot.name]["goal_position"]))return

3. 增加任务逻辑

下一步，驱动 jetbot 后退让出空间给franka去把cube捡起并放到指定位置。

from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
from omni.isaac.wheeled_robots.controllers.wheel_base_pose_controller import WheelBasePoseController
from omni.isaac.wheeled_robots.controllers.differential_controller import DifferentialController
from omni.isaac.core.utils.types import ArticulationAction
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):def __init__(self,name):super().__init__(name=name, offset=None)self._jetbot_goal_position = np.array([1.3, 0.3, 0]) # 初始化时，定义了 Jetbot 的目标位置# 增加任务逻辑标志--任务事件self._task_event = 0# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]))returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path="/World/Fancy_Robot",name="fancy_robot",wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。self._franka.set_world_pose(position=np.array([1.0, 0, 0]))self._franka.set_default_state(position=np.array([1.0, 0, 0]))return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {# 通过任务事件来控制 Jetbot 的行为"task_event": self._task_event,self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representation# 该方法在每个物理步之前调用，进行判断当前 Jetbot 应该做什么。# 它根据任务事件的状态来决定 Jetbot 的行为。def pre_step(self, control_index, simulation_time):# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if self._task_event == 0:current_jetbot_position, _ = self._jetbot.get_world_pose()# 计算 Jetbot 到目标位置的距离，如果距离小于 4cm，则认为 Jetbot 到达了目标位置。if np.mean(np.abs(current_jetbot_position[:2] - self._jetbot_goal_position[:2])) < 0.04:self._task_event += 1self._cube_arrive_step_index = control_indexelif self._task_event == 1:# Jetbot 有200个物理步长的时间来后退if control_index - self._cube_arrive_step_index == 200:self._task_event += 1 # 任务事件加1，表示 Jetbot 后退的时间结束。returndef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()returndef setup_scene(self):world = self.get_world()world.add_task(RobotsPlaying(name="awesome_task"))  # 添加自定义的任务returnasync def setup_post_load(self):self._world = self.get_world()task_params = self._world.get_task("awesome_task").get_params()self._jetbot = self._world.scene.get_object(task_params["jetbot_name"]["value"])self._cube_name = task_params["cube_name"]["value"]# 这里使用了 WheelBasePoseController 来控制 Jetbot 的运动self._jetbot_controller = WheelBasePoseController(name="cool_controller",open_loop_wheel_controller=DifferentialController(name="simple_control",wheel_radius=0.03,wheel_base=0.1125))self._world.add_physics_callback("sim_step", callback_fn=self.physics_step)await self._world.play_async()returnasync def setup_post_reset(self):self._jetbot_controller.reset()await self._world.play_async()returndef physics_step(self, step_size):current_observations = self._world.get_observations()# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if current_observations["task_event"] == 0:self._jetbot.apply_wheel_actions(self._jetbot_controller.forward(start_position=current_observations[self._jetbot.name]["position"],start_orientation=current_observations[self._jetbot.name]["orientation"],goal_position=current_observations[self._jetbot.name]["goal_position"]))# 如果任务事件为 1，表示 Jetbot 需要向后移动。elif current_observations["task_event"] == 1:# Go backwardsself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[-8, -8]))# 如果任务事件为 2，表示 Jetbot 停止移动。elif current_observations["task_event"] == 2:# Apply zero velocity to override the velocity applied before.# Note: target joint positions and target joint velocities will stay active unless changedself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[0.0, 0.0]))return

4. 机器人交接任务

现在该是franka去pickplace了

from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
from omni.isaac.wheeled_robots.controllers.wheel_base_pose_controller import WheelBasePoseController
from omni.isaac.wheeled_robots.controllers.differential_controller import DifferentialController
from omni.isaac.core.utils.types import ArticulationAction
from omni.isaac.franka.controllers import PickPlaceController
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):def __init__(self,name):super().__init__(name=name, offset=None)self._jetbot_goal_position = np.array([1.3, 0.3, 0]) # 初始化时，定义了 Jetbot 的目标位置# 增加任务逻辑标志--任务事件self._task_event = 0# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]))returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path="/World/Fancy_Robot",name="fancy_robot",wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。self._franka.set_world_pose(position=np.array([1.0, 0, 0]))self._franka.set_default_state(position=np.array([1.0, 0, 0]))return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {# 通过任务事件来控制 Jetbot 的行为"task_event": self._task_event,self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}# 获取 PickPlace 任务的观察值，并将其添加到 observations 字典中。observations.update(self._pick_place_task.get_observations())return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representation# 该方法在每个物理步之前调用，进行判断当前 Jetbot 应该做什么。# 它根据任务事件的状态来决定 Jetbot 的行为。def pre_step(self, control_index, simulation_time):# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if self._task_event == 0:current_jetbot_position, _ = self._jetbot.get_world_pose()# 计算 Jetbot 到目标位置的距离，如果距离小于 4cm，则认为 Jetbot 到达了目标位置。if np.mean(np.abs(current_jetbot_position[:2] - self._jetbot_goal_position[:2])) < 0.04:self._task_event += 1self._cube_arrive_step_index = control_indexelif self._task_event == 1:# Jetbot 有200个物理步长的时间来后退if control_index - self._cube_arrive_step_index == 200:self._task_event += 1 # 任务事件加1，表示 Jetbot 后退的时间结束。returndef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()returndef setup_scene(self):world = self.get_world()world.add_task(RobotsPlaying(name="awesome_task"))  # 添加自定义的任务returnasync def setup_post_load(self):self._world = self.get_world()# 获得任务参数task_params = self._world.get_task("awesome_task").get_params()# 获得jetbot和cube的名称和参数self._jetbot = self._world.scene.get_object(task_params["jetbot_name"]["value"])self._cube_name = task_params["cube_name"]["value"]# 获得franka机械臂的名称和参数self._franka = self._world.scene.get_object(task_params["franka_name"]["value"])# 这里使用了 WheelBasePoseController 来控制 Jetbot 的运动self._jetbot_controller = WheelBasePoseController(name="cool_controller",open_loop_wheel_controller=DifferentialController(name="simple_control",wheel_radius=0.03,wheel_base=0.1125))# 增加了 PickPlaceController 来控制 Franka Panda 机械臂的抓取和放置任务。self._franka_controller = PickPlaceController(name="pick_place_controller",gripper=self._franka.gripper,robot_articulation=self._franka)self._world.add_physics_callback("sim_step", callback_fn=self.physics_step)await self._world.play_async()returnasync def setup_post_reset(self):self._jetbot_controller.reset()# 重置 Franka Panda 的控制器self._franka_controller.reset()await self._world.play_async()returndef physics_step(self, step_size):current_observations = self._world.get_observations()# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if current_observations["task_event"] == 0:self._jetbot.apply_wheel_actions(self._jetbot_controller.forward(start_position=current_observations[self._jetbot.name]["position"],start_orientation=current_observations[self._jetbot.name]["orientation"],goal_position=current_observations[self._jetbot.name]["goal_position"]))# 如果任务事件为 1，表示 Jetbot 需要向后移动。elif current_observations["task_event"] == 1:# Go backwardsself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[-8, -8]))# 如果任务事件为 2，表示 Jetbot 停止移动。elif current_observations["task_event"] == 2:# Apply zero velocity to override the velocity applied before.# Note: target joint positions and target joint velocities will stay active unless changedself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[0.0, 0.0]))# jetbot后退完后，让 Franka Panda 机械臂执行抓取和放置任务actions = self._franka_controller.forward(picking_position=current_observations[self._cube_name]["position"],placing_position=current_observations[self._cube_name]["target_position"],current_joint_positions=current_observations[self._franka.name]["joint_positions"])self._franka.apply_action(actions)# Pause once the controller is doneif self._franka_controller.is_done():self._world.pause()return

5. 任务参数化

本教程描述了如何在 Omniverse Isaac Sim 中添加多个任务。它解释了如何向任务中添加参数以扩展你的仿真。

# 多个任务
from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
from omni.isaac.wheeled_robots.controllers.wheel_base_pose_controller import WheelBasePoseController
from omni.isaac.wheeled_robots.controllers.differential_controller import DifferentialController
from omni.isaac.core.utils.types import ArticulationAction
from omni.isaac.franka.controllers import PickPlaceController
# find_unique_string_name 用于生成一个唯一的字符串名称，通常用于 Prim 路径 或 场景名称，以确保这些名称不会重复。
from omni.isaac.core.utils.string import find_unique_string_name  
# 验证一个 Prim 路径是否有效，即路径是否已存在于当前仿真场景中。      
from omni.isaac.core.utils.prims import is_prim_path_valid     
from omni.isaac.core.objects.cuboid import VisualCuboid
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):# 位置被偏移量 (offset) 调整。def __init__(self,name,offset=None):super().__init__(name=name, offset=offset)self._jetbot_goal_position = np.array([1.3, 0.3, 0]) + self._offset # 初始化时，定义了 Jetbot 的目标位置# 增加任务逻辑标志--任务事件self._task_event = 0# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]),offset=offset)returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)# 为Jetbot创建一个唯一的名称和Prim路径jetbot_name = find_unique_string_name(initial_name="fancy_jetbot", is_unique_fn=lambda x: not self.scene.object_exists(x))jetbot_prim_path = find_unique_string_name(initial_name="/World/Fancy_Jetbot", is_unique_fn=lambda x: not is_prim_path_valid(x))assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path=jetbot_prim_path,name=jetbot_name,wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))# 将Jetbot添加到任务对象中self._task_objects[self._jetbot.name] = self._jetbot# 获取PickPlace任务中的Franka机器人pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。current_position, _ = self._franka.get_world_pose()self._franka.set_world_pose(position=current_position + np.array([1.0, 0, 0]))self._franka.set_default_state(position=current_position + np.array([1.0, 0, 0]))# 调用父类方法，将任务对象按照偏移量进行平移self._move_task_objects_to_their_frame()return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {# 通过任务事件来控制 Jetbot 的行为"task_event": self._task_event,self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}# 获取 PickPlace 任务的观察值，并将其添加到 observations 字典中。observations.update(self._pick_place_task.get_observations())return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representation# 该方法在每个物理步之前调用，进行判断当前 Jetbot 应该做什么。# 它根据任务事件的状态来决定 Jetbot 的行为。def pre_step(self, control_index, simulation_time):# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if self._task_event == 0:current_jetbot_position, _ = self._jetbot.get_world_pose()# 计算 Jetbot 到目标位置的距离，如果距离小于 4cm，则认为 Jetbot 到达了目标位置。if np.mean(np.abs(current_jetbot_position[:2] - self._jetbot_goal_position[:2])) < 0.04:self._task_event += 1self._cube_arrive_step_index = control_indexelif self._task_event == 1:# Jetbot 有200个物理步长的时间来后退if control_index - self._cube_arrive_step_index == 100:self._task_event += 1 # 任务事件加1，表示 Jetbot 后退的时间结束。returndef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()returndef setup_scene(self):world = self.get_world()world.add_task(RobotsPlaying(name="awesome_task", offset=np.array([0, -1.0, 0])))  # 添加自定义的任务# 仅用于可视化，展示Franka原本应该在的位置VisualCuboid(prim_path="/new_cube_1",name="visual_cube",position=np.array([1.0, 0, 0.05]),scale=np.array([0.1, 0.1, 0.1]))returnasync def setup_post_load(self):self._world = self.get_world()# 获得任务参数task_params = self._world.get_task("awesome_task").get_params()# 获得jetbot和cube的名称和参数self._jetbot = self._world.scene.get_object(task_params["jetbot_name"]["value"])self._cube_name = task_params["cube_name"]["value"]# 获得franka机械臂的名称和参数self._franka = self._world.scene.get_object(task_params["franka_name"]["value"])# 这里使用了 WheelBasePoseController 来控制 Jetbot 的运动self._jetbot_controller = WheelBasePoseController(name="cool_controller",open_loop_wheel_controller=DifferentialController(name="simple_control",wheel_radius=0.03,wheel_base=0.1125))# 增加了 PickPlaceController 来控制 Franka Panda 机械臂的抓取和放置任务。self._franka_controller = PickPlaceController(name="pick_place_controller",gripper=self._franka.gripper,robot_articulation=self._franka)self._world.add_physics_callback("sim_step", callback_fn=self.physics_step)await self._world.play_async()returnasync def setup_post_reset(self):robots_playing_task = self._world.get_task("awesome_task")robots_playing_task._task_event = 0  # 直接重置任务实例的事件self._jetbot_controller.reset()# 重置 Franka Panda 的控制器self._franka_controller.reset()await self._world.play_async()returndef physics_step(self, step_size):current_observations = self._world.get_observations()# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if current_observations["task_event"] == 0:self._jetbot.apply_wheel_actions(self._jetbot_controller.forward(start_position=current_observations[self._jetbot.name]["position"],start_orientation=current_observations[self._jetbot.name]["orientation"],goal_position=current_observations[self._jetbot.name]["goal_position"]))# 如果任务事件为 1，表示 Jetbot 需要向后移动。elif current_observations["task_event"] == 1:# Go backwardsself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[-2, -2]))# 如果任务事件为 2，表示 Jetbot 停止移动。elif current_observations["task_event"] == 2:# Apply zero velocity to override the velocity applied before.# Note: target joint positions and target joint velocities will stay active unless changedself._jetbot.apply_wheel_actions(ArticulationAction(joint_velocities=[0.0, 0.0]))# jetbot后退完后，让 Franka Panda 机械臂执行抓取和放置任务actions = self._franka_controller.forward(picking_position=current_observations[self._cube_name]["position"],placing_position=current_observations[self._cube_name]["target_position"],current_joint_positions=current_observations[self._franka.name]["joint_positions"])self._franka.apply_action(actions)# Pause once the controller is doneif self._franka_controller.is_done():self._world.pause()return

6. 扩展到多个任务

# 多个任务
from omni.isaac.examples.base_sample import BaseSample
from omni.isaac.franka.tasks import PickPlace
from omni.isaac.wheeled_robots.robots import WheeledRobot
from omni.isaac.core.utils.nucleus import get_assets_root_path
from omni.isaac.core.tasks import BaseTask
from omni.isaac.wheeled_robots.controllers.wheel_base_pose_controller import WheelBasePoseController
from omni.isaac.wheeled_robots.controllers.differential_controller import DifferentialController
from omni.isaac.core.utils.types import ArticulationAction
from omni.isaac.franka.controllers import PickPlaceController
# find_unique_string_name 用于生成一个唯一的字符串名称，通常用于 Prim 路径 或 场景名称，以确保这些名称不会重复。
from omni.isaac.core.utils.string import find_unique_string_name  
# 验证一个 Prim 路径是否有效，即路径是否已存在于当前仿真场景中。      
from omni.isaac.core.utils.prims import is_prim_path_valid     
from omni.isaac.core.objects.cuboid import VisualCuboid
import numpy as np# RobotsPlaying 继承自 BaseTask 类，表示一个包含多个机器人任务的复合任务。
class RobotsPlaying(BaseTask):# 位置被偏移量 (offset) 调整。def __init__(self,name,offset=None):super().__init__(name=name, offset=offset)# 随机化jetbot目标位置self._jetbot_goal_position = np.array([np.random.uniform(1.2, 1.6), 0.3, 0]) + self._offset # 初始化时，定义了 Jetbot 的目标位置# 增加任务逻辑标志--任务事件self._task_event = 0# 使用 PickPlace 任务类创建了一个 拾取和放置任务，并定义了立方体的初始位置和目标位置。self._pick_place_task = PickPlace(cube_initial_position=np.array([0.1, 0.3, 0.05]),target_position=np.array([0.7, -0.3, 0.0515 / 2.0]),offset=offset)returndef set_up_scene(self, scene):super().set_up_scene(scene)self._pick_place_task.set_up_scene(scene)# 为Jetbot创建一个唯一的名称和Prim路径jetbot_name = find_unique_string_name(initial_name="fancy_jetbot", is_unique_fn=lambda x: not self.scene.object_exists(x))jetbot_prim_path = find_unique_string_name(initial_name="/World/Fancy_Jetbot", is_unique_fn=lambda x: not is_prim_path_valid(x))assets_root_path = get_assets_root_path()jetbot_asset_path = assets_root_path + "/Isaac/Robots/Jetbot/jetbot.usd"# 在场景中添加 Jetbot 机器人# 通过 WheeledRobot 类来创建一个轮式机器人实例，并将其添加到场景中。# 这里指定了轮子关节的名称和机器人模型self._jetbot = scene.add(WheeledRobot(prim_path=jetbot_prim_path,name=jetbot_name,wheel_dof_names=["left_wheel_joint", "right_wheel_joint"],create_robot=True,usd_path=jetbot_asset_path,position=np.array([0, 0.3, 0])))# 将Jetbot添加到任务对象中self._task_objects[self._jetbot.name] = self._jetbot# 获取PickPlace任务中的Franka机器人pick_place_params = self._pick_place_task.get_params()self._franka = scene.get_object(pick_place_params["robot_name"]["value"])# 将 Franka Panda 的默认位置设置为 (1.0, 0, 0)，即重置后的位置。current_position, _ = self._franka.get_world_pose()self._franka.set_world_pose(position=current_position + np.array([1.0, 0, 0]))self._franka.set_default_state(position=current_position + np.array([1.0, 0, 0]))# 调用父类方法，将任务对象按照偏移量进行平移self._move_task_objects_to_their_frame()return# 该方法获取 Jetbot 的信息def get_observations(self):current_jetbot_position, current_jetbot_orientation = self._jetbot.get_world_pose()# Observations needed to drive the jetbot to push the cubeobservations= {# 通过任务事件来控制 Jetbot 的行为self.name + "_event": self._task_event, # 改变任务事件以使其唯一self._jetbot.name: {"position": current_jetbot_position,"orientation": current_jetbot_orientation,"goal_position": self._jetbot_goal_position}}# 获取 PickPlace 任务的观察值，并将其添加到 observations 字典中。observations.update(self._pick_place_task.get_observations())return observationsdef get_params(self):# 为了避免硬编码，获取了 PickPlace 任务的参数并将其返回，同时也添加了 Jetbot 和 Franka Panda 的名称作为任务的参数。pick_place_params = self._pick_place_task.get_params()params_representation = pick_place_paramsparams_representation["jetbot_name"] = {"value": self._jetbot.name, "modifiable": False}params_representation["franka_name"] = pick_place_params["robot_name"]return params_representation# 该方法在每个物理步之前调用，进行判断当前 Jetbot 应该做什么。# 它根据任务事件的状态来决定 Jetbot 的行为。def pre_step(self, control_index, simulation_time):# 如果任务事件为 0，表示 Jetbot 需要移动到目标位置。if self._task_event == 0:current_jetbot_position, _ = self._jetbot.get_world_pose()# 计算 Jetbot 到目标位置的距离，如果距离小于 4cm，则认为 Jetbot 到达了目标位置。if np.mean(np.abs(current_jetbot_position[:2] - self._jetbot_goal_position[:2])) < 0.04:self._task_event += 1self._cube_arrive_step_index = control_indexelif self._task_event == 1:# Jetbot 有200个物理步长的时间来后退if control_index - self._cube_arrive_step_index == 100:self._task_event += 1 # 任务事件加1，表示 Jetbot 后退的时间结束。returndef post_reset(self):self._franka.gripper.set_joint_positions(self._franka.gripper.joint_opened_positions)class MyHelloWorld(BaseSample): # 继承自basesampledef __init__(self) -> None:super().__init__()# Add lists for tasks,self._tasks = []self._num_of_tasks = 3#  Add lists for controllersself._franka_controllers = []self._jetbot_controllers = []# Add lists for variables needed for controlself._jetbots = []self._frankas = []self._cube_names = []returndef setup_scene(self):world = self.get_world()# 添加多个 RobotsPlaying 任务，每个任务都有一个偏移量# 这里使用了循环来创建多个任务实例，并将它们添加到世界中。for i in range(self._num_of_tasks):world.add_task(RobotsPlaying(name="my_awesome_task_" + str(i), offset=np.array([0, (i * 2) - 3, 0])))returnasync def setup_post_load(self):self._world = self.get_world()for i in range(self._num_of_tasks):self._tasks.append(self._world.get_task(name="my_awesome_task_" + str(i)))# 获取每个任务的变量task_params = self._tasks[i].get_params()self._frankas.append(self._world.scene.get_object(task_params["franka_name"]["value"]))self._jetbots.append(self._world.scene.get_object(task_params["jetbot_name"]["value"]))self._cube_names.append(task_params["cube_name"]["value"])# 每个任务都要定义控制器# 通过改变 dt，可以控制机器人在执行拾取和放置任务时的过渡速度，使其变得更慢，# 这对于在一个场景中处理多个物体（如多个立方体）时可能会非常有用，确保机器人能更精确地完成每个任务。self._franka_controllers.append(PickPlaceController(name="pick_place_controller",gripper=self._frankas[i].gripper,robot_articulation=self._frankas[i],events_dt=[0.008, 0.002, 0.5, 0.1, 0.05, 0.05, 0.0025, 1, 0.008, 0.08]))self._jetbot_controllers.append(WheelBasePoseController(name="cool_controller",open_loop_wheel_controller=DifferentialController(name="simple_control",wheel_radius=0.03,wheel_base=0.1125)))self._world.add_physics_callback("sim_step", callback_fn=self.physics_step)await self._world.play_async()returnasync def setup_post_reset(self):for i in range(len(self._tasks)):# Reset all controllersself._franka_controllers[i].reset()self._jetbot_controllers[i].reset()await self._world.play_async()returndef physics_step(self, step_size):current_observations = self._world.get_observations()for i in range(len(self._tasks)):# Apply actions for each taskif current_observations[self._tasks[i].name + "_event"] == 0:self._jetbots[i].apply_wheel_actions(self._jetbot_controllers[i].forward(start_position=current_observations[self._jetbots[i].name]["position"],start_orientation=current_observations[self._jetbots[i].name]["orientation"],goal_position=current_observations[self._jetbots[i].name]["goal_position"]))elif current_observations[self._tasks[i].name + "_event"] == 1:self._jetbots[i].apply_wheel_actions(ArticulationAction(joint_velocities=[-8.0, -8.0]))elif current_observations[self._tasks[i].name + "_event"] == 2:self._jetbots[i].apply_wheel_actions(ArticulationAction(joint_velocities=[0.0, 0.0]))actions = self._franka_controllers[i].forward(picking_position=current_observations[self._cube_names[i]]["position"],placing_position=current_observations[self._cube_names[i]]["target_position"],current_joint_positions=current_observations[self._frankas[i].name]["joint_positions"])self._frankas[i].apply_action(actions)return# This function is called after a hot reload or a clear# to delete the variables defined in this extension applicationdef world_cleanup(self):self._tasks = []self._franka_controllers = []self._jetbot_controllers = []self._jetbots = []self._frankas = []self._cube_names = []return

可以看出cube最终位置都不一样。