Safe lab preview

notebook

This is a sanitized, read-only preview. Nothing executes in this page.

Read-only

Notebook preview

notebook

Download notebook

Markdown cell[1]

> **Integrated runtime note:** This preview uses a small deterministic, rights-safe offline fixture for reproducible learning. Full-scale results require the lesson's documented dataset or model in an approved external environment. ## CartPole Skating > **Problem**: If Peter wants to escape from the wolf, he needs to be able to move faster than him. We will see how Peter can learn to skate, in particular, to keep balance, using Q-Learning. First, let's Gymnasium is supplied by this edition's pinned runtime environment.

Code cell[2]

import sys
# Dependencies are provisioned by this course edition's pinned runtime profile.

import gymnasium as gym
import matplotlib.pyplot as plt
import numpy as np
import random

np.random.seed(2026)
random.seed(2026)

Markdown cell[3]

## Create a cartpole environment

Code cell[4]

env = gym.make("CartPole-v1", render_mode="rgb_array")
print(env.action_space)
print(env.observation_space)
print(env.action_space.sample())

Markdown cell[5]

To see how the environment works, let's run a short simulation for 100 steps.

Code cell[6]

env.reset()

for _ in range(100):
    env.render()
    _, _, terminated, truncated, _ = env.step(env.action_space.sample())
    if terminated or truncated:
        break

Markdown cell[7]

During simulation, we need to get observations in order to decide how to act. In fact, `step` function returns us back current observations, reward function, and the `done` flag that indicates whether it makes sense to continue the simulation or not:

Code cell[8]

env.reset()

done = False
while not done:
   env.render()
   obs, rew, terminated, truncated, info = env.step(env.action_space.sample())
   done = terminated or truncated
   print(f"{obs} -> {rew}")

Markdown cell[9]

We can get min and max value of those numbers:

Code cell[10]

print(env.observation_space.low)
print(env.observation_space.high)

Markdown cell[11]

## State Discretization

Code cell[12]

def discretize(x):
    return tuple((x/np.array([0.25, 0.25, 0.01, 0.1])).astype(np.int64))

Markdown cell[13]

Let's also explore other discretization method using bins:

Code cell[14]

def create_bins(i,num):
    return np.arange(num+1)*(i[1]-i[0])/num+i[0]

print("Sample bins for interval (-5,5) with 10 bins\n",create_bins((-5,5),10))

ints = [(-5,5),(-2,2),(-0.5,0.5),(-2,2)] # intervals of values for each parameter
nbins = [20,20,10,10] # number of bins for each parameter
bins = [create_bins(ints[i],nbins[i]) for i in range(4)]

def discretize_bins(x):
    return tuple(np.digitize(x[i],bins[i]) for i in range(4))

Markdown cell[15]

Let's now run a short simulation and observe those discrete environment values.

Code cell[16]

env.reset()

done = False
while not done:
   #env.render()
   obs, rew, terminated, truncated, info = env.step(env.action_space.sample())
   done = terminated or truncated
   #print(discretize_bins(obs))
   print(discretize(obs))

Markdown cell[17]

## Q-Table Structure

Code cell[18]

Q = {}
actions = (0,1)

def qvalues(state):
    return [Q.get((state,a),0) for a in actions]

Markdown cell[19]

## Let's Start Q-Learning!

Code cell[20]

# hyperparameters
alpha = 0.3
gamma = 0.9
epsilon = 0.90

Code cell[21]

def probs(v,eps=1e-4):
    v = v-v.min()+eps
    v = v/v.sum()
    return v

Qmax = 0
cum_rewards = []
rewards = []
for epoch in range(200):
    obs, _ = env.reset(seed=2026 + epoch)
    done = False
    cum_reward=0
    # == do the simulation ==
    while not done:
        s = discretize(obs)
        if random.random()<epsilon:
            # exploitation - chose the action according to Q-Table probabilities
            v = probs(np.array(qvalues(s)))
            a = random.choices(actions,weights=v)[0]
        else:
            # exploration - randomly chose the action
            a = np.random.randint(env.action_space.n)

        obs, rew, terminated, truncated, info = env.step(a)

        done = terminated or truncated
        cum_reward+=rew
        ns = discretize(obs)
        bootstrap = 0.0 if terminated else gamma * max(qvalues(ns))
        Q[(s,a)] = (1 - alpha) * Q.get((s,a), 0) + alpha * (rew + bootstrap)
    cum_rewards.append(cum_reward)
    rewards.append(cum_reward)
    # == Periodically print results and calculate average reward ==
    if epoch % 50==0:
        print(f"{epoch}: {np.average(cum_rewards)}, alpha={alpha}, epsilon={epsilon}")
        if np.average(cum_rewards) > Qmax:
            Qmax = np.average(cum_rewards)
            Qbest = Q
        cum_rewards=[]

Markdown cell[22]

## Plotting Training Progress

Code cell[23]

plt.plot(rewards)

Markdown cell[24]

From this graph, it is not possible to tell anything, because due to the nature of stochastic training process the length of training sessions varies greatly. To make more sense of this graph, we can calculate **running average** over series of experiments, let's say 100. This can be done conveniently using `np.convolve`:

Code cell[25]

def running_average(x,window):
    return np.convolve(x,np.ones(window)/window,mode='valid')

