見出し画像

PCGRL OpenAI GYM Interface

以下の記事を参考に書いてます。

PCGRL OpenAI GYM Interface

1. PCGRL OpenAI GYM Interface

強化学習による「手続き型コンテンツ生成」(PCGRL)のための「OpenAI Gym環境」です。このフレームワークは、論文「 PCGRL: Procedural Content Generation via Reinforcement Learning」をカバーしています。

2. インストール

インストール方法は、次のとおりです。

(1) リポジトリをローカルマシンにクローン。

$ git clone https://github.com/amidos2006/gym-pcgrl.git

(2) Pythonの仮想環境(Anacondaなど)で「gym-pcgrlフォルダ」に移動し、以下のコマンドを入力。

$ pip install tensorflow==1.15
$ pip install stable-baselines
$ cd gym-pcgrl
$ pip install -e .

3. 使用方法

◎ 環境の命名規則
「PCGRL OpenAI GYM Interface」には複数の異なる「環境」があり、各環境は「問題」と「表現」の2つの部分で構成されています。

「環境」の命名規則は次のとおりです。

[問題名]-[表現名]-[バージョン]

◎ 環境一覧
「環境」の一覧を表示するには、次のコードを実行します。

from gym import envs
import gym_pcgrl

print([env.id for env in envs.registry.all() if "gym_pcgrl" in env.entry_point])
['binary-narrow-v0', 
'binary-narrowcast-v0', 
'binary-narrowmulti-v0', 
'binary-wide-v0', 
'binary-turtle-v0', 
'binary-turtlecast-v0', 
'ddave-narrow-v0', 
'ddave-narrowcast-v0', 
'ddave-narrowmulti-v0', 
'ddave-wide-v0', 
'ddave-turtle-v0', 
'ddave-turtlecast-v0', 
'mdungeon-narrow-v0', 
'mdungeon-narrowcast-v0', 
'mdungeon-narrowmulti-v0', 
'mdungeon-wide-v0', 
'mdungeon-turtle-v0', 
'mdungeon-turtlecast-v0', 
'sokoban-narrow-v0', 
'sokoban-narrowcast-v0', 
'sokoban-narrowmulti-v0', 
'sokoban-wide-v0', 
'sokoban-turtle-v0', 
'sokoban-turtlecast-v0', 
'zelda-narrow-v0', 
'zelda-narrowcast-v0', 
'zelda-narrowmulti-v0', 
'zelda-wide-v0', 
'zelda-turtle-v0', 
'zelda-turtlecast-v0']

◎ 環境の実行
OpenAI Gymインターフェースと同様の方法で、環境を実行できます。「PCGRL OpenAI GYM Interface」独自の追加インタフェースもあります。

「narrow」表現の「sokoban」環境を使用するコードは、次のとおりです。

import gym
import gym_pcgrl

env = gym.make('sokoban-narrow-v0')
obs = env.reset()
for t in range(1000):
  action = env.action_space.sample()
  obs, reward, done, info = env.step(action)
  env.render('human')
  if done:
    print("Episode finished after {} timesteps".format(t+1))
    break

画像2

4. 問題

問題」は、PCGRLを適用するゲーム環境です。
「PCGRL OpenAI GYM Interface」でサポートされている「問題」は、次のとおりです。

◎ binary
最長経路がしきい値以上の迷路を生成。

・タイル : 0:empty, 1:solid

◎ ddave
「Dangerous Dave」ライクなゲームのレベルの生成。

・タイル : 0:empty, 1:solid, 2:player, 3:exit, 4:diamonds, 5:trophy (出口の鍵), 6:spikes

◎ mdungeon
「MiniDungeons 1」ライクなゲームのレベルの生成。
プレイヤーは、クリア前に敵の50%を殺さなければならない。

・タイル : 0:empty, 1:solid, 2:player (最大体力5), 3:exit, 4:potion (体力2を回復), 5:treasure, 6:goblin (1ダメージ与える), 7:ogre (2ダメージ与える)

◎ sokoban
倉庫番のレベルの生成。

・タイル : 0:empty, 1:solid, 2:player, 3:crate (targetまで押す), 4: target (createの運び先)

◎ zelda
GVGAI版のゼルダのレベルの生成。
プレーヤーは、キーに到達してからドアに到達しなければならない。

・タイル : 0:empty, 1:solid, 2:player, 3:key (ドアに到達する前に取得), 4:door (ゴール), 5:bat (避けるべき敵), 6:scorpion (避けるべき敵), 7:spider (避けるべき敵)

