C pathfinding and chokes

MapAnalyzer/Debugger.py

@@ -209,6 +209,56 @@ def plot_map(
             tick.label1.set_fontweight("bold")
         plt.grid()
 
+
+    def plot_influenced_path_c(self, start: Union[Tuple[float, float], Point2],
+                               goal: Union[Tuple[float, float], Point2],
+                               weight_array: ndarray,
+                               smoothing: bool = False,
+                               name: Optional[str] = None,
+                               fontdict: dict = None) -> None:
+        import matplotlib.pyplot as plt
+        from mpl_toolkits.axes_grid1 import make_axes_locatable
+        from matplotlib.cm import ScalarMappable
+        if not fontdict:
+            fontdict = {"family": "serif", "weight": "bold", "size": 20}
+        plt.style.use(["ggplot", "bmh"])
+        org = "lower"
+        if name is None:
+            name = self.map_data.map_name
+        arr = weight_array.copy()
+        path = self.map_data.pathfind_c(start, goal,
+                                        grid=arr,
+                                        smoothing=smoothing,
+                                        sensitivity=1)
+        ax: plt.Axes = plt.subplot(1, 1, 1)
+        if path is not None:
+            path = np.flipud(path)  # for plot align
+            logger.info("Found")
+            x, y = zip(*path)
+            ax.scatter(x, y, s=3, c='green')
+        else:
+            logger.info("Not Found")
+
+            x, y = zip(*[start, goal])
+            ax.scatter(x, y)
+
+        influence_cmap = plt.cm.get_cmap("afmhot")
+        ax.text(start[0], start[1], f"Start {start}")
+        ax.text(goal[0], goal[1], f"Goal {goal}")
+        ax.imshow(self.map_data.path_arr, alpha=0.5, origin=org)
+        ax.imshow(self.map_data.terrain_height, alpha=0.5, origin=org, cmap='bone')
+        arr = np.where(arr == np.inf, 0, arr).T
+        ax.imshow(arr, origin=org, alpha=0.3, cmap=influence_cmap)
+        divider = make_axes_locatable(ax)
+        cax = divider.append_axes("right", size="5%", pad=0.05)
+        sc = ScalarMappable(cmap=influence_cmap)
+        sc.set_array(arr)
+        sc.autoscale()
+        cbar = plt.colorbar(sc, cax=cax)
+        cbar.ax.set_ylabel('Pathing Cost', rotation=270, labelpad=25, fontdict=fontdict)
+        plt.title(f"{name}", fontdict=fontdict, loc='right')
+        plt.grid()
+
     def plot_influenced_path(self, start: Union[Tuple[int, int], Point2],
                              goal: Union[Tuple[int, int], Point2],
                              weight_array: ndarray,

MapAnalyzer/MapData.py

@@ -13,13 +13,14 @@
 from MapAnalyzer.Debugger import MapAnalyzerDebugger
 from MapAnalyzer.Pather import MapAnalyzerPather
 from MapAnalyzer.Region import Region
-from MapAnalyzer.utils import get_sets_with_mutual_elements
+from MapAnalyzer.utils import get_sets_with_mutual_elements, fix_map_ramps
 
 from .constants import BINARY_STRUCTURE, CORNER_MIN_DISTANCE, MAX_REGION_AREA, MIN_REGION_AREA
 
 from .decorators import progress_wrapped
 from .exceptions import CustomDeprecationWarning
-from MapAnalyzer.constructs import ChokeArea, MDRamp, VisionBlockerArea, PathLibChoke
+from MapAnalyzer.constructs import ChokeArea, MDRamp, VisionBlockerArea, RawChoke
+from .cext import CMapInfo, CMapChoke
 
 try:
     __version__ = get_distribution('sc2mapanalyzer')
@@ -40,6 +41,11 @@ def __init__(self, bot: BotAI, loglevel: str = "ERROR", arcade: bool = False,
                  corner_distance: int = CORNER_MIN_DISTANCE) -> None:
         # store relevant data from api
         self.bot = bot
+        # temporary fix to set ramps correctly if they are broken in burnysc2 due to having
+        # destructables on them. ramp sides don't consider the destructables now,
+        # should update them during the game
+        self.bot.game_info.map_ramps, self.bot.game_info.vision_blockers = fix_map_ramps(self.bot)
+
         self.corner_distance = corner_distance  # the lower this value is,  the sharper the corners will be
         self.arcade = arcade
         self.version = __version__
@@ -63,17 +69,21 @@ def __init__(self, bot: BotAI, loglevel: str = "ERROR", arcade: bool = False,
         self.map_vision_blockers: list = []  # set later  on compile
         self.vision_blockers_labels: list = []  # set later  on compile
         self.vision_blockers_grid: list = []  # set later  on compile
+        self.overlord_spots: list = []
         self.resource_blockers = [Point2((m.position[0], m.position[1])) for m in self.bot.all_units if
                                   any(x in m.name.lower() for x in {"rich", "450"})]
-        self.pathlib_to_local_chokes = None
         self.overlapping_choke_ids = None
 
+        pathing_grid = np.fmax(self.path_arr, self.placement_arr)
+        self.c_ext_map = CMapInfo(pathing_grid.T, self.terrain_height.T, self.bot.game_info.playable_area)
+        self.overlord_spots = self.c_ext_map.overlord_spots
+
         # plugins
         self.log_level = loglevel
         self.debugger = MapAnalyzerDebugger(self, loglevel=self.log_level)
         self.pather = MapAnalyzerPather(self)
+
         self.connectivity_graph = None  # set by pather
-        self.pathlib_map = self.pather.pathlib_map
         self.pyastar = self.pather.pyastar
         self.nonpathable_indices_stacked = self.pather.nonpathable_indices_stacked
 
@@ -192,7 +202,7 @@ def find_lowest_cost_points(self, from_pos: Point2, radius: float, grid: np.ndar
     def get_climber_grid(self, default_weight: float = 1, include_destructables: bool = True) -> ndarray:
         """
         :rtype: numpy.ndarray
-        Climber grid is a grid modified by :mod:`sc2pathlibp`, and is used for units that can climb,
+        Climber grid is a grid modified by the c extension, and is used for units that can climb,
 
         such as Reaper, Colossus
 
@@ -302,6 +312,12 @@ def pathfind(self, start: Union[Tuple[float, float], Point2], goal: Union[Tuple[
         return self.pather.pathfind(start=start, goal=goal, grid=grid, allow_diagonal=allow_diagonal,
                                     sensitivity=sensitivity)
 
+    def pathfind_c(self, start: Union[Tuple[float, float], Point2], goal: Union[Tuple[float, float], Point2],
+                   grid: Optional[ndarray] = None, smoothing: bool = False,
+                   sensitivity: int = 1) -> Optional[List[Point2]]:
+        return self.pather.pathfind_c(start=start, goal=goal, grid=grid, smoothing=smoothing,
+                                      sensitivity=sensitivity)
+
     def add_cost(self, position: Tuple[float, float], radius: float, grid: ndarray, weight: float = 100, safe: bool = True,
                  initial_default_weights: float = 0) -> ndarray:
         """
@@ -600,17 +616,13 @@ def in_region_p(self, point: Union[Point2, tuple]) -> Optional[Region]:
 
     def _clean_plib_chokes(self) -> None:
         # needs to be called AFTER MDramp and VisionBlocker are populated
-        raw_chokes = self.pathlib_map.chokes
-        self.pathlib_to_local_chokes = []
-        for i, c in enumerate(raw_chokes):
-            self.pathlib_to_local_chokes.append(PathLibChoke(pathlib_choke=c, pk=i))
         areas = self.map_ramps.copy()
         areas.extend(self.map_vision_blockers)
-        self.overlapping_choke_ids = self._get_overlapping_chokes(local_chokes=self.pathlib_to_local_chokes,
+        self.overlapping_choke_ids = self._get_overlapping_chokes(local_chokes=self.c_ext_map.chokes,
                                                                   areas=areas)
 
     @staticmethod
-    def _get_overlapping_chokes(local_chokes: List[PathLibChoke],
+    def _get_overlapping_chokes(local_chokes: List[CMapChoke],
                                 areas: Union[List[MDRamp], List[Union[MDRamp, VisionBlockerArea]]]) -> Set[int]:
         li = []
         for area in areas:
@@ -684,10 +696,15 @@ def _calc_grid(self) -> None:
             self.vision_blockers_labels = np.unique(vb_labeled_array)
 
     def _set_map_ramps(self):
+        # some ramps coming from burnysc2 have broken data and the bottom_center and top_center
+        # may even be the same. by removing them they should be tagged as chokes in the c extension
+        # if they really are ones
+        viable_ramps = list(filter(lambda x: x.bottom_center.distance_to(x.top_center) >= 1,
+                            self.bot.game_info.map_ramps))
         self.map_ramps = [MDRamp(map_data=self,
                                  ramp=r,
                                  array=self.points_to_numpy_array(r.points))
-                          for r in self.bot.game_info.map_ramps]
+                          for r in viable_ramps]
 
     def _calc_vision_blockers(self) -> None:
         # compute VisionBlockerArea
@@ -713,7 +730,7 @@ def _calc_chokes(self) -> None:
         # compute ChokeArea
 
         self._clean_plib_chokes()
-        chokes = [c for c in self.pathlib_to_local_chokes if c.id not in self.overlapping_choke_ids]
+        chokes = [c for c in self.c_ext_map.chokes if c.id not in self.overlapping_choke_ids]
         self.map_chokes = self.map_ramps.copy()
         self.map_chokes.extend(self.map_vision_blockers)
 
@@ -726,7 +743,7 @@ def _calc_chokes(self) -> None:
                 cm = int(cm[0]), int(cm[1])
                 areas = self.where_all(cm)
 
-                new_choke = ChokeArea(
+                new_choke = RawChoke(
                         map_data=self, array=new_choke_array, pathlibchoke=choke
                 )
                 for area in areas:
@@ -835,6 +852,26 @@ def plot_influenced_path(self,
                                            fontdict=fontdict,
                                            allow_diagonal=allow_diagonal)
 
+    def plot_influenced_path_c(self,
+                               start: Union[Tuple[float, float], Point2],
+                               goal: Union[Tuple[float, float], Point2],
+                               weight_array: ndarray,
+                               smoothing: bool = False,
+                               name: Optional[str] = None,
+                               fontdict: dict = None) -> None:
+        """
+
+        A useful debug utility method for experimenting with the :mod:`.Pather` module
+
+        """
+
+        self.debugger.plot_influenced_path_c(start=start,
+                                             goal=goal,
+                                             weight_array=weight_array,
+                                             smoothing=smoothing,
+                                             name=name,
+                                             fontdict=fontdict)
+
     def _plot_regions(self, fontdict: Dict[str, Union[str, int]]) -> None:
         return self.debugger.plot_regions(fontdict=fontdict)
 

MapAnalyzer/Pather.py

@@ -2,6 +2,7 @@
 
 import numpy as np
 import pyastar.astar_wrapper as pyastar
+
 from loguru import logger
 from numpy import ndarray
 from sc2.ids.unit_typeid import UnitTypeId as UnitID
@@ -11,7 +12,7 @@
 from MapAnalyzer.constants import GEYSER_RADIUS_FACTOR, NONPATHABLE_RADIUS_FACTOR, RESOURCE_BLOCKER_RADIUS_FACTOR
 from MapAnalyzer.exceptions import OutOfBoundsException, PatherNoPointsException
 from MapAnalyzer.Region import Region
-from .sc2pathlibp import Sc2Map
+from .cext import astar_path
 
 if TYPE_CHECKING:
     from MapAnalyzer.MapData import MapData
@@ -23,13 +24,7 @@ class MapAnalyzerPather:
     def __init__(self, map_data: "MapData") -> None:
         self.map_data = map_data
         self.pyastar = pyastar
-        self.pathlib_map = None
-        self._set_pathlib_map()
-        if self.pathlib_map:
-            # noinspection PyProtectedMember
-            self._climber_grid = np.array(self.pathlib_map._map.reaper_pathing).astype(np.float32)
-        else:
-            logger.error('Could not set Pathlib Map')
+
         nonpathable_indices = np.where(self.map_data.bot.game_info.pathing_grid.data_numpy == 0)
         self.nonpathable_indices_stacked = np.column_stack(
                 (nonpathable_indices[1], nonpathable_indices[0])
@@ -63,17 +58,6 @@ def find_all_paths(self, start: Region, goal: Region, path: Optional[List[Region
                     paths.append(newpath)
         return paths
 
-    def _set_pathlib_map(self) -> None:
-        """
-        Will initialize the sc2pathlib `SC2Map` object for future use
-        """
-        self.pathlib_map = Sc2Map(
-                self.map_data.path_arr,
-                self.map_data.placement_arr,
-                self.map_data.terrain_height,
-                self.map_data.bot.game_info.playable_area,
-        )
-
     def _add_non_pathables_ground(self, grid: ndarray, include_destructables: bool = True) -> ndarray:
         nonpathables = self.map_data.bot.structures.not_flying
         nonpathables.extend(self.map_data.bot.enemy_structures.not_flying)
@@ -130,9 +114,11 @@ def get_base_pathing_grid(self) -> ndarray:
 
     def get_climber_grid(self, default_weight: float = 1, include_destructables: bool = True) -> ndarray:
         """Grid for units like reaper / colossus """
-        grid = self._climber_grid.copy()
+        grid = self.get_base_pathing_grid()
         grid = np.where(grid != 0, default_weight, np.inf).astype(np.float32)
+        grid = np.where(self.map_data.c_ext_map.climber_grid != 0, default_weight, grid).astype(np.float32)
         grid = self._add_non_pathables_ground(grid=grid, include_destructables=include_destructables)
+
         return grid
 
     def get_clean_air_grid(self, default_weight: float = 1) -> ndarray:
@@ -148,7 +134,7 @@ def get_air_vs_ground_grid(self, default_weight: float) -> ndarray:
         return air_vs_ground_grid
 
     def get_pyastar_grid(self, default_weight: float = 1, include_destructables: bool = True) -> ndarray:
-        grid = self.map_data.pather.get_base_pathing_grid()
+        grid = self.get_base_pathing_grid()
         grid = np.where(grid != 0, default_weight, np.inf).astype(np.float32)
         grid = self._add_non_pathables_ground(grid=grid, include_destructables=include_destructables)
         return grid
@@ -184,6 +170,39 @@ def pathfind(self, start: Tuple[float, float], goal: Tuple[float, float], grid:
             logger.debug(f"No Path found s{start}, g{goal}")
             return None
 
+    def pathfind_c(self, start: Tuple[float, float], goal: Tuple[float, float], grid: Optional[ndarray] = None,
+                   smoothing: bool = False,
+                   sensitivity: int = 1) -> ndarray:
+        if start is not None and goal is not None:
+            start = int(round(start[0])), int(round(start[1]))
+            goal = int(round(goal[0])), int(round(goal[1]))
+        else:
+            logger.warning(PatherNoPointsException(start=start, goal=goal))
+            return None
+        if grid is None:
+            logger.warning("Using the default pyastar grid as no grid was provided.")
+            grid = self.get_pyastar_grid()
+
+        path = astar_path(grid, start, goal, smoothing)
+
+        if path is not None:
+            path = list(map(Point2, path))[::sensitivity]
+            """
+            Edge case
+            EverDreamLE,  (81, 29) is considered in map bounds,  but it is not.
+            """
+            # `if point` is checking with burnysc2 that the point is in map bounds
+            if 'everdream' in self.map_data.map_name.lower():
+                legal_path = [point for point in path if point and point.x != 81 and point.y != 29]
+            else:  # normal case
+                legal_path = [point for point in path if point]
+
+            legal_path.pop(0)
+            return legal_path
+        else:
+            logger.debug(f"No Path found s{start}, g{goal}")
+            return None
+
     @staticmethod
     def add_cost(position: Tuple[float, float], radius: float, arr=ndarray, weight: float = 100,
                  safe: bool = True, initial_default_weights: float = 0) -> ndarray:

MapAnalyzer/cext/__init__.py

@@ -0,0 +1 @@
+from .wrapper import astar_path, CMapInfo, CMapChoke