Merge pull request #267 from tweakimp/develop

Update library

sc2/bot_ai.py

@@ -16,6 +16,7 @@
 from .ids.ability_id import AbilityId
 from .ids.unit_typeid import UnitTypeId
 from .ids.upgrade_id import UpgradeId
+from .pixel_map import PixelMap
 from .position import Point2, Point3
 from .unit import Unit
 from .units import Units
@@ -106,7 +107,6 @@ def main_base_ramp(self) -> "Ramp":
         )
         return self.cached_main_base_ramp
 
-
     @property_cache_forever
     def expansion_locations(self) -> Dict[Point2, Units]:
         """
@@ -183,7 +183,6 @@ def expansion_locations(self) -> Dict[Point2, Units]:
             centers[result] = resources
         return centers
 
-
     def _correct_zerg_supply(self):
         """ The client incorrectly rounds zerg supply down instead of up (see
             https://github.com/Blizzard/s2client-proto/issues/123), so self.supply_used
@@ -713,9 +712,13 @@ def _prepare_first_step(self):
             self._game_info.player_start_location = self.townhalls.first.position
         self._game_info.map_ramps = self._game_info._find_ramps()
 
-    def _prepare_step(self, state):
-        """Set attributes from new state before on_step."""
+    def _prepare_step(self, state, proto_game_info):
+        # Set attributes from new state before on_step."""
         self.state: GameState = state  # See game_state.py
+        # update pathing grid
+        self._game_info.pathing_grid: PixelMap = PixelMap(
+            proto_game_info.game_info.start_raw.pathing_grid, in_bits=True, mirrored=False
+        )
         # Required for events
         self._units_previous_map: Dict = {unit.tag: unit for unit in self.units}
         self.units: Units = state.own_units

sc2/client.py

@@ -1,5 +1,4 @@
 import logging
-import pickle
 from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union
 
 from s2clientprotocol import common_pb2 as common_pb
@@ -41,7 +40,9 @@ def in_game(self):
         return self._status == Status.in_game
 
     async def join_game(self, name=None, race=None, observed_player_id=None, portconfig=None, rgb_render_config=None):
-        ifopts = sc_pb.InterfaceOptions(raw=True, score=True, show_cloaked=True, raw_affects_selection=False, raw_crop_to_playable_area=False)
+        ifopts = sc_pb.InterfaceOptions(
+            raw=True, score=True, show_cloaked=True, raw_affects_selection=False, raw_crop_to_playable_area=False
+        )
 
         if rgb_render_config:
             assert isinstance(rgb_render_config, dict)

sc2/constants.py

@@ -27,3 +27,69 @@
     PURIFIERVESPENEGEYSER.value,
     SHAKURASVESPENEGEYSER.value,
 }
