Environment API

Base environment

gym_classics2.envs.abstract.base_env.BaseEnv

Bases: Env

Abstract base class for shared functionality between all environments.

Source code in gym_classics2/envs/abstract/base_env.py

class BaseEnv(Env, metaclass=ABCMeta):
    """Abstract base class for shared functionality between all environments."""

    # needs to provide the following Gymnasium Env functions:
    # - reset
    # - step
    # - render
    # - close

    # added for gym_classics
    # - action_space
    # - observation_space
    # - start states
    # - transitions
    # - rewards
    # - done conditions
    # - encode
    # - decode
    # - is_reachable
    # - actions
    # - model
    # - generate_transitions

    def __init__(self, starts, action_labels = None, tabular = True, reachable_states=None):   
        assert action_labels is not None, "action_labels must be provided"

        self.action_labels = action_labels

        n_actions = len(action_labels)
        self.action_space = Discrete(n_actions)

        # observation space is defined by the subclass
        #self.observation_space = Discrete(len(self._reachable_states))

        # Normalize sets and other iterables so reset() can sample by index and
        # callers see a read-only public representation.
        self._starts = tuple(starts)
        self.state = None

        self._transition_cache = {}

        if reachable_states is None:
            # Get reachable states by searching through the state space
            reachable_states = set()
            for s in self._starts:
                self._search(s, reachable_states)

        # organize the reachable states in a consistent order
        self._reachable_states = reachable_states

        # build encoder and decoder for tabular access
        self._decoder = [s for s in sorted(self._reachable_states)]
        self._encoder = {s: i for i, s in enumerate(self._decoder)}

        self.tabular = tabular
        if self.tabular:
            self.observation_space = Discrete(len(self._reachable_states))


    def reset(self, seed=None, options=None):
        """Start a new episode and return ``(observation, info)``."""
        super().reset(seed=seed)

        self.state = self._starts[self.np_random.integers(len(self._starts))]
        observation = self.state
        info = {}

        #if self.render_mode == "human":
        #    self._render_frame()

        if self.tabular:
            observation = self.state2id(observation)

        return observation, info

    def step(self, action):
        """Sample one transition and return the five-item Gymnasium result."""
        assert self.action_space.contains(action)

        # after the random elements are sampled, the environment transition is deterministic
        elements = self._sample_random_elements(self.state, action)

        next_state, reward, done, _ = self._deterministic_step(self.state, action, *elements)
        self.state = next_state
        info = {}

        if self.tabular:
            next_state = self.state2id(next_state)

        return next_state, reward, done, False, info


    ## render needs to be implemented in each environment, so we don't provide a default implementation here
    ## we use the default close, can be overwritten.

    ## Additional Interface

    @property
    def start_states(self):
        """Tuple of raw states from which an episode may start.

        These are raw environment states even when :attr:`tabular` is true. Use
        :meth:`state2id` to convert them to integer observations.
        """
        return tuple(self._starts)

    ## Tabular access
    def states(self):
        """Returns a generator over all possible environment states."""
        return range(len(self._encoder))

    def state2id(self, state):
        """Converts a raw state into a unique integer."""

        if isinstance(state, list):
            return [self._encoder[s] for s in state]

        return self._encoder[state]

    def id2state(self, i):
        """Reverts an encoded integer back to its raw state."""
        if isinstance(i, list):
            return [self._decoder[j] for j in i]

        return self._decoder[i]

    def is_reachable(self, state):
        """Returns True if the state can be reached from at least one start location,
        False otherwise."""
        return state in self._reachable_states

    def actions(self):
        """Returns a generator over all possible agent actions."""
        return range(self.action_space.n)

    def action2id(self, action_label):
        """Converts a action label into a numeric action ID."""
        action_ids = dict(zip(self.action_labels, range(len(self.action_labels))))
        id = action_ids.get(action_label, -1)
        return id

    def id2action(self, action, type="text"):
        """Converts a numeric action ID into a label. Choices for type are 'text' and 'arrow'."""
        action_labels = self.action_labels + [""]  # Add empty label for hidden actions

        return action_labels[int(action)]

    def model(self, state, action):
        """Return the complete transition model for a state-action pair.

        Args:
            state: Integer state ID when :attr:`tabular` is true, otherwise a raw
                state.
            action: Integer action ID.

        Returns:
            A four-item list ``[next_states, rewards, terminals, probabilities]``.
            ``next_states`` is a list of IDs in tabular mode and raw states
            otherwise. The remaining items are one-dimensional NumPy arrays in the
            same order.
        """

        if self.tabular:
            state = self.id2state(state)

        # caching the model for efficiency    
        sa_pair = (state, action)
        if sa_pair in self._transition_cache:
            return self._transition_cache[sa_pair]

        tr = list(self._generate_transitions(state, action))
        tr = [[t[0] for t in tr], np.array([t[1] for t in tr]), np.array([t[2] for t in tr]), np.array([t[3] for t in tr])]

        assert (tr[3] >= 0.0).all(), "transition probabilities must be nonnegative"
        assert np.sum(tr[3]) == 1.0, "transition probabilities must sum to 1"

        if self.tabular:
            tr[0] = self.state2id(tr[0])

        self._transition_cache[sa_pair] = tr
        return tr

    def _search(self, state, visited):
        """A recursive depth-first search that adds all reachable states to the visited set."""
        visited.add(state)
        for a in self.actions():
            for transition in self._generate_transitions(state, a):
                next_state, _, done, prob = transition
                if prob > 0.0:
                    if not done and next_state not in visited:
                        self._search(next_state, visited)
                    # MFH add the final state!
                    if done and next_state not in visited:
                        visited.add(next_state)

    # Do not overwrite!
    def _deterministic_step(self, state, action, *random_elements):
        """An environment step that is deterministic conditioned on the given values
        of the random variables (if there are any).

        Do not override.
        """
        next_state, prob = self._next_state(state, action, *random_elements)
        reward = self._reward(state, action, next_state)
        done = self._done(state, action, next_state)
        #if done:
        #    next_state = state
        return next_state, reward, done, prob

    # these need to be implemented in the subclass
    def _sample_random_elements(self, state, action):
        """Samples values for random elements (if any) that influence the environment
        transition from the current state-action pair (S, A).

        If the environment is deterministic, no need to override this method.
        """
        return ()

    @abstractmethod
    def _next_state(self, state, action, *random_elements):
        """Returns the next state S' induced by the state-action pair (S, A), which must
        be deterministic conditioned on the values of any random_elements. Also returns
        the probability that this particular transition occurred."""
        raise NotImplementedError

    @abstractmethod
    def _reward(self, state, action, next_state):
        """Returns the reward yielded by this (S,A,S') outcome."""
        raise NotImplementedError

    @abstractmethod
    def _done(self, state, action, next_state):
        """Returns True if this (S,A,S') outcome should terminate, False otherwise."""
        raise NotImplementedError

    @abstractmethod
    def _generate_transitions(self, state, action):
        """Returns a generator over all transitions from this state-action pair.

        Should be overridden in the subclass.
        """
        raise NotImplementedError

