# A* Search on 'Map of Romania' (Russel and Norvig ch3) in Python 3

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]

# 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)]

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:

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
Bucharest

Timisoara, Lugoj, 111
Sibiu, Fagaras, 99
Sibiu, Rimnicu Vilcea, 80
Fagaras, Bucharest, 211
Rimnicu Vilcea, Pitesti, 97
Rimnicu Vilcea, Craiova, 146
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'):

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

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]

• 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. Jun 12, 2017 at 8:53
• A closing bracket is missing in the first link, but I can't edit it as it's only a single character. Jun 12, 2017 at 11:02