plt.plot(running_average(rewards,100))

Markdown cell[26]

## Varying Hyperparameters and Seeing the Result in Action Now it would be interesting to actually see how the trained model behaves. Let's run the simulation, and we will be following the same action selection strategy as during training: sampling according to the probability distribution in Q-Table:

Code cell[27]

obs, _ = env.reset()
done = False
while not done:
   s = discretize(obs)
   env.render()
   v = probs(np.array(qvalues(s)))
   a = random.choices(actions,weights=v)[0]
   obs, _, terminated, truncated, _ = env.step(a)
   done = terminated or truncated

Markdown cell[28]

## Saving result to an animated GIF If you want to impress your friends, you may want to send them the animated GIF picture of the balancing pole. To do this, we can invoke `env.render` to produce an image frame, and then save those to animated GIF using PIL library:

Code cell[29]

from pathlib import Path
from PIL import Image

output_dir = Path("outputs")
output_dir.mkdir(parents=True, exist_ok=True)

obs, _ = env.reset()
done = False
i = 0
ims = []
while not done:
   s = discretize(obs)
   img = env.render()
   if img is None:
      raise RuntimeError("CartPole rgb_array rendering returned no frame")
   ims.append(Image.fromarray(img))
   v = probs(np.array([Qbest.get((s,a),0) for a in actions]))
   a = random.choices(actions,weights=v)[0]
   obs, _, terminated, truncated, _ = env.step(a)
   done = terminated or truncated
   i += 1
env.close()

if not ims:
   raise RuntimeError("CartPole produced no frames")
output_path = output_dir / "cartpole-balance.gif"
ims[0].save(output_path,save_all=True,append_images=ims[1::2],loop=0,duration=5)
print(f"Saved {i} frames to {output_path}")

Outputs, execution counts, widgets, and active content were removed during import. Run notebooks only in an external environment you trust.

Record your practice

Optional self-reporting helps you remember what you practiced and never gates course completion.

Back to lesson

import sys # Dependencies are provisioned by this course edition's pinned runtime profile. import gymnasium as gym import matplotlib.pyplot as plt import numpy as np import random np.random.seed(2026) random.seed(2026)

def create_bins(i,num): return np.arange(num+1)*(i[1]-i[0])/num+i[0] print("Sample bins for interval (-5,5) with 10 bins\n",create_bins((-5,5),10)) ints = [(-5,5),(-2,2),(-0.5,0.5),(-2,2)] # intervals of values for each parameter nbins = [20,20,10,10] # number of bins for each parameter bins = [create_bins(ints[i],nbins[i]) for i in range(4)] def discretize_bins(x): return tuple(np.digitize(x[i],bins[i]) for i in range(4))

env.reset() done = False while not done: #env.render() obs, rew, terminated, truncated, info = env.step(env.action_space.sample()) done = terminated or truncated #print(discretize_bins(obs)) print(discretize(obs))

def probs(v,eps=1e-4): v = v-v.min()+eps v = v/v.sum() return v Qmax = 0 cum_rewards = [] rewards = [] for epoch in range(200): obs, _ = env.reset(seed=2026 + epoch) done = False cum_reward=0 # == do the simulation == while not done: s = discretize(obs) if random.random()<epsilon: # exploitation - chose the action according to Q-Table probabilities v = probs(np.array(qvalues(s))) a = random.choices(actions,weights=v)[0] else: # exploration - randomly chose the action a = np.random.randint(env.action_space.n) obs, rew, terminated, truncated, info = env.step(a) done = terminated or truncated cum_reward+=rew ns = discretize(obs) bootstrap = 0.0 if terminated else gamma * max(qvalues(ns)) Q[(s,a)] = (1 - alpha) * Q.get((s,a), 0) + alpha * (rew + bootstrap) cum_rewards.append(cum_reward) rewards.append(cum_reward) # == Periodically print results and calculate average reward == if epoch % 50==0: print(f"{epoch}: {np.average(cum_rewards)}, alpha={alpha}, epsilon={epsilon}") if np.average(cum_rewards) > Qmax: Qmax = np.average(cum_rewards) Qbest = Q cum_rewards=[]

obs, _ = env.reset() done = False while not done: s = discretize(obs) env.render() v = probs(np.array(qvalues(s))) a = random.choices(actions,weights=v)[0] obs, _, terminated, truncated, _ = env.step(a) done = terminated or truncated

from pathlib import Path from PIL import Image output_dir = Path("outputs") output_dir.mkdir(parents=True, exist_ok=True) obs, _ = env.reset() done = False i = 0 ims = [] while not done: s = discretize(obs) img = env.render() if img is None: raise RuntimeError("CartPole rgb_array rendering returned no frame") ims.append(Image.fromarray(img)) v = probs(np.array([Qbest.get((s,a),0) for a in actions])) a = random.choices(actions,weights=v)[0] obs, _, terminated, truncated, _ = env.step(a) done = terminated or truncated i += 1 env.close() if not ims: raise RuntimeError("CartPole produced no frames") output_path = output_dir / "cartpole-balance.gif" ims[0].save(output_path,save_all=True,append_images=ims[1::2],loop=0,duration=5) print(f"Saved {i} frames to {output_path}")