天天看點

Deepmind星際争霸2平台使用第一輪-完成采礦

這篇博文簡單的介紹PySC2的基本使用以及分析一份網友提供的代碼(使用DQN來讓計算機玩星際2)。

##1-PySC2 ##

Deepmind公布的這段python與PySC2通信的源碼主要包含以下的幾個方面(原圖來于​​部落格​​):

Deepmind星際争霸2平台使用第一輪-完成采礦

這裡我們隻講解對于強化學習算法開發過程當中需要用到的一些内容:主要就是env

在env當中包裝了觀測量集合、動作集合、狀态推進功能、狀态重置功能等,它所具有的執行個體化參數如下原圖來于​​部落格​​:

Deepmind星際争霸2平台使用第一輪-完成采礦

觀測量集合

可以通過SC2Env的執行個體獲得,也可以利用step方法傳回的元組來獲得,内容都是以python當中的字典的方式進行組織的。

Deepmind星際争霸2平台使用第一輪-完成采礦

動作集合

有了觀測值,我們就具有了動作部分所需要的輸入,星際2當中的動作空間非常的大,就單純考慮其中的動作類型就具有524個動作函數:

python -m pysc2.bin.valid_vactions      
Deepmind星際争霸2平台使用第一輪-完成采礦

對于這些函數都是具有其函數ID的,我們可以通過FunctionCall來完成調用,傳回的是一個action。

狀态推進

既然有了動作值,那麼我們怎麼進一步的去推進遊戲程序呢?這裡使用的方法與Gym當中一樣,使用step方法就可以完成,傳回的内容是一個環境的字典(前面已經提到過了)。

狀态重置

星際争霸是一個沒有固定時限的測試環境,我們當然可以把時間設定成無窮大,讓遊戲結束的時候傳回,但往往為了效率,我們可以設定一個時間的長度,當到達這個時間長度以後對環境狀态進行reset即可。

2-挖礦代碼解讀

這是一個網友分享出來的代碼當中的一個部分,在這段代碼當中他使用了一個DQN網絡對星際争霸2當中的挖礦任務進行了測試:

#導入需要的子產品
import sys

from absl import flags
import baselines.common.tf_util as U
import numpy as np
from baselines import deepq
from pysc2.env import environment
from pysc2.env import sc2_env    #導入環境子產品
from pysc2.lib import actions
from pysc2.lib import actions as sc2_actions   #導入動作子產品
from pysc2.lib import features

import deepq_mineral_shards

#擷取螢幕上的個體資訊,其實就是一個矩陣,裡面0表示環境中可以移動的位置,1表示隊友,3表示礦産
_PLAYER_RELATIVE = features.SCREEN_FEATURES.player_relative.index
_PLAYER_FRIENDLY = 1
_PLAYER_NEUTRAL = 3  # beacon/minerals
_PLAYER_HOSTILE = 4
_NO_OP = actions.FUNCTIONS.no_op.id
#記錄相應動作函數的ID
_MOVE_SCREEN = actions.FUNCTIONS.Move_screen.id
_ATTACK_SCREEN = actions.FUNCTIONS.Attack_screen.id
_SELECT_ARMY = actions.FUNCTIONS.select_army.id
_NOT_QUEUED = [0]
_SELECT_ALL = [0]

step_mul = 16
steps = 400

FLAGS = flags.FLAGS


