alf.environments.metadrive#

alf.environments.metadrive.agent_perception#

class AgentPerception(fov, history_window_size, history_frame_skip=4, agent_limit=16)[source]#

Bases: object

A perception module that once initialized can produced the vectorized feature of the dynamic road users (agents) that are visible to the ego car in the driving scenario.

The essential method of AgentPerception is observe(), which is called upon every observation to generate the observation of the dynamic agents.

Useful Notations:

B - the batch size, a.k.a. the agent limit (see below) H - history_window_size, a.k.a. the number of historical steps F - the size of the feature for each of the agent at each step

The final feature is a 3D tensor of shape [B, H, F].

The feature for each agent at each step is a vector of 7 elements:

The distance between the centers of ego and the agent (1)

The unit vector point from the center of ego to that of the agent (2)

The width and length of the agent (2)

The heading of agent (w.r.t. ego’s heading) as an unit vector (2)

Construct an AgentPerception instance.

Parameters

fov (FieldOfView) – Describe the field of view (FOV) of the ego car. When generating the agent features, only those agent who is within the FOV result in the feature.
history_window_size (int) – The feature only tracks this number of historical frames (current frame included).
history_frame_skip (int) – Pick every this number of frames to form the historical trajectories for the agents.
agent_limit (int) – The maximum number of agents shown in the feature. If the number of visible agents exceeds this limit, the farthest ones are filtered out until this limit is satisfied.

property observation_spec#

observe()[source]#

Called upon every observation to produce the feature vectors describing the dynamic agents that are visible to the ego car. The vectors are transformed so that they are in ego car’s body frame.

Returns

A 3D feature tensor of shape [B, H, F]. See class docstring for the meaning of B, H and F.
An integer indicating how many agents are actually filled.

Return type

A tuple of 2

reset(engine, ego)[source]#

Initialize by creating the buffers for holding the dynamic agents information.

This is internally used by MetaDrive related Observation objects, called once when a new MetaDrive environment (which is required to produce agent related observations) is constructed.

alf.environments.metadrive.environments#

class BirdEyeTopDownEnv(config=None)[source]#

Bases: metadrive.envs.metadrive_env.MetaDriveEnv

This is the counterpart of the TopDownEnv from MetaDrive with vectorized input insead of raster input (BEV).

classmethod default_config()[source]#

The default config is identical to that of the raster TopDownEnv.

Return type: Config

get_single_observation(_=None)[source]#

Implements the get_single_observation for the base class MetaDriveEnv.

The base class is calling this function to acquire the sensor (typed ObservationBase) that is used for generating observations. Unlike the name may suggest, it is

actually only called once per environment
returning a sensor object instead of the actual observation

The sensor object is then used to produce the actual observation of each frame.

Return type: ObservationBase

property observation_spec#

render(observation=None)[source]#: This is a pseudo-render function, only used to update onscreen message when using panda3d backend :param mode: ‘rgb’/’human’ :param text:text to show :rtype: Optional[ndarray] :return: when mode is ‘rgb’, image array is returned

class VectorizedTopDownEnv(config=None)[source]#

Bases: metadrive.envs.metadrive_env.MetaDriveEnv

This is the counterpart of the TopDownEnv from MetaDrive with vectorized input insead of raster input (BEV).

classmethod default_config()[source]#

The default config is identical to that of the raster TopDownEnv.

Return type: Config

get_single_observation(_=None)[source]#

Implements the get_single_observation for the base class MetaDriveEnv.

The base class is calling this function to acquire the sensor (typed ObservationBase) that is used for generating observations. Unlike the name may suggest, it is

actually only called once per environment
returning a sensor object instead of the actual observation

The sensor object is then used to produce the actual observation of each frame.

Return type: ObservationBase

property observation_spec#

render(observation=None)[source]#: This is a pseudo-render function, only used to update onscreen message when using panda3d backend :param mode: ‘rgb’/’human’ :param text:text to show :rtype: Optional[ndarray] :return: when mode is ‘rgb’, image array is returned

alf.environments.metadrive.extra_rewards#

class CrashVehicleReward(cost=20.0)[source]#

