[fix] Fix gym_env error when loading scene without robots

Author: APX103

grutopia/core/gym_env.py

@@ -46,6 +46,8 @@ def _validate(self):
         log.debug(f'================ len(episodes): {len(episodes)} ==================')
 
         for runtime in self._runtime.task_runtime_manager.episodes:
+            if len(runtime.robots) == 0:
+                return
             if len(runtime.robots) != 1:
                 raise ValueError(f'Only support single agent now, but episode requires {len(runtime.robots)} agents')
             if robot_name is None:
@@ -76,6 +78,7 @@ def reset(self, *, seed=None, options=None) -> tuple[gym.Space, dict[str, Any]]:
             info (dictionary):  Contains the key `task_runtime` if there is an unfinished task
         """
         info = {}
+        obs = {}
 
         origin_obs, task_runtime = self.runner.reset(self._current_task_name)
         if task_runtime is None:
@@ -84,7 +87,8 @@ def reset(self, *, seed=None, options=None) -> tuple[gym.Space, dict[str, Any]]:
 
         self._current_task_name = task_runtime.name
         info[Env.RESET_INFO_TASK_RUNTIME] = task_runtime
-        obs = origin_obs[task_runtime.name][self._robot_name]
+        if self._robot_name:
+            obs = origin_obs[task_runtime.name][self._robot_name]
 
         return obs, info
 
@@ -124,7 +128,8 @@ def step(self, action: Any) -> tuple[Any, float, bool, bool, dict[str, Any]]:
         if rewards[self._current_task_name] != -1:
             reward = rewards[self._current_task_name]
 
-        obs = origin_obs[self._current_task_name][self._robot_name]
+        if self._robot_name:
+            obs = origin_obs[self._current_task_name][self._robot_name]
         terminated = terminated_status[self._current_task_name]
 
         return obs, reward, terminated, truncated, info
@@ -160,6 +165,8 @@ def get_observations(self) -> dict[Any, Any] | Any:
             return {}
 
         _obs = self._runner.get_obs()
+        if self._robot_name is None:
+            return {}
         return _obs[self._current_task_name][self._robot_name]
 
     def render(self, mode='human'):

tests/e2e_test.py

@@ -49,6 +49,12 @@ def test_h1_locomotion():
     common_body(start_command)
 
 
+@pytest.mark.P0
+def test_load_scene_without_robot():
+    start_command = 'python ./tests/load_scene_without_robot.py'
+    common_body(start_command)
+
+
 @pytest.mark.P0
 def test_rep_camera_pointcloud():
     start_command = 'python ./tests/rep_camera_pointcloud.py'

tests/load_scene_without_robot.py

@@ -0,0 +1,61 @@
+def main():
+    from grutopia.core.config import Config, SimConfig
+    from grutopia.core.gym_env import Env
+    from grutopia.core.runtime import SimulatorRuntime
+    from grutopia.core.util import has_display
+    from grutopia.macros import gm
+    from grutopia_extension import import_extensions
+    from grutopia_extension.configs.tasks import (
+        SingleInferenceEpisodeCfg,
+        SingleInferenceTaskCfg,
+    )
+
+    headless = False
+    if not has_display():
+        headless = True
+
+    config = Config(
+        simulator=SimConfig(physics_dt=1 / 240, rendering_dt=1 / 240, use_fabric=False),
+        task_config=SingleInferenceTaskCfg(
+            episodes=[
+                SingleInferenceEpisodeCfg(
+                    scene_asset_path=gm.ASSET_PATH + '/scenes/empty.usd',
+                    scene_scale=(0.01, 0.01, 0.01),
+                    robots=[],
+                ),
+            ],
+        ),
+    )
+
+    print(config.model_dump_json(indent=4))
+
+    sim_runtime = SimulatorRuntime(config_class=config, headless=headless, native=headless)
+
+    import_extensions()
+
+    env = Env(sim_runtime)
+    obs, _ = env.reset()
+    print(f'========INIT OBS{obs}=============')
+
+    i = 0
+
+    while env.simulation_app.is_running():
+        i += 1
+        obs, _, terminated, _, _ = env.step(action={})
+
+        if i == 2000:
+            break
+
+    env.close()
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except Exception as e:
+        print(f'exception is {e}')
+        import sys
+        import traceback
+
+        traceback.print_exc()
+        sys.exit(1)