Generalize to non-int-tuple vertices

Author: jobh

a_star/__init__.py

@@ -1,3 +1,3 @@
 __author__ = 'Pablo Galindo Salgado'
 
-from .a_star import a_star_search,heuristic,Node,DijkstraHeap
+from .a_star import a_star_search, DijkstraHeap, Node

a_star/a_star.py

@@ -2,105 +2,96 @@
 import heapq
 
 
-def heuristic(a, b):
-    """
-    The heuristic function of the A* algorithm. In this case the Manhattan distance.
-
-    :param a: Tuple of two ints ( Point A)
-    :param b: Tuple of two ints ( Point B)
-    :returns: integer ( Distance between A nd B )
-    """
-    (x1, y1) = a
-    (x2, y2) = b
-    return abs(x1 - x2) + abs(y1 - y2)
-
-
 class DijkstraHeap(list):
     """
     An augmented heap for the A* algorithm. This class encapsulated the residual logic of
-    the A* algorithm like for example how to manage elements already visited that remain
-    in the heap, elements already visited that are not in the heap and from where we came to
-    a visited element.
+    the A* algorithm like for example how to manage nodes already visited that remain
+    in the heap, nodes already visited that are not in the heap and from where we came to
+    a visited node.
 
     This class will have three main elements:
 
         - A heap that will act as a cost queue (self).
-        - A visited dict that will act as a visited set and as a mapping of the form  point:came_from
-        - A costs dict that will act as a mapping of the form point:cost_so_far
+        - A visited dict that will act as a visited set and as a mapping of the form  vertex:node
     """
-    def __init__(self, first_node = None):
+    def __init__(self):
         self.visited = dict()
-        self.costs = dict()
 
-        if first_node is not None:
-            self.insert(first_node)
-
-    def insert(self, element):
+    def insert(self, node):
         """
-        Insert an element into the Dijkstra Heap.
+        Insert a node into the Dijkstra Heap.
 
-        :param element: A Node object.
+        :param node: A Node object.
         :return: None
         """
 
-        if element.point not in self.visited:
-            heapq.heappush(self,element)
+        if node.vertex not in self.visited:
+            heapq.heappush(self, node)
 
     def pop(self):
         """
-        Pop an element from the Dijkstra Heap, adding it to the visited and cost dicts.
+        Pop a node from the Dijkstra Heap, adding it to the visited dict.
 
         :return: A Node object
         """
 
-        while self and self[0].point in self.visited:
+        while self and self[0].vertex in self.visited:
             heapq.heappop(self)
 
         if self:
             next_elem = heapq.heappop(self)
-            self.visited[next_elem.point] = next_elem.came_from
-            self.costs[next_elem.point] = next_elem.cost_estimate
+            self.visited[next_elem.vertex] = next_elem
             return next_elem
 
+    def backtrack(self, current):
+        """
+        Retrieve the backward path, starting from current.
 
+        :param current: The starting vertex of the backward path
+        :return: A generator of vertices; listify and reverse to get forward path
+        """
+        while current is not None:
+            yield current
+            current = self.visited[current].came_from
 
+Node = collections.namedtuple("Node", ["cost_estimate", "vertex", "came_from"])
 
-Node = collections.namedtuple("Node","cost_estimate point came_from")
-
-def a_star_search(graph, start, end):
+def a_star_search(graph, start, end, heuristic):
     """
     Calculates the shortest path from start to end.
 
-    :param graph: A graph object. The graph object can be anything that implements the following methods:
+    :param graph: A graph object. The graph object can be anything that implements the following methods for a vertex of any comparable and hashable type V:
 
-        graph.neighbors( (x:int, y:int) ) : Iterable( (x:int,y:int), (x:int,y:int), ...)
-        graph.cost( (x:int,y:int) ) : int
+        graph.neighbors( from_vertex:V ) : Iterable( to_vertex:V, to_vertex:V, ...)
+        graph.cost( from_vertex:V, to_vertex:V ) : float
 
-    :param start: Tuple of two ints representing the starting point.
-    :param end: Tuple of two ints representing the ending point.
+    :param start: The starting vertex, as type V.
+    :param end: The ending vertex, as type V.
+    :param heuristic: Heuristic lower-bound cost function taking arguments ( from_vertex:V, end:V ).
     :returns: A DijkstraHeap object.
 
     """
 
-    frontier = DijkstraHeap( Node(heuristic(start, end), start, None) )
+    frontier = DijkstraHeap()
+    frontier.insert( Node(heuristic(start, end), start, None) )
 
     while True:
 
         current_node = frontier.pop()
 
         if not current_node:
             raise ValueError("No path from start to end")
-        if current_node.point == end:
+        if current_node.vertex == end:
             return frontier
 
-        for neighbor in graph.neighbors( current_node.point ):
+        for neighbor in graph.neighbors( current_node.vertex ):
 
-            cost_so_far = current_node.cost_estimate - heuristic(current_node.point, end)
+            cost_so_far = current_node.cost_estimate - heuristic(current_node.vertex, end)
             new_cost = ( cost_so_far
-                         + graph.cost(current_node.point, neighbor)
+                         + graph.cost(current_node.vertex, neighbor)
                          + heuristic(neighbor, end) )
 
