Source code for highway_env.envs.parking_env

from __future__ import annotations

from abc import abstractmethod

import numpy as np
from gymnasium import Env

from highway_env.envs.common.abstract import AbstractEnv
from highway_env.envs.common.observation import (
    MultiAgentObservation,
    observation_factory,
)
from highway_env.road.lane import LineType, StraightLane
from highway_env.road.road import Road, RoadNetwork
from highway_env.vehicle.graphics import VehicleGraphics
from highway_env.vehicle.kinematics import Vehicle
from highway_env.vehicle.objects import Landmark, Obstacle


class GoalEnv(Env):
    """
    Interface for A goal-based environment.

    This interface is needed by agents such as Stable Baseline3's Hindsight Experience Replay (HER) agent.
    It was originally part of https://github.com/openai/gym, but was later moved
    to https://github.com/Farama-Foundation/gym-robotics. We cannot add gym-robotics to this project's dependencies,
    since it does not have an official PyPi package, PyPi does not allow direct dependencies to git repositories.
    So instead, we just reproduce the interface here.

    A goal-based environment. It functions just as any regular OpenAI Gym environment but it
    imposes a required structure on the observation_space. More concretely, the observation
    space is required to contain at least three elements, namely `observation`, `desired_goal`, and
    `achieved_goal`. Here, `desired_goal` specifies the goal that the agent should attempt to achieve.
    `achieved_goal` is the goal that it currently achieved instead. `observation` contains the
    actual observations of the environment as per usual.
    """

    @abstractmethod
    def compute_reward(
        self, achieved_goal: np.ndarray, desired_goal: np.ndarray, info: dict
    ) -> float:
        """Compute the step reward. This externalizes the reward function and makes
        it dependent on a desired goal and the one that was achieved. If you wish to include
        additional rewards that are independent of the goal, you can include the necessary values
        to derive it in 'info' and compute it accordingly.
        Args:
            achieved_goal (object): the goal that was achieved during execution
            desired_goal (object): the desired goal that we asked the agent to attempt to achieve
            info (dict): an info dictionary with additional information
        Returns:
            float: The reward that corresponds to the provided achieved goal w.r.t. to the desired
            goal. Note that the following should always hold true:
                ob, reward, done, info = env.step()
                assert reward == env.compute_reward(ob['achieved_goal'], ob['desired_goal'], info)
        """
        raise NotImplementedError


