Merge pull request #1 from tweakimp/develop

Develop

Author: manantt

sc2/bot_ai.py

@@ -133,7 +133,7 @@ def main_base_ramp(self) -> "Ramp":
         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_squared(r.top_center),
+            key=lambda r: self.start_location.distance_to(r.top_center),
         )
         return self.cached_main_base_ramp
 
@@ -148,7 +148,7 @@ def expansion_locations(self) -> Dict[Point2, Units]:
         # any resource in a group is closer than 6 to any resource of another group
 
         # Distance we group resources by
-        RESOURCE_SPREAD_THRESHOLD = 36
+        RESOURCE_SPREAD_THRESHOLD = 8.5
         minerals = self.state.mineral_field
         geysers = self.state.vespene_geyser
         all_resources = minerals | geysers
@@ -164,7 +164,7 @@ def expansion_locations(self) -> Dict[Point2, Units]:
             for group_a, group_b in itertools.combinations(resource_groups, 2):
                 # Check if any pair of resource of these groups is closer than threshold together
                 if any(
-                    resource_a.position._distance_squared(resource_b.position) <= RESOURCE_SPREAD_THRESHOLD
+                    resource_a.distance_to(resource_b) <= RESOURCE_SPREAD_THRESHOLD
                     for resource_a, resource_b in itertools.product(group_a, group_b)
                 ):
                     # Remove the single groups and add the merged group
@@ -179,7 +179,7 @@ def expansion_locations(self) -> Dict[Point2, Units]:
             (x, y)
             for x in range(-offset_range, offset_range + 1)
             for y in range(-offset_range, offset_range + 1)
-            if 49 >= x ** 2 + y ** 2 >= 16
+            if 4 <= math.hypot(x, y) <= 7
         ]
         # Dict we want to return
         centers = {}
@@ -199,17 +199,11 @@ def expansion_locations(self) -> Dict[Point2, Units]:
                 # Check if point can be built on
                 if self._game_info.placement_grid[point.rounded] != 0
                 # Check if all resources have enough space to point
-                and all(
-                    point._distance_squared(resource.position) >= (49 if resource in geysers else 36)
-                    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_squared(resource.position) for resource in resources),
-            )
+            result = min(possible_points, key=lambda point: sum(point.distance_to(resource) for resource in resources))
             centers[result] = resources
         return centers
 
@@ -271,7 +265,7 @@ async def get_next_expansion(self) -> Optional[Point2]:
         for el in self.expansion_locations:
 
             def is_near_to_expansion(t):
-                return t.position._distance_squared(el) < self.EXPANSION_GAP_THRESHOLD ** 2
+                return t.distance_to(el) < self.EXPANSION_GAP_THRESHOLD
 
             if any(map(is_near_to_expansion, self.townhalls)):
                 # already taken
@@ -288,16 +282,27 @@ def is_near_to_expansion(t):
 
         return closest
 
-    async def distribute_workers(self):
+    async def distribute_workers(self, resource_ratio: float = 2):
         """
         Distributes workers across all the bases taken.
+        Keyword `resource_ratio` takes a float. If the current minerals to gas
+        ratio is bigger than `resource_ratio`, this function prefer filling geysers
+        first, if it is lower, it will prefer sending workers to minerals first.
+        This is only for workers that need to be moved anyways, it will NOT will
+        geysers on its own.
+
+        NOTE: This function is far from optimal, if you really want to have
+        refined worker control, you should write your own distribution function.
+        For example long distance mining control and moving workers if a base was killed
+        are not being handled.
+
         WARNING: This is quite slow when there are lots of workers or multiple bases.
         """
-        if not self.state.mineral_field or not self.workers or not self.owned_expansions.ready:
+        if not self.state.mineral_field or not self.workers or not self.townhalls.ready:
             return
         actions = []
         worker_pool = [worker for worker in self.workers.idle]
-        bases = self.owned_expansions.ready
+        bases = self.townhalls.ready
         geysers = self.geysers.ready
 
         # list of places that need more workers
@@ -308,53 +313,79 @@ async def distribute_workers(self):
             # perfect amount of workers, skip mining place
             if not difference:
                 continue
