Building the AI Lunar Landing - Complete Code
#Installing the required packages and importing the libraries
!pip install gymnasium
!pip install "gymnasium[atari, accept-rom-license]"
!apt-get install -y swig
!pip install gymnasium[box2d]
import os
import random
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torch.autograd as autograd
from torch.autograd import Variable
from collections import deque, namedtuple
#Part 1 - Building the AI
#Creating the architecture of the Neural Network
class Network(nn.Module):
def __init__(self, state_size, action_size, seed = 42):
super(Network, self).__init__()
self.seed = torch.manual_seed(seed)
self.fc1 = nn.Linear(state_size, 64)
self.fc2 = nn.Linear(64, 64)
self.fc3 = nn.Linear(64, action_size)
def forward(self, state):
x = self.fc1(state)
x = F.relu(x)
x = self.fc2(x)
x = F.relu(x)
return self.fc3(x)
#Part 2 - Training the AI
#Setting up the environment
import gymnasium as gym
env = gym.make('LunarLander-v2')
state_shape = env.observation_space.shape
state_size = env.observation_space.shape[0]
number_actions = env.action_space.n
print('State shape: ', state_shape)
print('State size: ', state_size)
print('Number of actions: ', number_actions)
#Initializing the hyperparameters
learning_rate = 5e-4
minibatch_size = 100
discount_factor = 0.99
replay_buffer_size = int(1e5)
interpolation_parameter = 1e-3
#Implementing Experience Replay
class ReplayMemory(object):
def __init__(self, capacity):
self.device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
self.capacity = capacity
self.memory = []
def push(self, event):
self.memory.append(event)
if len(self.memory) > self.capacity:
del self.memory[0]
def sample(self, batch_size):
experiences = random.sample(self.memory, k = batch_size)
states = torch.from_numpy(np.vstack([e[0] for e in experiences if e is not None])).float().to(self.device)
actions = torch.from_numpy(np.vstack([e[1] for e in experiences if e is not None])).long().to(self.device)
rewards = torch.from_numpy(np.vstack([e[2] for e in experiences if e is not None])).float().to(self.device)
next_states = torch.from_numpy(np.vstack([e[3] for e in experiences if e is not None])).float().to(self.device)
dones = torch.from_numpy(np.vstack([e[4] for e in experiences if e is not None]).astype(np.uint8)).float().to(self.device)
return states, next_states, actions, rewards, dones
#Implementing the DQN class
class Agent():
def __init__(self, state_size, action_size):
self.device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
self.state_size = state_size
self.action_size = action_size
self.local_qnetwork = Network(state_size, action_size).to(self.device)
self.target_qnetwork = Network(state_size, action_size).to(self.device)
self.optimizer = optim.Adam(self.local_qnetwork.parameters(), lr = learning_rate)
self.memory = ReplayMemory(replay_buffer_size)
self.t_step = 0
def step(self, state, action, reward, next_state, done):
self.memory.push((state, action, reward, next_state, done))
self.t_step = (self.t_step + 1) % 4
if self.t_step == 0:
if len(self.memory.memory) > minibatch_size:
experiences = self.memory.sample(100)
self.learn(experiences, discount_factor)
def act(self, state, epsilon = 0.):
state = torch.from_numpy(state).float().unsqueeze(0).to(self.device)
self.local_qnetwork.eval()
with torch.no_grad():
action_values = self.local_qnetwork(state)
self.local_qnetwork.train()
if random.random() > epsilon:
return np.argmax(action_values.cpu().data.numpy())
else:
return random.choice(np.arange(self.action_size))
def learn(self, experiences, discount_factor):
states, next_states, actions, rewards, dones = experiences
next_q_targets = self.target_qnetwork(next_states).detach().max(1)[0].unsqueeze(1)
q_targets = rewards + discount_factor * next_q_targets * (1 - dones)
q_expected = self.local_qnetwork(states).gather(1, actions)
loss = F.mse_loss(q_expected, q_targets)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
self.soft_update(self.local_qnetwork, self.target_qnetwork, interpolation_parameter)
def soft_update(self, local_model, target_model, interpolation_parameter):
for target_param, local_param in zip(target_model.parameters(), local_model.parameters()):
target_param.data.copy_(interpolation_parameter * local_param.data + (1.0 - interpolation_parameter) * target_param.data)
#Initializing the DQN agent
agent = Agent(state_size, number_actions)
#Training the DQN agent
number_episodes = 2000
maximum_number_timesteps_per_episode = 1000
epsilon_starting_value = 1.0
epsilon_ending_value = 0.01
epsilon_decay_value = 0.995
epsilon = epsilon_starting_value
scores_on_100_episodes = deque(maxlen = 100)
for episode in range(1, number_episodes + 1):
state, _ = env.reset()
score = 0
for t in range(maximum_number_timesteps_per_episode):
action = agent.act(state, epsilon)
next_state, reward, done, _, _ = env.step(action)
agent.step(state, action, reward, next_state, done)
state = next_state
score += reward
if done:
break
scores_on_100_episodes.append(score)
epsilon = max(epsilon_ending_value, epsilon_decay_value * epsilon)
print('\rEpisode {}\tAverage Score: {:.2f}'.format(episode, np.mean(scores_on_100_episodes)), end = "")
if episode % 100 == 0:
print('\rEpisode {}\tAverage Score: {:.2f}'.format(episode, np.mean(scores_on_100_episodes)))
if np.mean(scores_on_100_episodes) >= 200.0:
print('\nEnvironment solved in {:d} episodes!\tAverage Score: {:.2f}'.format(episode - 100, np.mean(scores_on_100_episodes)))
torch.save(agent.local_qnetwork.state_dict(), 'checkpoint.pth')
break
#Part 3 - Visualizing the results
import glob
import io
import base64
import imageio
from IPython.display import HTML, display
from gym.wrappers.monitoring.video_recorder import VideoRecorder
def show_video_of_model(agent, env_name):
env = gym.make(env_name, render_mode='rgb_array')
state, _ = env.reset()
done = False
frames = []
while not done:
frame = env.render()
frames.append(frame)
action = agent.act(state)
state, reward, done, _, _ = env.step(action.item())
env.close()
imageio.mimsave('video.mp4', frames, fps=30)
show_video_of_model(agent, 'LunarLander-v2')
def show_video():
mp4list = glob.glob('*.mp4')
if len(mp4list) > 0:
mp4 = mp4list[0]
video = io.open(mp4, 'r+b').read()
encoded = base64.b64encode(video)
display(HTML(data='''<video alt="test" autoplay
loop controls style="height: 400px;">
<source src="data:video/mp4;base64,{0}" type="video/mp4" />
</video>'''.format(encoded.decode('ascii'))))
else:
print("Could not find video")
show_video()
Building the AI Pac-Man - Complete Code
#Part 0 - Installing the required packages and importing the libraries
!pip install gymnasium
!pip install "gymnasium[atari, accept-rom-license]"
!apt-get install -y swig
!pip install gymnasium[box2d]
import os
import random
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
from collections import deque
from torch.utils.data import DataLoader, TensorDataset
#Part 1 - Building the AI-Creating the architecture of the Neural Network
class Network(nn.Module):
def __init__(self, action_size, seed = 42):
super(Network, self).__init__()
self.seed = torch.manual_seed(seed)
self.conv1 = nn.Conv2d(3, 32, kernel_size = 8, stride = 4)
self.bn1 = nn.BatchNorm2d(32)
self.conv2 = nn.Conv2d(32, 64, kernel_size = 4, stride = 2)
self.bn2 = nn.BatchNorm2d(64)
self.conv3 = nn.Conv2d(64, 64, kernel_size = 3, stride = 1)
self.bn3 = nn.BatchNorm2d(64)
self.conv4 = nn.Conv2d(64, 128, kernel_size = 3, stride = 1)
self.bn4 = nn.BatchNorm2d(128)
self.fc1 = nn.Linear(10 * 10 * 128, 512)
self.fc2 = nn.Linear(512, 256)
self.fc3 = nn.Linear(256, action_size)
def forward(self, state):
x = F.relu(self.bn1(self.conv1(state)))
x = F.relu(self.bn2(self.conv2(x)))
x = F.relu(self.bn3(self.conv3(x)))
x = F.relu(self.bn4(self.conv4(x)))
x = x.view(x.size(0), -1)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
return self.fc3(x)
#Part 2 - Training the AI-Setting up the environment
import gymnasium as gym
env = gym.make('MsPacmanDeterministic-v0', full_action_space = False)
state_shape = env.observation_space.shape
state_size = env.observation_space.shape[0]
number_actions = env.action_space.n
print('State shape: ', state_shape)
print('State size: ', state_size)
print('Number of actions: ', number_actions)
#Initializing the hyperparameters
learning_rate = 5e-4
minibatch_size = 64
discount_factor = 0.99
#Preprocessing the frames
from PIL import Image
from torchvision import transforms
def preprocess_frame(frame):
frame = Image.fromarray(frame)
preprocess = transforms.Compose([transforms.Resize((128, 128)), transforms.ToTensor()])
return preprocess(frame).unsqueeze(0)
#Implementing the DCQN class
class Agent():
def __init__(self, action_size):
self.device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
self.action_size = action_size
self.local_qnetwork = Network(action_size).to(self.device)
self.target_qnetwork = Network(action_size).to(self.device)
self.optimizer = optim.Adam(self.local_qnetwork.parameters(), lr = learning_rate)
self.memory = deque(maxlen = 10000)
def step(self, state, action, reward, next_state, done):
state = preprocess_frame(state)
next_state = preprocess_frame(next_state)
self.memory.append((state, action, reward, next_state, done))
if len(self.memory) > minibatch_size:
experiences = random.sample(self.memory, k = minibatch_size)
self.learn(experiences, discount_factor)
def act(self, state, epsilon = 0.):
state = preprocess_frame(state).to(self.device)
self.local_qnetwork.eval()
with torch.no_grad():
action_values = self.local_qnetwork(state)
self.local_qnetwork.train()
if random.random() > epsilon:
return np.argmax(action_values.cpu().data.numpy())
else:
return random.choice(np.arange(self.action_size))
def learn(self, experiences, discount_factor):
states, actions, rewards, next_states, dones = zip(*experiences)
states = torch.from_numpy(np.vstack(states)).float().to(self.device)
actions = torch.from_numpy(np.vstack(actions)).long().to(self.device)
rewards = torch.from_numpy(np.vstack(rewards)).float().to(self.device)
next_states = torch.from_numpy(np.vstack(next_states)).float().to(self.device)
dones = torch.from_numpy(np.vstack(dones).astype(np.uint8)).float().to(self.device)
next_q_targets = self.target_qnetwork(next_states).detach().max(1)[0].unsqueeze(1)
q_targets = rewards + discount_factor * next_q_targets * (1 - dones)
q_expected = self.local_qnetwork(states).gather(1, actions)
loss = F.mse_loss(q_expected, q_targets)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
#Initializing the DCQN agent
agent = Agent(number_actions)
#Training the DCQN agent
number_episodes = 2000
maximum_number_timesteps_per_episode = 10000
epsilon_starting_value = 1.0
epsilon_ending_value = 0.01
epsilon_decay_value = 0.995
epsilon = epsilon_starting_value
scores_on_100_episodes = deque(maxlen = 100)
for episode in range(1, number_episodes + 1):
state, _ = env.reset()
score = 0
for t in range(maximum_number_timesteps_per_episode):
action = agent.act(state, epsilon)
next_state, reward, done, _, _ = env.step(action)
agent.step(state, action, reward, next_state, done)
state = next_state
score += reward
if done:
break
scores_on_100_episodes.append(score)
epsilon = max(epsilon_ending_value, epsilon_decay_value * epsilon)
print('\rEpisode {}\tAverage Score: {:.2f}'.format(episode, np.mean(scores_on_100_episodes)), end = "")
if episode % 100 == 0:
print('\rEpisode {}\tAverage Score: {:.2f}'.format(episode, np.mean(scores_on_100_episodes)))
if np.mean(scores_on_100_episodes) >= 500.0:
print('\nEnvironment solved in {:d} episodes!\tAverage Score: {:.2f}'.format(episode - 100, np.mean(scores_on_100_episodes)))
torch.save(agent.local_qnet
work.state_dict(), 'checkpoint.pth')
break
#Part 3 - Visualizing the results
import glob
import io
import base64
import imageio
from IPython.display import HTML, display
from gym.wrappers.monitoring.video_recorder import VideoRecorder
def show_video_of_model(agent, env_name):
env = gym.make(env_name, render_mode='rgb_array')
state, _ = env.reset()
done = False
frames = []
while not done:
frame = env.render()
frames.append(frame)
action = agent.act(state)
state, reward, done, _, _ = env.step(action)
env.close()
imageio.mimsave('video.mp4', frames, fps=30)
show_video_of_model(agent, 'MsPacmanDeterministic-v0')
def show_video():
mp4list = glob.glob('*.mp4')
if len(mp4list) > 0:
mp4 = mp4list[0]
video = io.open(mp4, 'r+b').read()
encoded = base64.b64encode(video)
display(HTML(data='''<video alt="test" autoplay
loop controls style="height: 400px;">
<source src="data:video/mp4;base64,{0}" type="video/mp4" />
</video>'''.format(encoded.decode('ascii'))))
else:
print("Could not find video")
show_video()
Building the AI KungFuMaster - Complete Code
#Part 0 - Installing the required packages and importing the libraries
!pip install gymnasium
!pip install "gymnasium[atari, accept-rom-license]"
!apt-get install -y swig
!pip install gymnasium[box2d]
import cv2
import math
import random
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torch.multiprocessing as mp
import torch.distributions as distributions
from torch.distributions import Categorical
import gymnasium as gym
from gymnasium import ObservationWrapper
from gymnasium.spaces import Box
#Part 1 - Building the AI Creating the architecture of the Neural Network
class Network(nn.Module):
def __init__(self, action_size):
super(Network, self).__init__()
self.conv1 = torch.nn.Conv2d(in_channels = 4, out_channels = 32, kernel_size = (3,3), stride = 2)
self.conv2 = torch.nn.Conv2d(in_channels = 32, out_channels = 32, kernel_size = (3,3), stride = 2)
self.conv3 = torch.nn.Conv2d(in_channels = 32, out_channels = 32, kernel_size = (3,3), stride = 2)
self.flatten = torch.nn.Flatten()
self.fc1 = torch.nn.Linear(512, 128)
self.fc2a = torch.nn.Linear(128, action_size)
self.fc2s = torch.nn.Linear(128, 1)
def forward(self, state):
x = self.conv1(state)
x = F.relu(x)
x = self.conv2(x)
x = F.relu(x)
x = self.conv3(x)
x = F.relu(x)
x = self.flatten(x)
x = self.fc1(x)
x = F.relu(x)
action_values = self.fc2a(x)
state_value = self.fc2s(x)[0]
return action_values, state_value
#Part 2 - Training the AI Setting up the environment
class PreprocessAtari(ObservationWrapper):
def __init__(self, env, height = 42, width = 42, crop = lambda img: img, dim_order = 'pytorch', color = False, n_frames = 4):
super(PreprocessAtari, self).__init__(env)
self.img_size = (height, width)
self.crop = crop
self.dim_order = dim_order
self.color = color
self.frame_stack = n_frames
n_channels = 3 * n_frames if color else n_frames
obs_shape = {'tensorflow': (height, width, n_channels), 'pytorch': (n_channels, height, width)}[dim_order]
self.observation_space = Box(0.0, 1.0, obs_shape)
self.frames = np.zeros(obs_shape, dtype = np.float32)
def reset(self):
self.frames = np.zeros_like(self.frames)
obs, info = self.env.reset()
self.update_buffer(obs)
return self.frames, info
def observation(self, img):
img = self.crop(img)
img = cv2.resize(img, self.img_size)
if not self.color:
if len(img.shape) == 3 and img.shape[2] == 3:
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
img = img.astype('float32') / 255.
if self.color:
self.frames = np.roll(self.frames, shift = -3, axis = 0)
else:
self.frames = np.roll(self.frames, shift = -1, axis = 0)
if self.color:
self.frames[-3:] = img
else:
self.frames[-1] = img
return self.frames
def update_buffer(self, obs):
self.frames = self.observation(obs)
def make_env():
env = gym.make("KungFuMasterDeterministic-v0", render_mode = 'rgb_array')
env = PreprocessAtari(env, height = 42, width = 42, crop = lambda img: img, dim_order = 'pytorch', color = False, n_frames = 4)
return env
env = make_env()
state_shape = env.observation_space.shape
number_actions = env.action_space.n
print("State shape:", state_shape)
print("Number actions:", number_actions)
print("Action names:", env.env.env.get_action_meanings())
#Initializing the hyperparameters
learning_rate = 1e-4
discount_factor = 0.99
number_environments = 10
#Implementing the A3C class
class Agent():
def __init__(self, action_size):
self.device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
self.action_size = action_size
self.network = Network(action_size).to(self.device)
self.optimizer = torch.optim.Adam(self.network.parameters(), lr = learning_rate)
def act(self, state):
if state.ndim == 3:
state = [state]
state = torch.tensor(state, dtype = torch.float32, device = self.device)
action_values, _ = self.network(state)
policy = F.softmax(action_values, dim = -1)
return np.array([np.random.choice(len(p), p = p) for p in policy.detach().cpu().numpy()])
def step(self, state, action, reward, next_state, done):
batch_size = state.shape[0]
state = torch.tensor(state, dtype = torch.float32, device = self.device)
next_state = torch.tensor(next_state, dtype = torch.float32, device = self.device)
reward = torch.tensor(reward, dtype = torch.float32, device = self.device)
done = torch.tensor(done, dtype = torch.bool, device = self.device).to(dtype = torch.float32)
action_values, state_value = self.network(state)
_, next_state_value = self.network(next_state)
target_state_value = reward + discount_factor * next_state_value * (1 - done)
advantage = target_state_value - state_value
probs = F.softmax(action_values, dim = -1)
logprobs = F.log_softmax(action_values, dim = -1)
entropy = -torch.sum(probs * logprobs, axis = -1)
batch_idx = np.arange(batch_size)
logp_actions = logprobs[batch_idx, action]
actor_loss = -(logp_actions * advantage.detach()).mean() - 0.001 * entropy.mean()
critic_loss = F.mse_loss(target_state_value.detach(), state_value)
total_loss = actor_loss + critic_loss
self.optimizer.zero_grad()
total_loss.backward()
self.optimizer.step()
#Initializing the A3C agent
agent = Agent(number_actions)
#Evaluating our A3C agent on a certain number of episodes
def evaluate(agent, env, n_episodes = 1):
episodes_rewards = []
for _ in range(n_episodes):
state, _ = env.reset()
total_reward = 0
while True:
action = agent.act(state)
state, reward, done, info, _ = env.step(action[0])
total_reward += reward
if done:
break
episodes_rewards.append(total_reward)
return episodes_rewards
#Managing multiple environments simultaneously
class EnvBatch:
def __init__(self, n_envs = 10):
self.envs = [make_env() for _ in range(n_envs)]
def reset(self):
_states = []
for env in self.envs:
_states.append(env.reset()[0])
return np.array(_states)
def step(self, actions):
next_states, rewards, dones, infos, _ = map(np.array, zip(*[env.step(a) for env, a in zip(self.envs, actions)]))
for i in range(len(self.envs)):
if dones[i]:
next_states[i] = self.envs[i].reset()[0]
return next_states, rewards, dones, infos
#Training the A3C agent
import tqdm
env_batch = EnvBatch(number_environments)
batch_states = env_batch.reset()
with tqdm.trange(0, 3001) as progress_bar:
for i in progress_bar:
batch_actions = agent.act(batch_states)
batch_next_states, batch_rewards, batch_dones, _ = env_batch.step(batch_actions)
batch_rewards *= 0.01
agent.step(batch_states, batch_actions, batch_rewards, batch_next_states, batch_dones)
batch_states = batch_next_states
if i % 1000 == 0:
print("Average agent reward: ", np.mean(evaluate(agent, env, n_episodes = 10)))
#Part 3 - Visualizing the results
import glob
import io
import base64
import imageio
from IPython.display import HTML, display
from gymnasium.wrappers.monitoring.video_recorder import VideoRecorder
def show_video_of_model(agent, env):
state, _ = env.reset()
done = False
frames = []
while not done:
frame = env.render()
frames.append(frame)
action = agent.act(state)
state, reward, done, _, _ = env.step(action[0])
env.close()
imageio.mimsave('video.mp4', frames, fps=30)
show_video_of_model(agent, env)
def show_video():
mp4list = glob.glob('*.mp4')
if len(mp4list) > 0:
mp4 = mp4list[0]
video = io.open(mp4, 'r+b').read()
encoded = base64.b64encode(video)
display(HTML(data='''<video alt="test" autoplay
loop controls style="height: 400px;">
<source src="data:video/mp4;base64,{0}" type="video/mp4" />
</video>'''.format(encoded.decode('ascii'))))
else:
print("Could not find video")
show_video()
No comments:
Post a Comment