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
|
