Gym Retro - シューティングゲームを実行する② 学習編

Gym Retro - シューティングゲームを実行する② 学習編

前回はシューティングゲームをランダムで実行しましたが、今回は強化学習アルゴリズムを実装してゲームを攻略してみます。

baselinesインストール

baselinesがインストールされていない場合は、下記コマンドを実行してインストールしておきます。

1
2
3
git clone https://github.com/openai/baselines.git
cd baselines
pip install -e .

Airstrikerを学習して実行

Airstrikerを学習して実行するコードは次のようになります。

  • コールバック
    10更新ごとにベストモデルの保存を行います。
  • Airstrikerラッパー
    行動パターンを4096通りから3通りに減らして学習効率を上げています。
  • CustomRewardAndDoneラッパー
    報酬を20で割って範囲を「0~1.0」に近づけます。報酬は「0~1.0」の範囲に収めると、学習が安定する傾向があります。

[コード]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
import retro
import time
import os
import gym
import numpy as np
import datetime
import pytz
from stable_baselines.results_plotter import load_results, ts2xy
from stable_baselines.common.vec_env import DummyVecEnv
from stable_baselines import PPO2
from stable_baselines.common import set_global_seeds
from stable_baselines.bench import Monitor
from baselines.common.retro_wrappers import *
from util import log_dir, callback, AirstrikerDiscretizer, CustomRewardAndDoneEnv

# ログフォルダの生成
log_dir = './logs/'
os.makedirs(log_dir, exist_ok=True)

# コールバック
best_mean_reward = -np.inf
nupdates = 1
def callback(_locals, _globals):
global nupdates
global best_mean_reward

# 10更新毎
if (nupdates + 1) % 10 == 0:
# 平均エピソード長、平均報酬の取得
x, y = ts2xy(load_results(log_dir), 'timesteps')
if len(y) > 0:
# 直近10件の平均報酬
mean_reward = np.mean(y[-10:])

# 平均報酬がベスト報酬以上の時はモデルを保存
update_model = mean_reward > best_mean_reward
if update_model:
best_mean_reward = mean_reward
_locals['self'].save('airstriker_model')

# ログ
print('time: {}, nupdates: {}, mean: {:.2f}, best_mean: {:.2f}, model_update: {}'.format(
datetime.datetime.now(pytz.timezone('Asia/Tokyo')),
nupdates, mean_reward, best_mean_reward, update_model))
nupdates += 1
return True

# Airstrikerラッパー
class AirstrikerDiscretizer(gym.ActionWrapper):
# 初期化
def __init__(self, env):
super(AirstrikerDiscretizer, self).__init__(env)
buttons = ['B', 'A', 'MODE', 'START', 'UP', 'DOWN', 'LEFT', 'RIGHT', 'C', 'Y', 'X', 'Z']
actions = [['LEFT'], ['RIGHT'], ['B']]
self._actions = []
for action in actions:
arr = np.array([False] * 12)
for button in action:
arr[buttons.index(button)] = True
self._actions.append(arr)
self.action_space = gym.spaces.Discrete(len(self._actions))
# 行動の取得
def action(self, a):
return self._actions[a].copy()

# CustomRewardAndDoneラッパー
class CustomRewardAndDoneEnv(gym.Wrapper):
# 初期化
def __init__(self, env):
super(CustomRewardAndDoneEnv, self).__init__(env)
# ステップ
def step(self, action):
state, rew, done, info = self.env.step(action)
# 報酬の変更
rew /= 20
# エピソード完了の変更
if info['gameover'] == 1:
done = True
return state, rew, done, info

# 環境の生成
env = retro.make(game='Airstriker-Genesis', state='Level1')
env = AirstrikerDiscretizer(env) # 行動空間を離散空間に変換
env = CustomRewardAndDoneEnv(env) # 報酬とエピソード完了の変更
env = StochasticFrameSkip(env, n=4, stickprob=0.25) # スティッキーフレームスキップ
env = Downsample(env, 2) # ダウンサンプリング
env = Rgb2gray(env) # グレースケール
env = FrameStack(env, 4) # フレームスタック
env = ScaledFloatFrame(env) # 状態の正規化
env = Monitor(env, log_dir, allow_early_resets=True)
print('行動空間: ', env.action_space)
print('状態空間: ', env.observation_space)

# シードの指定
env.seed(0)
set_global_seeds(0)

# ベクトル化環境の作成
env = DummyVecEnv([lambda: env])

# モデルの作成
model = PPO2('CnnPolicy', env, verbose=0)

# モデルの読み込み
#model = PPO2.load('airstriker_model', env=env, verbose=0)

# モデルの学習
model.learn(total_timesteps=128000, callback=callback)

# モデルのテスト
state = env.reset()
total_reward = 0
while True:
# 環境の描画
env.render()
# スリープ
time.sleep(1/60)
# モデルの推論
action, _ = model.predict(state)
# 1ステップ実行
state, reward, done, info = env.step(action)
total_reward += reward[0]
# エピソード完了
if done:
print('reward:', total_reward)
state = env.reset()
total_reward = 0

上記コードを実行すると最初に学習処理が実行されます。(私の環境ではGPUなしで60分以上かかりました。)

それが完了されると下記のようなウィンドウが表示され、学習したモデルでゲームが実行されます。

実行結果

ランダム行動よりまともにゲームができているようです。


Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×