-            if mining_place.has_vespene:
+            if mining_place.is_vespene_geyser:
                 # get all workers that target the gas extraction site
                 # or are on their way back from it
                 local_workers = self.workers.filter(
                     lambda unit: unit.order_target == mining_place.tag
                     or (unit.is_carrying_vespene and unit.order_target == bases.closest_to(mining_place).tag)
                 )
             else:
-                # get minerals around expansion
-                local_minerals = self.expansion_locations[mining_place.position].filter(
-                    lambda resource: resource.has_minerals
-                )
+                # get tags of minerals around expansion
+                local_minerals_tags = {
+                    mineral.tag for mineral in self.state.mineral_field if mineral.distance_to(mining_place) <= 8
+                }
                 # get all target tags a worker can have
                 # tags of the minerals he could mine at that base
                 # get workers that work at that gather site
                 local_workers = self.workers.filter(
-                    lambda unit: unit.order_target in local_minerals.tags
+                    lambda unit: unit.order_target in local_minerals_tags
                     or (unit.is_carrying_minerals and unit.order_target == mining_place.tag)
                 )
             # too many workers
             if difference > 0:
-                worker_pool.append(local_workers[:difference])
+                for worker in local_workers[:difference]:
+                    worker_pool.append(worker)
             # too few workers
             # add mining place to deficit bases for every missing worker
             else:
                 deficit_mining_places += [mining_place for _ in range(-difference)]
 
+        # prepare all minerals near a base if we have too many workers
+        # and need to send them to the closest patch
+        if len(worker_pool) > len(deficit_mining_places):
+            all_minerals_near_base = [
+                mineral
+                for mineral in self.state.mineral_field
+                if any(mineral.distance_to(base) <= 8 for base in self.townhalls.ready)
+            ]
         # distribute every worker in the pool
         for worker in worker_pool:
             # as long as have workers and mining places
             if deficit_mining_places:
-                # remove current place from the list for next loop
-                current_place = deficit_mining_places.pop(0)
+                # choose only mineral fields first if current mineral to gas ratio is less than target ratio
+                if self.vespene and self.minerals / self.vespene < resource_ratio:
+                    possible_mining_places = [place for place in deficit_mining_places if not place.vespene_contents]
+                # else prefer gas
+                else:
+                    possible_mining_places = [place for place in deficit_mining_places if place.vespene_contents]
+                # if preferred type is not available any more, get all other places
+                if not possible_mining_places:
+                    possible_mining_places = deficit_mining_places
+                # find closest mining place
+                current_place = min(deficit_mining_places, key=lambda place: place.distance_to(worker))
+                # remove it from the list
+                deficit_mining_places.remove(current_place)
                 # if current place is a gas extraction site, go there
-                if current_place.has_vespene:
+                if current_place.vespene_contents:
                     actions.append(worker.gather(current_place))
                 # if current place is a gas extraction site,
                 # go to the mineral field that is near and has the most minerals left
                 else:
-                    local_minerals = self.expansion_locations[current_place.position].filter(
-                        lambda resource: resource.has_minerals
-                    )
+                    local_minerals = [
+                        mineral for mineral in self.state.mineral_field if mineral.distance_to(current_place) <= 8
+                    ]
                     target_mineral = max(local_minerals, key=lambda mineral: mineral.mineral_contents)
                     actions.append(worker.gather(target_mineral))
             # more workers to distribute than free mining spots
-            # else:
-            #     pass
+            # send to closest if worker is doing nothing
+            elif worker.is_idle and all_minerals_near_base:
+                target_mineral = min(all_minerals_near_base, key=lambda mineral: mineral.distance_to(worker))
+                actions.append(worker.gather(target_mineral))
+            else:
+                # there are no deficit mining places and worker is not idle
+                # so dont move him
+                pass
 
         await self.do_actions(actions)
 
@@ -366,7 +397,7 @@ def owned_expansions(self) -> Dict[Point2, Unit]:
         for el in self.expansion_locations:
 
             def is_near_to_expansion(t):
