<jupyter_start><jupyter_text>STEP 1 - Making Basic PySC2 Agent<jupyter_code>%load_ext autoreload
%autoreload 2<jupyter_output><empty_output><jupyter_text>0. Runnning 'Agent code' on jupyter notebook<jupyter_code># unfortunately, PySC2 uses Abseil, which treats python code as if its run like an app
# This does not play well with jupyter notebook
# So we will need to monkeypatch sys.argv


import sys
#sys.argv = ["python", "--map", "AbyssalReef"]
sys.argv = ["python", "--map", "Simple64"]

# Copyright 2017 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Run an agent."""

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import importlib
import threading

from absl import app
from absl import flags
from future.builtins import range  # pylint: disable=redefined-builtin

from pysc2 import maps
from pysc2.env import available_actions_printer
from pysc2.env import run_loop
from pysc2.env import sc2_env
from pysc2.lib import point_flag
from pysc2.lib import stopwatch

FLAGS = flags.FLAGS

# because of Abseil's horrible design for running code underneath Colabs
# We have to pull out this ugly hack from the hat
if "flags_defined" not in globals():
    flags.DEFINE_bool("render", True, "Whether to render with pygame.")
    point_flag.DEFINE_point("feature_screen_size", "84",
                            "Resolution for screen feature layers.")
    point_flag.DEFINE_point("feature_minimap_size", "64",
                            "Resolution for minimap feature layers.")
    point_flag.DEFINE_point("rgb_screen_size", None,
                            "Resolution for rendered screen.")
    point_flag.DEFINE_point("rgb_minimap_size", None,
                            "Resolution for rendered minimap.")
    flags.DEFINE_enum("action_space", None, sc2_env.ActionSpace._member_names_,  # pylint: disable=protected-access
                      "Which action space to use. Needed if you take both feature "
                      "and rgb observations.")
    flags.DEFINE_bool("use_feature_units", True,
                      "Whether to include feature units.")
    flags.DEFINE_bool("disable_fog", False, "Whether to disable Fog of War.")

    flags.DEFINE_integer("max_agent_steps", 0, "Total agent steps.")
    flags.DEFINE_integer("game_steps_per_episode", None, "Game steps per episode.")
    flags.DEFINE_integer("max_episodes", 0, "Total episodes.")
    flags.DEFINE_integer("step_mul", 8, "Game steps per agent step.")
    flags.DEFINE_float("fps", 22.4, "Frames per second to run the game.")

    #flags.DEFINE_string("agent", "sc2.agent.BasicAgent.ZergBasicAgent",
    #                    "Which agent to run, as a python path to an Agent class.")
    #flags.DEFINE_enum("agent_race", "zerg", sc2_env.Race._member_names_,  # pylint: disable=protected-access
    #                  "Agent 1's race.")
    flags.DEFINE_string("agent", "TerranBasicAgent",
                        "Which agent to run, as a python path to an Agent class.")
    flags.DEFINE_enum("agent_race", "terran", sc2_env.Race._member_names_,  # pylint: disable=protected-access
                      "Agent 1's race.")

    flags.DEFINE_string("agent2", "Bot", "Second agent, either Bot or agent class.")
    flags.DEFINE_enum("agent2_race", "random", sc2_env.Race._member_names_,  # pylint: disable=protected-access
                      "Agent 2's race.")
    flags.DEFINE_enum("difficulty", "very_easy", sc2_env.Difficulty._member_names_,  # pylint: disable=protected-access
                      "If agent2 is a built-in Bot, it's strength.")

    flags.DEFINE_bool("profile", False, "Whether to turn on code profiling.")
    flags.DEFINE_bool("trace", False, "Whether to trace the code execution.")
    flags.DEFINE_integer("parallel", 1, "How many instances to run in parallel.")

    flags.DEFINE_bool("save_replay", True, "Whether to save a replay at the end.")

    flags.DEFINE_string("map", None, "Name of a map to use.")
    flags.mark_flag_as_required("map")