-            new_node = Node(new_cost, neighbor, current_node.point)
+            new_node = Node(new_cost, neighbor, current_node.vertex)
 
             frontier.insert(new_node)
 

a_star/tools.py

@@ -100,3 +100,15 @@ def cost(self, from_node, to_node):
         return self.weights.get(to_node, 1)
 
 
+def manhattan_distance(a, b):
+    """
+    The heuristic function of the A* algorithm. In this case the Manhattan distance,
+    to be used in the case where vertices are 2D coordinate tuples (a grid).
+
+    :param a: Tuple of two ints ( Point A)
+    :param b: Tuple of two ints ( Point B)
+    :returns: integer ( Distance between A nd B )
+    """
+    (x1, y1) = a
+    (x2, y2) = b
+    return abs(x1 - x2) + abs(y1 - y2)

examples/maze_solving_example.py

@@ -13,7 +13,7 @@ def from_id_width(point, width):
 if __name__ == '__main__':
 
      import a_star
-     from a_star.tools import GridWithWeights
+     from a_star.tools import GridWithWeights, manhattan_distance
      import random
 
      # Construct a cool wall collection from this aparently arbitraty points.
@@ -41,17 +41,16 @@ def from_id_width(point, width):
                                             (7, 3), (7, 4), (7, 5)]}
 
      # Call the A* algorithm and get the frontier
-     frontier = a_star.a_star_search(graph = graph, start=(1, 4), end=(7, 8))
+     frontier = a_star.a_star_search(graph = graph, start=(1, 4), end=(7, 8), heuristic=manhattan_distance)
 
      # Print the results
 
-     graph.draw(width=5, point_to = frontier.visited, start=(1, 4), goal=(7, 8))
-
-     print("[costs]")
-
-     costs_so_far = { k: v - a_star.heuristic(k, (7, 8)) for k,v in frontier.costs.items() }
-     graph.draw(width=5, number = costs_so_far, start=(1, 4), goal=(7, 8))
+     graph.draw(width=5, point_to = {k:v.came_from for k,v in frontier.visited.items()}, start=(1, 4), goal=(7, 8))
 
      print("[total cost estimates]")
+     costs = {k:v.cost_estimate for k,v in frontier.visited.items()}
+     graph.draw(width=5, number = costs, start=(1, 4), goal=(7, 8))
 
-     graph.draw(width=5, number = frontier.costs, start=(1, 4), goal=(7, 8)) #  cost estimates
+     print("[costs]")
+     costs_so_far = { k: v - manhattan_distance(k, (7, 8)) for k,v in costs.items() }
+     graph.draw(width=5, number = costs_so_far, start=(1, 4), goal=(7, 8))

tests/test_dijkstra_heap.py

@@ -66,7 +66,7 @@ def test_came_from(self):
         came_from_dic = { x:x+1 for x in range(10) }
 
 
-        self.assertTrue( came_from_dic == frontier.visited )
+        self.assertTrue( came_from_dic == {k:v.came_from for k,v in frontier.visited.items()} )
 
     def test_came_from_unique(self):
 
@@ -86,7 +86,7 @@ def test_came_from_unique(self):
         came_from_dic = { x:x+1 for x in range(5) }
 
 
-        self.assertTrue( came_from_dic == frontier.visited )
+        self.assertTrue( came_from_dic == {k:v.came_from for k,v in frontier.visited.items()} )
 
 if __name__ == "__main__":
     unittest.main()

tests/test_maze.py

@@ -2,17 +2,9 @@
 
 import unittest
 import a_star
-from a_star.tools import GridWithWeights
+from a_star.tools import GridWithWeights, manhattan_distance
 
 
-def backtrack( came_from_dict , start, end):
-
-    current = end
-    yield current
-    while current != start:
-        current = came_from_dict[current]
-        yield current
-
 class MazeTests(unittest.TestCase):
 
     def test_maze_1(self):
@@ -22,12 +14,13 @@ def test_maze_1(self):
         weights = {(1,0):20,(3,0) : 2}
         maze.weights = weights
         my_solution = [(3,0),(3,1),(3,2),(3,3),(2,3),(1,3),(0,3),(0,2)]
+
         end = (3,0)
         start = (0,2)
 
         # Call the A* algorithm and get the frontier
-        frontier = a_star.a_star_search(graph = maze, start=start, end=end)
-        solution = list(backtrack(frontier.visited,start,end))
+        frontier = a_star.a_star_search(graph = maze, start=start, end=end, heuristic=manhattan_distance)
+        solution = list(frontier.backtrack(end))
         self.assertTrue( solution == my_solution  )
 
 
@@ -45,9 +38,9 @@ def test_weights_instead_of_walls(self):
                        (2, 0), (1, 0), (0, 0), (0, 1),
                        (0, 2), (0, 3)]
         # Call the A* algorithm and get the frontier
-        frontier = a_star.a_star_search(graph = maze, start=start, end=end)
+        frontier = a_star.a_star_search(graph = maze, start=start, end=end, heuristic=manhattan_distance)
 
-        solution = list(backtrack(frontier.visited,start,end))
+        solution = list(frontier.backtrack(end))
 
         self.assertTrue( solution == my_solution  )