万字详述 | 全开源：python写小游戏+AI强化学习与传统DFS/BFS控制分别实现

amazing_brick / amazing_brick_utils.py              / wrapped_amazing_brick.pykeyboard_play.py

class Play:    def __init__(self):        self.x = ...        self.y = ...        self.x_= ...        self.y_= ...        # 如果你觉得游戏太难的话，可以改变这些物理参数        self.gravity = 0.35        self.dragForce = 0.01        self.velX = 0        self.velY = 0        self.AccX = 4.5        self.AccY = 2.5    def lFlap(self):        # 按下左边按钮时，玩家获得一个向左上的力        # 因此速度发生改变        self.velX -= self.AccX        self.velY -= (self.AccY - self.gravity)    def rFlap(self):        # 按下右边按钮时，玩家获得一个向右上的力        # 因此速度发生改变        self.velX += self.AccX        self.velY -= (self.AccY - self.gravity)    def noneDo(self):        # 没有按按钮        # 玩家因为横向空气阻力而减缓横向速度        # 此外，还因为重力向下加速        if self.velX > 0:            self.velX -= self.dragForce        elif self.velX < 0:            self.velX += self.dragForce        self.velY += self.gravity

class GameState:    def __init__(self, ifRender=True, fps=30):        ...    def frame_step(self, action):        ...        if action == 0:            self.player.noneDo()        elif action == 1:            self.player.lFlap()        elif action == 2:            self.player.rFlap()        ...        # player's movement        self.player.x += self.player.velX        self.player.x_ += self.player.velX        self.player.y += self.player.velY        self.player.y_ += self.player.velY

class ScreenCamera:    def __init__(self):        self.x = 0        self.y = 0        self.width = CONST['SCREEN_WIDTH']        self.height = CONST['SCREEN_HEIGHT']        self.x_ = self.x + self.width        self.y_ = self.y + self.height    def __call__(self, obj: Box):        # output the obj's (x, y) on screen        x_c = obj.x - self.x        y_c = obj.y - self.y        # 每个实体：玩家、障碍物都有一套相对坐标，即 x_c, y_c        # obj.set_camera(x_c, y_c) 将其在屏幕上的新位置告诉它        # 绘图时，就根据其 x_c, y_c 来将其绘制在屏幕上        obj.set_camera(x_c, y_c)        return obj    def move(self, obj: Player):        # 如果玩家此时在屏幕上的坐标将高于屏幕的 1/2        # 镜头上移        # 即不允许玩家跑到屏幕上半部分去        self(obj)        if obj.y_c < self.height / 2:            self.y -= (self.height / 2 - obj.y_c)        else:            pass

class GameState:    def __init__(self, ifRender=True, fps=30):        ...    def frame_step(self, action):        ...        # check for score        playerMidPos = self.s_c(self.player).y_c + self.player.height / 2        for ind, pipe in enumerate(self.pipes):            if ind % 2 == 1:                continue            self.s_c(pipe)            # 判断 Y 轴是否处于间隙中央            if pipe.y_c <= playerMidPos <= pipe.y_c + pipe.height:                if not pipe.scored:                    self.score += 1                    # 不能在一个间隙中得两次分                    pipe.scored = True                    # reward 用于强化学习                    reward = 1

class GameState:    def __init__(self, ifRender=True, fps=30):        ...        self.pipes = []        self.blocks = []    def frame_step(self, action):        ...        # 判断是否新增障碍物        low_pipe = self.pipes[0]        if self.s_c(low_pipe).y_c >= self.s_c.height - low_pipe.width \                and len(self.pipes) < 6:            # 满足条件，新增障碍物            self._getRandomPipe()        # 如果条形障碍物超出屏幕，则删除        if self.s_c(low_pipe).y_c >= self.s_c.height \                and len(self.pipes) > 4:            self.pipes.pop(0)            self.pipes.pop(0)        # 如果块状障碍物超出屏幕，则删除        for block in self.blocks:            self.s_c(block)            x_flag = - CONST['BLOCK_WIDTH'] <= block.x_c <= self.s_c.width            y_flag = block.y_c >= self.s_c.height