flags_defined = True

def run_thread(agent_classes, players, map_name, visualize):
  """Run one thread worth of the environment with agents."""
  with sc2_env.SC2Env(
      map_name=map_name,
      players=players,
      agent_interface_format=sc2_env.parse_agent_interface_format(
          feature_screen=FLAGS.feature_screen_size,
          feature_minimap=FLAGS.feature_minimap_size,
          rgb_screen=FLAGS.rgb_screen_size,
          rgb_minimap=FLAGS.rgb_minimap_size,
          action_space=FLAGS.action_space,
          use_feature_units=FLAGS.use_feature_units),
      step_mul=FLAGS.step_mul,
      game_steps_per_episode=FLAGS.game_steps_per_episode,
      disable_fog=FLAGS.disable_fog,
      visualize=visualize) as env:
    env = available_actions_printer.AvailableActionsPrinter(env)
    agents = [agent_cls() for agent_cls in agent_classes]
    run_loop.run_loop(agents, env, FLAGS.max_agent_steps, FLAGS.max_episodes)
    if FLAGS.save_replay:
      env.save_replay(agent_classes[0].__name__)

def main(unused_argv):
  """Run an agent."""
  #stopwatch.sw.enabled = FLAGS.profile or FLAGS.trace
  #stopwatch.sw.trace = FLAGS.trace

  map_inst = maps.get(FLAGS.map)

  agent_classes = []
  players = []

  #agent_module, agent_name = FLAGS.agent.rsplit(".", 1)
  #agent_cls = getattr(importlib.import_module(agent_module), agent_name)
  #agent_classes.append(agent_cls)
  agent_classes.append(TerranBasicAgent)
  players.append(sc2_env.Agent(sc2_env.Race[FLAGS.agent_race]))

  if map_inst.players >= 2:
    if FLAGS.agent2 == "Bot":
      players.append(sc2_env.Bot(sc2_env.Race[FLAGS.agent2_race],
                                 sc2_env.Difficulty[FLAGS.difficulty]))
    else:
      agent_module, agent_name = FLAGS.agent2.rsplit(".", 1)
      agent_cls = getattr(importlib.import_module(agent_module), agent_name)
      agent_classes.append(agent_cls)
      players.append(sc2_env.Agent(sc2_env.Race[FLAGS.agent2_race]))

  threads = []
  for _ in range(FLAGS.parallel - 1):
    t = threading.Thread(target=run_thread,
                         args=(agent_classes, players, FLAGS.map, False))
    threads.append(t)
    t.start()

  run_thread(agent_classes, players, FLAGS.map, FLAGS.render)

  for t in threads:
    t.join()

  if FLAGS.profile:
    pass
    #print(stopwatch.sw)<jupyter_output>pygame 1.9.6
Hello from the pygame community. https://www.pygame.org/contribute.html<jupyter_text>1. Creating a PySC2 Agent<jupyter_code>from pysc2.agents import base_agent
from pysc2.env import sc2_env
from pysc2.lib import actions, features, units
from absl import app
import random
import time
class TerranBasicAgent(base_agent.BaseAgent):
  def __init__(self):
    super(TerranBasicAgent, self).__init__()

  def step(self, obs):
    super(TerranBasicAgent, self).step(obs)

    time.sleep(0.5)

    return actions.FUNCTIONS.no_op()<jupyter_output><empty_output><jupyter_text>[run code]<jupyter_code>if __name__ == "__main__":
  app.run(main)<jupyter_output><empty_output><jupyter_text>2. Controlling SCVs<jupyter_code>from pysc2.agents import base_agent
