一文实践强化学习训练游戏ai--doom枪战游戏实践

一文实践强化学习训练游戏ai–doom枪战游戏实践
上次文章写道下载doom的环境并尝试了简单的操作，这次让我们来进行对象化和训练、验证，如果你有基础，可以直接阅读本文，不然请你先阅读Doom基础知识，其中包含了下载、动作等等的基础知识。
本次与之前的马里奥训练不同，马里奥是已经有做好的step等函数的，而这个doom没有，但也因此我们可以更好的一窥训练的过程。
完整代码在最后，可以复制执行。

文章目录

一、训练模型
- 1、vizdoom_train类
- - 1）__init__
  - 2）step
  - 3）togray
  - 4）其他函数
- 2、保存模型函数
- 3、训练模型
二、成果验收
完整代码
- 训练代码
- 测试代码

一、训练模型

1、vizdoom_train类

这是我们的训练基类，由于我们想用openai gym环境，因此我们需要手写此环境必须的__init__、step等函数

1）init

在这个函数中，我们要定义训练的观察空间（即游戏图像）、动作空间（即ai可以执行的操作）和一些基础设置。

        VizDoom_basic_cfg = r"C:/Users/tttiger/Desktop/ViZDoom-master/ViZDoom-master/scenarios/basic.cfg"
        self.game = vizdoom.DoomGame()
        self.game.load_config(VizDoom_basic_cfg)

        if render == False: 
            self.game.set_window_visible(False)
        else:
            self.game.set_window_visible(True)
        self.game.init()

此处，我们先指定游戏文件的位置，然后加一个判断，决定是否允许游戏窗口显示出来，最后初始化游戏。
初始化后，我们就要规定观察空间和动作空间了

        self.observation_space = Box(low=0, high=255, shape=(100,160,1), dtype=np.uint8) 
        self.action_space = Discrete(3)

观察空间是我们游戏的界面，这是一个灰化后的图像，所以维度为1，大小为100*160
动作空间为离散空间3，即可以选取动作0、1、2，我们只需要在后续的代码中指定数字指代的动作就可以了。

2）step

step函数非常关键，这是ai执行动作的时候会调用的函数。
首先，我们定义我们的动作

        actions = np.identity(3,dtype=np.uint8)
        reward = self.game.make_action(actions[action], 4)

这部分的详细解释在Doom基础知识中，事实上就是定义一个矩阵，调用游戏文件中的左移、右移和射击。
接下来，我们要获取当前的状态，比如得分，游戏图像等，这样我们才可以训练。

        try:
            state = self.game.get_state()
            img = state.screen_buffer
            img = self.togray(img)
            info = state.game_variables[0]
        except:
            img = np.zeros(self.observation_space.shape)
            info = 0 
        finally:
            info = {"info":info}
            done = self.game.is_episode_finished()
        #img_show(img)
        return img,reward,done,info

使用try，是因为gameover时有些内容获取不到，为了防止程序因此暂停，用try。最后，我们要把info变成字典形式，这是因为openai gym环境时这么要求的。为了方便理解，这里可以调用imgshow，查看目前的图像，其实现如下。

def img_show(img):
    plt.imshow(img)
    plt.show()
    time.sleep(5)

3）togray

灰度化图像，我们知道，彩色图像时由rgb三个颜色矩阵组成，但这么大的数据量给我们的训练增添的很多负担，于是我们采用灰度图。同时，我们缩小图像，这样可以训练的更快。

    def togray(self,observation):
        gray = cv2.cvtColor(np.moveaxis(observation, 0, -1), cv2.COLOR_BGR2GRAY)
        resize = cv2.resize(gray, (160,100), interpolation=cv2.INTER_CUBIC)
        state = np.reshape(resize, (100,160,1))
        return state

observation 是一个 NumPy 数组，通常表示图像数据。
np.moveaxis 是 NumPy 库中的一个函数，用于重新排列数组的轴。
参数 0 表示将第0轴（通常是颜色通道）移动到新位置的最后一个轴位置（即 -1）。
例如，如果 observation 的形状是 (C, H, W)（即颜色通道在第一个维度），经过 np.moveaxis 后，形状将变为 (H, W, C)，这对于 OpenCV 处理图像更为常见，因为 OpenCV 期望颜色通道是图像的最后一个维度。

cv2.cvtColor 是 OpenCV 中用于颜色空间转换的函数。
第一个参数是输入图像（在这里是经过 np.moveaxis 处理后的图像）。
第二个参数 cv2.COLOR_BGR2GRAY 指定了将图像从 BGR 颜色空间转换为灰度图像。

4）其他函数

我们要定义一个关闭游戏的函数close

   def close(self):
        self.game.close()

以及一个reset函数，用于在结束一个游戏后，重置状态，继续下一轮训练

    def reset(self):
        state = self.game.new_episode()
        state = self.game.get_state()
        return self.togray(state.screen_buffer)

至此，我们已经成功地把这个独立游戏包装成了可以使用openai gym的环境的游戏。

2、保存模型函数

class TrainAndLoggingCallback(BaseCallback):

    def __init__(self, check_freq, save_path, verbose=1):
        super(TrainAndLoggingCallback, self).__init__(verbose)
        self.check_freq = check_freq
        self.save_path = save_path

    def _init_callback(self):
        if self.save_path is not None:
            os.makedirs(self.save_path, exist_ok=True)

    def _on_step(self):
        if self.n_calls % self.check_freq == 0:
            model_path = os.path.join(self.save_path, 'best_model_{}'.format(self.n_calls))
            self.model.save(model_path)

        return True