Bases: alf.environments.metadrive.extra_rewards.ExtraReward

The EXTRA reward that penalizes ego car from crashing into another vehicle.

MetaDrive has already has a flat penalty towards all kinds of crashing (road boundary, vehicle, objects, etc). To further penalize crashing into vehicles over crashing into the road boundary, we added this extra reward.

Note that the episode will END when a crash happens on the ego car. This means that such reward is imposed at most ONCE per episode.

Parameters: cost (float) – the extra cost imposed when the crash is with another vehicle.

env_info_spec()[source]#

Returns the env_info_spec generated by this reward.

This is useful for extending the default env info spec of MetaDrive.

evaluate(engine)[source]#

Evaluate the reward.

This is the main API that generate the reward and related information based on the current states of the MetaDrive engine.

Parameters: engine (BaseEngine) – The MetaDrive simulator’s engine object. It provides access to various states of the simulator, including the ones of the ego vehicle.
Returns: A pair of dictionaries. The first dictionary is a map of invidual reward names to their values. The second dictionary are extra env info that will be added to the environment’s env info when this reward is turned on.

class EgoKinematicReward(harsh_brake_cost_func=<function squared_brake_cost>, lon_jerk_cost_func=<function squared_jerk_cost>, lat_jerk_cost_func=<function squared_jerk_cost>)[source]#

Bases: alf.environments.metadrive.extra_rewards.ExtraReward

The comfort rewards that are based on the kinematics of the ego vehicle.

Parameters

harsh_brake_cost_func – the function that converts lon acc to a reward
lon_jerk_cost_func – the function that converts lon jerk to a reward
lat_jerk_cost_func – the function that converts lat jerk to a reward

env_info_spec()[source]#

Returns the env_info_spec generated by this reward.

This is useful for extending the default env info spec of MetaDrive.

evaluate(engine)[source]#

Evaluate the reward.

This is the main API that generate the reward and related information based on the current states of the MetaDrive engine.

Parameters: engine (BaseEngine) – The MetaDrive simulator’s engine object. It provides access to various states of the simulator, including the ones of the ego vehicle.
Returns: A pair of dictionaries. The first dictionary is a map of invidual reward names to their values. The second dictionary are extra env info that will be added to the environment’s env info when this reward is turned on.

reset()[source]#

Reset the internal states of the reward.

Some of the reward maintains internal states for computing rewards (e.g. history buffer for computing the derivatives). Such internal states are reset at the end of each episode via overriding this method.

class ExtraReward[source]#

Bases: abc.ABC

Base class for MetaDrive extra rewards.

The required interface are defined here.

abstract env_info_spec()[source]#

Returns the env_info_spec generated by this reward.

This is useful for extending the default env info spec of MetaDrive.

abstract evaluate(engine)[source]#

Evaluate the reward.

This is the main API that generate the reward and related information based on the current states of the MetaDrive engine.

Parameters: engine (BaseEngine) – The MetaDrive simulator’s engine object. It provides access to various states of the simulator, including the ones of the ego vehicle.
Return type: Tuple[dict, dict]
Returns: A pair of dictionaries. The first dictionary is a map of invidual reward names to their values. The second dictionary are extra env info that will be added to the environment’s env info when this reward is turned on.

reset()[source]#

Reset the internal states of the reward.

class LaneKeepingReward(broken_line_cost=0.05)[source]#

Bases: alf.environments.metadrive.extra_rewards.ExtraReward

The reward that penalizes riding a broken line.

A broken line resides between two lanes that permits lane change. This reward particularly encourages the car to stay in the lane unless it wants to perform lane change or overtaking.

Parameters: broken_line_cost (float) – the penalty for one step if the ego car is riding on a broken line.

env_info_spec()[source]#

Returns the env_info_spec generated by this reward.

This is useful for extending the default env info spec of MetaDrive.

evaluate(engine)[source]#

Evaluate the reward.

This is the main API that generate the reward and related information based on the current states of the MetaDrive engine.