from pysc2.env import sc2_env
from pysc2.lib import actions, features, units
from absl import app
import random
import time
class TerranBasicAgent(base_agent.BaseAgent):
    def __init__(self):
        super(TerranBasicAgent, self).__init__()

        self.base_top_left = None
        self.supply_depot_built = False
        self.barracks_built = False

    def transformLocation(self, x, x_distance, y, y_distance):
        if not self.base_top_left:
            return [x - x_distance, y - y_distance]
        
        return [x + x_distance, y + y_distance]

    def getMeanLocation(self, unitList):
        sum_x = 0
        sum_y = 0
        for unit in unitList:
            sum_x += unit.x
            sum_y += unit.y
        mean_x = sum_x / len(unitList)
        mean_y = sum_y / len(unitList)
        
        return [mean_x, mean_y]
    
    def unit_type_is_selected(self, obs, unit_type):
        if (len(obs.observation.single_select) > 0 and
            obs.observation.single_select[0].unit_type == unit_type):
              return True

        if (len(obs.observation.multi_select) > 0 and
            obs.observation.multi_select[0].unit_type == unit_type):
              return True

        return False

    def get_units_by_type(self, obs, unit_type):
        return [unit for unit in obs.observation.feature_units
                if unit.unit_type == unit_type]

    def can_do(self, obs, action):
        return action in obs.observation.available_actions

    def step(self, obs):
        super(TerranBasicAgent, self).step(obs)
        
        time.sleep(0.5)
        
        if obs.first():
            self.base_top_left = None
            self.supply_depot_built = False
            self.barracks_built = False
            
            player_y, player_x = (obs.observation.feature_minimap.player_relative == features.PlayerRelative.SELF).nonzero()
            self.base_top_left = 1 if player_y.any() and player_y.mean() <= 31 else 0
            
        if not self.supply_depot_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_SupplyDepot_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 0, int(mean_y), 20)
                        self.supply_depot_built = True

                        return actions.FUNCTIONS.Build_SupplyDepot_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))
        elif not self.barracks_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_Barracks_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 20, int(mean_y), 0)
                        self.barracks_built = True

                        return actions.FUNCTIONS.Build_Barracks_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))
         
        return actions.FUNCTIONS.no_op()<jupyter_output><empty_output><jupyter_text>[run code]<jupyter_code>if __name__ == "__main__":
  app.run(main)<jupyter_output><empty_output><jupyter_text>3. Controlling Barracks<jupyter_code>from pysc2.agents import base_agent
from pysc2.env import sc2_env
from pysc2.lib import actions, features, units
from absl import app
import random
import time
class TerranBasicAgent(base_agent.BaseAgent):
    def __init__(self):
        super(TerranBasicAgent, self).__init__()
        
        self.base_top_left = None
        self.supply_depot_built = False
        self.barracks_built = False
        self.barracks_rallied = False

    def transformLocation(self, x, x_distance, y, y_distance):
        if not self.base_top_left:
            return [x - x_distance, y - y_distance]
        
        return [x + x_distance, y + y_distance]

    def getMeanLocation(self, unitList):
        sum_x = 0
        sum_y = 0
        for unit in unitList:
            sum_x += unit.x
            sum_y += unit.y
        mean_x = sum_x / len(unitList)
        mean_y = sum_y / len(unitList)
        
        return [mean_x, mean_y]
    
    def unit_type_is_selected(self, obs, unit_type):
        if (len(obs.observation.single_select) > 0 and
            obs.observation.single_select[0].unit_type == unit_type):
              return True

        if (len(obs.observation.multi_select) > 0 and
            obs.observation.multi_select[0].unit_type == unit_type):
              return True

        return False

    def get_units_by_type(self, obs, unit_type):
        return [unit for unit in obs.observation.feature_units
                if unit.unit_type == unit_type]

    def can_do(self, obs, action):
        return action in obs.observation.available_actions

    def step(self, obs):
        super(TerranBasicAgent, self).step(obs)
        
        time.sleep(0.5)
        
        if obs.first():
            self.base_top_left = None
            self.supply_depot_built = False
            self.barracks_built = False
            self.barracks_rallied = False
            
            player_y, player_x = (obs.observation.feature_minimap.player_relative == features.PlayerRelative.SELF).nonzero()
            self.base_top_left = 1 if player_y.any() and player_y.mean() <= 31 else 0
            
        if not self.supply_depot_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_SupplyDepot_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 0, int(mean_y), 20)
                        self.supply_depot_built = True

                        return actions.FUNCTIONS.Build_SupplyDepot_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))
        elif not self.barracks_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_Barracks_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 20, int(mean_y), 0)
                        self.barracks_built = True

                        return actions.FUNCTIONS.Build_Barracks_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))
        
        elif not self.barracks_rallied:
            if self.unit_type_is_selected(obs, units.Terran.Barracks):
                self.barracks_rallied = True
                
                if self.base_top_left:
                    return actions.FUNCTIONS.Rally_Units_minimap("now", [29, 21])
                else:
                    return actions.FUNCTIONS.Rally_Units_minimap("now", [29, 46])
            barracks = self.get_units_by_type(obs, units.Terran.Barracks)
            if len(barracks) > 0:
                barrack = random.choice(barracks)
                return actions.FUNCTIONS.select_point("select", (barrack.x,
                                                                          barrack.y))
        
        elif obs.observation.player.food_cap - obs.observation.player.food_used:
            if self.can_do(obs, actions.FUNCTIONS.Train_Marine_quick.id):
                return actions.FUNCTIONS.Train_Marine_quick("queued")
        
        return actions.FUNCTIONS.no_op()<jupyter_output><empty_output><jupyter_text>[run code]<jupyter_code>if __name__ == "__main__":
  app.run(main)<jupyter_output><empty_output><jupyter_text>4. Attack<jupyter_code>from pysc2.agents import base_agent