5. 表現

表現」とは、手続き型コンテンツ生成問題を強化学習に利用できるように「マルコフ決定過程」で表現したものです。すべての問題はサポートされている「表現」を使って表現することができます。
「PCGRL OpenAI GYM Interface」でサポートされている「表現」は、次のとおりです。

◎ narrow
ランダムなXY座標に移動し、タイル番号を変更。
・観察空間 :
 0: タイル番号の2次元配列
 1: 現在のXY座標
・行動空間 :
 0: タイル番号 or 変更なし

◎ narrowcast
ランダムなXY座標に移動し、タイル番号を変更。
・観察空間 :
 0: タイル番号の2次元配列
 1: 現在のXY座標
・行動空間 :
 0: 変更種別(変更なし, 単一, 3x3)
 1: タイル番号

◎ narrowmulti
ランダムなXY座標に移動し、タイル番号を変更。
・観察空間 :
 0: タイル番号の2次元配列
 1: 現在のXY座標
・行動空間 :
 0: タイル番号 or 変更なし
   :
 8: タイル番号 or 変更なし

◎ wide
任意のXY座標に移動し、タイル番号を変更。
・観察空間 : 
 0: タイル番号の2次元配列
・行動空間 :
 0: X座標
 1: Y座標
 2: タイル番号

◎ turtle
1マス移動し、タイル番号を変更。
・観察空間 :
 0: タイル番号の2次元配列
 1: 現在のXY座標
・行動空間 :
 0: 移動方向(上下左右) or タイル番号

◎ turtlecast
1マス移動し、タイル番号を変更。
・観察空間 :
 0: タイル番号の2次元配列
 1: 現在のXY座標
・行動空間 :
 0: 移動方向(上下左右)or 変更種別(単一, 3x3)
 1: タイル番号

「narrow」「wide」「turtle」の表現は、Bhaumikらによる「Tree Search vs Optimization Approaches for Map Generation」から採用しています。

6. 独自の問題の作成

gym_pcgrl.envs.probsに新しい「問題クラス」を作成し、gym_pcgrl.envs.probs.problemの「Problemクラス」を継承します。

◎ オーバーライド関数
「問題クラス」は、次の関数を実装する必要があります。

# 初期化
def __init__(self):
  super().__init__()
  ...

# タイル名の取得
def get_tile_types(self):
  ...

# マップの取得
def get_stats(self, map):
  ...

# 報酬の取得
def get_reward(self, new_stats, old_stats):
  ...

# エピソード完了かどうかの取得
def get_episode_over(self, new_stats, old_stats):
  ...

# デバッグ情報の取得
def get_debug_info(self, new_stats, old_stats):
  ...

◎ コンストラクタ
コンストラクタでは、次のパラメータを設定する必要があります。設定しない場合は、「Problemクラス」の設定が適用されます。

self._width = 9 # 幅
self._height = 9 # 高さ
tiles = self.get_tile_types() # [タイル名,...]のリスト
self._prob = [] # [タイル名:初期化時の表示確率]の辞書
for _ in range(len(tiles)):
   self._prob.append(1.0/len(tiles))

self._border_size = 1 # ボーダーサイズ
self._border_tile = tiles[0] # ボーダータイル名
self._tile_size=16 # タイルサイズ
self._graphics = None [タイル名:画像]の辞書
すべての問題において、レンダリングしない限り、システムがグラフィックをロードしないようにしてください。render()をオーバーライドして、コンストラクタの代わりにrender()の先頭でself._graphicsを初期化します。__init__.pyの辞書「gym_pcgrl.envs.probs.PROBLEMS」に問題名とクラスを追加する必要があります。

7. 独自の表現の作成

gym_pcgrl.envs.repsに新しい「表現クラス」を作成し、gym_pcgrl.envs.reps.representationの「Representationクラス」を継承します。

◎ オーバーライド関数
「表現クラス」は、次の関数を実装する必要があります。

# 初期化
def __init__(self, width, height, prob):
  super().__init__(width, height, prob)
  ...

# 行動空間の取得
def get_action_space(self):
  ...

# 観察空間の取得
def get_observation_space(self):
  ...

# 観察の取得
def get_observation(self):
  ...

# 更新時に呼ばれる
def update(self, action):
  ...
  # 変更が発生したことを示すbool値と、変更が発生した場所を示すxy座標
  return change, x, y

