IDA* finder

Author: brean

pathfinding/core/node.py

@@ -29,11 +29,15 @@ def __init__(self, x=0, y=0, walkable=True):
         # used for backtracking to the start point
         self.parent = None
 
+        # used for recurion tracking of IDA*
+        self.retain_count = 0
+        self.tested = False
+
     def __lt__(self, other):
         """
         nodes are sorted by f value (see a_star.py)
 
         :param other: compare Node
         :return:
         """
-        return self.f < other.f
+        return self.f < other.f

pathfinding/finder/a_star.py

@@ -2,13 +2,16 @@
 import math
 import heapq # used for the so colled "open list" that stores known nodes
 import logging
+import time # for time limitation
 from pathfinding.core.heuristic import manhatten, octile
 from pathfinding.core.util import backtrace, bi_backtrace
 from pathfinding.core.diagonal_movement import DiagonalMovement
 
 
-# max. amount of tries until we abort the search
+# max. amount of tries we iterate until we abort the search
 MAX_RUNS = float('inf')
+# max. time after we until we abort the search (in seconds)
+TIME_LIMIT = float('inf')
 
 # square root of 2
 SQRT2 = math.sqrt(2)
@@ -20,16 +23,25 @@
 
 class AStarFinder(object):
     def __init__(self, heuristic=None, weight=1,
-                 diagonal_movement=DiagonalMovement.never):
+                 diagonal_movement=DiagonalMovement.never,
+                 time_limit=TIME_LIMIT,
+                 max_runs=MAX_RUNS):
         """
         find shortest path using A* algorithm
         :param heuristic: heuristic used to calculate distance of 2 points
             (defaults to manhatten)
         :param weight: weight for the edges
         :param diagonal_movement: if diagonal movement is allowed
             (see enum in diagonal_movement)
-        :return:
+        :param time_limit: max. runtime in seconds
+        :param max_runs: max. amount of tries until we abort the search
+            (optional, only if we enter huge grids and have time constrains)
+            <=0 means there are no constrains and the code might run on any
+            large map.
         """
+        self.time_limit = time_limit
+        self.max_runs = max_runs
+
         self.diagonal_movement = diagonal_movement
         self.weight = weight
 
@@ -60,8 +72,9 @@ def apply_heuristic(self, node_a, node_b):
         """
         helper function to calculate heuristic
         """