Parameters: engine (BaseEngine) – The MetaDrive simulator’s engine object. It provides access to various states of the simulator, including the ones of the ego vehicle.
Returns: A pair of dictionaries. The first dictionary is a map of invidual reward names to their values. The second dictionary are extra env info that will be added to the environment’s env info when this reward is turned on.

is_harsh_brake(lon_acc, speed)[source]#: Simple empirical thresholds for harsh brake.

squared_brake_cost(lon_acc, speed, harsh_brake_limit=- 1.2, speed_deadband=2.0, scale=2.0, cap=1.0)[source]#

Produce a cost based on the (harsh) brake.

The function is a squared cost of the violation that only activate when the lon acceleration is beyond the harsh brake limit and the speed is also above its deadband.

Also note that the cost is always POSITIVE or zero.

Parameters

lon_acc (float) – the longitudinal acceleration
speed (float) – the value of the speed
harsh_brake_limit (float) – the threshold used to determine whether the lon_acc is considered a harsh brake to be penalized.
speed_deadband (float) – cost is 0.0 if speed is below this threshold
scale (float) – the scale of the squared violation
cap (float) – if the scaled squared violation will be capped by this value

squared_jerk_cost(jerk, speed, jerk_deadband=4.0, speed_deadband=1.5, scale=0.001, cap=0.8)[source]#

Produce a cost based on the jerk.

The function is a squared cost of the violation that only activate when the jerk is above its deadband and the speed is also above its deadband.

Also note that the cost is always POSITIVE or zero.

Parameters

jerk (float) – the value of the jerk
speed (float) – the value of the speed
jerk_deadband (float) – cost is 0.0 if jerk’s abs value is below this threshold
speed_deadband (float) – cost is 0.0 if speed is below this threshold
scale (float) – the scale of the squared violation
cap (float) – if the scaled squared violation will be capped by this value

alf.environments.metadrive.geometry#

class CategoryEncoder[source]#

Bases: object

A category encoder can

Convert integer categories into their corresponding one-hot encoding.
Translate the type of driving scenario objects into its integer category.

encode_line_type(line_type)[source]#

Translate the line type into its corresponding category index.

Parameters: line_type (LineType) – the line type specifying the category of the line, e.g. broken line, continuous line.
Return type: int
Returns: An integer within [0, self.size - 1] denoting the index of the category corresponding to the line type.

encode_navigation()[source]#

Get the category index of the navigation.

Return type: int
Returns: An integer within [0, self.size - 1] denoting the index of the category representing the navigation category.

get_codes(indices)[source]#

Convert category indices to the onehot encoding vectors.

Parameters: indices (ndarray) – A tensor of category indices.
Return type: ndarray
Returns: A tensor with one extra dimension compared to the input, with each integer in the input replaced by its corresponding onehot encoding vector.

property size: int#

Get the number of categories.

Return type: int
Returns: An integer representing the total number of categories.

class FieldOfView(front=60.0, rear=40.0, lateral=30.0)[source]#

Bases: object

Describe the area with respect to the origin (0, 0) that are visible.

Under the hood the FieldOfView object is a rectangular bounding box.

Construct a FieldOfView object by specifying the relative metrics. Note that this is in car-body coordinate frame, where (1, 0) points to the car’s orientation direction.

Parameters

front (float) – Defines how far away are visible to the front of the car.
rear (float) – Defines how far away are visible to the back of the car.
lateral (float) – Defines how far away are visible to the left and right of the car.

property bbox#

within(points)[source]#

Returns for each of the input points, whether they are within the field of view or not.

