Initial commit

Author: okrusch

Protokoll.txt

@@ -0,0 +1,6 @@
+Gruppe: 
+
+Bearbeitet von:
+Aufgabe 1 (Markov Decision Process) ->
+Aufgabe 2 (Basic Agent) ->
+Aufgabe 3 (Q-Learning) ->

agents/AbstractAgent.py

@@ -0,0 +1,85 @@
+from abc import abstractmethod
+
+import numpy as np
+from pysc2.lib import actions, features
+
+
+class AbstractAgent:
+    _DIRECTIONS = {'N': [0, -1],
+                   'NE': [1, -1],
+                   'E': [1, 0],
+                   'SE': [1, 1],
+                   'S': [0, 1],
+                   'SW': [-1, 1],
+                   'W': [-1, 0],
+                   'NW': [-1, -1]}
+
+    _DIRECTIONS_REVERSE = {"[0, -1]": 'N',
+                           "[1, -1]": 'NE',
+                           "[1, 0]": 'E',
+                           "[1, 1]": 'SE',
+                           "[0, 1]": 'S',
+                           "[-1, 1]": 'SW',
+                           "[-1, 0]": 'W',
+                           "[-1, -1]": 'NW'}
+
+    """Sc2 Actions"""
+    _MOVE_SCREEN = actions.FUNCTIONS.Move_screen
+    _NO_OP = actions.FUNCTIONS.no_op()
+    _SELECT_ARMY = actions.FUNCTIONS.select_army("select")
+
+    def __init__(self, screen_size):
+        self.screen_size = screen_size
+
+    @abstractmethod
+    def step(self, obs):
+        ...
+
+    @abstractmethod
+    def save_model(self, path):
+        ...
+
+    @abstractmethod
+    def load_model(self, path):
+        ...
+
+    def _get_beacon(self, obs):
+        """Returns the unit obj representation of the beacon"""
+        beacon = next(unit for unit in obs.observation.feature_units
+                      if unit.alliance == features.PlayerRelative.NEUTRAL)
+        return beacon
+
+    def _get_marine(self, obs):
+        """Returns the unit obj representation of the marine"""
+        marine = next(unit for unit in obs.observation.feature_units
+                      if unit.alliance == features.PlayerRelative.SELF)
+        return marine
+
+    def _get_unit_pos(self, unit):
+        """Returns the (x, y) position of a unit obj"""
+        return np.array([unit.x, unit.y])
+
+    def _dir_to_sc2_action(self, d, marine_center):
+        """Takes the direction the marine should walk and outputs an action for PySC2"""
+
+        def _xy_offset(start, offset_x, offset_y):
+            """Return point (x', y') offset from start.
+               Pays attention to not set the point off beyond the screen border"""
+            dest = start + np.array([offset_x, offset_y])
+            if dest[0] < 0:
+                dest[0] = 0
+            elif dest[0] >= self.screen_size:
+                dest[0] = self.screen_size - 1
+            if dest[1] < 0:
+                dest[1] = 0
+            elif dest[1] >= self.screen_size:
+                dest[1] = self.screen_size - 1
+            return dest
+
+        if d in self._DIRECTIONS.keys():
+            next_pos = _xy_offset(marine_center,
+                                  self.screen_size * self._DIRECTIONS[d][0],
+                                  self.screen_size * self._DIRECTIONS[d][1])
+            return self._MOVE_SCREEN("now", next_pos)
+        else:
+            return self._NO_OP

agents/BasicAgent.py

