Deep Reinforcement Learning¶

!pip install gym
!conda install -y JSAnimation

%matplotlib inline
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

import re

from keras.models import Sequential, Model
from keras.layers import Dense, Input, Dot
from keras.models import load_model, model_from_json
from keras.optimizers import Adam

import gym

from collections import deque

import time

%matplotlib inline

# Imports specifically so we can render outputs in Jupyter.
from JSAnimation.IPython_display import display_animation
from matplotlib import animation
from IPython.display import display

env = gym.make('CartPole-v0')
STATES, ACTIONS = env.observation_space.shape[0], env.action_space.n
model, out = create_model(STATES, ACTIONS)
INITIAL_EPSILON = 1e-1
FINAL_EPSILON = 1e-4
DECAY = 0.9
GAMMA = 0.9 # decay rate of past observations
OBSERVE = 5000. # timesteps to observe before training
REPLAY_MEMORY = 5000 # number of previous transitions to remember
TIME_LIMIT = 100000
BATCH = 128
# open up a game state to communicate with emulator
state = env.reset()

# store the previous observations in replay memory
D = deque(maxlen=REPLAY_MEMORY)
loss = []

a_t = np.zeros(ACTIONS)
a_t[np.random.choice(ACTIONS)] = 1
s_t, r_0, terminal, _ = env.step(np.argmax(a_t))

# start training
epsilon = INITIAL_EPSILON

up_time = [0]
start = time.time()
for t in range(TIME_LIMIT):
    # choose an action
    readout_t = out.predict(s_t[None, :])
    a_t = np.zeros([ACTIONS])
    if np.random.random() <= epsilon:
        a_t[np.random.choice(ACTIONS)] = 1
    else:
        a_t[np.argmax(readout_t)] = 1

    # scale down epsilon
    if epsilon > FINAL_EPSILON and t > OBSERVE:
        epsilon *= DECAY

    # run the selected action and observe next state and reward
    # store the transition in D
    s_t1, r_t, terminal, _ = env.step(np.argmax(a_t))
    D.append((s_t, a_t, r_t, s_t1, terminal))
    
    if terminal:
        up_time.append(0)
        env.reset()
    else:
        up_time[-1] += r_t
    
    # only train if done observing
    if t > OBSERVE:
        # sample a minibatch to train on
        idx = np.random.choice(REPLAY_MEMORY, BATCH, replace=False)
        minibatch = [D[i] for i in idx]
        # get the batch variables
        s_t_batch, a_batch, y_batch = train_data(minibatch, out)
        # perform gradient step
        loss.append(model.train_on_batch([s_t_batch, a_batch], y_batch))

    # update the old state
    s_t = s_t1
        
    if t%(TIME_LIMIT/20)==0:
        print('Episode :', t, ', time taken: ', time.time() - start, 's, average up time: ',np.mean(up_time[-100:]))
        start = time.time()

plt.figure(figsize=(12, 5))
plt.plot(up_time)
plt.show()

plt.figure(figsize=(12, 5))
plt.plot(loss)
plt.show()

env = gym.make('CartPole-v0')

# Run a demo of the environment
observation = env.reset()
cum_reward = 0
frames = []
for t in range(5000):
    # Render into buffer. 
    frames.append(env.render(mode = 'rgb_array'))
    action = env.action_space.sample()
    observation, reward, done, info = env.step(action)
    if done:
        break
env.render(close=True)
display_frames_as_gif(frames)
print(t)

# Run a demo of the environment
observation = env.reset()
cum_reward = 0
frames = []
a_t = np.zeros([ACTIONS])
a_t[env.action_space.sample()] = 1
for t in range(5000):
    # Render into buffer.  
    s_t, reward, done, info = env.step(np.argmax(a_t))
    frames.append(env.render(mode = 'rgb_array'))
    readout_t = out.predict(s_t[None, :])
    a_t = np.zeros([ACTIONS])
    a_t[np.argmax(readout_t)] = 1
    
    if done:
        break
env.render(close=True)
display_frames_as_gif(frames)
print(t)