• 用深度强化学习来玩Flappy Bird

    目录

    演示视频

    核心代码

    演示视频

    用深度强化学习来玩Flappy Bird

    核心代码

    1. import torch.nn as nn
    2.  
    3. class DeepQNetwork(nn.Module):
    4.     def __init__(self):
    5.         super(DeepQNetwork, self).__init__()
    6.  
    7.         self.conv1 = nn.Sequential(nn.Conv2d(4, 32, kernel_size=8, stride=4), nn.ReLU(inplace=True))
    8.         self.conv2 = nn.Sequential(nn.Conv2d(32, 64, kernel_size=4, stride=2), nn.ReLU(inplace=True))
    9.         self.conv3 = nn.Sequential(nn.Conv2d(64, 64, kernel_size=3, stride=1), nn.ReLU(inplace=True))
    10.  
    11.         self.fc1 = nn.Sequential(nn.Linear(7 * 7 * 64, 512), nn.ReLU(inplace=True))
    12.         self.fc2 = nn.Linear(512, 2)
    13.         self._create_weights()
    14.  
    15.     def _create_weights(self):
    16.         for m in self.modules():
    17.             if isinstance(m, nn.Conv2d) or isinstance(m, nn.Linear):
    18.                 nn.init.uniform_(m.weight, -0.01, 0.01)
    19.                 nn.init.constant_(m.bias, 0)
    20.  
    21.     def forward(self, input):
    22.         output = self.conv1(input)
    23.         output = self.conv2(output)
    24.         output = self.conv3(output)
    25.         output = output.view(output.size(0), -1)
    26.         output = self.fc1(output)
    27.         output = self.fc2(output)
    28.  
    29.         return output
    1. from itertools import cycle
    2. from numpy.random import randint
    3. from pygame import Rect, init, time, display
    4. from pygame.event import pump
    5. from pygame.image import load
    6. from pygame.surfarray import array3d, pixels_alpha
    7. from pygame.transform import rotate
    8. import numpy as np
    9.  
    10.  
    11. class FlappyBird(object):
    12.     init()
    13.     fps_clock = time.Clock()
    14.     screen_width = 288
    15.     screen_height = 512
    16.     screen = display.set_mode((screen_width, screen_height))
    17.     display.set_caption('Deep Q-Network Flappy Bird')
    18.     base_image = load('assets/sprites/base.png').convert_alpha()
    19.     background_image = load('assets/sprites/background-black.png').convert()
    20.  
    21.     pipe_images = [rotate(load('assets/sprites/pipe-green.png').convert_alpha(), 180),
    22.                    load('assets/sprites/pipe-green.png').convert_alpha()]
    23.     bird_images = [load('assets/sprites/redbird-upflap.png').convert_alpha(),
    24.                    load('assets/sprites/redbird-midflap.png').convert_alpha(),
    25.                    load('assets/sprites/redbird-downflap.png').convert_alpha()]
    26.     # number_images = [load('assets/sprites/{}.png'.format(i)).convert_alpha() for i in range(10)]
    27.  
    28.     bird_hitmask = [pixels_alpha(image).astype(bool) for image in bird_images]
    29.     pipe_hitmask = [pixels_alpha(image).astype(bool) for image in pipe_images]
    30.  
    31.     fps = 30
    32.     pipe_gap_size = 100
    33.     pipe_velocity_x = -4
    34.  
    35.     # parameters for bird
    36.     min_velocity_y = -8
    37.     max_velocity_y = 10
    38.     downward_speed = 1
    39.     upward_speed = -9
    40.  
    41.     bird_index_generator = cycle([0, 1, 2, 1])
    42.  
    43.     def __init__(self):
    44.  
    45.         self.iter = self.bird_index = self.score = 0
    46.  
    47.         self.bird_width = self.bird_images[0].get_width()
    48.         self.bird_height = self.bird_images[0].get_height()
    49.         self.pipe_width = self.pipe_images[0].get_width()
    50.         self.pipe_height = self.pipe_images[0].get_height()
    51.  
    52.         self.bird_x = int(self.screen_width / 5)
    53.         self.bird_y = int((self.screen_height - self.bird_height) / 2)
    54.  
    55.         self.base_x = 0
    56.         self.base_y = self.screen_height * 0.79
    57.         self.base_shift = self.base_image.get_width() - self.background_image.get_width()
    58.  
    59.         pipes = [self.generate_pipe(), self.generate_pipe()]
    60.         pipes[0]["x_upper"] = pipes[0]["x_lower"] = self.screen_width
    61.         pipes[1]["x_upper"] = pipes[1]["x_lower"] = self.screen_width * 1.5
    62.         self.pipes = pipes
    63.  
    64.         self.current_velocity_y = 0
    65.         self.is_flapped = False
    66.  
    67.     def generate_pipe(self):
    68.         x = self.screen_width + 10
    69.         gap_y = randint(2, 10) * 10 + int(self.base_y / 5)
    70.         return {"x_upper": x, "y_upper": gap_y - self.pipe_height, "x_lower": x, "y_lower": gap_y + self.pipe_gap_size}
    71.  
    72.     def is_collided(self):
    73.         # Check if the bird touch ground
    74.         if self.bird_height + self.bird_y + 1 >= self.base_y:
    75.             return True
    76.         bird_bbox = Rect(self.bird_x, self.bird_y, self.bird_width, self.bird_height)
    77.         pipe_boxes = []
    78.         for pipe in self.pipes:
    79.             pipe_boxes.append(Rect(pipe["x_upper"], pipe["y_upper"], self.pipe_width, self.pipe_height))
    80.             pipe_boxes.append(Rect(pipe["x_lower"], pipe["y_lower"], self.pipe_width, self.pipe_height))
    81.             # Check if the bird's bounding box overlaps to the bounding box of any pipe
    82.             if bird_bbox.collidelist(pipe_boxes) == -1:
    83.                 return False
    84.             for i in range(2):
    85.                 cropped_bbox = bird_bbox.clip(pipe_boxes[i])
    86.                 min_x1 = cropped_bbox.x - bird_bbox.x
    87.                 min_y1 = cropped_bbox.y - bird_bbox.y
    88.                 min_x2 = cropped_bbox.x - pipe_boxes[i].x
    89.                 min_y2 = cropped_bbox.y - pipe_boxes[i].y
    90.                 if np.any(self.bird_hitmask[self.bird_index][min_x1:min_x1 + cropped_bbox.width,
    91.                        min_y1:min_y1 + cropped_bbox.height] * self.pipe_hitmask[i][min_x2:min_x2 + cropped_bbox.width,
    92.                                                               min_y2:min_y2 + cropped_bbox.height]):
    93.                     return True
    94.         return False
    95.  
    96.     def next_frame(self, action):
    97.         pump()
    98.         reward = 0.1
    99.         terminal = False
    100.         # Check input action
    101.         if action == 1:
    102.             self.current_velocity_y = self.upward_speed
    103.             self.is_flapped = True
    104.  
    105.         # Update score
    106.         bird_center_x = self.bird_x + self.bird_width / 2
    107.         for pipe in self.pipes:
    108.             pipe_center_x = pipe["x_upper"] + self.pipe_width / 2
    109.             if pipe_center_x < bird_center_x < pipe_center_x + 5:
    110.                 self.score += 1
    111.                 reward = 1
    112.                 break
    113.  
    114.         # Update index and iteration
    115.         if (self.iter + 1) % 3 == 0:
    116.             self.bird_index = next(self.bird_index_generator)
    117.             self.iter = 0
    118.         self.base_x = -((-self.base_x + 100) % self.base_shift)
    119.  
    120.         # Update bird's position
    121.         if self.current_velocity_y < self.max_velocity_y and not self.is_flapped:
    122.             self.current_velocity_y += self.downward_speed
    123.         if self.is_flapped:
    124.             self.is_flapped = False
    125.         self.bird_y += min(self.current_velocity_y, self.bird_y - self.current_velocity_y - self.bird_height)
    126.         if self.bird_y < 0:
    127.             self.bird_y = 0
    128.  
    129.         # Update pipes' position
    130.         for pipe in self.pipes:
    131.             pipe["x_upper"] += self.pipe_velocity_x
    132.             pipe["x_lower"] += self.pipe_velocity_x
    133.         # Update pipes
    134.         if 0 < self.pipes[0]["x_lower"] < 5:
    135.             self.pipes.append(self.generate_pipe())
    136.         if self.pipes[0]["x_lower"] < -self.pipe_width:
    137.             del self.pipes[0]
    138.         if self.is_collided():
    139.             terminal = True
    140.             reward = -1
    141.             self.__init__()
    142.  
    143.         # Draw everything
    144.         self.screen.blit(self.background_image, (0, 0))
    145.         self.screen.blit(self.base_image, (self.base_x, self.base_y))
    146.         self.screen.blit(self.bird_images[self.bird_index], (self.bird_x, self.bird_y))
    147.         for pipe in self.pipes:
    148.             self.screen.blit(self.pipe_images[0], (pipe["x_upper"], pipe["y_upper"]))
    149.             self.screen.blit(self.pipe_images[1], (pipe["x_lower"], pipe["y_lower"]))
    150.  
    151.  
    152.         image = array3d(display.get_surface())
    153.         display.update()
    154.         self.fps_clock.tick(self.fps)
    155.         return image, reward, terminal
    1. import argparse
    2. import torch
    3.  
    4. from src.deep_q_network import DeepQNetwork
    5. from src.flappy_bird import FlappyBird
    6. from src.utils import pre_processing
    7.  
    8.  
    9. def get_args():
    10.     parser = argparse.ArgumentParser(
    11.         """Implementation of Deep Q Network to play Flappy Bird""")
    12.     parser.add_argument("--image_size", type=int, default=84, help="The common width and height for all images")
    13.     parser.add_argument("--saved_path", type=str, default="trained_models")
    14.  
    15.     args = parser.parse_args()
    16.     return args
    17.  
    18.  
    19. def q_test(opt):
    20.     if torch.cuda.is_available():
    21.         torch.cuda.manual_seed(123)
    22.     else:
    23.         torch.manual_seed(123)
    24.     if torch.cuda.is_available():
    25.         model = torch.load("{}/flappy_bird".format(opt.saved_path))
    26.     else:
    27.         model = torch.load("{}/flappy_bird".format(opt.saved_path), map_location=lambda storage, loc: storage)
    28.     model.eval()
    29.     game_state = FlappyBird()
    30.     image, reward, terminal = game_state.next_frame(0)
    31.     image = pre_processing(image[:game_state.screen_width, :int(game_state.base_y)], opt.image_size, opt.image_size)
    32.     image = torch.from_numpy(image)
    33.     if torch.cuda.is_available():
    34.         model.cuda()
    35.         image = image.cuda()
    36.     state = torch.cat(tuple(image for _ in range(4)))[None, :, :, :]
    37.  
    38.     while True:
    39.         prediction = model(state)[0]
    40.         action = torch.argmax(prediction)
    41.  
    42.         next_image, reward, terminal = game_state.next_frame(action)
    43.         next_image = pre_processing(next_image[:game_state.screen_width, :int(game_state.base_y)], opt.image_size,
    44.                                     opt.image_size)
    45.         next_image = torch.from_numpy(next_image)
    46.         if torch.cuda.is_available():
    47.             next_image = next_image.cuda()
    48.         next_state = torch.cat((state[0, 1:, :, :], next_image))[None, :, :, :]
    49.  
    50.         state = next_state
    51.  
    52.  
    53. if __name__ == "__main__":
    54.     opt = get_args()
    55.     q_test(opt)
    1. def get_args():
    2.     parser = argparse.ArgumentParser(
    3.         """Implementation of Deep Q Network to play Flappy Bird""")
    4.     parser.add_argument("--image_size", type=int, default=84, help="The common width and height for all images")
    5.     parser.add_argument("--batch_size", type=int, default=32, help="The number of images per batch")
    6.     parser.add_argument("--optimizer", type=str, choices=["sgd", "adam"], default="adam")
    7.     parser.add_argument("--lr", type=float, default=1e-6)
    8.     parser.add_argument("--gamma", type=float, default=0.99)
    9.     parser.add_argument("--initial_epsilon", type=float, default=0.1)
    10.     parser.add_argument("--final_epsilon", type=float, default=1e-4)
    11.     parser.add_argument("--num_iters", type=int, default=2000000)
    12.     parser.add_argument("--replay_memory_size", type=int, default=50000,
    13.                         help="Number of epoches between testing phases")
    14.     parser.add_argument("--log_path", type=str, default="tensorboard")
    15.     parser.add_argument("--saved_path", type=str, default="trained_models")
    16.  
    17.     args = parser.parse_args()
    18.     return args
    19.  
    20.  
    21. def train(opt):
    22.     if torch.cuda.is_available():
    23.         torch.cuda.manual_seed(123)
    24.     else:
    25.         torch.manual_seed(123)
    26.     model = DeepQNetwork()
    27.     if os.path.isdir(opt.log_path):
    28.         shutil.rmtree(opt.log_path)
    29.     os.makedirs(opt.log_path)
    30.     writer = SummaryWriter(opt.log_path)
    31.     optimizer = torch.optim.Adam(model.parameters(), lr=opt.lr)
    32.     criterion = nn.MSELoss()
    33.     game_state = FlappyBird()
    34.     image, reward, terminal = game_state.next_frame(0)
    35.     image = pre_processing(image[:game_state.screen_width, :int(game_state.base_y)], opt.image_size, opt.image_size)
    36.     image = torch.from_numpy(image)
    37.     if torch.cuda.is_available():
    38.         model.cuda()
    39.         image = image.cuda()
    40.     state = torch.cat(tuple(image for _ in range(4)))[None, :, :, :]
    41.  
    42.     replay_memory = []
    43.     iter = 0
    44.     while iter < opt.num_iters:
    45.         prediction = model(state)[0]
    46.         # Exploration or exploitation
    47.         epsilon = opt.final_epsilon + (
    48.                 (opt.num_iters - iter) * (opt.initial_epsilon - opt.final_epsilon) / opt.num_iters)
    49.         u = random()
    50.         random_action = u <= epsilon
    51.         if random_action:
    52.             print("Perform a random action")
    53.             action = randint(0, 1)
    54.         else:
    55.  
    56.             action = torch.argmax(prediction)
    57.  
    58.         next_image, reward, terminal = game_state.next_frame(action)
    59.         next_image = pre_processing(next_image[:game_state.screen_width, :int(game_state.base_y)], opt.image_size,
    60.                                     opt.image_size)
    61.         next_image = torch.from_numpy(next_image)
    62.         if torch.cuda.is_available():
    63.             next_image = next_image.cuda()
    64.         next_state = torch.cat((state[0, 1:, :, :], next_image))[None, :, :, :]
    65.         replay_memory.append([state, action, reward, next_state, terminal])
    66.         if len(replay_memory) > opt.replay_memory_size:
    67.             del replay_memory[0]
    68.         batch = sample(replay_memory, min(len(replay_memory), opt.batch_size))
    69.         state_batch, action_batch, reward_batch, next_state_batch, terminal_batch = zip(*batch)
    70.  
    71.         state_batch = torch.cat(tuple(state for state in state_batch))
    72.         action_batch = torch.from_numpy(
    73.             np.array([[1, 0] if action == 0 else [0, 1] for action in action_batch], dtype=np.float32))
    74.         reward_batch = torch.from_numpy(np.array(reward_batch, dtype=np.float32)[:, None])
    75.         next_state_batch = torch.cat(tuple(state for state in next_state_batch))
    76.  
    77.         if torch.cuda.is_available():
    78.             state_batch = state_batch.cuda()
    79.             action_batch = action_batch.cuda()
    80.             reward_batch = reward_batch.cuda()
    81.             next_state_batch = next_state_batch.cuda()
    82.         current_prediction_batch = model(state_batch)
    83.         next_prediction_batch = model(next_state_batch)
    84.  
    85.         y_batch = torch.cat(
    86.             tuple(reward if terminal else reward + opt.gamma * torch.max(prediction) for reward, terminal, prediction in
    87.                   zip(reward_batch, terminal_batch, next_prediction_batch)))
    88.  
    89.         q_value = torch.sum(current_prediction_batch * action_batch, dim=1)
    90.         optimizer.zero_grad()
    91.         # y_batch = y_batch.detach()
    92.         loss = criterion(q_value, y_batch)
    93.         loss.backward()
    94.         optimizer.step()
    95.  
    96.         state = next_state
    97.         iter += 1
    98.         print("Iteration: {}/{}, Action: {}, Loss: {}, Epsilon {}, Reward: {}, Q-value: {}".format(
    99.             iter + 1,
    100.             opt.num_iters,
    101.             action,
    102.             loss,
    103.             epsilon, reward, torch.max(prediction)))
    104.         writer.add_scalar('Train/Loss', loss, iter)
    105.         writer.add_scalar('Train/Epsilon', epsilon, iter)
    106.         writer.add_scalar('Train/Reward', reward, iter)
    107.         writer.add_scalar('Train/Q-value', torch.max(prediction), iter)
    108.         if (iter+1) % 1000000 == 0:
    109.             torch.save(model, "{}/flappy_bird_{}".format(opt.saved_path, iter+1))
    110.     torch.save(model, "{}/flappy_bird".format(opt.saved_path))
    111.  
    112.  
    113. if __name__ == "__main__":
    114.     opt = get_args()
    115.     train(opt)

  • 相关阅读:
    DP27 跳跃游戏(二)
    PPT基础：表格
    区块链的生成与基本操作
    ​力扣解法汇总1374-生成每种字符都是奇数个的字符串
    如何在Mac上启用蓝牙，这里提供几个方法
    没有英语要求的中国人大女王金融硕士有多香你可能还不知道
    Redis 持久化之RDB操作
    大型Web网站高并发架构方案
    EFK部署centos7.9（三）Kibana部署
    DC综合基本概念：set_compile_directives
  • 原文地址：https://blog.csdn.net/timberman666/article/details/132590406