通常の動作とは異なる動作が必要な場合は、他の関数をオーバーライドしてください。

8. train.pyの実行

マップの生成の学習の実行手順は、次のとおりです。

(1) 「runsフォルダ」がない場合は生成。

$ mkdir runs

(2) 「train.py」内のパラメータを設定。

game = 'binary' # 問題
representation = 'narrow' # 表現
experiment = None # 実験
steps = 1e8 # ステップ数
render = False # レンダリング
logging = True # ログ出力
n_cpu = 50 # CPU数
kwargs = { # 環境に渡すパラメータ
   'resume': False
}

(3) 以下のコマンドを実行。

$ python train.py

「runsフォルダ」には、「TensorBoardログ」と「モデル」が出力されます。

9. inference.pyの実行

学習済みエージェントを実行し、マップを生成する手順は次のとおりです。

(1) 「inference.py」内のパラメータを設定。

game = 'binary' # 問題
representation = 'narrow' # 表現
model_path = 'models/{}/{}/model_1.pkl'.format(game, representation) # モデルパス
kwargs = { # 環境に渡すパラメータ
   'change_percentage': 0.4,
   'trials': 1,
   'verbose': True
}

(2) 以下のコマンドを実行。

$ python inference.py

学習済みエージェントを実行し、マップが生成されます。

画像1

【注意】Sokoban NarrowとTurtleのすべてのモデル、およびZelda Turtleの3番目のモデルは、python 3.6および3.7とは異なるシリアル化方法を持つpython 3.5を使用して保存しているため、python 3.6または3.7にロードしようとすると、不明なopコードエラーが発生します。

【おまけ1】  「binary」の問題クラスの実装例

(1) 「問題クラス」は「Problemクラス」を継承。

import os
import numpy as np
from PIL import Image
from gym_pcgrl.envs.probs.problem import Problem
from gym_pcgrl.envs.helper import get_range_reward, get_tile_locations, calc_num_regions, calc_longest_path

class BinaryProblem(Problem):

(2) コンストラクタの実装。
「6. 独自の問題の作成」で紹介したパラメータと、この問題独自のパラメータ(_target_path, _random_probs, _rewards)を定義します。

    """
    コンストラクタ
    """
    def __init__(self):
        super().__init__()
        self._width = 14 # 幅
        self._height = 14 # 高さ
        self._prob = {"empty": 0.5, "solid":0.5}  # [タイル名:初期化時の表示確率]の辞書
        self._border_tile = "solid" # ボーダータイル名

        self._target_path = 20 # エピソード完了とする最大経路のしきい値
        self._random_probs = True # self._probのランダム化

        self._rewards = { # 報酬の重み
            "regions": 5,
            "path-length": 1
        }

(3) タイル名のリストを取得するget_tile_types()の実装。
「binary」のタイル名は、「empty」(空)と「solid」(固)です。

    """
    タイル名のリストの取得

    戻り値:
        list: タイル名のリスト
    """
    def get_tile_types(self):
        return ["empty", "solid"]

(4) パラメータの調整を行うadjust_param()の拡張。
この問題独自のパラメータも調整できるように拡張します。

    """
    パラメータの調整
   
    パラメータ:
        width (int): 幅
        height (int): 高さ
        probs (dict(string, float)): [タイル名:初期化時の表示確率]の辞書
        target_path (int): エピソード完了とする最大経路のしきい値
        rewards (dict(string,float)): 報酬の重み
    """
    def adjust_param(self, **kwargs):
        super().adjust_param(**kwargs)

        self._target_path = kwargs.get('target_path', self._target_path)
        self._random_probs = kwargs.get('random_probs', self._random_probs)

        rewards = kwargs.get('rewards')
        if rewards is not None:
            for t in rewards:
                if t in self._rewards:
                    self._rewards[t] = rewards[t]

(5) 問題のリセットを行うreset()の拡張。
self._probのランダム化を行うように拡張します。

    """
    問題のリセット

    パラメータ:
        start_stats (dict(string,any)): 最初のマップ統計
    """
    def reset(self, start_stats):
        super().reset(start_stats)
        if self._random_probs:
            self._prob["empty"] = self._random.random()
            self._prob["solid"] = 1 - self._prob["empty"]