start_states property

start_states

Tuple of raw states from which an episode may start.

These are raw environment states even when :attr:tabular is true. Use :meth:state2id to convert them to integer observations.

states

states()

Returns a generator over all possible environment states.

Source code in gym_classics2/envs/abstract/base_env.py

def states(self):
    """Returns a generator over all possible environment states."""
    return range(len(self._encoder))

state2id

state2id(state)

Converts a raw state into a unique integer.

Source code in gym_classics2/envs/abstract/base_env.py

def state2id(self, state):
    """Converts a raw state into a unique integer."""

    if isinstance(state, list):
        return [self._encoder[s] for s in state]

    return self._encoder[state]

id2state

id2state(i)

Reverts an encoded integer back to its raw state.

Source code in gym_classics2/envs/abstract/base_env.py

def id2state(self, i):
    """Reverts an encoded integer back to its raw state."""
    if isinstance(i, list):
        return [self._decoder[j] for j in i]

    return self._decoder[i]

is_reachable

is_reachable(state)

Returns True if the state can be reached from at least one start location, False otherwise.

Source code in gym_classics2/envs/abstract/base_env.py

def is_reachable(self, state):
    """Returns True if the state can be reached from at least one start location,
    False otherwise."""
    return state in self._reachable_states

actions

actions()

Returns a generator over all possible agent actions.

Source code in gym_classics2/envs/abstract/base_env.py

def actions(self):
    """Returns a generator over all possible agent actions."""
    return range(self.action_space.n)

action2id

action2id(action_label)

Converts a action label into a numeric action ID.

Source code in gym_classics2/envs/abstract/base_env.py

def action2id(self, action_label):
    """Converts a action label into a numeric action ID."""
    action_ids = dict(zip(self.action_labels, range(len(self.action_labels))))
    id = action_ids.get(action_label, -1)
    return id

id2action

id2action(action, type='text')

Converts a numeric action ID into a label. Choices for type are 'text' and 'arrow'.

Source code in gym_classics2/envs/abstract/base_env.py

def id2action(self, action, type="text"):
    """Converts a numeric action ID into a label. Choices for type are 'text' and 'arrow'."""
    action_labels = self.action_labels + [""]  # Add empty label for hidden actions

    return action_labels[int(action)]

model

model(state, action)

Return the complete transition model for a state-action pair.

Parameters:

Name	Type	Description	Default
state		Integer state ID when :attr:tabular is true, otherwise a raw state.	required
action		Integer action ID.	required

Returns:

Type	Description
	A four-item list [next_states, rewards, terminals, probabilities].
	next_states is a list of IDs in tabular mode and raw states
	otherwise. The remaining items are one-dimensional NumPy arrays in the
	same order.

Source code in gym_classics2/envs/abstract/base_env.py