class GameState:    ...    def _getRandomPipe(self, init=False):        if self.score % 5 == 4:            self.color_ind = (self.color_ind + 1) % 5        gap_left_topXs = list(range(100, 190, 20))        if init:            index = random.randint(0, len(gap_left_topXs)-1)            x = gap_left_topXs[index]            y = CONST['SCREEN_HEIGHT'] / 2 - CONST['PIPE_WIDTH'] / 2            first_pipes = pipes(x, y, self.color_ind)            self.pipes.append(first_pipes[0])            self.pipes.append(first_pipes[1])            self._addBlocks()        index = random.randint(0, len(gap_left_topXs)-1)        x = self.s_c.x + gap_left_topXs[index]        y = self.pipes[-1].y - CONST['SCREEN_HEIGHT'] / 2        pipe = pipes(x, y, self.color_ind)        self.pipes.append(pipe[0])        self.pipes.append(pipe[1])        self._addBlocks()    def _addBlocks(self):        x = (self.pipes[-2].x_ + self.pipes[-1].x) / 2        y = (self.pipes[-2].y + self.pipes[-2].y_) / 2        for i in range(2, 0, -1):            y_block = y + i * CONST['BLOCK_SPACE']            x_block = x + np.random.normal() * CONST['PIPE_GAPSIZE'] / 2.5            block = Block(x_block, y_block, self.color_ind)            self.blocks.append(block)

final_s_a_list = []def dfs_forward(root_state, show=False):    # 最后需要返回的就是这个(状态、动作)列表    global final_s_a_list    final_s_a_list = []    # 在内部定义 dfs ，用于递归    # 在递归过程中，修改 final_s_a_list 的值    # 总是保留目前最优解    def dfs(state, s_a_list):        global final_s_a_list        # a trick        # 每次结点的排列都不一样        # 这样搜索速度更快        # 能更快地找到可行解        if len(s_a_list) % 2 == 1:            ACTIONS_tmp = (2, 1)        else:            ACTIONS_tmp = (1, 2)        for action in ACTIONS_tmp:            if len(final_s_a_list) > 0:                break            new_state = move_forward(state, action)            new_s_a_list = s_a_list.copy()            new_s_a_list.append((new_state, action))            if check_crash(new_state):                if show:                    # 绘图部分                    pygame.draw.rect(SCREEN, (255, 0, 0), \                            (new_state['x'] - game_state.s_c.x, new_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                    pygame.display.update()                del new_state                del new_s_a_list            else:                if show:                    # 绘图部分                    pygame.draw.rect(SCREEN, (100, 100, 100), \                            (new_state['x'] - game_state.s_c.x, new_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                    pygame.display.update()                if check_for_score(new_state):                    if show:                        # 绘图部分                        pygame.draw.rect(SCREEN, (0, 0, 255), \                                (new_state['x'] - game_state.s_c.x, new_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                        pygame.display.update()                    final_s_a_list = new_s_a_list                    break                dfs(new_state, new_s_a_list)    # 开始递归    dfs(root_state, [])    return final_s_a_list

python dfs_play.py

python dfs_play.py --display

Node = namedtuple("Node", ['sta' , 'act', 'father'])game_state = GameState(True)# 为了避免搜索空间过大# 这里调高了游戏的力学参数game_state.player.velMaxY = 20game_state.player.AccY = 5ACTIONS = (0, 1, 2)def bfs_forward(root_state, show=False):    # 保存结点的队列    q = Queue()    for action in ACTIONS:        node = Node(root_state.copy(), action, None)        q.put(node)    while True:        # 如果队列为空        # 则说明所有可行结点已经遍历        if q.empty():            break        father_node = q.get()        father_state = father_node.sta        if check_for_score(father_state):            # 如果得分            # 说明可行路径已经找到            # 跳出循环            if show:                pygame.draw.rect(SCREEN, (0, 0, 255), \                        (father_state['x'] - game_state.s_c.x, father_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                pygame.display.update()            break        # 只考虑动作 1 和 2        for action in ACTIONS[1:]:            # father_state = move_forward(father_state, ACTIONS[0])            new_state = move_forward(father_state, action)            if check_crash(new_state):                if show:                    pygame.draw.rect(SCREEN, (255, 0, 0), \                            (new_state['x'] - game_state.s_c.x, new_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                    pygame.display.update()            else:                if show:                    pygame.draw.rect(SCREEN, (100, 100, 100), \                            (new_state['x'] - game_state.s_c.x, new_state['y'] - game_state.s_c.y, game_state.player.width, game_state.player.height))                    pygame.display.update()                node = Node(new_state, action, father_node)                q.put(node)    return father_node

