# A* search algorithm: classes

So, this is a ton of code, but that's what I came up with for an efficient and extendable implementation of the A* search algorithm.

The first four classes can be seen as interfaces to show the user what the interface looks like. (Is this a good way of doing that?) I can also provide examples if needed or wanted.

"""
Tools and classes for finding the best path between two points.
"""

from functools import total_ordering

from utils.classes import Abstract
from utils.priority_queue import PriorityQueue
from utils.iterable import CacheDict

@total_ordering
class Position(Abstract):
def __hash__(self):
pass

def __eq__(self, other):
pass

def __lt__(self, other):
pass

def __repr__(self):
pass

class TiledMap(Abstract):
pass

def is_accessible(self, position, for_mover):
pass

def get_cost(self, from_position, to_position, for_mover):
pass

class Heuristic(Abstract):
@staticmethod
def get_cost(from_position, to_position):
pass

class Mover(Abstract):
def can_move_on(self, tile):
pass

class Node(object):
def __init__(self, position):
self.position = position
self.movement_cost = 0
self.heuristic_cost = 0

self._distance_in_tiles = 0
self._predecessor = None

def set_predecessor(self, new_predecessor):
self._distance_in_tiles = new_predecessor.get_distance_in_tiles() + 1
self._predecessor = new_predecessor

def get_distance_in_tiles(self):
return self._distance_in_tiles

def get_predecessor(self):
return self._predecessor

def get_path_score(self):
return self.movement_cost + self.heuristic_cost

class AStarPathFinder(object):
"""
A reusable implementation of the A* search algorithm.
"""
def __init__(self, tiled_map, heuristic, max_distance_in_tiles):
self._tiled_map = tiled_map
self._heuristic = heuristic
self._max_distance_in_tiles = max_distance_in_tiles

self.set_observe_function(lambda node: None)
self._reset_path_data()

def set_observe_function(self, func):
self._observe = func

def _reset_path_data(self):
self._closed = set()
self._open = PriorityQueue(key=lambda node: node.get_path_score())
self._nodes = CacheDict(Node)

def get_path(self, for_mover, start_position, target_position):
self._for_mover = for_mover
self._start_position = start_position
self._target_position = target_position

path = []
if self._is_target_accessible():
if self._has_found_path():
path = self._get_retraced_path()
self._reset_path_data()
return path

def _is_target_accessible(self):
return self._tiled_map.is_accessible(self._target_position, self._for_mover)

def _has_found_path(self):
current_node = self._nodes[self._start_position]

while self._open and self._is_within_max_range(current_node):
current_node = self._open.pop()

if self._target_position in self._closed:
return True

return False

def _is_within_max_range(self, current_node):
return current_node.get_distance_in_tiles() < self._max_distance_in_tiles

def _get_retraced_path(self):
path = []
node = self._nodes[self._target_position]
while node.position != self._start_position:
path.append(node.position)
node = node.get_predecessor()
path.reverse()
return path

node = self._nodes[position]

tentative_movement_cost = self._get_movement_cost(current_node, node)
if tentative_movement_cost < node.movement_cost:
if node in self._open:
self._open.remove(node)
elif position in self._closed:
self._closed.remove(position)

if position not in self._closed and node not in self._open:
node.set_predecessor(current_node)
node.movement_cost = tentative_movement_cost
node.heuristic_cost = self._get_heuristic_cost(position)

self._observe(node)

def _get_movement_cost(self, from_node, to_node):
new_movement_cost = self._tiled_map.get_cost(from_node.position, to_node.position, self._for_mover)
return from_node.movement_cost + new_movement_cost

def _get_heuristic_cost(self, position):
return self._heuristic.get_cost(position, self._target_position)


The code for CacheDict that is used by AStarPathFinder:

class CacheDict(dict):
"""
Acts like a normal dict, but when trying to get an item by a key that
is not contained, the provided function is called with the key as the
argument(s) and it's return value is stored with the key.
value that is returned by the invocation of the
provided function with the key as the argument is stored for the key.

If the key already exists in the dictionary, the stored value is returned.

>>> def very_slow_function(x):
...     a, b = x
...     print('very_slow_function is executed')
...     return (a * b) + a + b
>>> cd = CacheDict(very_slow_function)
>>> cd[(3,4)]
very_slow_function is executed
19
>>> cd[(3,4)]
19
>>> cd
{(3, 4): 19}
"""
def __init__(self, func):
self.func = func

def __missing__(self, key):
ret = self[key] = self.func(key)
return ret


The PriorityQueue looks like this: (Mostly the rewritten version by Gareth Rees from my previous code review)

from heapq import heappush, heappop

class PriorityQueue(object):
"""
A priority queue with O(log n) addition, O(1) membership test and
amortized O(log n) removal.

The key argument specifies a function that returns the score for an
element in the priority queue. (If not supplied, an element is its own score).

>>> q = PriorityQueue([3, 1, 4])
>>> q.pop()
1
2
>>> q.remove(3); q.pop()
4
>>> list(q)
[]
>>> bool(q)
False
>>> q.pop()
Traceback (most recent call last):
...
IndexError: index out of range
>>> q = PriorityQueue('length of text'.split(), key = lambda s:len(s))
>>> q.pop()
'of'
"""