def model(self, state, action):
    """Return the complete transition model for a state-action pair.

    Args:
        state: Integer state ID when :attr:`tabular` is true, otherwise a raw
            state.
        action: Integer action ID.

    Returns:
        A four-item list ``[next_states, rewards, terminals, probabilities]``.
        ``next_states`` is a list of IDs in tabular mode and raw states
        otherwise. The remaining items are one-dimensional NumPy arrays in the
        same order.
    """

    if self.tabular:
        state = self.id2state(state)

    # caching the model for efficiency    
    sa_pair = (state, action)
    if sa_pair in self._transition_cache:
        return self._transition_cache[sa_pair]

    tr = list(self._generate_transitions(state, action))
    tr = [[t[0] for t in tr], np.array([t[1] for t in tr]), np.array([t[2] for t in tr]), np.array([t[3] for t in tr])]

    assert (tr[3] >= 0.0).all(), "transition probabilities must be nonnegative"
    assert np.sum(tr[3]) == 1.0, "transition probabilities must sum to 1"

    if self.tabular:
        tr[0] = self.state2id(tr[0])

    self._transition_cache[sa_pair] = tr
    return tr

Gridworld

gym_classics2.envs.abstract.gridworld.Gridworld

Bases: BaseEnv

Abstract class for creating gridworld-type environments.

Source code in gym_classics2/envs/abstract/gridworld.py

