1472

pending completion

Build # 1472

Build Type

push

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/AboudyKreidieh/h-baselines/commit/<a class=hub.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://git"><a class=hub.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://git"><a class=hub.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://git"><a class=hub.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://git"><a class=hub.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://git"><a class=hub.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">6dee72e9a<a href="https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">&lt;a href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">&lt;a href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;lt;a href=&amp;amp;amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;quot;&amp;amp;amp;gt;Point2d bipedal envs (#130)

* added new envs

* Rendered TeX expressions in &amp;amp;amp;lt;/a&amp;amp;amp;gt;&amp;amp;amp;lt;a class=&amp;amp;amp;quot;double-link&amp;amp;amp;quot; href=&amp;amp;amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/&amp;amp;amp;lt;a class=&amp;amp;amp;quot;double-link&amp;amp;amp;quot; href=&amp;amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/24c4a2f50b8f29ca6721d201c9bb3fd245002292&amp;amp;quot;&amp;amp;gt;24c4a2f50&amp;amp;lt;/a&amp;amp;gt;&amp;amp;quot;&amp;amp;gt;24c4a2f50&amp;amp;lt;/a&amp;amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&amp;amp;quot;&amp;amp;gt;

* flake8 and pydocstyle (mostly)

* added template for tests and multiworld to travis

* some cleanup

* some addition to tests

* added tests for point envs

* Rendered TeX expressions in &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/AboudyKreidieh/h-baselines/commit/8084cb65d2bce16ad66f55aeb93cc75b6a95d94a&amp;quot;&amp;gt;8084cb65d&lt;/a&gt;&quot;&gt;8084cb65d&lt;/a&gt;&lt;a href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151&quot;&gt;, &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/9c5a71fc29482367a65a8d1bc12918ba7883d84a&quot;&gt;9c5a71fc2&lt;/a&gt;&quot;&gt;9c5a71fc2&lt;/a&gt;&lt;a href=&quot;https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">, </a><a class="double-link" href="https://github.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://github.com/AboudyKreidieh/h-baselines/commit/4103c7d07c7a6c878fdab8e4e2f5e0a57b52a2cc">4103c7d07</a>">4103c7d07</a><a href="https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">, </a><a class="double-link" href="https://github.com/AboudyKreidieh/h-baselines/commit/<a class="double-link" href="https://github.com/AboudyKreidieh/h-baselines/commit/89231e60a37de1d3d1f2f7df39fd317cc5d25b38">89231e60a</a>">89231e60a</a><a href="https://github.com/AboudyKreidieh/h-baselines/commit/6dee72e9ad46dd997016eb1b80d50e5631cca151">, a2f2231cf">a2f2231cf, 
66.85

/hbaselines/envs/point2d.py

"""Script containing the Point2DEnv object."""
import logging
import numpy as np
from gym import spaces
from pygame import Color
import matplotlib.pyplot as plt
from multiworld.core.multitask_env import MultitaskEnv
from multiworld.core.serializable import Serializable
from multiworld.envs.pygame.pygame_viewer import PygameViewer


