Skip to content

Utility API

Performance plots

gym_classics2.performance

simple_moving_average 

simple_moving_average(data, window_size=100)

Return a centered simple moving average padded with NaN values.

Parameters:

Name Type Description Default
data

One-dimensional numeric sequence.

 required
window_size

Number of observations in the averaging window.

 100

Returns:

Type Description

NumPy array with the same length as data.
Source code in gym_classics2/performance.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
def simple_moving_average(data, window_size = 100):
    """Return a centered simple moving average padded with ``NaN`` values.

    Args:
        data: One-dimensional numeric sequence.
        window_size: Number of observations in the averaging window.

    Returns:
        NumPy array with the same length as ``data``.
    """
    weights = np.ones(window_size) / window_size
    sma = np.convolve(data, weights, mode='valid')
    sma = np.concatenate((np.full((window_size)//2, np.nan),sma,np.full(len(data)-len(sma)-(window_size)//2, np.nan)))  # Pad the beginning with NaN for alignment
    return sma

cum_avg 

cum_avg(data)

Return the cumulative average at every position in a numeric sequence.

Source code in gym_classics2/performance.py 
20
21
22
def cum_avg(data):
    """Return the cumulative average at every position in a numeric sequence."""
    return np.cumsum(data) / np.arange(1, len(data) + 1)

plot_returns 

plot_returns(returns, y_label='Episode Return', title='', window_size=100)

Plot episode returns with a moving average and cumulative average.

Source code in gym_classics2/performance.py 
24
25
26
27
28
29
30
31
32
33
34
35
def plot_returns(returns, y_label = "Episode Return", title = "", window_size = 100):
    """Plot episode returns with a moving average and cumulative average."""
    x = range(len(returns))
    plt.plot(x, returns, label="Episode")
    plt.plot(x, simple_moving_average(returns, window_size), label="Moving Average (100)")
    plt.plot(x, cum_avg(returns), label="Cumulative Average")

    plt.xlabel("Episode")
    plt.ylabel(y_label)
    plt.title(title)
    plt.legend()
    plt.show()

plot_ep_lens 

plot_ep_lens(ep_lens, y_label='Episode Length', title='', window_size=100)

Plot episode lengths with a moving average.

Source code in gym_classics2/performance.py 
37
38
39
40
41
42
43
44
45
46
47
def plot_ep_lens(ep_lens, y_label = "Episode Length", title = "", window_size = 100):
    """Plot episode lengths with a moving average."""
    x = range(len(ep_lens))
    plt.plot(x, ep_lens, label="Episode Length")
    plt.plot(x, simple_moving_average(ep_lens, window_size), label="Moving Average (100)")

    plt.xlabel("Episode")
    plt.ylabel(y_label)
    plt.title(title)
    plt.legend()
    plt.show()

Gridworld animation

gym_classics2.animation

gridworld_animate 

gridworld_animate(env, Vs, policies=None, interval=1000, repeat=False, cmap='coolwarm', clim=None, origin='lower')

Create an animation showing the evolution of value functions in a gridworld.

:param env: The gridworld environment. :param Vs: A list of value functions to animate. :param repeat: Whether the animation should repeat. :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. :return: An animation object.

Source code in gym_classics2/animation.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def gridworld_animate(env, Vs, policies = None, interval = 1000, repeat=False, cmap = "coolwarm", clim = None, origin='lower'):
    """
    Create an animation showing the evolution of value functions in a gridworld.

    :param env: The gridworld environment.
    :param Vs: A list of value functions to animate.
    :param repeat: Whether the animation should repeat.
    :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.
    :return: An animation object.
    """

    if clim is None:
        vmin = None
        vmax = None
    else:
        vmin = clim[0]
        vmax = clim[1]

    cmap = plt.colormaps[cmap].copy()
    cmap.set_bad(color='black')

    mazes = [env.to_matrix(V) for V in Vs]

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

    fig, ax = plt.subplots()

    # use last slide for clims 
    im = ax.imshow(mazes[-1], cmap=cmap, origin=origin, vmin = vmin, vmax=vmax)
    title = ax.set_title("")
    labels = []
    if not policies is None:
        for s in env.states():
            (i,j) = env.id2state(s)
            labels.append(ax.text(i, j, '', ha='center', va='center', color='black', fontsize=10))

    plt.colorbar(im, ax=ax)

    def init():
        ax.set_xticks(np.arange(mazes[0].shape[1]))
        ax.set_yticks(np.arange(mazes[0].shape[0]))

        num_rows, num_cols = mazes[0].shape
        ax.set_xticks(np.arange(-.5, num_cols, 1), minor=True)
        ax.set_yticks(np.arange(-.5, num_rows, 1), minor=True)
        ax.tick_params(which='minor', bottom=False, left=False)
        ax.grid(which='minor', color='black', linestyle='-', linewidth=1)

        return im, title, *labels,

    if not policies is None:
        for s in env.states():
            (i,j) = env.id2state(s)
            labels.append(ax.text(i, j, '', ha='center', va='center', color='black', fontsize=10))

    def step(i):
        im = ax.imshow(mazes[i], cmap=cmap, origin=origin, vmin = vmin, vmax=vmax)
        title.set_text(f'After Iteration {i}')

        if not policies is None:
            for pos, a in enumerate(policies[i]):
                labels[pos].set_text(a)

        return im, title, *labels,

    ani = animation.FuncAnimation(
        fig,
        step,
        frames = len(mazes),
        init_func = init,
        interval = interval,
        repeat = repeat,
        blit = True
    )

    plt.close()

    return ani

General utilities

gym_classics2.utils

get_rng 

get_rng(rng=None)

Return rng as a NumPy random generator.

rng may be a :class:numpy.random.Generator, an integer seed, or None. Passing a generator lets callers share one reproducible random stream across an algorithm and all of its helpers.

Source code in gym_classics2/utils.py 
 4
 5
 6
 7
 8
 9
10
11
def get_rng(rng=None):
    """Return *rng* as a NumPy random generator.

    ``rng`` may be a :class:`numpy.random.Generator`, an integer seed, or
    ``None``. Passing a generator lets callers share one reproducible random
    stream across an algorithm and all of its helpers.
    """
    return np.random.default_rng(rng)

clip 

clip(x, low, high)

A scalar version of numpy.clip. Much faster because it avoids memory allocation.

Source code in gym_classics2/utils.py 
13
14
15
def clip(x, low, high):
    """A scalar version of numpy.clip. Much faster because it avoids memory allocation."""
    return min(max(x, low), high)

random_argmax 

random_argmax(x, axis=None, rng=None)

Argmax that breaks ties randomly. If axis is None, returns a single index. If axis is specified, returns an array of indices along that axis. rng may be a NumPy generator or an integer seed.

Source code in gym_classics2/utils.py 
19
20
21
22
23
24
25
26
27
28
29
def random_argmax(x, axis=None, rng=None):
    """
    Argmax that breaks ties randomly. If axis is None, returns a single index.
    If axis is specified, returns an array of indices along that axis. ``rng``
    may be a NumPy generator or an integer seed.
    """
    rng = get_rng(rng)
    if axis is None:
        return rng.choice(np.where(x == np.max(x))[0])
    else:
        return np.apply_along_axis(lambda values: random_argmax(values, rng=rng), axis, x)