class Gridworld(BaseEnv):
    """Abstract class for creating gridworld-type environments."""
    metadata = {"render_modes": ["human", "rgb_array"], "render_fps": 4}

    def __init__(self, layout_string, action_labels = ["up", "right", "down", "left"], 
                 goal_reward = 1.0, step_reward = 0.0, 
                 tabular = True, render_mode=None):
        """Initializes the gridworld environment from a layout string. The layout string should be a rectangular grid of characters, where each character represents a type of cell:
        - 'S': Start (may be more than one)
        - 'G': Goal (may be more than one)
        - 'X': Block (agent cannot occupy these cells)
        - ' ': Empty (agent can occupy these cells)
        - All other characters are treated as empty cells that the agent can occupy.

        param layout_string: The string representation of the gridworld layout.
        param action_labels: The labels for the actions. Defaults to ["up", "right", "down", "left"]. You can specify additional labels for extra actions.
        param tabular: If True, the environment will use a tabular state representation (i.e., states are represented as integer IDs). 
            If False, states will be represented as their (x,y) coordinates. Defaults to True.
        """

        self._goal_reward = goal_reward
        self._step_reward = step_reward

        self.dims, starts, self._goals, self._blocks, self._extra_labels = parse_gridworld(layout_string)

        assert render_mode is None or render_mode in self.metadata["render_modes"]
        self.render_mode = render_mode
        self.PyGame_window_size = 512  # The size of the PyGame window in pixels
        self.PyGame_window = None
        self.PyGame_clock = None

        super().__init__(starts, action_labels = action_labels, tabular = tabular, reachable_states = None)

        if not tabular:
            self.observation_space = MultiDiscrete(self.dims)

    @property
    def goal_states(self):
        """Tuple containing the raw terminal goal coordinates."""
        return tuple(sorted(self._goals))

    def _next_state(self, state, action, *random_elements):
        next_state = self._move(state, action)
        if self._is_blocked(next_state):
            next_state = state
        return self._clamp(next_state), 1.0

    def _move(self, state, action):
        x, y = state
        return {
            0: (x,   y+1),  # Up
            1: (x+1, y),    # Right
            2: (x,   y-1),  # Down
            3: (x-1, y)     # Left
        }[action]

    def _clamp(self, state):
        """Clamps the state within the grid dimensions."""
        x, y = state
        x = max(0, min(x, self.dims[0] - 1))
        y = max(0, min(y, self.dims[1] - 1))
        return (x, y)

    def _is_blocked(self, state):
        """Returns True if this state cannot be occupied, False otherwise."""
        return state in self._blocks

    def _generate_transitions(self, state, action):
        yield self._deterministic_step(state, action)

    def _reward(self, state, action, next_state):      
        if next_state in self._goals: 
            return self._goal_reward 
        return self._step_reward

    def _done(self, state, action, next_state):
        return next_state in self._goals

    def step(self, action):
        """Advance the environment and render a frame in human mode."""
        next_state, reward, done, x, info = super().step(action)

        if self.render_mode == "human":
            self._render_frame()

        return next_state, reward, done, x, info

    def reset(self, seed=None, options=None):
        """Reset to a start cell and render a frame in human mode."""
        observation, info = super().reset(seed=seed, options=options)

        if self.render_mode == "human":
            self._render_frame()

        return observation, info

    def close(self):
        """Release any PyGame display resources."""
        if self.PyGame_window is not None:
            pygame.display.quit()
            pygame.quit()
            self.PyGame_window = None
            self.PyGame_clock = None

        super().close()

    def render(self):
        """Return an RGB frame when ``render_mode='rgb_array'``."""
        if self.render_mode == "rgb_array":
            return self._render_frame()

    def _render_frame(self):
        pix_square_size = self.PyGame_window_size / max(self.dims)
        display_size = np.array(pix_square_size) * self.dims

        if self.PyGame_window is None and self.render_mode == "human":
            pygame.init()
            pygame.display.init()
            self.PyGame_window = pygame.display.set_mode(display_size)
        if self.PyGame_clock is None and self.render_mode == "human":
            self.PyGame_clock = pygame.time.Clock()

        canvas = pygame.Surface(display_size)
        canvas.fill((255, 255, 255))

        # draw the goal
        for g in self._goals:
            pos = np.array(g)
            pos[1] = self.dims[1] - pos[1] - 1
            pygame.draw.rect(
                canvas,
                (0, 128, 0),
                pygame.Rect(
                    pix_square_size * pos,
                    (pix_square_size, pix_square_size),
                ),
            )

        # unreachable squares
        for g in np.argwhere(self.to_matrix() == -1):
            pos = np.array(g)
            pos = np.flip(pos)
            pos[1] = self.dims[1] - pos[1] - 1
            pygame.draw.rect(
                canvas,
                (72, 72, 72),
                pygame.Rect(
                    pix_square_size * pos,
                    (pix_square_size, pix_square_size),
                ),
            )

        # draw the agent
        pos = np.array(self.state)
        pos[1] = self.dims[1] - pos[1] - 1
        pygame.draw.circle(
            canvas,
            (0, 0, 255),
            (pos + .5) * pix_square_size,
            pix_square_size / 3,
        )

        # Finally, add some gridlines
        for x in range(self.dims[1] + 1):
            pygame.draw.line(
                canvas,
                0,
                (0, pix_square_size * x),
                (display_size[0], pix_square_size * x),
                width=3,
            )

        for x in range(self.dims[0] + 1):
            pygame.draw.line(
                canvas,
                0,
                (pix_square_size * x, 0),
                (pix_square_size * x, display_size[1]),
                width=3,
            )

        if self.render_mode == "human":
            # The following line copies our drawings from `canvas` to the visible window
            self.PyGame_window.blit(canvas, canvas.get_rect())
            pygame.event.pump()
            pygame.display.update()

            # We need to ensure that human-rendering occurs at the predefined framerate.
            # The following line will automatically add a delay to
            # keep the framerate stable.
            self.PyGame_clock.tick(self.metadata["render_fps"])
        else:  # rgb_array
            return np.transpose(
                np.array(pygame.surfarray.pixels3d(canvas)), axes=(1, 0, 2)
            )

    ### Addition to the interface

    ## overwrite so we have arrows
    def id2action(self, action, type="text"):
        """Converts a numeric action ID into a label. Choices for type are 'text' and 'arrow'."""
        action = int(action)

        action_labels = self.action_labels + [""]  # Add empty label for hidden actions

        # empty has index 4 and is used to hide actions
        if (type == "arrow"):
             action_labels[0:4] = ['↑', '→', '↓', '←']

        return action_labels[action]

    def to_matrix(self, value = None):
        """Converts a vector with values for states in a gridworld to a matrix for display. Values can be a value function, policy, etc.

        param value: The value function as a vector.

        return: The value function as a matrix.
        """

        if value is None:
            value = list(self.states())

        value = np.array(value)

        if np.issubdtype(value.dtype, np.integer):
            m = np.full(self.dims, -1, dtype=value.dtype)
        elif np.issubdtype(value.dtype, np.str_):
            m = np.full(self.dims, "", dtype=value.dtype)
        elif np.issubdtype(value.dtype, np.floating):
            m = np.full(self.dims, np.nan, dtype=value.dtype)
        else:
            m = np.zeros(self.dims, dtype=value.dtype)

        for y in range(self.dims[1]):
            for x in range(self.dims[0]):
                state = (x, y)
                if self.is_reachable(state):
                    m[x,y] = value[self.state2id(state)]
                else:
                    pass

        return m.transpose() 

    def print(self, array, decimals=2, separator=' ' * 2, signed=True, transpose=False):
        """Prints a gridworld array in a human-readable format. The array should be a vector with values for states in the gridworld, 
        such as a value function or policy."""

        def formatter(x):
            string = '{:' + ('+' if signed else '') + '.' + str(decimals) + 'f}'
            return string.format(x)
        maxlen = max([len(formatter(x)) for x in array])

        # Now we can actually print the values
        for y in reversed(range(self.dims[1])):
            for x in range(self.dims[0]):
                state = (x, y) if not transpose else (y, x)
                if self.is_reachable(state):
                    s = self.state2id(state)
                    print(formatter(array[s]).rjust(maxlen), end=separator)
                else:
                    print(' ' * maxlen, end=separator)
            print()

    def image(self, V=None, policy=None, episode = None, labels=None, title=None, cmap = 'auto', origin='lower', clim = None):
        """
        Display the a gridworld as an image.

        :param V: The value (e.g., a value function) to display. If None, display state indices.
        :param labels: The labels to show on the grid cells in the same order as the value function. If True, show rounded values from V.
        :param policy: The policy to display. If not None, show the policy.
        :param episode: Show an episode
        :param title: Title of the plot.
        :param cmap: Colormap to use for the value function.
        :param origin: 'lower' means (0,0) is at the bottom-left, 'upper' means (0,0) is at the top-left.
        """

        colorbar = True

        if not V is None:
            m = self.to_matrix(V)
        else:
            m = np.zeros(self.dims).transpose()
            # missing positions have -1
            m[self.to_matrix(labels) == -1] = np.nan
            labels = self.states()
            colorbar = False

        if not episode is None:
            # start with policy that hides all actions with an index past the last action.
            policy = np.full(len(self.states()), len(self.actions()))
            for step in episode:
                policy[step[0]] = step[1]

        if not policy is None:
            labels = [self.id2action(a, type = "arrow") for a in policy]

        if isinstance(labels, bool) and labels:
                labels = np.round(V, 2)

        if not labels is None:
            labels = self.to_matrix(labels)

        extra = [""] * self.dims[0] * self.dims[1]
        for s in self._starts:
            extra[self.state2id(s)] = "S"
        for s in self._goals:
            extra[self.state2id(s)] = "G"
        for s, label in self._extra_labels:
            extra[self.state2id(s)] = label
        extra = self.to_matrix(extra)

        _image(m, title=title, labels=labels, extra=extra, cmap=cmap, clim = clim, origin=origin, colorbar=colorbar)  


    def image_list(self, Vs = None, policies = None, episodes = None, cmap = 'auto', clim = None, origin='lower'):
        """
        Creates a sequence of images, one for each episode.
        """
        n_states = len(self.states())
        if Vs is not None:
            iterations = len(Vs)
        elif policies is not None:
            iterations = len(policies)
        else:
            iterations = len(episodes)

        V = None
        policy = None
        episode = None

        for i in range(iterations):
            if not Vs is None:
                V = Vs[i]
            if not policies is None:
                policy = policies[i]
            if not episodes is None:
                episode = episodes[i]    

            self.image(V, policy=policy, episode=episode, title=f'After Iteration {i}', cmap=cmap, clim = clim, origin=origin)  

