8
\$\begingroup\$

I've implemented A* search using Python 3 in order to find the shortest path from 'Arad' to 'Bucharest'. The graph is the map of Romania as found in chapter 3 of the book: "Artificial Intelligence: A Modern Approach" by Stuart J. Russel and Peter Norvig. Please have a look at my code and provide your feedback.

from priority_queue import *
import colorama
from colorama import Fore, Back, Style

visualize is for showing the progress of the algorithm:

# HELPER
def visualize(frontier):
    colorama.init()
    for i in range(len(frontier._queue)):
        text = str(frontier._queue[i])
        if i == 0:
            print(Back.RED + Fore.WHITE + text + Style.RESET_ALL)
        else:
            print(text)
    print()

I used this to build a graph:

def build_graph_weighted(file):
    """Builds a weighted, undirected graph"""
    graph = {}
    for line in file:
        v1, v2, w = line.split(',')
        v1, v2 = v1.strip(), v2.strip()
        w = int(w.strip())
        if v1 not in graph:
            graph[v1] = []
        if v2 not in graph:
            graph[v2] = []
        graph[v1].append((v2,w))
        graph[v2].append((v1,w))
    return graph

For the heuristics, I've used a text file containing the straight line distance from a particular city to Bucharest

# Helper methods for A*
def build_heuristic_dict():
    h = {}
    with open("sld_to_bucharest", 'r') as file:
        for line in file:
            line = line.strip().split(",")
            node = line[0].strip()
            sld = int(line[1].strip())
            h[node] = sld
    return h

def heuristic(node, values):
    return values[node]

Here's the main algorithm:

# A* search
def a_star(graph, start, dest, visualization=False):
    """Performs a* search on graph 'graph' with
        'start' as the beginning node and 'dest' as the goal.
        Returns shortest path from 'start' to 'dest'.
        If 'visualization' is set to True, then progress of
        algorithm is shown."""

    frontier = PriorityQueue()

    # uses helper function for heuristics
    h = build_heuristic_dict()

    # path is a list of tuples of the form ('node', 'cost')
    frontier.insert([(start, 0)], 0)
    explored = set()

    while not frontier.is_empty():

        # show progress of algorithm
        if visualization:
            visualize(frontier)

        # shortest available path
        path = frontier.remove()

        # frontier contains paths with final node unexplored
        node = path[-1][0]
        g_cost = path[-1][1]
        explored.add(node)

        # goal test:
        if node == dest:
            # return only path without cost
            return [x for x, y in path]

        for neighbor, distance in graph[node]:
            cumulative_cost = g_cost + distance
            f_cost = cumulative_cost + heuristic(neighbor, h)
            new_path = path + [(neighbor, cumulative_cost)]

            # add new_path to frontier
            if neighbor not in explored:
                frontier.insert(new_path, f_cost)

            # update cost of path in frontier
            elif neighbor in frontier._queue:
                frontier.insert(new_path, f_cost)
                print(path)
    return False

For running the algorithm:

with open('graph.txt', 'r') as file:
    lines = file.readlines()

start = lines[1].strip()
dest = lines[2].strip()
graph = build_graph_weighted(lines[4:])
print(a_star(graph, start, dest, True), "\n\n")

And here are the files:

Graph.txt:

20 23
Arad
Bucharest

Arad, Zerind, 75
Arad, Sibiu, 140
Arad, Timisoara, 118
Zerind, Oradea, 71
Oradea, Sibiu, 151
Timisoara, Lugoj, 111
Sibiu, Fagaras, 99
Sibiu, Rimnicu Vilcea, 80
Lugoj, Mehadia, 70
Fagaras, Bucharest, 211
Rimnicu Vilcea, Pitesti, 97
Rimnicu Vilcea, Craiova, 146
Mehadia, Dobreta, 75
Bucharest, Pitesti, 101
Bucharest, Urziceni, 85
Bucharest, Giurglu, 90
Pitesti, Craiova, 138
Craiova, Dobreta, 120
Urziceni, Hirsova, 98
Urziceni, Vaslui, 142
Hirsova, Eforie, 86
Vaslui, Lasi, 92
Lasi, Neamt, 87

And finally the heuristics doc ('sld_to_bucharest.txt'):

Arad, 366
Bucharest, 0
Craiova, 160
Dobreta, 242
Eforie, 161
Fagaras, 176
Giurgiu, 77
Hirsowa, 151
Lasi, 226
Lugoj, 244
Mehadia, 241
Neamt, 234
Oradea, 380
Pitesti, 100
Rimnicu Vilcea, 193
Sibiu, 253
Timisoara, 329
Urziceni, 80
Vaslui, 199
Zerind, 374
\$\endgroup\$
1
  • \$\begingroup\$ it is showing an error with priority_queue module, is it implemented separately? \$\endgroup\$
    – mansi
    Mar 15, 2018 at 16:28

1 Answer 1

5
\$\begingroup\$

Just a couple of observations:

  1. There's a comment documenting the elements of the priority queue:

    # path is a list of tuples of the form ('node', 'cost')
    frontier.insert([(start, 0)], 0)
    

    This kind of data structure can be made clearer using collections.namedtuple:

    from collections import namedtuple
    
    # Position in A* search: reached 'node' having paid cumulative 'cost'.
    SearchPos = namedtuple('SearchPos', 'node cost')
    

    Then you can write:

    node = path[-1].node
    g_cost = path[-1].cost
    
  2. Each position in the A* search is stored as a list of nodes visited along the path to that position. The trouble with this approach is that constructing the new path requires copying the old path:

    new_path = path + [(neighbor, cumulative_cost)]
    

    The longer the path gets, the longer it takes longer to copy it out, and the more memory is needed to store all the paths in the queue. This leads to quadratic runtime performance.

    Instead of copying the path, remember the previous position on the path:

    # Position in A* search: reached 'node' having paid cumulative 'cost';
    # 'prev' is the previous position on the lowest-cost path to 'node'
    # (or None if this is the first node on the path).
    SearchPos = namedtuple('SearchPos', 'node cost prev')
    

    Then constructing the new position looks like this:

    new_position = SearchPos(neighbour, cumulative_cost, position)
    

    and when you find the goal, you can reconstruct the path by working backwards along the chain of positions:

    if position.node == dest:
        path = []
        while position is not None:
            path.append(position.node)
            position = position.prev
        return path[::-1]
    
\$\endgroup\$
2
  • \$\begingroup\$ Thank you so much. I added the changes and learned something new. Just one point I need a little clarification on. The position in SearchPos is a reference to a previous node, so in the frontier, we are essentially adding a linked list? Also, it would be very helpful if you could give me some feedback on my code structure, variable names, etc. and how I can improve in that department. Thank you once again. \$\endgroup\$ Jun 12, 2017 at 8:53
  • \$\begingroup\$ A closing bracket is missing in the first link, but I can't edit it as it's only a single character. \$\endgroup\$
    – mkrieger1
    Jun 12, 2017 at 11:02

Not the answer you're looking for? Browse other questions tagged or ask your own question.