Parameters: points (ndarray) – A n-d tensor with shape of […, 2] describing a batch of input 2D points.
Returns: -1]``. Each cell in the result is True (the point is within the FOV) or False (the point is not within the FOV).
Return type: A (n-1)-d tensor with shape ``points.shape[

class Polyline(point: np.ndarray, category: Optional[np.ndarray] = None)[source]#

Bases: tuple

Hold a single polyline or a batch of polylines.

A single 2D polyline of S segments can be represented as an ordered sequence of (S + 1) 2D points. A batch of B 2D polylines of S segments can be represented as B x such point sequences. Each polyline can optionally have a category attached to them, which is denoted as an integer.

When the Polyline instance is used to store a batch of polylines: point will be of shape [B, S + 1, 2], and category will be of shape [B,] if it is not None
When the Polyline instance is used to store a single polyline: point will be of shape [S + 1, 2], and category will be a single integer if it is not None

The Polyline class provides useful utility methods to transform the polyline(s) within it, and to extract features from it.

Create new instance of Polyline(point, category)

property batched: bool#

Returns true if the instance represents a batch of polylines, as opposed to a single polyline.

Return type: bool

category: Optional[numpy.ndarray]#: Alias for field number 1

static from_lane(lane, lateral, category, segment_resolution, polyline_size)[source]#

Constructs a Polyline instance from a MetaDrive lane.

The constructed Polyline instance will contain a set of polylines. The input MetaDrive lane (tegother with lateral) describes a curve. It will be divided into a set of polylines where each polyline will have polyline_size segments, and each segment will be targeting a length of segment_resolution.

This means that the length of each polyline will be the product of segment_resolution and polyline_size unless it is the last polyline. The last polyline is usually longer or smaller, depending on the length of curve described by the lane.

Parameters

lane (MetaDriveLane) – a MetaDrive lane as the reference for the target curve.
lateral (float) – describes the lateral offset (ratio) of the target curve with respect to the lane. It should be within [-0.5, 0.5]. For example, if it is set to -0.5, it means that the target curve is the right side boundary of the lane. Similarly, 0.5 is for the left side boungdary, and 0.0 is for the cetner line of the lane.
category (int) – an integer representing the category of the curve. For example, whether it is a broken line or a solid line. Not all polylines need to have a category, but it is necessary here since the polyline represents an object from the map.
segment_resolution (float) – together with polyline_size it describes how to divided the target curve into polylines. See method docstring for details. Unit in meters.
polyline_size (int) – together with segment_resolution it describes how to divided the target curve into polylines. See method docstring for details. Unit in meters.

Return type

Polyline

Returns

A Polyline instance containing a batch of polylines extracted from the input lane.

keep_closest_n(n)[source]#

Filter the polylines so that only the closest n polylines are kept. The distances are measured as L2 distance with respect to (0, 0).

Return type: Polyline

point: numpy.ndarray#: Alias for field number 0

to_feature(required_batch_size=None, category_encoder=None)[source]#

Convert the polyline(s) to their corresponding feature vectors.

Suppose there are B polylines, each with S segments (and therefore S + 1 points). The feature construction is described below.

The feature vector of a single segment is of 6d - the direction of the middle point of the segment (unit vector) - the direction along the segment (unit vector) - the distance between (0, 0) and the middle point - the length of the segment
The feature vector of a polyline is a concatenation (ordered) of all the feature vectors of its segments plus one-hot encoding of the polyline’s category if present. Each polyline feature vector is of size F = (S * 6 + numumber of categories).
The feature vector of the batch of polylines (i.e. the Polyline instance itself) is a stack of all the polyline feature vectors. The shape is therefore [B, F].

Note that if the Polyline instance is NOT batched, ignore No. 3 above.

Parameters

required_batch_size (Optional[int]) – The returned feature should have a batch size of this. If the number of polylines does not match this number, employ zero padding to make it so. ONLY USEFUL for batched case.
category_encoder (Optional[CategoryEncoder]) – an encoder that can convert integer category to its corresponding one-hot encoding. When not provided, category one-hot encoding WILL NOT be appended to the polyline feature.

Return type

ndarray

Returns

A tensor as the feature representation of the polylines in the instance.

transformed(center, orientation)[source]#

Returns a transformed (set of) polyline(s) so that the resulting points are in the coordinate frame defined by the pose (center and the orientation).

Parameters

center (ndarray) – A 2D point denoting the origin of the new coordinate frame.
orientation (float) – orientation (x-axis direction) of the new coordinate in radian.

Return type

Polyline

Returns

A NEW Polyline instance where all the polylines within are transformed.

transformed_within_fov(position, heading, fov)[source]#

Transform the polylines into the car-body coordinate frame defined by the car’s position and heading, and filtered out the polylines that are not within the field of view.

Parameters

position (ndarray) – the position of the car serving as the observer.
heading (float) – heading of the car serving as the observer.
fov (FieldOfView) – the field of view of the car defining the area that are visible.

Return type

Polyline

Returns

A NEW Polyline instance where only the polylines within the specified FOV are kept.

alf.environments.metadrive.map_perception#

class MapPolylinePerception(fov, segment_resolution=1.0, polyline_size=1, polyline_limit=256)[source]#

Bases: object

A perception module that once initialized can produced the polyline-based vectorized feature of the semantic map and navigation of a MetaDrive scenario.

The essential method of MapPolylinePerception is observe(), which is called upon every observation to generate the part of the observation from the map.

Under the hood, upon initialization we will create polylines for all of the map objects. Whenever observe() is called, it will crop and rotate the polylines (with batch vectorized operations) to produce the feature vectors in a fast manner.

Construct a MapPolylinePerception instance.

Parameters

fov (FieldOfView) – Defines the field of view with respect to the ego car. Map object and agents outside of the field of view will not appear in the observation. Usually the bigger the field of view, the more expensive the computation of the obsrevation (and the training) will be.
segment_resolution (float) – The length of each line segment in the polylines during sampling. The smaller the value, the more segments and polylines. As a result, it also implies more expensive training.
polyline_size (int) – Specify the number of segments in one polyline. Putting more segments in a polyline can reduce the number of polylines for each observation.
polyline_limit (int) – Specify the maximum number of polylines in the observation for map and navigation. If the actual number of polylines goes beyond this limit, farthest polylines (from the ego car) will be filtered out until the limit is satisfied. If the actual number of polylines is below this limit, zero padding will be employed to bring fill the vacancies.

property observation_spec#

observe(position, heading)[source]#

Called upon every observation to get a rotated and cropped view of the map objects and navigation. Returns the feature vector of the observation.

The position and heading of the observer car needs to be provided to define the coordinate frame of the resulting features.

Parameters

position (tuple) – A 2D vector denoting the current world frame position of the ego car.
heading (float) – A radian denoting the current world frame heading of the ego car.

Returns

A feature tensor of shape [polyline_limit, feature_size], where feature size is determined by the number of segments in each polyline (polyline_size).
An integer indicating among the polyline_limit of polylines, how many of them are actually filled. If the feature has 128 polylines and oly 120 are filled, feature’s [120:] will be all zeros.

Return type

A Tuple of 2

reset(road_network, navigation)[source]#

Initialize by creating the polylines for all the map objects and navigation.

Parameters

road_network – The road network (map) of the current MetaDrive environment.
navigation – The navigatiton of the current MetaDrive environment.

alf.environments.metadrive.renderer#

class Renderer(observation_renderer=None)[source]#

Bases: metadrive.obs.top_down_renderer.TopDownRenderer

Specialized TopDownRenderer for MetaDrive that adds extra information by

rendering actions and some other internal info
rendering observations

The original MetaDrive top-down renderer renders on an 1000 x 1000 canvas. This specialized renderer extends the area to be 1000 x 1200 so that the bottom part of size 1000 x 200 can be used for extra information.

Construct a Renderer instance.

Please refer to make_vectorized_observation_renderer nad make_bird_eye_observation_renderer below for available observaion renderers.

Parameters: observation_renderer (Optional[Callable[[Surface, Any], None]]) – A function that takes in a canvas (typed as pygame.Surface) and an observation, and renders the observation on the canvas. When specified as None, this Renderer will not render observation.

render(observation=None)[source]#

Renders the current frame.

This function is designed to be called once per frame. It dras the map, the dynamic objects (ego and other cars), while also visualizing the observation given

An observation renderer is specified upon construction
An observation is passed in

Parameters: observation – The observation of the current frame. If None, nothing will be rendered about the observation.
Returns: The canvas with everything rendered. The return value is provided for recording purposes, and the render on screen does not rely on having this return value.

make_bird_eye_observation_renderer()[source]#

Create a renderer for the BEV observation.

Each channel from the BEV will be drawn on the canvas in a row.

make_vectorized_observation_renderer(sensor)[source]#

Create a renderer for the vectorized observation.

The created renderer is a closure that draws vectorized observation on a pygame Surface. The parameters about the observation is retrieved from the input sensor.

Parameters: sensor (VectorizedObservation) – A vectorized observation sensor providing properties about the observation, such as the number of polylines and the number of segments within the polylines.

alf.environments.metadrive.sensors#

class BirdEyeObservation(env_config, velocity_steps=1, velocity_normalization=100.0)[source]#

Bases: metadrive.obs.top_down_obs_multi_channel.TopDownMultiChannel

This implements a MetaDrive Observation that produces BEV, together with historical velocities.

The implementation is simply adding velocity alongside the original BEV produced by MetaDrive’s TopDownMultiChannel.

Note that the velocity histories are not encoded as images. The model will have the flexibility to choose to how to encode it.

Construct a BirdEyeObservation instance.

Parameters

env_config (Config) – MetaDrive’s environment configuration.
velocity_steps (int) – The number of historical steps for the velocity.
velocity_normalization (float) – The velocities (in m/s) will be normalized by this factor before producing the feature.

property observation_spec#

observe(vehicle)[source]#

The main API to generate the feature given the current ego car status.

Parameters: vehicle (BaseVehicle) – The MetaDrive vehicle object is the container for all the ego car related information.

class VectorizedObservation(vehicle_config, fov=<alf.environments.metadrive.geometry.FieldOfView object>, segment_resolution=2.0, polyline_size=4, polyline_limit=64, history_window_size=8, agent_limit=16)[source]#

Bases: metadrive.obs.observation_base.ObservationBase

This implements a customized observation for the MetaDrive environment that produces vectorized inputs (as opposed to raster inputs such as BEV).

The main API observe() is designed to produces a dictionary of vectorized inputs. Depending on the configuration the dictionary may consist of:

‘map’: polyline based features of road network (lanes) and navigations
‘ego’: polyline based feature of ego car history
‘agents’: polyline based feature of dynamic road users for interaction

All polylines are within a specified field of view, and transformed so that they are in the ego car’s body frame, where x-axis points to the heading direction of the ego car.

Construct a VectorizedObservation instance.

Parameters

vehicle_config (Config) – A MetaDrive global configuration of the MetaDrive environment that this observation is attached to.
fov (FieldOfView) – Defines the field of view with respect to the ego car. Map object and agents outside of the field of view will not appear in the observation. Usually the bigger the field of view, the more expensive the computation of the obsrevation (and the training) will be.
segment_resolution (float) – The length of each line segment in the polylines during sampling. The smaller the value, the more segments and polylines. As a result, it also implies more expensive training.
polyline_size (int) – Specify the number of segments in one polyline. Putting more segments in a polyline can reduce the number of polylines for each observation.
polyline_limit (int) – Specify the maximum number of polylines in the observation for map and navigation. If the actual number of polylines goes beyond this limit, farthest polylines (from the ego car) will be filtered out until the limit is satisfied. If the actual number of polylines is below this limit, zero padding will be employed to bring fill the vacancies.
history_window_size (int) – The past positions of the most recent this number of frames will be recorded and used for the ego history feature.
agent_limit (int) – the maximum number of agents appears in the agent feature.

property observation_space#

The base class ObservationBase requires that we have observation_space() implemented and returns a gym.spaces.Box object.

THIS IS A HACK.

This is just a dummy function and it does not return the actual observation spec (because the actual observation spec is a dictionary, which defies the requirement of the base class. We do this nonetheless because in our implementation we are actually using observation_spec() below.

property observation_spec#

observe(vehicle)[source]#

The main API to generate the vectorized input, given the current ego state.

All the static polylines are generated during pre-compuation in reset() (see below). The call to observe() basically crops the pre-computed polylines within the field of view, and does the coordinate frame transformation into the ego car’s body frame.

Parameters: vehicle (BaseVehicle) – The MetaDrive vehicle object is the container for all the ego car related information.

reset(env, vehicle=None)[source]#