def __init__(self, *args, **kwargs):
self._key = kwargs.pop('key', lambda x:x)
self._heap = []
self._dict = {}
if args:
for elem in args[0]:

def __nonzero__(self):
return bool(self._dict)

def __iter__(self):
return iter(self._dict)

def __contains__(self, element):
return element in self._dict

"""
Add an element to the priority queue.
"""

e = PriorityQueueElement(element, self._key(element))
self._dict[element] = e
heappush(self._heap, e)

def remove(self, element):
"""
Remove an element from the priority queue.
If the element is not a member, raise KeyError.
"""

e = self._dict.pop(element)
e.removed = True

def pop(self):
"""
Remove and return the element with the smallest score from the
priority queue.
"""

while True:
e = heappop(self._heap)
if not e.removed:
del self._dict[e.element]
return e.element

class PriorityQueueElement(object):
"""
A proxy for an element in a priority queue that remembers (and
compares according to) its score.
"""

def __init__(self, element, score):
self.element = element
self._score = score
self.removed = False

def __lt__(self, other):
return self._score < other._score


What could be simpler, more efficient or worded better?

• This is a lot of code. Would you be interested in putting on github and making a link? That might make it easier for people to recommend changes/improvements, too. – user809695 Oct 17 '13 at 5:17
• Good idea, I'll probably do that later. Bitbucket is okay too, right? (hg is easier to use.) – Joschua Oct 26 '13 at 18:10
• Yeah of course; any common version control system would be fine. – user809695 Oct 29 '13 at 4:23

1. You write, "The [abstract] classes can be seen as interfaces to show the user what the interface looks like." But none of these abstract classes have any documentation, so how are people expected to figure out how to use them? I mean, in some cases we might be able to figure it out by reading the code, but what about this method:

def iter_accessible_adjacent(self, position, for_mover):
pass


Presumably this is supposed to return an iterator over the positions that are adjacent to position, and accessible from it, but what are we supposed to pass for the for_mover argument?

2. Why do Position objects need to have a total order? You shouldn't need to compare them, and coordinates have no natural ordering anyway. Also, why do they need a __repr__ method? You don't seem to call repr anywhere in this code.

3. In several cases you could provide actual implementations of the methods, instead of just writing pass. For example, if the idea of the for_mover argument to is_accessible is that it's supposed to be a Mover object, you could write:

def is_accessible(self, position, for_mover):
return for_mover.can_move_on(position)


Indeed, it's not clear to me why anyone would want to implement it any other way. Similarly, Heuristic.get_cost could be:

def get_cost(from_position, to_position):
return 0


Since 0 is always an admissible estimate.

4. Is it important that Mover.can_move_on takes a tile argument rather than a position. What is a tile, anyway?

5. The Heuristic class seems useless: it doesn't have any instance methods. Why not just use a function?

6. The Heuristic.get_cost method seems poorly named: the A* algorithm uses an admissible estimate of the cost, not the cost itself.

7. The TiledMap class seems poorly named: it doesn't seem to have anything to do with tiles or maps. What it knows about is which positions are adjacent, and the cost of travelling between adjacent positions. This kind of data structure is normally known as a weighted graph.

8. Similarly, the Node and AStarPathFinder classes talk about "distance in tiles" but really you are talking about the number of edges in a path in a weighted graph.

9. Your A* algorithm takes a max_distance_in_tiles. But what if I don't want to specify a maximum distance? For the class to be really re-usable, there ought to be a way of avoiding having to specify this.

10. Instead of writing your own Abstract class, why not use the built-in abc.ABCMeta? Similarly, why not decorate your abstract methods with the abc.abstractmethod decorator? Then you'd get useful exceptions if anyone ever tried to instantiate an abstract base class.

11. Position could inherit from collections.abc.Hashable instead of defining its own abstract __hash__ method.

12. In the PriorityQueue class, it would be more general to implement __len__ than __nonzero__ (also, the latter is Python 2 only):

def __len__(self):
return len(self._heap)

13. The name TiledMap suggests that you are going to be looking for paths in a unit-cost grid (as in many 2-dimensional tile-based games). But in this use case, jump point search is a much better algorithm than A*.

• Thank you so much for your review. 4. A tile is something that defines the space at the given position (e.g. there's a Stone so it's inaccessible). I could of course pass the position to the mover, but then the mover would have to ask the Map what's there. – Joschua Nov 15 '13 at 14:38
• 2, 3, 5-12 are done. Also, I'm unsure about 7. - see question in previous comment. Regarding 6. I think about putting get_estimate (""" Returns the admissible heuristic estimate for the given positions. """) into the Position class - is this a good idea? – Joschua Nov 15 '13 at 15:18