+transforming = {
+    # terran structures
+    BARRACKS: LAND_BARRACKS,
+    BARRACKSFLYING: LAND_BARRACKS,
+    COMMANDCENTER: LAND_COMMANDCENTER,
+    COMMANDCENTERFLYING: LAND_COMMANDCENTER,
+    ORBITALCOMMAND: LAND_ORBITALCOMMAND,
+    ORBITALCOMMANDFLYING: LAND_ORBITALCOMMAND,
+    FACTORY: LAND_FACTORY,
+    FACTORYFLYING: LAND_FACTORY,
+    STARPORT: LAND_STARPORT,
+    STARPORTFLYING: LAND_STARPORT,
+    SUPPLYDEPOT: MORPH_SUPPLYDEPOT_RAISE,
+    SUPPLYDEPOTLOWERED: MORPH_SUPPLYDEPOT_LOWER,
+    # terran units
+    HELLION: MORPH_HELLION,
+    HELLIONTANK: MORPH_HELLBAT,
+    LIBERATOR: MORPH_LIBERATORAAMODE,
+    LIBERATORAG: MORPH_LIBERATORAGMODE,
+    SIEGETANK: UNSIEGE_UNSIEGE,
+    SIEGETANKSIEGED: SIEGEMODE_SIEGEMODE,
+    THOR: MORPH_THOREXPLOSIVEMODE,
+    THORAP: MORPH_THORHIGHIMPACTMODE,
+    VIKINGASSAULT: MORPH_VIKINGASSAULTMODE,
+    VIKINGFIGHTER: MORPH_VIKINGFIGHTERMODE,
+    WIDOWMINE: BURROWUP,
+    WIDOWMINEBURROWED: BURROWDOWN,
+    # protoss structures
+    GATEWAY: MORPH_GATEWAY,
+    WARPGATE: MORPH_WARPGATE,
+    # protoss units
+    OBSERVER: MORPH_OBSERVERMODE,
+    OBSERVERSIEGEMODE: MORPH_SURVEILLANCEMODE,
+    WARPPRISM: MORPH_WARPPRISMTRANSPORTMODE,
+    WARPPRISMPHASING: MORPH_WARPPRISMPHASINGMODE,
+    # zerg structures
+    SPINECRAWLER: SPINECRAWLERROOT_SPINECRAWLERROOT,
+    SPINECRAWLERUPROOTED: SPINECRAWLERUPROOT_SPINECRAWLERUPROOT,
+    SPORECRAWLER: SPORECRAWLERROOT_SPORECRAWLERROOT, 
+    SPORECRAWLERUPROOTED: SPORECRAWLERUPROOT_SPORECRAWLERUPROOT,
+    # zerg units
+    BANELING: BURROWUP_BANELING,
+    BANELINGBURROWED: BURROWDOWN_BANELING,
+    DRONE: BURROWUP_DRONE,
+    DRONEBURROWED: BURROWDOWN_DRONE,
+    HYDRALISK: BURROWUP_HYDRALISK,
+    HYDRALISKBURROWED: BURROWDOWN_HYDRALISK,
+    INFESTOR: BURROWUP_INFESTOR,
+    INFESTORBURROWED: BURROWDOWN_INFESTOR,
+    INFESTORTERRAN: BURROWUP_INFESTORTERRAN,
+    INFESTORTERRANBURROWED: BURROWDOWN_INFESTORTERRAN,
+    LURKERMP: BURROWUP_LURKER,
+    LURKERMPBURROWED: BURROWDOWN_LURKER,
+    OVERSEER: MORPH_OVERSEERMODE,
+    OVERSEERSIEGEMODE: MORPH_OVERSIGHTMODE,
+    QUEEN: BURROWUP_QUEEN,
+    QUEENBURROWED: BURROWDOWN_QUEEN,
+    ROACH: BURROWUP_ROACH,
+    ROACHBURROWED: BURROWDOWN_ROACH,
+    SWARMHOSTBURROWEDMP: BURROWDOWN_SWARMHOST,
+    SWARMHOSTMP: BURROWUP_SWARMHOST,
+    ULTRALISK: BURROWUP_ULTRALISK,
+    ULTRALISKBURROWED: BURROWDOWN_ULTRALISK,
+    ZERGLING: BURROWUP_ZERGLING,
+    ZERGLINGBURROWED: BURROWDOWN_ZERGLING,
+}

sc2/game_state.py

@@ -124,37 +124,47 @@ def __init__(self, response_observation):
         # https://github.com/Blizzard/s2client-proto/blob/33f0ecf615aa06ca845ffe4739ef3133f37265a9/s2clientprotocol/score.proto#L31
         self.score: ScoreDetails = ScoreDetails(self.observation.score)
         self.abilities = self.observation.abilities  # abilities of selected units
-
         # Fix for enemy units detected by my sensor tower, as blips have less unit information than normal visible units
-        visibleUnits, blipUnits, minerals, geysers, destructables, enemy, own = ([] for _ in range(7))
+        blipUnits, minerals, geysers, destructables, enemy, own, watchtowers = ([] for _ in range(7))
 
         for unit in self.observation_raw.units:
             if unit.is_blip:
                 blipUnits.append(unit)
             else:
-                visibleUnits.append(unit)
-                if unit.alliance == Alliance.Neutral.value:
+                alliance = unit.alliance
+                # Alliance.Neutral.value = 3
+                if alliance == 3:
+                    unit_type = unit.unit_type
+                    # XELNAGATOWER = 149
+                    if unit_type == 149:
+                        watchtowers.append(unit)
                     # all destructable rocks except the one below the main base ramps