goal_states property

goal_states

Tuple containing the raw terminal goal coordinates.

reset

reset(seed=None, options=None)

Reset to a start cell and render a frame in human mode.

Source code in gym_classics2/envs/abstract/gridworld.py

def reset(self, seed=None, options=None):
    """Reset to a start cell and render a frame in human mode."""
    observation, info = super().reset(seed=seed, options=options)

    if self.render_mode == "human":
        self._render_frame()

    return observation, info

step

step(action)

Advance the environment and render a frame in human mode.

Source code in gym_classics2/envs/abstract/gridworld.py

def step(self, action):
    """Advance the environment and render a frame in human mode."""
    next_state, reward, done, x, info = super().step(action)

    if self.render_mode == "human":
        self._render_frame()

    return next_state, reward, done, x, info

render

render()

Return an RGB frame when render_mode='rgb_array'.

Source code in gym_classics2/envs/abstract/gridworld.py

def render(self):
    """Return an RGB frame when ``render_mode='rgb_array'``."""
    if self.render_mode == "rgb_array":
        return self._render_frame()

print

print(array, decimals=2, separator=' ' * 2, signed=True, transpose=False)

Prints a gridworld array in a human-readable format. The array should be a vector with values for states in the gridworld, such as a value function or policy.

Source code in gym_classics2/envs/abstract/gridworld.py

def print(self, array, decimals=2, separator=' ' * 2, signed=True, transpose=False):
    """Prints a gridworld array in a human-readable format. The array should be a vector with values for states in the gridworld, 
    such as a value function or policy."""

    def formatter(x):
        string = '{:' + ('+' if signed else '') + '.' + str(decimals) + 'f}'
        return string.format(x)
    maxlen = max([len(formatter(x)) for x in array])

    # Now we can actually print the values
    for y in reversed(range(self.dims[1])):
        for x in range(self.dims[0]):
            state = (x, y) if not transpose else (y, x)
            if self.is_reachable(state):
                s = self.state2id(state)
                print(formatter(array[s]).rjust(maxlen), end=separator)
            else:
                print(' ' * maxlen, end=separator)
        print()

image

image(V=None, policy=None, episode=None, labels=None, title=None, cmap='auto', origin='lower', clim=None)

Display the a gridworld as an image.

:param V: The value (e.g., a value function) to display. If None, display state indices. :param labels: The labels to show on the grid cells in the same order as the value function. If True, show rounded values from V. :param policy: The policy to display. If not None, show the policy. :param episode: Show an episode :param title: Title of the plot. :param cmap: Colormap to use for the value function. :param origin: 'lower' means (0,0) is at the bottom-left, 'upper' means (0,0) is at the top-left.

Source code in gym_classics2/envs/abstract/gridworld.py

def image(self, V=None, policy=None, episode = None, labels=None, title=None, cmap = 'auto', origin='lower', clim = None):
    """
    Display the a gridworld as an image.

    :param V: The value (e.g., a value function) to display. If None, display state indices.
    :param labels: The labels to show on the grid cells in the same order as the value function. If True, show rounded values from V.
    :param policy: The policy to display. If not None, show the policy.
    :param episode: Show an episode
    :param title: Title of the plot.
    :param cmap: Colormap to use for the value function.
    :param origin: 'lower' means (0,0) is at the bottom-left, 'upper' means (0,0) is at the top-left.
    """

    colorbar = True

    if not V is None:
        m = self.to_matrix(V)
    else:
        m = np.zeros(self.dims).transpose()
        # missing positions have -1
        m[self.to_matrix(labels) == -1] = np.nan
        labels = self.states()
        colorbar = False

    if not episode is None:
        # start with policy that hides all actions with an index past the last action.
        policy = np.full(len(self.states()), len(self.actions()))
        for step in episode:
            policy[step[0]] = step[1]

    if not policy is None:
        labels = [self.id2action(a, type = "arrow") for a in policy]

    if isinstance(labels, bool) and labels:
            labels = np.round(V, 2)

    if not labels is None:
        labels = self.to_matrix(labels)

    extra = [""] * self.dims[0] * self.dims[1]
    for s in self._starts:
        extra[self.state2id(s)] = "S"
    for s in self._goals:
        extra[self.state2id(s)] = "G"
    for s, label in self._extra_labels:
        extra[self.state2id(s)] = label
    extra = self.to_matrix(extra)

    _image(m, title=title, labels=labels, extra=extra, cmap=cmap, clim = clim, origin=origin, colorbar=colorbar)  

