How to simulate using custom strategies

[CarlDegio]

I'm using this to replay nuplan dataset. And I wander how to input a user built policy and simulate it like nuplan dev toolkit.

More precisely, I want to input a specific planned trajectory, then all vehicles simulated one step (either through control or direct displacement), and then input a new plan. But now I find that nuplan can only be replayed in simulation by ReplayEgoCarPolicy. The steering wheel commands I input will not be executed, and I can't find how to input the desired trajectory like nuplan input.

[QuanyiLi]

You can develop a new policy like EnvInputPolicy, which allows you to access the external input from env.step(). This new policy can have a dict action space or just a box action space. Supposing you are building a box action space, you can make the shape of this space = num_of_vehicles_to_control * 2. Then the action from env.step(action) could be the actions for multiple agents.

I have a working-in-progress version about the policy interface of MetaDrive: https://metadrive-simulator.readthedocs.io/en/new-doc/action.html You can look into it to see if there are anything helpful

[CarlDegio]

Thanks for your development, I will check this.

[abastola0]

You can develop a new policy like EnvInputPolicy, which allows you to access the external input from env.step(). This new policy can have a dict action space or just a box action space. Supposing you are building a box action space, you can make the shape of this space = num_of_vehicles_to_control * 2. Then the action from env.step(action) could be the actions for multiple agents.

I have a working-in-progress version about the policy interface of MetaDrive: https://metadrive-simulator.readthedocs.io/en/new-doc/action.html You can look into it to see if there are anything helpful

Hi Quanyi. Just a followup on this. I tried developing a policy for this but I see the observation space to be (336, 1). Can you please brief out where this shape came from. I was expecting (84, 84, 3) just like other metadrive observations but i couldn't get it to work this way. I tried modifying the configuration with topdown and other similar configurations but the observation space remains the same. I tried it for Waymo dataset.

[QuanyiLi]

Hi Abastola,

Have you checked this page: https://metadrive-simulator.readthedocs.io/en/latest/obs.html#customization-multisensor It allows you to customize your observation and its shape. Just pass your new observation class to the config dict. Then changes will be applied accordingly.

[abastola0]

Yeah I tried this approach as well. I created observation like so:

class MyObservation(BaseObservation):
    def __init__(self, config):
        super(MyObservation, self).__init__(config)
        self.rgb = ImageObservation(config, 'rgb', config['norm_pixel'])
        self.depth = ImageObservation(config, "depth", config["norm_pixel"])
        self.semantic = ImageObservation(config, "semantic", config["norm_pixel"])
        self.state = StateObservation(config)

    @property
    def observation_space(self):
        os={o: getattr(self, o).observation_space for o in ["rgb", "state", "depth", "semantic"]}
        return gym.spaces.Dict(os)

    def observe(self, vehicle):
        os={o: getattr(self, o).observe() for o in ["rgb", "state", "depth", "semantic"]}
        return os

and configuration like so:

    cfg = {
        "use_render": args.render == "3D" or args.render == "advanced",
        #"agent_observation": MyObservation,
        #"image_observation": True,
        "agent_policy": ReplayEgoCarPolicy,
        #"agent_policy": IDMPolicy,
        #"manual_control": True,
        "render_pipeline": args.render == "advanced",
        "show_interface": True,
        "reactive_traffic": True,
        "show_logo": False,
        "show_fps": False,
        "debug": True,
        "log_level": logging.CRITICAL,
        "num_scenarios": num_scenario,
        "interface_panel": [],
        "horizon": 1000,
        "vehicle_config": dict(
            show_navi_mark=True,
            show_line_to_dest=False,
            show_dest_mark=False,
            image_source="main_camera",
            no_wheel_friction=True,
            #no_wheel_friction=False,
            lidar=dict(num_lasers=120, distance=50, num_others=4),
            lane_line_detector=dict(num_lasers=12, distance=50),
            side_detector=dict(num_lasers=160, distance=50)
        ),
        'sensors':{"rgb": (RGBCamera, *sensor_size),
                  "depth": (DepthCamera, *sensor_size),
                  "semantic": (SemanticCamera, *sensor_size)},
        "data_directory": database_path,
    }

and finally created env like this:

    env = ScenarioEnv(cfg)

when i do env.engine.dict, i don't see sensors sensors within the namespace so am not sure if i'm missing something.

My goal here is to get all three types of sensor data as single observation.

[QuanyiLi]

Could you set image_observation=True in the env config?

also, please update the observe function to

    def observe(self, vehicle):
        os = {o: getattr(self, o).observe(vehicle if o=="state" else vehicle.origin) for o in ["rgb", "state", "depth", "semantic"]}
        return os

Then it should work

[abastola0]

It works. Thanks!