我们使用这段代码来存档数据，这部分复制即可

3、训练模型

首先，我们指定训练结果保存的路径

    CHECKPOINT_DIR = './train/train_basic'
    LOG_DIR = './logs/log_basic'
    callback = TrainAndLoggingCallback(check_freq=10000, save_path=CHECKPOINT_DIR)    ```

然后我们调用训练函数进行训练

    env = vizdoom_train(render=False)
    model = PPO('CnnPolicy', env, tensorboard_log=LOG_DIR, verbose=1, learning_rate=0.0001, n_steps=2048)
    model.learn(total_timesteps=100000, callback=callback)

这里我们使用PPO这个强化学习算法。

经过几十分钟的等待，就得到训练好的模型了

二、成果验收

训练好模型后，我们要使用模型，看看效果。
首先，我们载入训练好的模型

model = PPO.load('./train/train_basic/best_model_100000')

然后测试以下模型的平均得分

mean_reward, _ = evaluate_policy(model, env, n_eval_episodes=5)

这样还不够直观，我们让ai玩给我们看


for episode in range(100): 
    obs = env.reset()
    done = False
    total_reward = 0
    while not done: 
        action, _ = model.predict(obs)
        obs, reward, done, info = env.step(action)
        time.sleep(0.20)
        total_reward += reward
    print('Total Reward for episode {} is {}'.format(total_reward, episode))
    time.sleep(2)

我们首先让ai模型预测，然后将预测的动作输入step函数，然后展示页面。

如图所示，平均得分非常高，基本能快速索敌，然后一枪秒杀。
至此，我们完成了ai的训练。

完整代码

训练代码

from gym import Env
from gym.spaces import Discrete,Box
import cv2
from vizdoom import *
import vizdoom
import random
import time
import numpy as np
#DIS离散空间 作用类似于random
#box用来装游戏图像
from matplotlib import pyplot as plt
class vizdoom_train(Env):
    def __init__(self, render=True):
        super().__init__()
        VizDoom_basic_cfg = r"C:/Users/tttiger/Desktop/ViZDoom-master/ViZDoom-master/scenarios/basic.cfg"
        self.game = vizdoom.DoomGame()
        self.game.load_config(VizDoom_basic_cfg)

        if render == False: 
            self.game.set_window_visible(False)
        else:
            self.game.set_window_visible(True)
        self.game.init()

        self.observation_space = Box(low=0, high=255, shape=(100,160,1), dtype=np.uint8) 
        self.action_space = Discrete(3)
    def step(self,action):
        actions = np.identity(3,dtype=np.uint8)
        reward = self.game.make_action(actions[action], 4) 
        try:
            state = self.game.get_state()
            img = state.screen_buffer
            img = self.togray(img)
            info = state.game_variables[0]
        except:
            img = np.zeros(self.observation_space.shape)
            info = 0 
        finally:
            info = {"info":info}
            done = self.game.is_episode_finished()
        #img_show(img)
        return img,reward,done,info
    
    def close(self):
        self.game.close()

    def reset(self):
        state = self.game.new_episode()
        state = self.game.get_state()
        return self.togray(state.screen_buffer)
    
    def togray(self,observation):
        gray = cv2.cvtColor(np.moveaxis(observation, 0, -1), cv2.COLOR_BGR2GRAY)
        resize = cv2.resize(gray, (160,100), interpolation=cv2.INTER_CUBIC)
        state = np.reshape(resize, (100,160,1))
        return state
    
def img_show(img):
    plt.imshow(img)
    plt.show()
    time.sleep(5)


# Import os for file nav
import os 
# Import callback class from sb3
from stable_baselines3.common.callbacks import BaseCallback
class TrainAndLoggingCallback(BaseCallback):

    def __init__(self, check_freq, save_path, verbose=1):
        super(TrainAndLoggingCallback, self).__init__(verbose)
        self.check_freq = check_freq
        self.save_path = save_path

    def _init_callback(self):
        if self.save_path is not None:
            os.makedirs(self.save_path, exist_ok=True)

    def _on_step(self):
        if self.n_calls % self.check_freq == 0:
            model_path = os.path.join(self.save_path, 'best_model_{}'.format(self.n_calls))
            self.model.save(model_path)

        return True
    
if __name__ == "__main__":

    CHECKPOINT_DIR = './train/train_basic'
    LOG_DIR = './logs/log_basic'
    callback = TrainAndLoggingCallback(check_freq=10000, save_path=CHECKPOINT_DIR)    

    train = vizdoom_train()

    from stable_baselines3 import PPO

    env = vizdoom_train(render=False)
    model = PPO('CnnPolicy', env, tensorboard_log=LOG_DIR, verbose=1, learning_rate=0.0001, n_steps=2048)
    model.learn(total_timesteps=100000, callback=callback)

测试代码

from stable_baselines3.common.evaluation import evaluate_policy
from stable_baselines3 import PPO
import time
model = PPO.load('./train/train_basic/best_model_100000')

from second_gym import vizdoom_train
env = vizdoom_train(render=True)

mean_reward, _ ,_= evaluate_policy(model, env, n_eval_episodes=5)

print(mean_reward)

for episode in range(100): 
    obs = env.reset()
    done = False
    total_reward = 0
    while not done: 
        action, _ = model.predict(obs)
        obs, reward, done, info = env.step(action)
        time.sleep(0.20)
        total_reward += reward
    print('Total Reward for episode {} is {}'.format(total_reward, episode))
    time.sleep(2)