(6) マップ統計を取得するget_stats()の実装。
マップ統計には、「reigons」(空領域数)と「path-length」(最長経路)を含みます。

    """
    現在のマップ統計の取得

    戻り値:
        dict(string,any): 現在のマップ統計
            reigons: 空領域数(隣接する空は1つにカウント)
            path-length: 最長経路
    """
    def get_stats(self, map):
        map_locations = get_tile_locations(map, self.get_tile_types())
        return {
            "regions": calc_num_regions(map, map_locations, ["empty"]),
            "path-length": calc_longest_path(map, map_locations, ["empty"])
        }

(5) 報酬を取得するget_reward()の実装。
2つのマップ統計の差分で報酬を計算します。空領域数が減るほど、最長経路が増えるほど、報酬が大きくなります。空領域数と最長経路の報酬のバランスは報酬の重み「_rewards」で調整しています。

    """
    報酬の取得

    パラメータ:
        new_stats (dict(string,any)): 行動後のマップ統計
        old_stats (dict(string,any)): 行動前のマップ統計
    戻り値:
        float: 報酬
    """
    def get_reward(self, new_stats, old_stats):
        # 空領域数と最長経路の報酬の計算
        rewards = {
            "regions": get_range_reward(new_stats["regions"], old_stats["regions"], 1, 1),
            "path-length": get_range_reward(new_stats["path-length"],old_stats["path-length"], np.inf, np.inf)
        }
        # 総報酬の計算
        return rewards["regions"] * self._rewards["regions"] + rewards["path-length"] * self._rewards["path-length"]

(6) エピソード完了を取得するget_episode_over()の実装。
空領域数が1かつ、最長経路がしきい値以上増えた時に、エピソード完了になります。

    """
    エピソード完了の取得

    パラメータ:
        new_stats (dict(string,any)): 行動後のマップ統計
        old_stats (dict(string,any)): 行動前のマップ統計

    戻り値:
        boolean: エピソード完了
    """
    def get_episode_over(self, new_stats, old_stats):
        return new_stats["regions"] == 1 and new_stats["path-length"] - self._start_stats["path-length"] >= self._target_path

(7) デバッグ情報を取得するget_debug_info()の実装。
デバッグ情報には、「regions」(空領域数)「path-length」(最長経路)「path-imp」(最長経路の増加数)を含みます。

    """
    デバッグ情報の取得

    パラメータ:
        new_stats (dict(string,any)): 行動後のマップ統計
        old_stats (dict(string,any)): 行動前のマップ統計

    戻り値:
        dict(any,any): デバッグ情報
    """
    def get_debug_info(self, new_stats, old_stats):
        return {
            "regions": new_stats["regions"],
            "path-length": new_stats["path-length"],
            "path-imp": new_stats["path-length"] - self._start_stats["path-length"]
        }

(8) レンダリング画像を取得するrender()の実装。
_graphicsが空の時、[タイル名 : Pillow画像]の辞書を生成する必要があります。引数のmapを親のrender()に渡すことで、レンダリング画像を生成できます。

    """
    レンダリング画像の取得

    パラメータ:
        map (string[][]): 現在のマップ

    戻り値:
        Image: マップ画像(Pillow)
    """
    def render(self, map):
        if self._graphics == None:
            self._graphics = {
                "empty": Image.open(os.path.dirname(__file__) + "/binary/empty.png").convert('RGBA'),
                "solid": Image.open(os.path.dirname(__file__) + "/binary/solid.png").convert('RGBA')
            }
        return super().render(map)

【おまけ2】  「narrow」の表現クラスの実装例

(1) 「表現クラス」は「Representationクラス」を継承。

from gym_pcgrl.envs.reps.representation import Representation
from PIL import Image
from gym import spaces
import numpy as np
from collections import OrderedDict

class NarrowRepresentation(Representation):

(2) コンストラクタの実装。
この問題独自のパラメータ(_random_tile)を定義します。

    """
    コンストラクタ
    """
    def __init__(self):
        super().__init__()
        self._random_tile = True # タイル間の移動(ランダム or 順次)

(3) 表現のリセットを行うreset()の拡張。
XY座標のリセットを拡張しています。

    """
    リセット

    パラメータ:
        width (int): 幅
        height (int): 高さ
        prob (dict(int,float)): [タイル名:初期化時の表示確率]の辞書
    """
    def reset(self, width, height, prob):
        super().reset(width, height, prob)
        self._x = self._random.randint(width)
        self._y = self._random.randint(height)