-                    if unit.radius > 1.5:
+                    elif unit.radius > 1.5:
                         destructables.append(unit)
                     # mineral field enums
-                    elif unit.unit_type in mineral_ids:
+                    elif unit_type in mineral_ids:
                         minerals.append(unit)
                     # geyser enums
-                    elif unit.unit_type in geyser_ids:
+                    elif unit_type in geyser_ids:
                         geysers.append(unit)
-                elif unit.alliance == Alliance.Self.value:
+                # Alliance.Self.value = 1
+                elif alliance == 1:
                     own.append(unit)
-                elif unit.alliance == Alliance.Enemy.value:
+                # Alliance.Enemy.value = 4
+                elif alliance == 4:
                     enemy.append(unit)
 
+        resources = minerals + geysers
+        visible_units = resources + destructables + enemy + own + watchtowers
+
         self.own_units: Units = Units.from_proto(own)
         self.enemy_units: Units = Units.from_proto(enemy)
         self.mineral_field: Units = Units.from_proto(minerals)
         self.vespene_geyser: Units = Units.from_proto(geysers)
-        self.resources: Units = Units.from_proto(minerals + geysers)
+        self.resources: Units = Units.from_proto(resources)
         self.destructables: Units = Units.from_proto(destructables)
-        self.units: Units = Units.from_proto(visibleUnits)
+        self.watchtowers: Units = Units.from_proto(watchtowers)
+        self.units: Units = Units.from_proto(visible_units)
         self.upgrades: Set[UpgradeId] = {UpgradeId(upgrade) for upgrade in self.observation_raw.player.upgrade_ids}
 
         # Set of unit tags that died this step

sc2/helpers/devtools.py

@@ -0,0 +1,23 @@
+import time
+from contextlib import contextmanager
+
+
+@contextmanager
+def time_this(label):
+    start = time.perf_counter_ns()
+    try:
+        yield
+    finally:
+        end = time.perf_counter_ns()
+        print(f"TIME {label}: {(end-start)/1000000000} sec")
+
+
+# Use like this
+if __name__ == "__main__":
+    with time_this("square rooting"):
+        for n in range(10 ** 7):
+            x = n ** 0.5
+
+
+# returns:
+# TIME square rooting: 2.307249782 sec

sc2/main.py

@@ -3,6 +3,7 @@
 import time
 
 import async_timeout
+from s2clientprotocol import sc2api_pb2 as sc_pb
 
 from .client import Client
 from .data import CreateGameError, Result
