Update expansionlocations again, add dump data, unit.can_be_attacked

Author: tweakimp

sc2/bot_ai.py

@@ -5,8 +5,6 @@
 from collections import Counter
 from typing import Any, Dict, List, Optional, Set, Tuple, Union  # mypy type checking
 
-from s2clientprotocol import common_pb2 as common_pb
-
 from .cache import property_cache_forever, property_cache_once_per_frame
 from .data import ActionResult, Alert, Race, Result, Target, race_gas, race_townhalls, race_worker
 from .game_data import AbilityData, GameData
@@ -126,15 +124,24 @@ def known_enemy_structures(self) -> Units:
 
     @property
     def main_base_ramp(self) -> "Ramp":
-        """ Returns the Ramp instance of the closest main-ramp to start location. Look in game_info.py for more information """
+        """ Returns the Ramp instance of the closest main-ramp to start location.
+        Look in game_info.py for more information """
         if hasattr(self, "cached_main_base_ramp"):
             return self.cached_main_base_ramp
-        """ The reason for len(ramp.upper) in {2, 5} is:
-        ParaSite map has 5 upper points, and most other maps have 2 upper points at the main ramp. The map Acolyte has 4 upper points at the wrong ramp (which is closest to the start position) """
-        self.cached_main_base_ramp = min(
-            (ramp for ramp in self.game_info.map_ramps if len(ramp.upper) in {2, 5}),
-            key=lambda r: self.start_location.distance_to(r.top_center),
-        )
+        # The reason for len(ramp.upper) in {2, 5} is:
+        # ParaSite map has 5 upper points, and most other maps have 2 upper points at the main ramp.
+        # The map Acolyte has 4 upper points at the wrong ramp (which is closest to the start position).
+        try:
+            self.cached_main_base_ramp = min(
+                (ramp for ramp in self.game_info.map_ramps if len(ramp.upper) in {2, 5}),
+                key=lambda r: self.start_location.distance_to(r.top_center),
+            )
+        except ValueError:
+            # Hardcoded hotfix for Honorgrounds LE map, as that map has a large main base ramp with inbase natural
+            self.cached_main_base_ramp = min(
+                (ramp for ramp in self.game_info.map_ramps if len(ramp.upper) in {4, 9}),
+                key=lambda r: self.start_location.distance_to(r.top_center),
+            )
         return self.cached_main_base_ramp
 
     @property_cache_forever
@@ -149,13 +156,9 @@ def expansion_locations(self) -> Dict[Point2, Units]:
 
         # Distance we group resources by
         RESOURCE_SPREAD_THRESHOLD = 8.5
-        minerals = self.state.mineral_field
         geysers = self.state.vespene_geyser
-        all_resources = minerals | geysers
-        # Presort resources to get faster clustering
-        all_resources.sort(key=lambda resource: (resource.position.x, resource.position.y))
         # Create a group for every resource
-        resource_groups = [[resource] for resource in all_resources]
+        resource_groups = [[resource] for resource in self.state.resources]
         # Loop the merging process as long as we change something
         found_something = True
         while found_something:
@@ -177,9 +180,8 @@ def expansion_locations(self) -> Dict[Point2, Units]:
         offset_range = 7
         offsets = [
             (x, y)
-            for x in range(-offset_range, offset_range + 1)
-            for y in range(-offset_range, offset_range + 1)
-            if 4 <= math.hypot(x, y) <= 7
+            for x, y in itertools.product(range(-offset_range, offset_range + 1), repeat=2)
+            if math.hypot(x, y) <= 8
         ]
         # Dict we want to return
         centers = {}
@@ -189,20 +191,19 @@ def expansion_locations(self) -> Dict[Point2, Units]:
             amount = len(resources)
             # Calculate center, round and add 0.5 because expansion location will have (x.5, y.5)
             # coordinates because bases have size 5.
-            center_x = round(sum(resource.position.x for resource in resources) / amount) + 0.5
-            center_y = round(sum(resource.position.y for resource in resources) / amount) + 0.5
+            center_x = int(sum(resource.position.x for resource in resources) / amount) + 0.5
+            center_y = int(sum(resource.position.y for resource in resources) / amount) + 0.5
             possible_points = (Point2((offset[0] + center_x, offset[1] + center_y)) for offset in offsets)
             # Filter out points that are too near
             possible_points = (
                 point
                 for point in possible_points
                 # Check if point can be built on
-                if self._game_info.placement_grid[point.rounded] != 0
+                if self._game_info.placement_grid[point.rounded] == 1
                 # Check if all resources have enough space to point
-                and all(point.distance_to(resource) >= (7 if resource in geysers else 6) for resource in resources)
+                and all(point.distance_to(resource) > (7 if resource in geysers else 6) for resource in resources)
             )
             # Choose best fitting point
-            # TODO can we improve this by calculating the distance only one time?
             result = min(possible_points, key=lambda point: sum(point.distance_to(resource) for resource in resources))
             centers[result] = resources
         return centers

sc2/client.py

@@ -138,6 +138,26 @@ async def get_game_data(self) -> GameData:
         )
         return GameData(result.data)
 
+    async def dump_data(self, ability_id=True, unit_type_id=True, upgrade_id=True, buff_id=True, effect_id=True):
+        """
+        Dump the game data files
+        choose what data to dump in the keywords
+        this function writes to a text file
+        call it one time in on_step with:
+        await self._client.dump_data()
+        """
+        result = await self._execute(
+            data=sc_pb.RequestData(
+                ability_id=ability_id,
+                unit_type_id=unit_type_id,
+                upgrade_id=upgrade_id,
+                buff_id=buff_id,
+                effect_id=effect_id,
+            )
+        )
+        with open("data_dump.txt", "a") as file:
+            file.write(str(result.data))
+
     async def get_game_info(self) -> GameInfo:
         result = await self._execute(game_info=sc_pb.RequestGameInfo())
         return GameInfo(result.game_info)