def main():
  FLAGS(sys.argv)
  #執行個體化環境
  with sc2_env.SC2Env(
      map_name="CollectMineralShards",
      step_mul=step_mul, #推進的速度,通俗了解就是人類玩家的每秒的有效操作
      visualize=True,  #是否可視化
      game_steps_per_episode=steps * step_mul #每輪的運作步長,None則表示沒有時間限制
      ) as env:

    #調用Baselines中的deepq,以後會再做講解
    model = deepq.models.cnn_to_mlp(
      convs=[(32, 8, 4), (64, 4, 2), (64, 3, 1)],
      hiddens=[256],
      dueling=True)

    def make_obs_ph(name):
      return U.BatchInput((64, 64), name=name)

    act_params = {
      'make_obs_ph': make_obs_ph,
      'q_func': model,
      'num_actions': 4,
    }
    #導入訓練好的模型參數
    act = deepq_mineral_shards.load(
      "mineral_shards.pkl", act_params=act_params)

    while True:
      #環境初始化
      obs = env.reset()
      episode_rew = 0

      done = False
      #這裡固定了第一步的操作:選擇所有的個體,一共就兩個農民,利用step執行該指令,獲得新的環境
      step_result = env.step(actions=[
        sc2_actions.FunctionCall(_SELECT_ARMY, [_SELECT_ALL])
      ])

      while not done:
        #檢視傳回的字典中螢幕中的目标關系分布圖:1表示着地圖中個體的位置,3表示着礦物的位置
        #這裡面因為一個個體可能占好幾個像素點,是以個體可能是有1構成塊狀體表示
        player_relative = step_result[0].observation["screen"][
          _PLAYER_RELATIVE]

        obs = player_relative
        #篩選裡面個體是盟軍的整列,變成了一個0-1矩陣,盟軍的位置表示為1
        player_y, player_x = (
          player_relative == _PLAYER_FRIENDLY).nonzero()
        #計算平均位置player_x表示各個1所在的位置x坐标集合,player_y則是y坐标上的集合
        #注意這裡是一個位置矩陣,其實就是行列坐标
        player = [int(player_x.mean()), int(player_y.mean())]
        #shift函數就是對螢幕視角進行中心化移動,然後将溢出矩陣的部分資料無效化(指派為2)
        #這裡是相當于DQN的輸入的設定
        if (player[0] > 32):
          obs = shift(LEFT, player[0] - 32, obs)
        elif (player[0] < 32):
          obs = shift(RIGHT, 32 - player[0], obs)

        if (player[1] > 32):
          obs = shift(UP, player[1] - 32, obs)
        elif (player[1] < 32):
          obs = shift(DOWN, 32 - player[1], obs)

        #将觀察情況輸入到Baselines中的act函數裡,獲得傳回的行為
        action = act(obs[None])[0]
        coord = [player[0], player[1]]
        #根據輸出的action讓個體在64*64的地圖内進行移動,移動到礦物所在的位置就可以進行采礦
        #這裡設定移動的步長都是16,這裡是DQN生成的行為的解碼
        if (action == 0):  #UP

          if (player[1] >= 16):
            coord = [player[0], player[1] - 16]
          elif (player[1] > 0):
            coord = [player[0], 0]

        elif (action == 1):  #DOWN

          if (player[1] <= 47):
            coord = [player[0], player[1] + 16]
          elif (player[1] > 47):
            coord = [player[0], 63]

        elif (action == 2):  #LEFT

          if (player[0] >= 16):
            coord = [player[0] - 16, player[1]]
          elif (player[0] < 16):
            coord = [0, player[1]]

        elif (action == 3):  #RIGHT

          if (player[0] <= 47):
            coord = [player[0] + 16, player[1]]
          elif (player[0] > 47):
            coord = [63, player[1]]

        #在這裡我們需要對DQN生成的行為轉化為星際争霸裡面的操作
        new_action = [
          sc2_actions.FunctionCall(_MOVE_SCREEN,
                                   [_NOT_QUEUED, coord])
        ]

        step_result = env.step(actions=new_action)
        #擷取遊戲回報的回報值,也就是礦物的采集情況
        rew = step_result[0].reward
        #判斷是否需要進入下一輪遊戲
        done = step_result[0].step_type == environment.StepType.LAST

        episode_rew += rew
      print("Episode reward", episode_rew)


UP, DOWN, LEFT, RIGHT = 'up', 'down', 'left', 'right'


def shift(direction, number, matrix):
  ''' shift given 2D matrix in-place the given number of rows or columns
    in the specified (UP, DOWN, LEFT, RIGHT) direction and return it
'''
  if direction in (UP):
    matrix = np.roll(matrix, -number, axis=0)
    matrix[number:, :] = -2
    return matrix
  elif direction in (DOWN):
    matrix = np.roll(matrix, number, axis=0)
    matrix[:number, :] = -2
    return matrix
  elif direction in (LEFT):
    matrix = np.roll(matrix, -number, axis=1)
    matrix[:, number:] = -2
    return matrix
  elif direction in (RIGHT):
    matrix = np.roll(matrix, number, axis=1)
    matrix[:, :number] = -2
    return matrix
  else:
    return matrix


if __name__ == '__main__':
  main()      

---------------------------------------------------------------------------------------------------