(4) 行動空間を取得するget_action_space()の実装。
行動空間の型は「Discrete」になります。

    """
    行動空間の取得

    パラメータ:
        width: 幅
        height: 高さ
        num_tiles: タイル数

    戻り値:
        Discrete: 行動空間
    """
    def get_action_space(self, width, height, num_tiles):
        return spaces.Discrete(num_tiles + 1)

(5) 観察空間を取得するget_observation_space()の実装。
観察空間の型は辞書で、「pos」(現在の場所のXY座標)と「map」(タイル番号の2次元配列)を含みます。

    """
    観察空間の取得

    パラメータ:
        width: 幅
        height: 高さ
        num_tiles: タイル数

    戻り値:
        Dict: 観察空間の辞書
             pos: 現在の場所のXY座標。
             map:  タイル番号の2次元配列
    """
    def get_observation_space(self, width, height, num_tiles):
        return spaces.Dict({
            "pos": spaces.Box(low=np.array([0, 0]), high=np.array([width-1, height-1]), dtype=np.uint8),
            "map": spaces.Box(low=0, high=num_tiles-1, dtype=np.uint8, shape=(height, width))
        })

(6) 現在の観察を取得するget_observationの実装。

    """
    現在の観察の取得

    Returns:
        observation: 観察の辞書
             pos: 現在の場所のXY座標。
             map:  タイル番号の2次元配列
    """
    def get_observation(self):
        return OrderedDict({
            "pos": np.array([self._x, self._y], dtype=np.uint8),
            "map": self._map.copy()
        })

(7) パラメータの調整を行うadjust_param()の拡張。
この表現独自のパラメータも調整できるように拡張します。

    """
   パラメータの調整

   パラメータ:
       random_start (boolean): リセット時のマップ(新規(true) or 以前(false))
       random_tile (boolean): タイル間の移動(ランダム(true) or 順次(false))
   """
   def adjust_param(self, **kwargs):
       super().adjust_param(**kwargs)
       self._random_tile = kwargs.get('random_tile', self._random_tile)

(8) 行動で表現を更新するの実装。
行動に応じて、タイルの変更とXY座標の移動を行います。

    """
    行動で表現を更新

    パラメータ:
        action: 行動

    戻り値:
        boolean: マップ変更(あり(true) or なし(false)), 現在のX座標, 現在のY座標
    """
    def update(self, action):
        change = 0

        # タイルの変更
        if action > 0:
            change += [0,1][self._map[self._y][self._x] != action-1]
            self._map[self._y][self._x] = action-1

        # XY座標の移動
        if self._random_tile:
            self._x = self._random.randint(self._map.shape[1])
            self._y = self._random.randint(self._map.shape[0])
        else:
            self._x += 1
            if self._x >= self._map.shape[1]:
                self._x = 0
                self._y += 1
                if self._y >= self._map.shape[0]:
                    self._y = 0
        return change, self._x, self._y

(9) レンダリングするrender()の実装。
変更タイルの周りに赤い長方形を描画します。

    """
    変更タイルの周りに赤い長方形を描画

    パラメータ:
        lvl_image (img): 現在のレンダリング画像
        tile_size (int): lvl_imageで使用されるピクセル単位のタイルサイズ
        border_size (int): ボーダーがレベルの一部でない場合のタイルオフセット

    Returns:
        img: レンダリング画像
    """
    def render(self, lvl_image, tile_size, border_size):
        x_graphics = Image.new("RGBA", (tile_size,tile_size), (0,0,0,0))
        for x in range(tile_size):
            x_graphics.putpixel((0,x),(255,0,0,255))
            x_graphics.putpixel((1,x),(255,0,0,255))
            x_graphics.putpixel((tile_size-2,x),(255,0,0,255))
            x_graphics.putpixel((tile_size-1,x),(255,0,0,255))
        for y in range(tile_size):
            x_graphics.putpixel((y,0),(255,0,0,255))
            x_graphics.putpixel((y,1),(255,0,0,255))
            x_graphics.putpixel((y,tile_size-2),(255,0,0,255))
            x_graphics.putpixel((y,tile_size-1),(255,0,0,255))
        lvl_image.paste(x_graphics, ((self._x+border_size)*tile_size, (self._y+border_size)*tile_size,
            (self._x+border_size+1)*tile_size,(self._y+border_size+1)*tile_size), x_graphics)
        return lvl_image



この記事が気に入ったらサポートをしてみませんか?