from pysc2.env import sc2_env
from pysc2.lib import actions, features, units
from absl import app
import random
import time
class TerranBasicAgent(base_agent.BaseAgent):
    def __init__(self):
        super(TerranBasicAgent, self).__init__()

        self.base_top_left = None  
        self.supply_depot_built = False
        self.barracks_built = False
        self.barracks_rallied = False
        self.army_rallied = False

    def transformLocation(self, x, x_distance, y, y_distance):
        if not self.base_top_left:
            return [x - x_distance, y - y_distance]

        return [x + x_distance, y + y_distance]
    
    def getMeanLocation(self, unitList):
        sum_x = 0
        sum_y = 0
        for unit in unitList:
            sum_x += unit.x
            sum_y += unit.y
        mean_x = sum_x / len(unitList)
        mean_y = sum_y / len(unitList)
        
        return [mean_x, mean_y]
    
    def unit_type_is_selected(self, obs, unit_type):
        if (len(obs.observation.single_select) > 0 and
            obs.observation.single_select[0].unit_type == unit_type):
              return True

        if (len(obs.observation.multi_select) > 0 and
            obs.observation.multi_select[0].unit_type == unit_type):
              return True

        return False

    def get_units_by_type(self, obs, unit_type):
        return [unit for unit in obs.observation.feature_units
                if unit.unit_type == unit_type]

    def can_do(self, obs, action):
        return action in obs.observation.available_actions

    def step(self, obs):
        super(TerranBasicAgent, self).step(obs)

        #time.sleep(0.5)

        if obs.first():
            self.base_top_left = None  
            self.supply_depot_built = False
            self.barracks_built = False
            self.barracks_rallied = False
            self.army_rallied = False
            
            player_y, player_x = (obs.observation.feature_minimap.player_relative == features.PlayerRelative.SELF).nonzero()
            self.base_top_left = 1 if player_y.any() and player_y.mean() <= 31 else 0

        if not self.supply_depot_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_SupplyDepot_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 0, int(mean_y), 20)
                        self.supply_depot_built = True

                        return actions.FUNCTIONS.Build_SupplyDepot_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))
        elif not self.barracks_built:
            if self.unit_type_is_selected(obs, units.Terran.SCV):
                if self.can_do(obs, actions.FUNCTIONS.Build_Barracks_screen.id):
                    ccs = self.get_units_by_type(obs, units.Terran.CommandCenter)
                    if len(ccs) > 0:
                        mean_x, mean_y = self.getMeanLocation(ccs)
                        target = self.transformLocation(int(mean_x), 20, int(mean_y), 0)
                        self.barracks_built = True

                        return actions.FUNCTIONS.Build_Barracks_screen("now", target)
            scvs = self.get_units_by_type(obs, units.Terran.SCV)
            if len(scvs) > 0:
                scv = random.choice(scvs)
                return actions.FUNCTIONS.select_point("select", (scv.x,
                                                                          scv.y))

        elif not self.barracks_rallied:
            if self.unit_type_is_selected(obs, units.Terran.Barracks):
                self.barracks_rallied = True

                if self.base_top_left:
                    return actions.FUNCTIONS.Rally_Units_minimap("now", [29, 21])
                else:
                    return actions.FUNCTIONS.Rally_Units_minimap("now", [29, 46])
            barracks = self.get_units_by_type(obs, units.Terran.Barracks)
            if len(barracks) > 0:
                barrack = random.choice(barracks)
                return actions.FUNCTIONS.select_point("select", (barrack.x,
                                                                          barrack.y))
        elif obs.observation.player.food_cap - obs.observation.player.food_used:
            if self.can_do(obs, actions.FUNCTIONS.Train_Marine_quick.id):
                return actions.FUNCTIONS.Train_Marine_quick("queued")
            
        elif not self.army_rallied:
            if self.can_do(obs, actions.FUNCTIONS.Attack_minimap.id):
                self.army_rallied = True
                
                if self.base_top_left:
                    return actions.FUNCTIONS.Attack_minimap("now", [39, 45])
                else:
                    return actions.FUNCTIONS.Attack_minimap("now", [21, 24])

            if self.can_do(obs, actions.FUNCTIONS.select_army.id):
                return actions.FUNCTIONS.select_army("select")

        return actions.FUNCTIONS.no_op()<jupyter_output><empty_output><jupyter_text>[run code]<jupyter_code>if __name__ == "__main__":
  app.run(main)<jupyter_output>I0625 22:35:10.589324 4359478720 sc_process.py:135] Launching SC2: /Applications/StarCraft II/Versions/Base80188/SC2.app/Contents/MacOS/SC2 -listen 127.0.0.1 -port 19243 -dataDir /Applications/StarCraft II/ -tempDir /var/folders/kl/h0d5qxj551x0d2y091w17l1h0000gn/T/sc-5wpehtuy/ -displayMode 0 -windowwidth 640 -windowheight 480 -windowx 50 -windowy 50