@@ -0,0 +1,43 @@
+from agents.AbstractAgent import AbstractAgent
+
+
+class BasicAgent(AbstractAgent):
+    def __init__(self, train, screen_size):
+        super(BasicAgent, self).__init__(screen_size)
+
+    def step(self, obs):
+        if self._MOVE_SCREEN.id in obs.observation.available_actions:
+            marine = self._get_marine(obs)
+            if marine is None:
+                return self._NO_OP
+            marine_coordinates = self._get_unit_pos(marine)
+            marine_x, marine_y = marine_coordinates[0], marine_coordinates[1]
+
+            beacon = self._get_beacon(obs)
+            if beacon is None:
+                return self._NO_OP
+            beacon_coordinates = self._get_unit_pos(beacon)
+            beacon_x, beacon_y = beacon_coordinates[0], beacon_coordinates[1]
+
+            move = ""
+            if(marine_y < beacon_y):
+                move += "S"
+            elif (marine_y > beacon_y):
+                move += "N"
+
+            if(marine_x < beacon_x):
+                move += "E"
+            elif(marine_x > beacon_x):
+                move += "W"
+
+            assert move != ""
+
+            return self._dir_to_sc2_action(move, marine_coordinates)
+        else:
+            return self._SELECT_ARMY
+
+    def save_model(self, filename):
+        pass
+
+    def load_model(self, filename):
+        pass

agents/QLearningAgent.py

@@ -0,0 +1,101 @@
+from agents.AbstractAgent import AbstractAgent
+import pandas as pd
+import numpy as np
+
+def get_row_index_in_string_format(state):
+    """
+        Returns a state (row index) as q-table row index string.
+
+                Parameters:
+                        state ([int, int]): The distance between agent (marine) and beacon.
+
+                Returns:
+                        state (str): Transformed state, so it can be used as index in the q-table.
+    """
+    return "("+str(state[0]) + "," + str(state[1]) + ")"
+
+
+class QLearningAgent(AbstractAgent):
+    def __init__(self, train, screen_size, explore=1):
+        super(QLearningAgent, self).__init__(screen_size)
+        # TODO Initialize all hyperparameter and the q-table (with the helper function below)
+        self.train = train
+        self.explore = explore
+        self.actions = self._DIRECTIONS.keys()
+        self.states = []
+        for x in range(-64, 65):
+            for y in range(-64, 65):
+                self.states.append((x, y))
+        self.q_table = self.init_q_table()
+
+        pass
+
+
+    def step(self, obs):
+        # TODO step method
+        if self._MOVE_SCREEN.id in obs.observation.available_actions:
+            pass
+        else:
+            return self._SELECT_ARMY    # initialize army in first step
+
+    def save_model(self, path):
+        self.q_table.to_pickle(path)
+
+    def load_model(self, path):
+        self.q_table = pd.read_pickle(path)
+
+    def get_new_action(self, state):
+        """
+            Returns the action to execute.
+
+                    Parameters:
+                            state ([int, int]): A row index (a state) of the q-table.
+
+                    Returns:
+                            action (str): e.g. 'N', 'NW', 'NO', ...
+        """
+        # TODO get_new_action method
+        ...
+
+    def get_q_value(self, q_table_column_index, q_table_row_index):
+        """
+            Returns a q-value.
+
+                    Parameters:
+                            q_table_column_index (str): The column index of the searched value (the action).
+                            q_table_row_index (str): The row index of the searched value (the state).
+
+                    Returns:
+                            action (float): The value for the given indices.
+        """
+        # TODO get_new_action method
+        ...
+
+    def update_q_value(self, old_state, old_action, new_state, reward, terminal):
+        # TODO update_q_value method
+        ...
+
+    def get_q_state_from_position(self, marine_position, beacon_position):
+        """
+            Transforms the position of agent (marine) and beacon into a q-table row index.
+
+                    Parameters:
+                            marine_position ([int, int]): The position of the agent (marine).
+                            beacon_position ([int, int]): The position of the beacon.
+
+                    Returns:
+                            state ([int, int]): A row index (a state) of the q-table.
+        """
+        # TODO get_q_state_from_position method
+        ...
+
+    def init_q_table(self):
+        """
+            Initializes the q-table
+
+                    Returns:
+                            q_table (panda.Dataframe): The q-table.
+                                                       The row indices must be in the format '(x,y)'
+                                                       The column indices must be in the format 'action' (e.g. 'W')
+        """
+        return pd.DataFrame(np.random.rand(len(self.states), len(self.actions)), index=self.states, columns=self.actions)

agents/RandomAgent.py