image_list

image_list(Vs=None, policies=None, episodes=None, cmap='auto', clim=None, origin='lower')

Creates a sequence of images, one for each episode.

Source code in gym_classics2/envs/abstract/gridworld.py

def image_list(self, Vs = None, policies = None, episodes = None, cmap = 'auto', clim = None, origin='lower'):
    """
    Creates a sequence of images, one for each episode.
    """
    n_states = len(self.states())
    if Vs is not None:
        iterations = len(Vs)
    elif policies is not None:
        iterations = len(policies)
    else:
        iterations = len(episodes)

    V = None
    policy = None
    episode = None

    for i in range(iterations):
        if not Vs is None:
            V = Vs[i]
        if not policies is None:
            policy = policies[i]
        if not episodes is None:
            episode = episodes[i]    

        self.image(V, policy=policy, episode=episode, title=f'After Iteration {i}', cmap=cmap, clim = clim, origin=origin)  

Concrete environments

gym_classics2.envs.gym_classics2.classic_gridworld_v1.ClassicGridworld

Bases: Gridworld

A 4x3 pedagogical gridworld. The agent starts in the bottom-left cell. Actions are noisy; with a 10% chance each, a move action may be rotated by 90 degrees clockwise or counter-clockwise (the "80-10-10 rule"). Cell (1, 1) is blocked and cannot be occupied by the agent.

reference: cite{1} (page 646).

state: Grid location.

actions: Move up/right/down/left.

rewards: +1 for taking any action in cell (3, 2). -1 for taking any action in cell (3, 1). NOTE: v1 uses the original -0.04 penalty for each state.

termination: Earning a nonzero reward.

Source code in gym_classics2/envs/gym_classics2/classic_gridworld_v1.py

class ClassicGridworld(Gridworld):
    """A 4x3 pedagogical gridworld. The agent starts in the bottom-left cell. Actions
    are noisy; with a 10% chance each, a move action may be rotated by 90 degrees
    clockwise or counter-clockwise (the "80-10-10 rule"). Cell (1, 1) is blocked and
    cannot be occupied by the agent.

    **reference:** cite{1} (page 646).

    **state**: Grid location.

    **actions**: Move up/right/down/left.

    **rewards**: +1 for taking any action in cell (3, 2). -1 for taking any
    action in cell (3, 1). *NOTE:*  v1 uses the original -0.04 penalty for each state.

    **termination**: Earning a nonzero reward.
    """

    layout = """
|   G|
| X G|
|S   |
"""

    def __init__(self, goal_reward = 1.0, trap_reward = -1.0, step_reward = -0.04, **args):
        self._trap_reward = trap_reward
        super().__init__(ClassicGridworld.layout, goal_reward = goal_reward, step_reward = step_reward, **args)

    def _reward(self, state, action, next_state):
        if state in self._goals:
            return 0.0
        return {(3, 1): self._trap_reward, (3, 2): self._goal_reward}.get(next_state, self._step_reward)

    def _done(self, state, action, next_state):
        return next_state in self._goals

    # Implement the non-deterministic actions

    # This method is called in the step function to create random events.
    # Here, the random event is that the environment executes a different
    # noisy action instead of the action the agent asked for.  
    # We return actually executed action as a list of random elements.
    def _sample_random_elements(self, state, action):
        noisy_action = (action + self.np_random.choice([-1, 0, 1], p=[.1,.8,.1])) % self.action_space.n
        return [noisy_action]

    # Returns an iterator for all possible outcomes. The random element is that
    # we have a noisy action, that may not be the intended action. 
    def _generate_transitions(self, state, action):
        # goal state is absorbing
        if state in self._goals:
            yield state, 0, True, 1.0

        else:
            for i in [-1, 0, 1]:
                noisy_action = (action + i) % self.action_space.n
                yield self._deterministic_step(state, action, noisy_action)

    # execute the noisy action and the probability
    def _next_state(self, state, action, noisy_action):
        next_state, _ = super()._next_state(state, noisy_action)
        p = 0.8 if action == noisy_action else 0.1
        return next_state, p

gym_classics2.envs.gym_classics2.cliff_walk_v1.CliffWalk

Bases: Gridworld

The Cliff Walking task, a 12x4 gridworld often used to contrast Sarsa with Q-Learning. The agent begins in the bottom-left cell and must navigate to the goal (bottom-right cell) without entering the region along the bottom ("The Cliff").

v1 follows the textbook and does not end episodes when the cliff is reached. Also, the goal is a real state.

reference: cite{3} (page 132, example 6.6).