I0625 22:35:10.670936 4359478720 remote_controller.py:166] Connecting to: ws://127.0.0.1:19243/sc2api, attempt: 0, running: True
I0625 22:35:11.677583 4359478720 remote_controller.py:166] Connecting to: ws://127.0.0.1:19243/sc2api, attempt: 1, running: True
I0625 22:35:12.684371 4359478720 remote_controller.py:166] Connecting to: ws://127.0.0.1:19243/sc2api, attempt: 2, running: True
I0625 22:35:13.688807 4359478720 remote_controller.py:166] Connecting to: ws://127.0.0.1:19243/sc2api, attempt: 3, running: True
I0625 22:35:14.693933 4359478720 remote_controller.py:166] Connecting to: ws://127.0.0.1:19243/sc2api, attempt: 4, running: True
I06[...]
{'fraction_non_alphanumeric': 0.09521150955550783, 'fraction_numerical': 0.020227614343998283, 'mean_word_length': 2.3006378454996455, 'pattern_counts': {'":': 5, '<': 42, '<?xml version=': 0, '>': 60, 'https://': 1, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 25, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '576670', 'license': 'MIT', 'n_tokens_mistral': 7654, 'n_tokens_neox': 7164, 'n_words': 1504, 'path': 's10071/STEP1-making-basic-pysc2-agent.ipynb', 'repo_name': 'parksurk/skcc-drl-sc2-course-2020_1st', 'size': '23285'}