-                return t.position._distance_squared(el) < self.EXPANSION_GAP_THRESHOLD ** 2
+                return t.distance_to(el) < self.EXPANSION_GAP_THRESHOLD
 
             th = next((x for x in self.townhalls if is_near_to_expansion(x)), None)
             if th:
@@ -425,21 +456,21 @@ async def can_cast(
                 ability_target == 1
                 or ability_target == Target.PointOrNone.value
                 and isinstance(target, (Point2, Point3))
-                and unit.position._distance_squared(target.position) <= cast_range ** 2
+                and unit.distance_to(target) <= cast_range
             ):  # cant replace 1 with "Target.None.value" because ".None" doesnt seem to be a valid enum name
                 return True
             # Check if able to use ability on a unit
             elif (
                 ability_target in {Target.Unit.value, Target.PointOrUnit.value}
                 and isinstance(target, Unit)
-                and unit.position._distance_squared(target.position) <= cast_range ** 2
+                and unit.distance_to(target) <= cast_range
             ):
                 return True
             # Check if able to use ability on a position
             elif (
                 ability_target in {Target.Point.value, Target.PointOrUnit.value}
                 and isinstance(target, (Point2, Point3))
-                and unit.position._distance_squared(target) <= cast_range ** 2
+                and unit.distance_to(target) <= cast_range
             ):
                 return True
         return False