state: Grid location.

actions: Move up/right/down/left.

rewards: -100 for entering The Cliff. -1 for all other transitions.

termination: reaching the goal.

Source code in gym_classics2/envs/gym_classics2/cliff_walk_v1.py

class CliffWalk(Gridworld):
    """The Cliff Walking task, a 12x4 gridworld often used to contrast Sarsa with
    Q-Learning. The agent begins in the bottom-left cell and must navigate to the goal
    (bottom-right cell) without entering the region along the bottom ("The Cliff").

    v1 follows the textbook and does not end episodes when the cliff is reached. Also, the goal is 
    a real state.

    **reference:** cite{3} (page 132, example 6.6).

    **state**: Grid location.

    **actions**: Move up/right/down/left.

    **rewards**: -100 for entering The Cliff. -1 for all other transitions.

    **termination**: reaching the goal.
    """

    layout = """
|            |
|            |
|            |
|S          G|
"""

    def __init__(self, goal_reward=0.0, step_reward=-1.0, cliff_reward=-100.0, **args):
        self._cliff = frozenset((x, 0) for x in range(1, 11))
        self._cliff_reward = cliff_reward
        super().__init__(CliffWalk.layout, goal_reward=goal_reward, step_reward=step_reward, **args)

    # cliff is unreachable. Leads to the start state    
    def _next_state(self, state, action):
        state, _ = super()._next_state(state, action)
        if (state in self._cliff):
            state = self._starts[0]
        return state, 1.0

    def _reward(self, state, action, next_state):
        if next_state in self._goals: 
            return self._goal_reward

        n_state, _ = super()._next_state(state, action)
        if n_state in self._cliff: 
            return self._cliff_reward

        return self._step_reward

    def _done(self, state, action, next_state):
        return next_state in self._goals

gym_classics2.envs.gym_classics2.dyna_maze.DynaMaze

Bases: Gridworld

A 9x6 deterministic gridworld with barriers to make navigation more challenging. The agent starts in cell (0, 3); the goal is the top-right cell.

reference: cite{3} (page 164, example 8.1).

state: Grid location.

actions: Move up/right/down/left.

rewards: +1 for episode termination.

termination: Reaching the goal.

Source code in gym_classics2/envs/gym_classics2/dyna_maze.py

class DynaMaze(Gridworld):
    """A 9x6 deterministic gridworld with barriers to make navigation more challenging.
    The agent starts in cell (0, 3); the goal is the top-right cell.

    **reference:** cite{3} (page 164, example 8.1).

    **state**: Grid location.

    **actions**: Move up/right/down/left.

    **rewards**: +1 for episode termination.

    **termination**: Reaching the goal.
    """

    layout = """
|       XG|
|  X    X |
|S X    X |
|  X      |
|     X   |
|         |
"""

    def __init__(self, **args):
        super().__init__(DynaMaze.layout, **args)

gym_classics2.envs.gym_classics2.four_rooms.FourRooms

Bases: Gridworld

An 11x11 gridworld segmented into four rooms. The agent begins in the bottom-left cell; the goal is in the top-right cell.

reference: cite{2} (page 192).

state: Grid location.

actions: Move up/right/down/left.

rewards: +1 for episode termination.

termination: Taking any action in the goal.

Sutton, Precup and Singh: Between MDPs and semi-MDPs: A framework for temporal abstraction in reinforcement learning.

Artificial Intelligence, 112(1-2):181-211, 1999. [https://hdl.handle.net/20.500.14394/9879]

Source code in gym_classics2/envs/gym_classics2/four_rooms.py

class FourRooms(Gridworld):
    """An 11x11 gridworld segmented into four rooms. The agent begins in the bottom-left
    cell; the goal is in the top-right cell.

    **reference:** cite{2} (page 192).

    **state**: Grid location.

    **actions**: Move up/right/down/left.

    **rewards**: +1 for episode termination.

    **termination**: Taking any action in the goal.

    Reference: Sutton, Precup and Singh: Between MDPs and semi-MDPs: A framework for temporal abstraction in reinforcement learning. 
        Artificial Intelligence, 112(1-2):181-211, 1999. [https://hdl.handle.net/20.500.14394/9879] 
    """

    layout = """
|     X     |
|     X   G |
|           |
|     X     |
|     X     |
|X XXXX     |
|     XXX XX|
|     X     |
|     X     |
|           |
|S    X     |
"""

    def __init__(self, **args):
        super().__init__(FourRooms.layout, **args)

gym_classics2.envs.gym_classics2.L_maze.LMazeGridworld

Bases: Gridworld

A deterministic 10x10 maze separated by an L-shaped barrier.

The agent begins below the horizontal barrier and must travel around it to reach the goal near the upper-right corner.

Reference: Maze used to demonstrate Dijkstra's algorithm, Wikipedia [https://en.wikipedia.org/wiki/Dijkstra's_algorithm]

Source code in gym_classics2/envs/gym_classics2/L_maze.py