-        return self.weight * \
-            self.heuristic(abs(node_a.x - node_b.x), abs(node_a.y - node_b.y))
+        return self.heuristic(
+            abs(node_a.x - node_b.x),
+            abs(node_a.y - node_b.y))
 
 
     def check_neighbors(self, start, end, grid, open_list,
@@ -100,7 +113,8 @@ def check_neighbors(self, start, end, grid, open_list,
             # can be reached with smaller cost from the current node
             if not neighbor.opened or ng < neighbor.g:
                 neighbor.g = ng
-                neighbor.h = neighbor.h or self.apply_heuristic(neighbor, end)
+                neighbor.h = neighbor.h or \
+                    self.apply_heuristic(neighbor, end) *  self.weight
                 # f is the estimated total cost from start to goal
                 neighbor.f = neighbor.g + neighbor.h
                 neighbor.parent = node
@@ -118,35 +132,47 @@ def check_neighbors(self, start, end, grid, open_list,
         # the end has not been reached (yet) keep the find_path loop running
         return None
 
-
-    def find_path(self, start, end, grid, max_runs=MAX_RUNS):
+    def keep_running(self):
+        """
+        check, if we run into time or iteration constrains.
+        """
+        if self.runs >= self.max_runs:
+            logging.error('{} run into barrier of {} iterations without '
+                          'finding the destination'.format(
+                            self.__name__, self.max_runs))
+            return False
+        if time.time() - self.start_time >= self.time_limit:
+            logging.error('{} took longer than {} '
+                          'seconds, aborting!'.format(
+                            self.__name__, self.time_limit))
+            return False
+        return True
+
+
+    def find_path(self, start, end, grid):
         """
         find a path from start to end node on grid using the A* algorithm
         :param start: start node
         :param end: end node
         :param grid: grid that stores all possible steps/tiles as 2D-list
-        :param max_runs: max. amount of tries until we abort the search
-            (optional, only if we enter huge grids and have time constrains)
-            <=0 means there are no constrains and the code might run on any
-            large map.
         :return:
         """
+        self.start_time = time.time() # execution time limitation
+        self.runs = 0 # count number of iterations
+
         open_list = []
         start.g = 0
         start.f = 0
         heapq.heappush(open_list, start)
 
-        runs = 0 # count number of iterations
         while len(open_list) > 0:
-            runs += 1
-            if max_runs <= runs:
-                logging.error('A* run into barrier of {} iterations without '
-                              'finding the destination'.format(max_runs))
+            self.runs += 1
+            if not self.keep_running():
                 break
 
             path = self.check_neighbors(start, end, grid, open_list)
             if path:
-                return path, runs
+                return path, self.runs
 
         # failed to find path
-        return [], runs
+        return [], self.runs

pathfinding/finder/bi_a_star.py

@@ -10,48 +10,43 @@ class BiAStarFinder(AStarFinder):
     Similar to the default A* algorithm from a_star.
     """
 
-    def find_path(self, start, end, grid, max_runs=MAX_RUNS):
+    def find_path(self, start, end, grid):
         """
         find a path from start to end node on grid using the A* algorithm
         :param start: start node
         :param end: end node
         :param grid: grid that stores all possible steps/tiles as 2D-list
-        :param max_runs: max. amount of tries until we abort the search
-            (optional, only if we enter huge grids and have time constrains)
-            <=0 means there are no constrains and the code might run on any
-            large map.
         :return:
         """
+        self.start_time = time.time() # execution time limitation
+        self.runs = 0 # count number of iterations
+
         start_open_list = []
-        end_open_list = []
         start.g = 0
         start.f = 0
         heapq.heappush(start_open_list, start)
         start.opened = BY_START
 
+        end_open_list = []
         end.g = 0
         end.f = 0
         heapq.heappush(end_open_list, end)
         end.opened = BY_END
 
-        runs = 0 # count number of iterations
         while len(start_open_list) > 0 and len(end_open_list) > 0:
-            runs += 1
-            if 0 < max_runs <= runs:
-                logging.error('Bi-Directional A* run into barrier of {} '
-                              'iterations without finding the '
-                              'destination'.format(max_runs))
+            self.runs += 1
+            if not self.keep_running():
                 break
 
             path = self.check_neighbors(start, end, grid, start_open_list,
                 open_value=BY_START, backtrace_by=BY_END)
             if path:
-                return path, runs
+                return path, self.runs
 
             path = self.check_neighbors(end, start, grid, end_open_list,
                 open_value=BY_END, backtrace_by=BY_START)
             if path:
-                return path, runs
+                return path, self.runs
 
         # failed to find path
-        return [], runs
+        return [], self.runs

pathfinding/finder/dijkstra.py

@@ -1,16 +1,22 @@
-from .a_star import AStarFinder
+from .a_star import AStarFinder, MAX_RUNS, TIME_LIMIT
 from pathfinding.core.diagonal_movement import DiagonalMovement
 
 
 class DijkstraFinder(AStarFinder):
-    def __init__(self, weight=1, diagonal_movement=DiagonalMovement.never):
+    def __init__(self, weight=1,
+            diagonal_movement=DiagonalMovement.never,
+            time_limit=TIME_LIMIT,
+            max_runs=MAX_RUNS):
+        self.time_limit = time_limit
+        self.max_runs = max_runs
+
         def heuristic(dx, dy):
             # return 0, so node.h will always be calculated as 0,
-            # distance cost (node.f) is calculated only from 
+            # distance cost (node.f) is calculated only from
             # start to current point (node.g)
             return 0
 
         super(DijkstraFinder, self).__init__(
             weight=weight, diagonal_movement=diagonal_movement)
 
-        self.heuristic = heuristic
+        self.heuristic = heuristic

pathfinding/finder/ida_star.py

@@ -0,0 +1,116 @@
+from .a_star import *
+from pathfinding.core.node import Node
+
+class IDAStarFinder(AStarFinder):
+    """
+    Iterative Deeping A Star (IDA*) path-finder.
+
+    Recursion based on:
+       http://www.apl.jhu.edu/~hall/AI-Programming/IDA-Star.html
+
+    Path retracing based on:
+     V. Nageshwara Rao, Vipin Kumar and K. Ramesh
+     "A Parallel Implementation of Iterative-Deeping-A*", January 1987.
+     ftp://ftp.cs.utexas.edu/.snapshot/hourly.1/pub/AI-Lab/tech-reports/UT-AI-TR-87-46.pdf
+
+    based on the JavaScript implementation by Gerard Meier (www.gerardmeier.com)
+    """
+    def __init__(self, heuristic=None, weight=1,
+                 diagonal_movement=DiagonalMovement.never,
+                 time_limit=TIME_LIMIT,
+                 max_runs=MAX_RUNS,
+                 track_recursion=True):
+        super(IDAStarFinder, self).__init__(heuristic=heuristic, weight=weight,
+            diagonal_movement=diagonal_movement,
+            time_limit=time_limit,
+            max_runs=max_runs)
+        self.track_recursion = track_recursion
+
+    def search(self, node, g, cutoff, path, depth, end, grid):
+        self.nodes_visited += 1
+
+        if not self.keep_running():
+            # time or iteration limit
+            return
+
+        f = g + self.apply_heuristic(node, end) * self.weight
+
+        # We've searched too deep for this iteration.
+        if f > cutoff:
+            return f
+
+        if node == end:
+            if len(path) < depth:
+                path += [None] * (depth - len(path)  + 1)
+            path[depth] = node
+            return node
+
+        neighbors = grid.neighbors(node, self.diagonal_movement)
+
+        # Sort the neighbors, gives nicer paths. But, this deviates
+        # from the original algorithm - so I left it out
+        # TODO: make this an optional parameter
+        #    def sort_neighbors(a, b):
+        #        return self.apply_heuristic(a, end) - \
+        #            self.apply_heuristic(b, end)
+        #    sorted(neighbors, sort_neighbors)
+        min_t = float('inf')
+        k = 0
+        for neighbor in neighbors:
+            if self.track_recursion:
+                # Retain a copy for visualisation. Due to recursion, this
+                # node may be part of other paths too.
+                neighbor.retain_count += 1;
+                neighbor.tested = True
+
+            t = self.search(neighbor, g + self.calc_cost(node, neighbor),
+                cutoff, path, depth + 1, end, grid)
+
+            if isinstance(t, Node):
+                if len(path) < depth:
+                    path += [None] * (depth - len(path)  + 1)
+                path[depth] = node
+                return t
+
+            # Decrement count, then determine whether it's actually closed.
+            if self.track_recursion:
+                neighbor.retain_count -= 1
+                if neighbor.retain_count == 0:
+                    neighbor.tested = False
+
+            if t < min_t:
+                min_t = t
+
+        return min_t
+
+
+    def find_path(self, start, end, grid):
+        self.start_time = time.time() # execution time limitation
+        self.runs = 0 # count number of iterations
+
+        self.nodes_visited = 0 # for statistics
+
+        self.runs = 0 # count number of iterations
+
+        # initial search depth, given the typical heuristic contraints,
+        # there should be no cheaper route possible.
+        cutoff = self.apply_heuristic(start, end)
+
+        while True:
+            self.runs += 1
+
+            path = []
+
+            # search till cut-off depth:
+            t = self.search(start, 0, cutoff, path, 0, end, grid)
+
+            # If t is a node, it's also the end node. Route is now
+            # populated with a valid path to the end node.
+            if isinstance(t, Node):
+                return path, self.runs
+
+            # Try again, this time with a deeper cut-off. The t score
+            # is the closest we got to the end node.
+            cutoff = t
+
+        return [], self.runs

test/path_test.py

@@ -4,6 +4,7 @@
 from pathfinding.finder.a_star import AStarFinder
 from pathfinding.finder.dijkstra import DijkstraFinder
 from pathfinding.finder.bi_a_star import BiAStarFinder
+from pathfinding.finder.ida_star import IDAStarFinder
 from pathfinding.core.grid import Grid
 from pathfinding.core.diagonal_movement import DiagonalMovement
 
@@ -13,8 +14,9 @@
 # test scenarios from Pathfinding.JS
 scenarios = os.path.join(BASE_PATH, 'path_test_scenarios.json')
 data = json.load(open(scenarios, 'r'))
-finders = [AStarFinder, BiAStarFinder, DijkstraFinder]
+finders = [AStarFinder, BiAStarFinder, DijkstraFinder, IDAStarFinder]
 
+TIME_LIMIT = 10 # give it a 10 second limit.
 
 def test_path():
     """
@@ -25,7 +27,7 @@ def test_path():
             grid = Grid(matrix=scenario['matrix'])
             start = grid.node(scenario['startX'], scenario['startY'])
             end = grid.node(scenario['endX'], scenario['endY'])
-            finder = find()
+            finder = find(time_limit=TIME_LIMIT)
             path, runs = finder.find_path(start, end, grid)
             print(find.__name__)
             print(grid.grid_str(path=path, start=start, end=end))
@@ -40,7 +42,8 @@ def test_path_diagonal():
             grid = Grid(matrix=scenario['matrix'])
             start = grid.node(scenario['startX'], scenario['startY'])
             end = grid.node(scenario['endX'], scenario['endY'])
-            finder = find(diagonal_movement=DiagonalMovement.always)
+            finder = find(diagonal_movement=DiagonalMovement.always,
+                time_limit=TIME_LIMIT)
             path, runs = finder.find_path(start, end, grid)
             print(find.__name__, runs, len(path))
             print(grid.grid_str(path=path, start=start, end=end))