@@ -0,0 +1,26 @@
+from agents.AbstractAgent import AbstractAgent
+import numpy as np
+
+
+class RandomAgent(AbstractAgent):
+
+    def __init__(self, train, screen_size):
+        super(RandomAgent, self).__init__(screen_size)
+
+    def step(self, obs):
+        if self._MOVE_SCREEN.id in obs.observation.available_actions:
+            marine = self._get_marine(obs)
+            if marine is None:
+                return self._NO_OP
+            marine_coordinates = self._get_unit_pos(marine)
+
+            d = np.random.choice(list(self._DIRECTIONS.keys()))
+            return self._dir_to_sc2_action(d, marine_coordinates)
+        else:
+            return self._SELECT_ARMY
+
+    def save_model(self, filename):
+        pass
+
+    def load_model(self, filename):
+        pass

env.py

@@ -0,0 +1,26 @@
+from pysc2.env import sc2_env
+from pysc2.lib import features
+
+
+class Env:
+    def __init__(self, screen_size=32, minimap_size=32, visualize=False):
+        self.sc2_env = sc2_env.SC2Env(
+            map_name="MoveToBeacon",
+            players=[sc2_env.Agent(sc2_env.Race.terran)],
+            agent_interface_format=features.AgentInterfaceFormat(
+                feature_dimensions=features.Dimensions(screen=screen_size, minimap=minimap_size),
+                use_feature_units=True
+            ),
+            step_mul=8,
+            visualize=visualize
+        )
+
+    def reset(self):
+        return self.preprocess_obs(self.sc2_env.reset())
+
+    def step(self, action):
+        return self.preprocess_obs(self.sc2_env.step([action]))
+
+    def preprocess_obs(self, timsteps):
+        # Any kind of preprocessing can take place here
+        return timsteps[0]

runScripts/graphs/BasicAgent/240425_1854/run/events.out.tfevents.1714064067.Leonhards-MacBook-Pro.local.5957.0

@@ -0,0 +1,39 @@
+from absl import app
+
+from env import Env
+from runners.basic_runner import Runner
+from agents.BasicAgent import BasicAgent
+
+_CONFIG = dict(
+    episodes=100,
+    screen_size=64,
+    minimap_size=64,
+    visualize=True,
+    train=False,
+    agent=BasicAgent
+)
+
+def main(unused_argv):
+
+    agent = _CONFIG['agent'](
+        train=_CONFIG['train'],
+        screen_size=_CONFIG['screen_size']
+    )
+
+    env = Env(
+        screen_size=_CONFIG['screen_size'],
+        minimap_size=_CONFIG['minimap_size'],
+        visualize=_CONFIG['visualize']
+    )
+
+    runner = Runner(
+        agent=agent,
+        env=env,
+        train=_CONFIG['train']
+    )
+
+    runner.run(episodes=_CONFIG['episodes'])
+
+
+if __name__ == "__main__":
+    app.run(main)

runScripts/graphs/BasicAgent/240425_1856/run/events.out.tfevents.1714064216.Leonhards-MacBook-Pro.local.5991.0

@@ -0,0 +1,40 @@
+# TODO Write the runQLAgent script
+from absl import app
+
+from env import Env
+from runners.basic_runner import Runner
+from agents.QLearningAgent import QLearningAgent
+
+_CONFIG = dict(
+    episodes=100,
+    screen_size=64,
+    minimap_size=64,
+    visualize=True,
+    train=False,
+    agent=QLearningAgent
+)
+
+def main(unused_argv):
+
+    agent = _CONFIG['agent'](
+        train=_CONFIG['train'],
+        screen_size=_CONFIG['screen_size']
+    )
+
+    env = Env(
+        screen_size=_CONFIG['screen_size'],
+        minimap_size=_CONFIG['minimap_size'],
+        visualize=_CONFIG['visualize']
+    )
+
+    runner = Runner(
+        agent=agent,
+        env=env,
+        train=_CONFIG['train']
+    )
+
+    runner.run(episodes=_CONFIG['episodes'])
+
+
+if __name__ == "__main__":
+    app.run(main)