class LMazeGridworld(Gridworld):
    """A deterministic 10x10 maze separated by an L-shaped barrier.

    The agent begins below the horizontal barrier and must travel around it to
    reach the goal near the upper-right corner.

    Reference: Maze used to demonstrate Dijkstra's algorithm, Wikipedia [https://en.wikipedia.org/wiki/Dijkstra's_algorithm]
    """
    layout = """
|          |
|        G |
|          |
| XXXXXX   |
|      X   |
|      X   |
|      X   |
|   S  X   |
|          |
|          |
"""

    def __init__(self, **args):
        super().__init__(self.layout, **args)

gym_classics2.envs.gym_classics2.sparse_gridworld.SparseGridworld

Bases: NoisyGridworld

A 10x8 featureless gridworld. The agent starts in cell (1, 3) and the goal is at cell (6, 3). To make it more challenging, the same 80-10-10 transition probabilities from ClassicGridworld are used. Great for testing various forms of credit assignment in the presence of noise.

reference: cite{3} (page 147, figure 7.4).

states: Grid location.

actions: Move up/right/down/left.

rewards: +1 for episode termination.

termination: Reaching the goal.

Source code in gym_classics2/envs/gym_classics2/sparse_gridworld.py

class SparseGridworld(NoisyGridworld):
    """A 10x8 featureless gridworld. The agent starts in cell (1, 3) and the goal is at
    cell (6, 3). To make it more challenging, the same 80-10-10 transition probabilities
    from `ClassicGridworld` are used. Great for testing various forms of credit
    assignment in the presence of noise.

    **reference:** cite{3} (page 147, figure 7.4).

    **states:** Grid location.

    **actions:** Move up/right/down/left.

    **rewards:** +1 for episode termination.

    **termination:** Reaching the goal.
    """

    layout = """
|          |
|          |
|          |
|          |
| S    G   |
|          |
|          |
|          |
"""

    def __init__(self, **args):
        super().__init__(SparseGridworld.layout, **args)

gym_classics2.envs.gym_classics2.windy_gridworld.WindyGridworld

Bases: Gridworld

A 10x7 deterministic gridworld where some columns are affected by an upward wind. The agent starts in cell (0, 3) and the goal is at cell (7, 3). If an agent executes an action from a cell with wind, the resulting position is given by the vector sum of the action's effect and the wind.

reference: cite{3} (page 130, example 6.5).

state: Grid location.

actions: Move up/right/down/left.

rewards: -1 for all transitions unless the episode terminates.

termination: Reaching the goal.

Source code in gym_classics2/envs/gym_classics2/windy_gridworld.py

class WindyGridworld(Gridworld):
    """A 10x7 deterministic gridworld where some columns are affected by an upward wind.
    The agent starts in cell (0, 3) and the goal is at cell (7, 3). If an agent executes
    an action from a cell with wind, the resulting position is given by the vector sum
    of the action's effect and the wind.

    **reference:** cite{3} (page 130, example 6.5).

    **state:** Grid location.

    **actions:** Move up/right/down/left.

    **rewards:** -1 for all transitions unless the episode terminates.

    **termination:** Reaching the goal.
    """

    layout = """
|          |
|          |
|          |
|S      G  |
|          |
|          |
|          |
"""

    def __init__(self, **args):
        super().__init__(WindyGridworld.layout, **args)

    def _next_state(self, state, action):
        wind_strength = self._wind_strength(state)
        state, _ = super()._next_state(state, action)
        state = self._apply_wind(state, wind_strength)
        return self._clamp(state), 1.0

    def _apply_wind(self, state, strength):
        x, y = state
        return (x, y + strength)

    def _wind_strength(self, state):
        """Returns wind strength in the given state."""
        x, _ = state
        if x in {3, 4, 5, 8}:
            return 1
        elif x in {6, 7}:
            return 2
        else:
            return 0

gym_classics2.envs.gym_classics2.linear_walks.Walk5

Bases: LinearWalk

A 5-state deterministic linear walk. Ideal for implementing random walk experiments.

reference: cite{3} (page 125).

state: Discrete position {0, ..., 4} on the number line.

actions: Move left/right.

rewards: +1 for moving right in the extreme right state.

termination: Moving right in the extreme right state or moving left in the extreme left state.

Source code in gym_classics2/envs/gym_classics2/linear_walks.py

class Walk5(LinearWalk):
    """A 5-state deterministic linear walk. Ideal for implementing random walk
    experiments.

    **reference:** cite{3} (page 125).

    **state:** Discrete position {0, ..., 4} on the number line.

    **actions:** Move left/right.

    **rewards:** +1 for moving right in the extreme right state.

    **termination:** Moving right in the extreme right state or moving left in the
    extreme left state.
    """

    def __init__(self):
        super().__init__(length=5, left_reward=0.0, right_reward=1.0)

gym_classics2.envs.gym_classics2.linear_walks.Walk19

Bases: LinearWalk

Same as 5Walk but with 19 states and an additional -1 reward for moving left in the extreme left state.

reference: cite{3} (page 145).

Source code in gym_classics2/envs/gym_classics2/linear_walks.py

class Walk19(LinearWalk):
    """Same as `5Walk` but with 19 states and an additional -1 reward for moving left
    in the extreme left state.

    **reference:** cite{3} (page 145).
    """

    def __init__(self):
        super().__init__(length=19, left_reward=-1.0, right_reward=1.0)