[docs] class ParkingEnv(AbstractEnv, GoalEnv): """ A continuous control environment. It implements a reach-type task, where the agent observes their position and speed and must control their acceleration and steering so as to reach a given goal. Credits to Munir Jojo-Verge for the idea and initial implementation. """ # For parking env with GrayscaleObservation, the env need # this PARKING_OBS to calculate the reward and the info. # Bug fixed by Mcfly(https://github.com/McflyWZX) PARKING_OBS = { "observation": { "type": "KinematicsGoal", "features": ["x", "y", "vx", "vy", "cos_h", "sin_h"], "scales": [100, 100, 5, 5, 1, 1], "normalize": False, } } def __init__(self, config: dict = None, render_mode: str | None = None) -> None: super().__init__(config, render_mode) self.observation_type_parking = None
[docs] @classmethod def default_config(cls) -> dict: config = super().default_config() config.update( { "observation": { "type": "KinematicsGoal", "features": ["x", "y", "vx", "vy", "cos_h", "sin_h"], "scales": [100, 100, 5, 5, 1, 1], "normalize": False, }, "action": {"type": "ContinuousAction"}, "reward_weights": [1, 0.3, 0, 0, 0.02, 0.02], "success_goal_reward": 0.12, "collision_reward": -5, "steering_range": np.deg2rad(45), "simulation_frequency": 15, "policy_frequency": 5, "duration": 100, "screen_width": 600, "screen_height": 300, "centering_position": [0.5, 0.5], "scaling": 7, "controlled_vehicles": 1, "vehicles_count": 0, "add_walls": True, } ) return config
[docs] def define_spaces(self) -> None: """ Set the types and spaces of observation and action from config. """ super().define_spaces() self.observation_type_parking = observation_factory( self, self.PARKING_OBS["observation"] )
def _info(self, obs, action) -> dict: info = super()._info(obs, action) if isinstance(self.observation_type, MultiAgentObservation): success = tuple( self._is_success(agent_obs["achieved_goal"], agent_obs["desired_goal"]) for agent_obs in obs ) else: obs = self.observation_type_parking.observe() success = self._is_success(obs["achieved_goal"], obs["desired_goal"]) info.update({"is_success": success}) return info def _reset(self): self._create_road() self._create_vehicles() def _create_road(self, spots: int = 14) -> None: """ Create a road composed of straight adjacent lanes. :param spots: number of spots in the parking """ net = RoadNetwork() width = 4.0 lt = (LineType.CONTINUOUS, LineType.CONTINUOUS) x_offset = 0 y_offset = 10 length = 8 for k in range(spots): x = (k + 1 - spots // 2) * (width + x_offset) - width / 2 net.add_lane( "a", "b", StraightLane( [x, y_offset], [x, y_offset + length], width=width, line_types=lt ), ) net.add_lane( "b", "c", StraightLane( [x, -y_offset], [x, -y_offset - length], width=width, line_types=lt ), ) self.road = Road( network=net, np_random=self.np_random, record_history=self.config["show_trajectories"], ) def _create_vehicles(self) -> None: """Create some new random vehicles of a given type, and add them on the road.""" empty_spots = list(self.road.network.lanes_dict().keys()) # Controlled vehicles self.controlled_vehicles = [] for i in range(self.config["controlled_vehicles"]): x0 = (i - self.config["controlled_vehicles"] // 2) * 10 vehicle = self.action_type.vehicle_class( self.road, [x0, 0], 2 * np.pi * self.np_random.uniform(), 0 ) vehicle.color = VehicleGraphics.EGO_COLOR self.road.vehicles.append(vehicle) self.controlled_vehicles.append(vehicle) empty_spots.remove(vehicle.lane_index) # Goal for vehicle in self.controlled_vehicles: lane_index = empty_spots[self.np_random.choice(np.arange(len(empty_spots)))] lane = self.road.network.get_lane(lane_index) vehicle.goal = Landmark( self.road, lane.position(lane.length / 2, 0), heading=lane.heading ) self.road.objects.append(vehicle.goal) empty_spots.remove(lane_index) # Other vehicles for i in range(self.config["vehicles_count"]): if not empty_spots: continue lane_index = empty_spots[self.np_random.choice(np.arange(len(empty_spots)))] v = Vehicle.make_on_lane(self.road, lane_index, 4, speed=0) self.road.vehicles.append(v) empty_spots.remove(lane_index) # Walls if self.config["add_walls"]: width, height = 70, 42 for y in [-height / 2, height / 2]: obstacle = Obstacle(self.road, [0, y]) obstacle.LENGTH, obstacle.WIDTH = (width, 1) obstacle.diagonal = np.sqrt(obstacle.LENGTH**2 + obstacle.WIDTH**2) self.road.objects.append(obstacle) for x in [-width / 2, width / 2]: obstacle = Obstacle(self.road, [x, 0], heading=np.pi / 2) obstacle.LENGTH, obstacle.WIDTH = (height, 1) obstacle.diagonal = np.sqrt(obstacle.LENGTH**2 + obstacle.WIDTH**2) self.road.objects.append(obstacle)
[docs] def compute_reward( self, achieved_goal: np.ndarray, desired_goal: np.ndarray, info: dict, p: float = 0.5, ) -> float: """ Proximity to the goal is rewarded We use a weighted p-norm :param achieved_goal: the goal that was achieved :param desired_goal: the goal that was desired :param dict info: any supplementary information :param p: the Lp^p norm used in the reward. Use p<1 to have high kurtosis for rewards in [0, 1] :return: the corresponding reward """ return -np.power( np.dot( np.abs(achieved_goal - desired_goal), np.array(self.config["reward_weights"]), ), p, )
def _reward(self, action: np.ndarray) -> float: obs = self.observation_type_parking.observe() obs = obs if isinstance(obs, tuple) else (obs,) reward = sum( self.compute_reward( agent_obs["achieved_goal"], agent_obs["desired_goal"], {} ) for agent_obs in obs ) reward += self.config["collision_reward"] * sum( v.crashed for v in self.controlled_vehicles ) return reward def _is_success(self, achieved_goal: np.ndarray, desired_goal: np.ndarray) -> bool: return ( self.compute_reward(achieved_goal, desired_goal, {}) > -self.config["success_goal_reward"] ) def _is_terminated(self) -> bool: """The episode is over if the ego vehicle crashed or the goal is reached or time is over.""" crashed = any(vehicle.crashed for vehicle in self.controlled_vehicles) obs = self.observation_type_parking.observe() obs = obs if isinstance(obs, tuple) else (obs,) success = all( self._is_success(agent_obs["achieved_goal"], agent_obs["desired_goal"]) for agent_obs in obs ) return bool(crashed or success) def _is_truncated(self) -> bool: """The episode is truncated if the time is over.""" return self.time >= self.config["duration"]
class ParkingEnvActionRepeat(ParkingEnv): def __init__(self): super().__init__({"policy_frequency": 1, "duration": 20}) class ParkingEnvParkedVehicles(ParkingEnv): def __init__(self): super().__init__({"vehicles_count": 10})