• 用深度强化学习来玩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)

  • 相关阅读:
    Linux 驱动开发 六十六：多点触控(MT)协议
    对SPA的理解、对 vue组件化的理解
    （数据科学学习手札137）orjson：Python中最好用的json库
    mac openssl 版本到底怎么回事 已解决
    公钥加密如何确保数据的完整性
    一座 “数智桥梁”，华为助力“天堑变通途”
    数据分析--数据预处理
    DETRs Beat YOLOs on Real-time Object Detection
    如何使用IDE端通义灵码
    sql：SQL优化知识点记录（十二）
  • 原文地址：https://blog.csdn.net/timberman666/article/details/132590406