@@ -101,7 +102,8 @@ async def _play_game_ai(client, player_id, ai, realtime, step_time_limit, game_t
     ai._prepare_start(client, player_id, game_info, game_data)
     state = await client.observation()
     gs = GameState(state.observation)
-    ai._prepare_step(gs)
+    proto_game_info = await client._execute(game_info=sc_pb.RequestGameInfo())
+    ai._prepare_step(gs, proto_game_info)
     ai._prepare_first_step()
     try:
         ai.on_start()
@@ -126,8 +128,8 @@ async def _play_game_ai(client, player_id, ai, realtime, step_time_limit, game_t
             if game_time_limit and (gs.game_loop * 0.725 * (1 / 16)) > game_time_limit:
                 ai.on_end(Result.Tie)
                 return Result.Tie
-
-            ai._prepare_step(gs)
+            proto_game_info = await client._execute(game_info=sc_pb.RequestGameInfo())
+            ai._prepare_step(gs, proto_game_info)
 
         logger.debug(f"Running AI step, it={iteration} {gs.game_loop * 0.725 * (1 / 16):.2f}s")
 

sc2/position.py

@@ -54,8 +54,6 @@ def sort_by_distance(self, ps: Union["Units", List["Point2"]]) -> List["Point2"]
         """ This returns the target points sorted as list.
         You should not pass a set or dict since those are not sortable.
         If you want to sort your units towards a point, use 'units.sorted_by_distance_to(point)' instead. """
-        # if ps and all(isinstance(p, Point2) for p in ps):
-        #     return sorted(ps, key=lambda p: self._distance_squared(p))
         return sorted(ps, key=lambda p: self._distance_squared(p.position))
 
     def closest(self, ps: Union["Units", List["Point2"], Set["Point2"]]) -> Union["Unit", "Point2"]:

sc2/unit.py

@@ -3,6 +3,7 @@
 
 from . import unit_command
 from .cache import property_immutable_cache, property_mutable_cache
+from .constants import transforming
 from .data import Alliance, Attribute, CloakState, DisplayType, Race, TargetType, warpgate_abilities
 from .ids.ability_id import AbilityId
 from .ids.buff_id import BuffId
@@ -131,19 +132,25 @@ def unit_alias(self) -> Optional[UnitTypeId]:
         For SCV, this returns None """
         return self._type_data.unit_alias
 
-    @property
+    @property_immutable_cache
     def _weapons(self):
         """ Returns the weapons of the unit. """
-        if hasattr(self._type_data._proto, "weapons"):
+        try:
             return self._type_data._proto.weapons
-        return None
+        except:
+            return None
 
-    @property
+    @property_immutable_cache
     def can_attack(self) -> bool:
         """ Checks if the unit can attack at all. """
         # TODO BATTLECRUISER doesnt have weapons in proto?!
         return bool(self._weapons) or self.type_id == UnitTypeId.BATTLECRUISER
 
+    @property_immutable_cache
+    def can_attack_both(self) -> bool:
+        """ Checks if the unit can attack both ground and air units. """
+        return self.can_attack_ground and self.can_attack_air
+
     @property_immutable_cache
     def can_attack_ground(self) -> bool:
         """ Checks if the unit can attack ground units. """
@@ -583,6 +590,11 @@ def is_constructing_scv(self) -> bool:
             }
         )
 
+    @property_immutable_cache
+    def is_transforming(self) -> bool:
+        """ Checks if the unit transforming. """
+        return self.type_id in transforming and self.is_using_ability(transforming[self.type_id])
+
     @property_immutable_cache
     def is_repairing(self) -> bool:
         """ Checks if the unit is an SCV or MULE that is currently repairing. """

sc2/units.py

@@ -53,10 +53,6 @@ def copy(self):
         return self.subgroup(self)
 
     def __or__(self, other: "Units") -> "Units":
-        if self is None:
-            return other
-        if other is None:
-            return self
         return Units(
             chain(
                 iter(self),
@@ -65,17 +61,9 @@ def __or__(self, other: "Units") -> "Units":
         )
 
     def __and__(self, other: "Units") -> "Units":
-        if self is None:
-            return other
-        if other is None:
-            return self
         return Units(other_unit for other_unit in other if other_unit.tag in (self_unit.tag for self_unit in self))
 
     def __sub__(self, other: "Units") -> "Units":
-        if self is None:
-            return Units([])
-        if other is None:
-            return self
         return Units(self_unit for self_unit in self if self_unit.tag not in (other_unit.tag for other_unit in other))
 
     def __hash__(self):
@@ -200,24 +188,18 @@ def sorted_by_distance_to(self, position: Union[Unit, Point2], reverse: bool = F
     def tags_in(self, other: Union[Set[int], List[int], Dict[int, Any]]) -> "Units":
         """ Filters all units that have their tags in the 'other' set/list/dict """
         # example: self.units(QUEEN).tags_in(self.queen_tags_assigned_to_do_injects)
-        if isinstance(other, list):
-            other = set(other)
         return self.filter(lambda unit: unit.tag in other)
 
     def tags_not_in(self, other: Union[Set[int], List[int], Dict[int, Any]]) -> "Units":
         """ Filters all units that have their tags not in the 'other' set/list/dict """
         # example: self.units(QUEEN).tags_not_in(self.queen_tags_assigned_to_do_injects)
-        if isinstance(other, list):
-            other = set(other)
         return self.filter(lambda unit: unit.tag not in other)
 
     def of_type(self, other: Union[UnitTypeId, Set[UnitTypeId], List[UnitTypeId], Dict[UnitTypeId, Any]]) -> "Units":
         """ Filters all units that are of a specific type """
         # example: self.units.of_type([ZERGLING, ROACH, HYDRALISK, BROODLORD])
         if isinstance(other, UnitTypeId):
             other = {other}
-        if isinstance(other, list):
-            other = set(other)
         return self.filter(lambda unit: unit.type_id in other)
 
     def exclude_type(