python bfs_play.py

python bfs_play.py --display

DQN_train/gym_warpper.pyDQN_train/dqn_train3.pyDQN_train/dqn_render3.py

tianshoupytorch > 1.40gym

python DQN_train/dqn_train3.py

python DQN_train/dqn_render3.py 3

python DQN_train/dqn_render3.py 25

python DQN_train/dqn_render3.py 30

class Net(nn.Module):    def __init__(self):        super().__init__()        self.fc1 = nn.Linear(16, 128)        self.fc2 = nn.Linear(128, 256)        self.fc3 = nn.Linear(256, 128)        self.fc4 = nn.Linear(128, 3)    def forward(self, obs, state=None, info={}):        if not isinstance(obs, torch.Tensor):            obs = torch.tensor(obs, dtype=torch.float)        x = F.relu(self.fc1(obs))        x = F.relu(self.fc2(x))        x = F.relu(self.fc3(x))        x = self.fc4(x)        return x, state

dqn2_path = osp.join(path, 'DQN_train/dqn_weights/')if __name__ == '__main__':    try:        with open(dqn3_path + 'dqn3_log.json', 'r') as f:            jlist = json.load(f)        log_dict = jlist[-1]        round = log_dict['round']        policy.load_state_dict(torch.load(dqn3_path + 'dqn3round_' + str(int(round)) + '.pth'))        del jlist    except FileNotFoundError as identifier:        print('\n\nWe shall train a bright new net.\n')        # 第一次训练时，新建一个 .json 文件        # 声明一个列表        # 以后每次写入 json 文件，向列表新增一个字典对象        with open(dqn3_path + 'dqn3_log.json', 'a+') as f:            f.write('[]')            round = 0    while True:        round += 1        print('\n\nround:{}\n\n'.format(round))        result = ts.trainer.offpolicy_trainer(            policy, train_collector, test_collector,            max_epoch=max_epoch, step_per_epoch=step_per_epoch,            collect_per_step=collect_per_step,            episode_per_test=30, batch_size=64,            # 如下，每新一轮训练才更新 epsilon            train_fn=lambda e: policy.set_eps(0.1 / round),            test_fn=lambda e: policy.set_eps(0.05 / round), writer=None)        print(f'Finished training! Use {result["duration"]}')        torch.save(policy.state_dict(), dqn3_path + 'dqn3round_' + str(int(round)) + '.pth')        policy.load_state_dict(torch.load(dqn3_path + 'dqn3round_' + str(int(round)) + '.pth'))        log_dict = {}        log_dict['round'] = round        log_dict['last_train_time'] = datetime.datetime.now().strftime('%y-%m-%d %I:%M:%S %p %a')        log_dict['best_reward'] = result['best_reward']        with open(dqn3_path + 'dqn3_log.json', 'r') as f:            """dqn3_log.json should be inited as []"""            jlist = json.load(f)        jlist.append(log_dict)        with open(dqn3_path + 'dqn3_log.json', 'w') as f:            json.dump(jlist, f)        del jlist

import os.path as ospimport sysdirname = osp.dirname(__file__)path = osp.join(dirname, '..')sys.path.append(path)from amazing_brick.game.wrapped_amazing_brick import GameStatefrom amazing_brick.game.amazing_brick_utils import CONSTfrom DQN_train.gym_wrapper import AmazingBrickEnv3import tianshou as tsimport torch, numpy as npfrom torch import nnimport torch.nn.functional as Fimport jsonimport datetimetrain_env = AmazingBrickEnv3()test_env = AmazingBrickEnv3()state_shape = 16action_shape = 1net = Net()optim = torch.optim.Adam(net.parameters(), lr=1e-3)'''args for rl'''estimation_step = 3max_epoch = 5step_per_epoch = 300collect_per_step = 50policy = ts.policy.DQNPolicy(net, optim,    discount_factor=0.9, estimation_step=estimation_step,    use_target_network=True, target_update_freq=320)train_collector = ts.data.Collector(policy, train_env, ts.data.ReplayBuffer(size=2000))test_collector = ts.data.Collector(policy, test_env)