class Point2DEnv(MultitaskEnv, Serializable):
    """A little 2D point whose life goal is to reach a target.

    States
        TODO

    TODO: everything...

    Attributes
    ----------
    render_dt_msec : float
        seconds before the next frame in the image is rendered
    action_l2norm_penalty : float
        penalty scale for the actions by the agent
    render_onscreen : bool
        whether to include the rendering visually (instead of simply using the
        image for the observation)
    render_size : int
        width/length number of pixels in the rendered image
    reward_type : str
        the reward type. Must be one of: "sparse", "dense", or
        "vectorized_dense"
    action_scale : float
        the multiple from action to velocity
    target_radius : float
        the radius of the targeted position when being rendered
    boundary_dist : float
        the distance from the center to the boundary
    ball_radius : float
        the radius of the agent when being rendered
    walls : TODO
        TODO
    fixed_goal : [float, float] or None
        the goal to use. If set to None, it is picked randomly.
    randomize_position_on_reset : bool
        whether to initialize the position of the agent randomly
    images_are_rgb : bool
        specifies whether the image is RGB. Otherwise, it's black and white
    show_goal : bool
        whether to render the goal(s)
    images_in_obs : bool
        whether to use the image in the observation
    image_size : int
        number of elements in the image. Set to 0 if images are not being used.
    max_target_distance : float
        TODO
    action_space : gym.spaces.*
        the action space of the environment
    obs_range : gym.spaces.*
        the range of the initial position of the agent
    context_space  : gym.spaces.*
        the context space of the environment
    observation_space : gym.spaces.*
        the observation space of the environment
    drawer : multiworld.envs.pygame.pygame_viewer.PygameViewer or None
        The drawer for the images of the environment. Set to None if images are
        not being used.
    render_drawer : multiworld.envs.pygame.pygame_viewer.PygameViewer or None
        The drawer for the images of the environment if the environment is
        being rendered. Set to None if images are not being used.
    horizon : int
        environment time horizon
    t : int
        number of steps since the start of the most recent episode
    """

    def __init__(self,
                 render_dt_msec=0,
                 action_l2norm_penalty=0,  # disabled for now
                 render_onscreen=False,
                 render_size=32,
                 reward_type="dense",
                 action_scale=1.0,
                 target_radius=0.60,
                 boundary_dist=4,
                 ball_radius=0.50,
                 walls=None,
                 fixed_goal=None,
                 randomize_position_on_reset=True,
                 images_are_rgb=False,  # else black and white
                 show_goal=True,
                 images_in_obs=True,
                 **kwargs):
        """Instantiate the environment.

        Parameters
        ----------
        render_dt_msec : float
            seconds before the next frame in the image is rendered
        action_l2norm_penalty : float
            penalty scale for the actions by the agent
        render_onscreen : bool
            whether to include the rendering visually (instead of simply using
            the image for the observation)
        render_size : int
            width/length number of pixels in the rendered image
        reward_type : str
            the reward type. Must be one of: "sparse", "dense", or
            "vectorized_dense"
        action_scale : float
            the multiple from action to velocity
        target_radius : float
            the radius of the targeted position when being rendered
        boundary_dist : float
            the distance from the center to the boundary
        ball_radius : float
            the radius of the agent when being rendered
        walls : TODO
            TODO
        fixed_goal : [float, float] or None
            the goal to use. If set to None, it is picked randomly.
        randomize_position_on_reset : bool
            whether to initialize the position of the agent randomly
        images_are_rgb : bool
            specifies whether the image is RGB. Otherwise, it's black and white
        show_goal : bool
            whether to render the goal(s)
        images_in_obs : bool
            whether to use the image in the obsevation
        kwargs : dict
            additional arguments. Unused here.
        """
        if walls is None:
            walls = []
        if fixed_goal is not None:
            fixed_goal = np.array(fixed_goal)
        if len(kwargs) > 0:
            logger = logging.getLogger(__name__)
            logger.log(logging.WARNING, "WARNING, ignoring kwargs:", kwargs)

        self.quick_init(locals())
        self.render_dt_msec = render_dt_msec
        self.action_l2norm_penalty = action_l2norm_penalty
        self.render_onscreen = render_onscreen
        self.render_size = render_size
        self.reward_type = reward_type
        self.action_scale = action_scale
        self.target_radius = target_radius
        self.boundary_dist = boundary_dist
        self.ball_radius = ball_radius
        self.walls = walls
        self.fixed_goal = fixed_goal
        self.randomize_position_on_reset = randomize_position_on_reset
        self.images_are_rgb = images_are_rgb
        self.show_goal = show_goal
        self.images_in_obs = images_in_obs
        self.image_size = 1024 * (3 if self.images_are_rgb else 1)
        if not self.images_in_obs:
            self.image_size = 0

        self.max_target_distance = self.boundary_dist - self.target_radius

        self._target_position = None
        self._position = np.zeros(2)

        u = np.ones(2)
        self.action_space = spaces.Box(-u, u, dtype=np.float32)

        o = self.boundary_dist * np.ones(2)
        self.obs_range = spaces.Box(
            -o, o, dtype='float32')
        self.context_space = spaces.Box(
            -o, o, dtype='float32')
        self.observation_space = spaces.Box(
            np.concatenate([np.zeros([self.image_size]), -o], 0),
            np.concatenate([np.ones([self.image_size]), o], 0),
            dtype='float32')

        self.drawer = None
        self.render_drawer = None
        self.horizon = 200
        self.t = 0

    @property
    def current_context(self):
        """Return the current goal by the environment."""
        return self._target_position

    def step(self, velocities):
        """Advance the simulation by one step.

        Parameters
        ----------
        velocities : array_like
            the action by the agent, defined as its velocities in the x and y
            directions

        Returns
        -------
        array_like
            agent's observation of the current environment
        float
            amount of reward associated with the previous state/action pair
        bool
            indicates whether the episode has ended
        dict
            contains other diagnostic information from the previous action
        """
        assert self.action_scale <= 1.0
        velocities = np.clip(
            velocities, a_min=-1, a_max=1) * self.action_scale
        new_position = self._position + velocities
        orig_new_pos = new_position.copy()
        for wall in self.walls:
            new_position = wall.handle_collision(
                self._position, new_position
            )
        if sum(new_position != orig_new_pos) > 1:
            # Hack: sometimes you get caught on two walls at a time. If you
            # process the input in the other direction, you might only get
            # caught on one wall instead.
            new_position = orig_new_pos.copy()
            for wall in self.walls[::-1]:
                new_position = wall.handle_collision(
                    self._position, new_position
                )

        self.t += 1

        self._position = new_position
        self._position = np.clip(
            self._position,
            a_min=-self.boundary_dist,
            a_max=self.boundary_dist,
        )
        distance_to_target = np.linalg.norm(
            self._position - self._target_position
        )
        is_success = distance_to_target < self.target_radius

        ob = self._get_obs()
        reward = self.compute_reward(velocities, {'ob': ob})
        info = {
            'radius': self.target_radius,
            'target_position': self._target_position,
            'distance_to_target': distance_to_target,
            'velocity': velocities,
            'speed': np.linalg.norm(velocities),
            'is_success': is_success,
        }
        done = self.t >= self.horizon
        return ob, reward, done, info

    def reset(self):
        """Reset the environment."""
        self.t = 0
        self._target_position = self.sample_goal()['goals']
        if self.randomize_position_on_reset:
            self._position = self._sample_position(
                self.obs_range.low,
                self.obs_range.high,
            )

        return self._get_obs()

    def _position_inside_wall(self, pos):
        """Return True if the agent is in a wall."""
        for wall in self.walls:
            if wall.contains_point(pos):
                return True
        return False

    def _sample_position(self, low, high):
        """Sample a starting position for the agent."""
        pos = np.random.uniform(low, high)
        while self._position_inside_wall(pos) is True:
            pos = np.random.uniform(low, high)
        return pos

    def _get_obs(self):
        """Return the observation of the agent.

        See States in the description of the environment for more.
        """
        if self.images_in_obs:
            img = self.get_image(
                32, 32).reshape([-1]).astype(np.float32) / 255.0
            return np.concatenate([img, self._position.copy()], 0)
        else:
            return self._position.copy()

    def compute_rewards(self, actions, obs):
        """See parent class.

        The rewards are described in the Rewards section of the description of
        the environment.
        """
        achieved_goals = obs['ob'][:, self.image_size:]
        desired_goals = self._target_position[np.newaxis, :]
        d = np.linalg.norm(achieved_goals - desired_goals, axis=-1)
        if self.reward_type == "sparse":
            return -(d > self.target_radius).astype(np.float32)
        elif self.reward_type == "dense":
            return -d
        elif self.reward_type == 'vectorized_dense':
            return -np.abs(achieved_goals - desired_goals)
        else:
            raise NotImplementedError()

    def get_goal(self):
        """See parent class."""
        return self._target_position.copy()

    def sample_goals(self, batch_size):
        """See parent class.

        The goal is the desired x,y coordinates.
        """
        if self.fixed_goal is not None:
            goals = np.repeat(
                self.fixed_goal.copy()[None],
                batch_size,
                0
            )
        else:
            goals = np.zeros((batch_size, self.obs_range.low.size))
            for b in range(batch_size):
                if batch_size > 1:
                    logging.warning("This is very slow!")
                goals[b, :] = self._sample_position(
                    self.obs_range.low,
                    self.obs_range.high,
                )
        return {'goals': goals}

    # ======================================================================= #
    #                     Functions for ImageEnv wrapper                      #
    # ======================================================================= #

    def get_image(self, width=None, height=None):
        """Return a black and white image."""
        if self.drawer is None:
            if width != height:
                raise NotImplementedError()
            self.drawer = PygameViewer(
                screen_width=width,
                screen_height=height,
                x_bounds=(-self.boundary_dist - self.ball_radius,
                          self.boundary_dist + self.ball_radius),
                y_bounds=(-self.boundary_dist - self.ball_radius,
                          self.boundary_dist + self.ball_radius),
                render_onscreen=self.render_onscreen,
            )
        self.draw(self.drawer)
        img = self.drawer.get_image()
        if self.images_are_rgb:
            return img.transpose((1, 0, 2))
        else:
            r, b = img[:, :, 0], img[:, :, 2]
            img = (-r + b).transpose().flatten()
            return img

    def draw(self, drawer):
        """Create the image corresponding to the current state."""
        drawer.fill(Color('white'))
        if self.show_goal:
            drawer.draw_solid_circle(
                self._target_position,
                self.target_radius,
                Color('green'),
            )
        drawer.draw_solid_circle(
            self._position,
            self.ball_radius,
            Color('blue'),
        )

        for wall in self.walls:
            drawer.draw_segment(
                wall.endpoint1,
                wall.endpoint2,
                Color('black'),
            )
            drawer.draw_segment(
                wall.endpoint2,
                wall.endpoint3,
                Color('black'),
            )
            drawer.draw_segment(
                wall.endpoint3,
                wall.endpoint4,
                Color('black'),
            )
            drawer.draw_segment(
                wall.endpoint4,
                wall.endpoint1,
                Color('black'),
            )
        drawer.render()

    def render(self, mode='human', close=False):
        """Render the environment state."""
        if mode == 'rgb_array':
            return self.get_image(self.render_size, self.render_size)

        if close:
            self.render_drawer = None
            return

        if self.render_drawer is None or self.render_drawer.terminated:
            self.render_drawer = PygameViewer(
                self.render_size,
                self.render_size,
                x_bounds=(-self.boundary_dist-self.ball_radius,
                          self.boundary_dist+self.ball_radius),
                y_bounds=(-self.boundary_dist-self.ball_radius,
                          self.boundary_dist+self.ball_radius),
                render_onscreen=True,
            )
        self.draw(self.render_drawer)
        self.render_drawer.tick(self.render_dt_msec)
        if mode != 'interactive':
            self.render_drawer.check_for_exit()

    # ======================================================================= #
    #                  Static visualization/utility methods                   #
    # ======================================================================= #

    @staticmethod
    def true_model(state, action):
        """Return the next position by the agent.

        Parameters
        ----------
        state : array_like
            the state by the agent
        action : array_like
            the action by the agent

        Returns
        -------
        array_like
            the next position.
        """
        velocities = np.clip(action, a_min=-1, a_max=1)
        position = state
        new_position = position + velocities
        return np.clip(
            new_position,
            a_min=-Point2DEnv.boundary_dist,
            a_max=Point2DEnv.boundary_dist,
        )

    @staticmethod
    def true_states(state, actions):
        """Return the next states given a set of states and actions.

        Parameters
        ----------
        state : array_like
            the states by the agent
        actions : array_like
            the actions by the agent

        Returns
        -------
        list of array_like
            the next states
        """
        real_states = [state]
        for action in actions:
            next_state = Point2DEnv.true_model(state, action)
            real_states.append(next_state)
            state = next_state
        return real_states

    def plot_trajectory(self, ax, states, actions, goal=None):
        """Plot the trajectory of an agent.

        Parameters
        ----------
        ax : matplotlib.axes.Axes
            the axis object to plot the figure on
        states : array_like
            the states by the agent
        actions : array_like
            the actions by the agent
        goal : [int, int]
            the x,y coordinates of the goal
        """
        assert len(states) == len(actions) + 1
        x = states[:, 0]
        y = -states[:, 1]
        num_states = len(states)
        plasma_cm = plt.get_cmap('plasma')
        for i, state in enumerate(states):
            color = plasma_cm(float(i) / num_states)
            ax.plot(state[0], -state[1],
                    marker='o', color=color, markersize=10,
                    )

        actions_x = actions[:, 0]
        actions_y = -actions[:, 1]

        ax.quiver(x[:-1], y[:-1], x[1:] - x[:-1], y[1:] - y[:-1],
                  scale_units='xy', angles='xy', scale=1, width=0.005)
        ax.quiver(x[:-1], y[:-1], actions_x, actions_y, scale_units='xy',
                  angles='xy', scale=1, color='r',
                  width=0.0035, )
        ax.plot(
            [
                -self.boundary_dist,
                -self.boundary_dist,
            ],
            [
                self.boundary_dist,
                -self.boundary_dist,
            ],
            color='k', linestyle='-',
        )
        ax.plot(
            [
                self.boundary_dist,
                -self.boundary_dist,
            ],
            [
                self.boundary_dist,
                self.boundary_dist,
            ],
            color='k', linestyle='-',
        )
        ax.plot(
            [
                self.boundary_dist,
                self.boundary_dist,
            ],
            [
                self.boundary_dist,
                -self.boundary_dist,
            ],
            color='k', linestyle='-',
        )
        ax.plot(
            [
                self.boundary_dist,
                -self.boundary_dist,
            ],
            [
                -self.boundary_dist,
                -self.boundary_dist,
            ],
            color='k', linestyle='-',
        )

        if goal is not None:
            ax.plot(goal[0], -goal[1], marker='*', color='g', markersize=15)
        ax.set_ylim(
            -self.boundary_dist - 1,
            self.boundary_dist + 1,
        )
        ax.set_xlim(
            -self.boundary_dist - 1,
            self.boundary_dist + 1,
        )

1	"""Script containing the Point2DEnv object."""
2	import logging	1✔
3	import numpy as np	1✔
4	from gym import spaces	1✔
5	from pygame import Color	1✔
6	import matplotlib.pyplot as plt	1✔
7	from multiworld.core.multitask_env import MultitaskEnv	1✔
8	from multiworld.core.serializable import Serializable	1✔
9	from multiworld.envs.pygame.pygame_viewer import PygameViewer	1✔
10
11
12	class Point2DEnv(MultitaskEnv, Serializable):	1✔
13	"""A little 2D point whose life goal is to reach a target.
14
15	States
16	TODO
17
18	TODO: everything...
19
20	Attributes
21	----------
22	render_dt_msec : float
23	seconds before the next frame in the image is rendered
24	action_l2norm_penalty : float
25	penalty scale for the actions by the agent
26	render_onscreen : bool
27	whether to include the rendering visually (instead of simply using the
28	image for the observation)
29	render_size : int
30	width/length number of pixels in the rendered image
31	reward_type : str
32	the reward type. Must be one of: "sparse", "dense", or
33	"vectorized_dense"
34	action_scale : float
35	the multiple from action to velocity
36	target_radius : float
37	the radius of the targeted position when being rendered
38	boundary_dist : float
39	the distance from the center to the boundary
40	ball_radius : float
41	the radius of the agent when being rendered
42	walls : TODO
43	TODO
44	fixed_goal : [float, float] or None
45	the goal to use. If set to None, it is picked randomly.
46	randomize_position_on_reset : bool
47	whether to initialize the position of the agent randomly
48	images_are_rgb : bool
49	specifies whether the image is RGB. Otherwise, it's black and white
50	show_goal : bool
51	whether to render the goal(s)
52	images_in_obs : bool
53	whether to use the image in the observation
54	image_size : int
55	number of elements in the image. Set to 0 if images are not being used.
56	max_target_distance : float
57	TODO
58	action_space : gym.spaces.*
59	the action space of the environment
60	obs_range : gym.spaces.*
61	the range of the initial position of the agent
62	context_space : gym.spaces.*
63	the context space of the environment
64	observation_space : gym.spaces.*
65	the observation space of the environment
66	drawer : multiworld.envs.pygame.pygame_viewer.PygameViewer or None
67	The drawer for the images of the environment. Set to None if images are
68	not being used.
69	render_drawer : multiworld.envs.pygame.pygame_viewer.PygameViewer or None
70	The drawer for the images of the environment if the environment is
71	being rendered. Set to None if images are not being used.
72	horizon : int
73	environment time horizon
74	t : int
75	number of steps since the start of the most recent episode
76	"""
77
78	def __init__(self,	1✔
79	render_dt_msec=0,
80	action_l2norm_penalty=0, # disabled for now
81	render_onscreen=False,
82	render_size=32,
83	reward_type="dense",
84	action_scale=1.0,
85	target_radius=0.60,
86	boundary_dist=4,
87	ball_radius=0.50,
88	walls=None,
89	fixed_goal=None,
90	randomize_position_on_reset=True,
91	images_are_rgb=False, # else black and white
92	show_goal=True,
93	images_in_obs=True,
94	**kwargs):
95	"""Instantiate the environment.
96
97	Parameters
98	----------
99	render_dt_msec : float
100	seconds before the next frame in the image is rendered
101	action_l2norm_penalty : float
102	penalty scale for the actions by the agent
103	render_onscreen : bool
104	whether to include the rendering visually (instead of simply using
105	the image for the observation)
106	render_size : int
107	width/length number of pixels in the rendered image
108	reward_type : str
109	the reward type. Must be one of: "sparse", "dense", or
110	"vectorized_dense"
111	action_scale : float
112	the multiple from action to velocity
113	target_radius : float
114	the radius of the targeted position when being rendered
115	boundary_dist : float
116	the distance from the center to the boundary
117	ball_radius : float
118	the radius of the agent when being rendered
119	walls : TODO
120	TODO
121	fixed_goal : [float, float] or None
122	the goal to use. If set to None, it is picked randomly.
123	randomize_position_on_reset : bool
124	whether to initialize the position of the agent randomly
125	images_are_rgb : bool
126	specifies whether the image is RGB. Otherwise, it's black and white
127	show_goal : bool
128	whether to render the goal(s)
129	images_in_obs : bool
130	whether to use the image in the obsevation
131	kwargs : dict
132	additional arguments. Unused here.
133	"""
134	if walls is None:	1✔
135	walls = []	1✔
136	if fixed_goal is not None:	1✔
137	fixed_goal = np.array(fixed_goal)	1✔
138	if len(kwargs) > 0:	1✔
139	logger = logging.getLogger(__name__)	×
140	logger.log(logging.WARNING, "WARNING, ignoring kwargs:", kwargs)	×
141
142	self.quick_init(locals())	1✔
143	self.render_dt_msec = render_dt_msec	1✔
144	self.action_l2norm_penalty = action_l2norm_penalty	1✔
145	self.render_onscreen = render_onscreen	1✔
146	self.render_size = render_size	1✔
147	self.reward_type = reward_type	1✔
148	self.action_scale = action_scale	1✔
149	self.target_radius = target_radius	1✔
150	self.boundary_dist = boundary_dist	1✔
151	self.ball_radius = ball_radius	1✔
152	self.walls = walls	1✔
153	self.fixed_goal = fixed_goal	1✔
154	self.randomize_position_on_reset = randomize_position_on_reset	1✔
155	self.images_are_rgb = images_are_rgb	1✔
156	self.show_goal = show_goal	1✔
157	self.images_in_obs = images_in_obs	1✔
158	self.image_size = 1024 * (3 if self.images_are_rgb else 1)	1✔
159	if not self.images_in_obs:	1✔
160	self.image_size = 0	1✔
161
162	self.max_target_distance = self.boundary_dist - self.target_radius	1✔
163
164	self._target_position = None	1✔
165	self._position = np.zeros(2)	1✔
166
167	u = np.ones(2)	1✔
168	self.action_space = spaces.Box(-u, u, dtype=np.float32)	1✔
169
170	o = self.boundary_dist * np.ones(2)	1✔
171	self.obs_range = spaces.Box(	1✔
172	-o, o, dtype='float32')
173	self.context_space = spaces.Box(	1✔
174	-o, o, dtype='float32')
175	self.observation_space = spaces.Box(	1✔
176	np.concatenate([np.zeros([self.image_size]), -o], 0),
177	np.concatenate([np.ones([self.image_size]), o], 0),
178	dtype='float32')
179
180	self.drawer = None	1✔
181	self.render_drawer = None	1✔
182	self.horizon = 200	1✔
183	self.t = 0	1✔
184
185	@property	1✔
186	def current_context(self):
187	"""Return the current goal by the environment."""
188	return self._target_position	1✔
189
190	def step(self, velocities):	1✔
191	"""Advance the simulation by one step.
192
193	Parameters
194	----------
195	velocities : array_like
196	the action by the agent, defined as its velocities in the x and y
197	directions
198
199	Returns
200	-------
201	array_like
202	agent's observation of the current environment
203	float
204	amount of reward associated with the previous state/action pair
205	bool
206	indicates whether the episode has ended
207	dict
208	contains other diagnostic information from the previous action
209	"""
210	assert self.action_scale <= 1.0	1✔
211	velocities = np.clip(	1✔
212	velocities, a_min=-1, a_max=1) * self.action_scale
213	new_position = self._position + velocities	1✔
214	orig_new_pos = new_position.copy()	1✔
215	for wall in self.walls:	1✔
216	new_position = wall.handle_collision(	×
217	self._position, new_position
218	)
219	if sum(new_position != orig_new_pos) > 1:	1✔
220	# Hack: sometimes you get caught on two walls at a time. If you
221	# process the input in the other direction, you might only get
222	# caught on one wall instead.
223	new_position = orig_new_pos.copy()	×
224	for wall in self.walls[::-1]:	×
225	new_position = wall.handle_collision(	×
226	self._position, new_position
227	)
228
229	self.t += 1	1✔
230
231	self._position = new_position	1✔
232	self._position = np.clip(	1✔
233	self._position,
234	a_min=-self.boundary_dist,
235	a_max=self.boundary_dist,
236	)
237	distance_to_target = np.linalg.norm(	1✔
238	self._position - self._target_position
239	)
240	is_success = distance_to_target < self.target_radius	1✔
241
242	ob = self._get_obs()	1✔
243	reward = self.compute_reward(velocities, {'ob': ob})	1✔
244	info = {	1✔
245	'radius': self.target_radius,
246	'target_position': self._target_position,
247	'distance_to_target': distance_to_target,
248	'velocity': velocities,
249	'speed': np.linalg.norm(velocities),
250	'is_success': is_success,
251	}
252	done = self.t >= self.horizon	1✔
253	return ob, reward, done, info	1✔
254
255	def reset(self):	1✔
256	"""Reset the environment."""
257	self.t = 0	1✔
258	self._target_position = self.sample_goal()['goals']	1✔
259	if self.randomize_position_on_reset:	1✔
260	self._position = self._sample_position(	1✔
261	self.obs_range.low,
262	self.obs_range.high,
263	)
264
265	return self._get_obs()	1✔
266
267	def _position_inside_wall(self, pos):	1✔
268	"""Return True if the agent is in a wall."""
269	for wall in self.walls:	1✔
270	if wall.contains_point(pos):	×
271	return True	×
272	return False	1✔
273
274	def _sample_position(self, low, high):	1✔
275	"""Sample a starting position for the agent."""
276	pos = np.random.uniform(low, high)	1✔
277	while self._position_inside_wall(pos) is True:	1✔
278	pos = np.random.uniform(low, high)	×
279	return pos	1✔
280
281	def _get_obs(self):	1✔
282	"""Return the observation of the agent.
283
284	See States in the description of the environment for more.
285	"""
286	if self.images_in_obs:	1✔
287	img = self.get_image(	1✔
288	32, 32).reshape([-1]).astype(np.float32) / 255.0
289	return np.concatenate([img, self._position.copy()], 0)	1✔
290	else:
291	return self._position.copy()	1✔
292
293	def compute_rewards(self, actions, obs):	1✔
294	"""See parent class.
295
296	The rewards are described in the Rewards section of the description of
297	the environment.
298	"""
299	achieved_goals = obs['ob'][:, self.image_size:]	1✔
300	desired_goals = self._target_position[np.newaxis, :]	1✔
301	d = np.linalg.norm(achieved_goals - desired_goals, axis=-1)	1✔
302	if self.reward_type == "sparse":	1✔
303	return -(d > self.target_radius).astype(np.float32)	×
304	elif self.reward_type == "dense":	1✔
305	return -d	1✔
306	elif self.reward_type == 'vectorized_dense':	×
307	return -np.abs(achieved_goals - desired_goals)	×
308	else:
309	raise NotImplementedError()
310
311	def get_goal(self):	1✔
312	"""See parent class."""
313	return self._target_position.copy()	×
314
315	def sample_goals(self, batch_size):	1✔
316	"""See parent class.
317
318	The goal is the desired x,y coordinates.
319	"""
320	if self.fixed_goal is not None:	1✔
321	goals = np.repeat(	1✔
322	self.fixed_goal.copy()[None],
323	batch_size,
324	0
325	)
326	else:
327	goals = np.zeros((batch_size, self.obs_range.low.size))	1✔
328	for b in range(batch_size):	1✔
329	if batch_size > 1:	1✔
330	logging.warning("This is very slow!")	×
331	goals[b, :] = self._sample_position(	1✔
332	self.obs_range.low,
333	self.obs_range.high,
334	)
335	return {'goals': goals}	1✔
336
337	# ======================================================================= #
338	# Functions for ImageEnv wrapper #
339	# ======================================================================= #
340
341	def get_image(self, width=None, height=None):	1✔
342	"""Return a black and white image."""
343	if self.drawer is None:	1✔
344	if width != height:	1✔
345	raise NotImplementedError()
346	self.drawer = PygameViewer(	1✔
347	screen_width=width,
348	screen_height=height,
349	x_bounds=(-self.boundary_dist - self.ball_radius,
350	self.boundary_dist + self.ball_radius),
351	y_bounds=(-self.boundary_dist - self.ball_radius,
352	self.boundary_dist + self.ball_radius),
353	render_onscreen=self.render_onscreen,
354	)
355	self.draw(self.drawer)	1✔
356	img = self.drawer.get_image()	1✔
357	if self.images_are_rgb:	1✔
358	return img.transpose((1, 0, 2))	×
359	else:
360	r, b = img[:, :, 0], img[:, :, 2]	1✔
361	img = (-r + b).transpose().flatten()	1✔
362	return img	1✔
363
364	def draw(self, drawer):	1✔
365	"""Create the image corresponding to the current state."""
366	drawer.fill(Color('white'))	1✔
367	if self.show_goal:	1✔
368	drawer.draw_solid_circle(	1✔
369	self._target_position,
370	self.target_radius,
371	Color('green'),
372	)
373	drawer.draw_solid_circle(	1✔
374	self._position,
375	self.ball_radius,
376	Color('blue'),
377	)
378
379	for wall in self.walls:	1✔
380	drawer.draw_segment(	×
381	wall.endpoint1,
382	wall.endpoint2,
383	Color('black'),
384	)
385	drawer.draw_segment(	×
386	wall.endpoint2,
387	wall.endpoint3,
388	Color('black'),
389	)
390	drawer.draw_segment(	×
391	wall.endpoint3,
392	wall.endpoint4,
393	Color('black'),
394	)
395	drawer.draw_segment(	×
396	wall.endpoint4,
397	wall.endpoint1,
398	Color('black'),
399	)
400	drawer.render()	1✔
401
402	def render(self, mode='human', close=False):	1✔
403	"""Render the environment state."""
404	if mode == 'rgb_array':	×
405	return self.get_image(self.render_size, self.render_size)	×
406
407	if close:	×
408	self.render_drawer = None	×
409	return	×
410
411	if self.render_drawer is None or self.render_drawer.terminated:	×
412	self.render_drawer = PygameViewer(	×
413	self.render_size,
414	self.render_size,
415	x_bounds=(-self.boundary_dist-self.ball_radius,
416	self.boundary_dist+self.ball_radius),
417	y_bounds=(-self.boundary_dist-self.ball_radius,
418	self.boundary_dist+self.ball_radius),
419	render_onscreen=True,
420	)
421	self.draw(self.render_drawer)	×
422	self.render_drawer.tick(self.render_dt_msec)	×
423	if mode != 'interactive':	×
424	self.render_drawer.check_for_exit()	×
425
426	# ======================================================================= #
427	# Static visualization/utility methods #
428	# ======================================================================= #
429
430	@staticmethod	1✔
431	def true_model(state, action):
432	"""Return the next position by the agent.
433
434	Parameters
435	----------
436	state : array_like
437	the state by the agent
438	action : array_like
439	the action by the agent
440
441	Returns
442	-------
443	array_like
444	the next position.
445	"""
446	velocities = np.clip(action, a_min=-1, a_max=1)	×
447	position = state	×
448	new_position = position + velocities	×
449	return np.clip(	×
450	new_position,
451	a_min=-Point2DEnv.boundary_dist,
452	a_max=Point2DEnv.boundary_dist,
453	)
454
455	@staticmethod	1✔
456	def true_states(state, actions):
457	"""Return the next states given a set of states and actions.
458
459	Parameters
460	----------
461	state : array_like
462	the states by the agent
463	actions : array_like
464	the actions by the agent
465
466	Returns
467	-------
468	list of array_like
469	the next states
470	"""
471	real_states = [state]	×
472	for action in actions:	×
473	next_state = Point2DEnv.true_model(state, action)	×
474	real_states.append(next_state)	×
475	state = next_state	×
476	return real_states	×
477
478	def plot_trajectory(self, ax, states, actions, goal=None):	1✔
479	"""Plot the trajectory of an agent.
480
481	Parameters
482	----------
483	ax : matplotlib.axes.Axes
484	the axis object to plot the figure on
485	states : array_like
486	the states by the agent
487	actions : array_like
488	the actions by the agent
489	goal : [int, int]
490	the x,y coordinates of the goal
491	"""
492	assert len(states) == len(actions) + 1	×
493	x = states[:, 0]	×
494	y = -states[:, 1]	×
495	num_states = len(states)	×
496	plasma_cm = plt.get_cmap('plasma')	×
497	for i, state in enumerate(states):	×
498	color = plasma_cm(float(i) / num_states)	×
499	ax.plot(state[0], -state[1],	×
500	marker='o', color=color, markersize=10,
501	)
502
503	actions_x = actions[:, 0]	×
504	actions_y = -actions[:, 1]	×
505
506	ax.quiver(x[:-1], y[:-1], x[1:] - x[:-1], y[1:] - y[:-1],	×
507	scale_units='xy', angles='xy', scale=1, width=0.005)
508	ax.quiver(x[:-1], y[:-1], actions_x, actions_y, scale_units='xy',	×
509	angles='xy', scale=1, color='r',
510	width=0.0035, )
511	ax.plot(	×
512	[
513	-self.boundary_dist,
514	-self.boundary_dist,
515	],
516	[
517	self.boundary_dist,
518	-self.boundary_dist,
519	],
520	color='k', linestyle='-',
521	)
522	ax.plot(	×
523	[
524	self.boundary_dist,
525	-self.boundary_dist,
526	],
527	[
528	self.boundary_dist,
529	self.boundary_dist,
530	],
531	color='k', linestyle='-',
532	)
533	ax.plot(	×
534	[
535	self.boundary_dist,
536	self.boundary_dist,
537	],
538	[
539	self.boundary_dist,
540	-self.boundary_dist,
541	],
542	color='k', linestyle='-',
543	)
544	ax.plot(	×
545	[
546	self.boundary_dist,
547	-self.boundary_dist,
548	],
549	[
550	-self.boundary_dist,
551	-self.boundary_dist,
552	],
553	color='k', linestyle='-',
554	)
555
556	if goal is not None:	×
557	ax.plot(goal[0], -goal[1], marker='*', color='g', markersize=15)	×
558	ax.set_ylim(	×
559	-self.boundary_dist - 1,
560	self.boundary_dist + 1,
561	)
562	ax.set_xlim(	×
563	-self.boundary_dist - 1,
564	self.boundary_dist + 1,
565	)