@@ -514,7 +545,7 @@ async def find_placement(
             if random_alternative:
                 return random.choice(possible)
             else:
-                return min(possible, key=lambda p: p._distance_squared(near))
+                return min(possible, key=lambda p: p.distance_to_point2(near))
         return None
 
     def already_pending_upgrade(self, upgrade_type: UpgradeId) -> Union[int, float]:
@@ -752,7 +783,7 @@ def _prepare_first_step(self):
         """First step extra preparations. Must not be called before _prepare_step."""
         if self.townhalls:
             self._game_info.player_start_location = self.townhalls.first.position
-        self._game_info.map_ramps = self._game_info._find_ramps()
+        self._game_info.map_ramps, self._game_info.vision_blockers = self._game_info._find_ramps_and_vision_blockers()
 
     def _prepare_step(self, state, proto_game_info):
         # Set attributes from new state before on_step."""

sc2/cache.py

@@ -20,7 +20,7 @@ def property_cache_once_per_frame(f):
     then clears it if it is accessed in a different game loop.
     Only works on properties of the bot object, because it requires
     access to self.state.game_loop """
-    
+
     @wraps(f)
     def inner(self):
         property_cache = "_cache_" + f.__name__

sc2/game_info.py

@@ -1,6 +1,8 @@
 from collections import deque
 from typing import Any, Deque, Dict, FrozenSet, Generator, List, Optional, Sequence, Set, Tuple, Union
 
+import numpy as np
+
 from .cache import property_immutable_cache, property_mutable_cache
 from .pixel_map import PixelMap
 from .player import Player
@@ -57,7 +59,7 @@ def upper2_for_ramp_wall(self) -> Set[Point2]:
             return set()  # HACK: makes this work for now
             # FIXME: please do
 
-        upper2 = sorted(list(self.upper), key=lambda x: x._distance_squared(self.bottom_center), reverse=True)
+        upper2 = sorted(list(self.upper), key=lambda x: x.distance_to_point2(self.bottom_center), reverse=True)
         while len(upper2) > 2:
             upper2.pop()
         return set(upper2)
@@ -99,7 +101,7 @@ def barracks_in_middle(self) -> Point2:
             # Offset from top point to barracks center is (2, 1)
             intersects = p1.circle_intersection(p2, 5 ** 0.5)
             anyLowerPoint = next(iter(self.lower))
-            return max(intersects, key=lambda p: p._distance_squared(anyLowerPoint))
+            return max(intersects, key=lambda p: p.distance_to_point2(anyLowerPoint))
         raise Exception("Not implemented. Trying to access a ramp that has a wrong amount of upper points.")
 
     @property_immutable_cache
@@ -112,7 +114,7 @@ def depot_in_middle(self) -> Point2:
             # Offset from top point to depot center is (1.5, 0.5)
             intersects = p1.circle_intersection(p2, 2.5 ** 0.5)
             anyLowerPoint = next(iter(self.lower))
-            return max(intersects, key=lambda p: p._distance_squared(anyLowerPoint))
+            return max(intersects, key=lambda p: p.distance_to_point2(anyLowerPoint))
         raise Exception("Not implemented. Trying to access a ramp that has a wrong amount of upper points.")
 
     @property_mutable_cache
@@ -122,7 +124,7 @@ def corner_depots(self) -> Set[Point2]:
             points = self.upper2_for_ramp_wall
             p1 = points.pop().offset((self.x_offset, self.y_offset))  # still an error with pixelmap?
             p2 = points.pop().offset((self.x_offset, self.y_offset))
-            center = p1.towards(p2, p1.distance_to(p2) / 2)
+            center = p1.towards(p2, p1.distance_to_point2(p2) / 2)
             depotPosition = self.depot_in_middle
             # Offset from middle depot to corner depots is (2, 1)
             intersects = center.circle_intersection(depotPosition, 5 ** 0.5)
@@ -168,22 +170,38 @@ def __init__(self, proto):
         self.playable_area = Rect.from_proto(self._proto.start_raw.playable_area)
         self.map_center = self.playable_area.center
         self.map_ramps: List[Ramp] = None  # Filled later by BotAI._prepare_first_step
+        self.vision_blockers: Set[Point2] = None  # Filled later by BotAI._prepare_first_step
         self.player_races: Dict[int, "Race"] = {
             p.player_id: p.race_actual or p.race_requested for p in self._proto.player_info
         }
         self.start_locations: List[Point2] = [Point2.from_proto(sl) for sl in self._proto.start_raw.start_locations]
         self.player_start_location: Point2 = None  # Filled later by BotAI._prepare_first_step
 
-    def _find_ramps(self) -> List[Ramp]:
-        """Calculate (self.pathing_grid - self.placement_grid) (for sets) and then find ramps by comparing heights."""
+    def _find_ramps_and_vision_blockers(self) -> Tuple[List[Ramp], Set[Point2]]:
+        """ Calculate points that are pathable but not placeable.
+        Then devide them into ramp points if not all points around the points are equal height
+        and into vision blockers if they are. """
+
+        def equal_height_around(tile):
+            # mask to slice array 1 around tile
+            sliced = self.terrain_height.data_numpy[tile[1] - 1 : tile[1] + 2, tile[0] - 1 : tile[0] + 2]
+            return len(np.unique(sliced)) == 1
+
         map_area = self.playable_area
-        rampPoints = (
-            Point2((x, y))
-            for x in range(map_area.x, map_area.x + map_area.width)
-            for y in range(map_area.y, map_area.y + map_area.height)
-            if self.placement_grid[(x, y)] == 0 and self.pathing_grid[(x, y)] == 1
-        )
-        return [Ramp(group, self) for group in self._find_groups(rampPoints)]
+        # all points in the playable area that are pathable but not placable
+        points = [
+            Point2((b, a))
+            for (a, b), value in np.ndenumerate(self.pathing_grid.data_numpy)
+            if value == 1
+            and map_area.x <= a < map_area.x + map_area.width
+            and map_area.y <= b < map_area.y + map_area.height
+            and self.placement_grid[(b, a)] == 0
+        ]
+        # devide points into ramp points and vision blockers
+        rampPoints = [point for point in points if not equal_height_around(point)]
+        visionBlockers = set(point for point in points if equal_height_around(point))
+        ramps = [Ramp(group, self) for group in self._find_groups(rampPoints)]
+        return ramps, visionBlockers
 
     def _find_groups(self, points: Set[Point2], minimum_points_per_group: int = 8):
         """
@@ -224,7 +242,7 @@ def paint(pt: Point2) -> None:
                     if picture[py][px] != NOT_COLORED_YET:
                         continue
                     point: Point2 = Point2((px, py))
-                    remaining.remove(point)
+                    remaining.discard(point)
                     paint(point)
                     queue.append(point)
                     currentGroup.add(point)