sc2/unit.py

@@ -43,9 +43,7 @@ def __repr__(self) -> str:
         return f"UnitOrder({self.ability}, {self.target}, {self.progress})"
 
 
-class PassengerUnit:
-    """ Is inherited by the Unit class. Everything in here is also available in Unit. """
-
+class Unit:
     def __init__(self, proto_data):
         self._proto = proto_data
         self.cache = {}
@@ -302,11 +300,6 @@ def energy_percentage(self) -> Union[int, float]:
             return 0
         return self._proto.energy / self._proto.energy_max
 
-
-class Unit(PassengerUnit):
-
-    # All type data is in PassengerUnit.
-
     @property_immutable_cache
     def is_snapshot(self) -> bool:
         """ Checks if the unit is only available as a snapshot for the bot.
@@ -396,6 +389,11 @@ def is_revealed(self) -> bool:
         """ Checks if the unit is revealed. """
         return self._proto.cloak is CloakState.CloakedDetected.value
 
+    @property_immutable_cache
+    def can_be_attacked(self) -> bool:
+        """ Checks if the unit is revealed or not cloaked and therefore can be attacked """
+        return self._proto.cloak in {CloakState.NotCloaked.value, CloakState.CloakedDetected.value}
+
     @property_immutable_cache
     def buffs(self) -> Set:
         """ Returns the set of current buffs the unit has. """
@@ -558,7 +556,8 @@ def is_idle(self) -> bool:
         return not self.orders
 
     def is_using_ability(self, abilities: Union[AbilityId, Set[AbilityId]]) -> bool:
-        """ Check if the unit is using one of the given abilities. """
+        """ Check if the unit is using one of the given abilities.
+        Only works for own units. """
         if not self.orders:
             return False
         if isinstance(abilities, AbilityId):
@@ -567,12 +566,14 @@ def is_using_ability(self, abilities: Union[AbilityId, Set[AbilityId]]) -> bool:
 
     @property_immutable_cache
     def is_moving(self) -> bool:
-        """ Checks if the unit is moving. """
+        """ Checks if the unit is moving.
+        Only works for own units. """
         return self.is_using_ability(AbilityId.MOVE)
 
     @property_immutable_cache
     def is_attacking(self) -> bool:
-        """ Checks if the unit is attacking. """
+        """ Checks if the unit is attacking.
+        Only works for own units. """
         return self.is_using_ability(
             {
                 AbilityId.ATTACK,
@@ -585,27 +586,32 @@ def is_attacking(self) -> bool:
 
     @property_immutable_cache
     def is_patrolling(self) -> bool:
-        """ Checks if a unit is patrolling. """
+        """ Checks if a unit is patrolling.
+        Only works for own units. """
         return self.is_using_ability(AbilityId.PATROL)
 
     @property_immutable_cache
     def is_gathering(self) -> bool:
-        """ Checks if a unit is on its way to a mineral field or vespene geyser to mine. """
+        """ Checks if a unit is on its way to a mineral field or vespene geyser to mine.
+        Only works for own units. """
         return self.is_using_ability(AbilityId.HARVEST_GATHER)
 
     @property_immutable_cache
     def is_returning(self) -> bool:
-        """ Checks if a unit is returning from mineral field or vespene geyser to deliver resources to townhall. """
+        """ Checks if a unit is returning from mineral field or vespene geyser to deliver resources to townhall.
+        Only works for own units. """
         return self.is_using_ability(AbilityId.HARVEST_RETURN)
 
     @property_immutable_cache
     def is_collecting(self) -> bool:
-        """ Checks if a unit is gathering or returning. """
+        """ Checks if a unit is gathering or returning.
+        Only works for own units. """
         return self.is_using_ability({AbilityId.HARVEST_GATHER, AbilityId.HARVEST_RETURN})
 
     @property_immutable_cache
     def is_constructing_scv(self) -> bool:
-        """ Checks if the unit is an SCV that is currently building. """
+        """ Checks if the unit is an SCV that is currently building.
+        Only works for own units. """
         return self.is_using_ability(
             {
                 AbilityId.TERRANBUILD_ARMORY,
@@ -626,12 +632,14 @@ def is_constructing_scv(self) -> bool:
 
     @property_immutable_cache
     def is_transforming(self) -> bool:
-        """ Checks if the unit transforming. """
+        """ Checks if the unit transforming.
+        Only works for own units. """
         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. """
+        """ Checks if the unit is an SCV or MULE that is currently repairing.
+        Only works for own units. """
         return self.is_using_ability(
             {AbilityId.EFFECT_REPAIR, AbilityId.EFFECT_REPAIR_MULE, AbilityId.EFFECT_REPAIR_SCV}
         )
@@ -653,9 +661,9 @@ def add_on_land_position(self) -> Point2:
         return self.position.offset(Point2((-2.5, 0.5)))
 
     @property_mutable_cache
-    def passengers(self) -> Set["PassengerUnit"]:
+    def passengers(self) -> Set["Unit"]:
         """ Returns the units inside a Bunker, CommandCenter, PlanetaryFortress, Medivac, Nydus, Overlord or WarpPrism. """
-        return {PassengerUnit(unit) for unit in self._proto.passengers}
+        return {Unit(unit) for unit in self._proto.passengers}
 
     @property_mutable_cache
     def passengers_tags(self) -> Set[int]: