In Python, I am implementing the A* search algorithm for solving the Tile Problem. I have the following Node class which holds the state as a tuple of tuple. For example the initial state is:
initial= ((7,2,4),(5,0,6),(8,3,1));#empty tile is marked with value 0
Below is the Node class,
class Node:
def __init__(self, state, parent=None, action=None, pathCost=0, emptyTileI=1,emptyTileJ=1):
self.state = state
self.parent = parent
self.action = action
self.pathCost = pathCost
self.emptyTileI=emptyTileI;#row index of empty tile
self.emptyTileJ=emptyTileJ;#column index of empty tile
def expand(self, problem):
children=[];#empty list of children nodes initially.
#after doing some work....
return children #a list of Node objects, one for each successor states, parent of the created nodes is self
def getState(self):
return self.state;
def getPathCost(self):
return self.pathCost;
def getParent(self):
return self.parent;
I also have a TileProblem class which holds the heuristics that will be used by A* as the following,
## very generic Problem superclass, subclasses can overwrite the goal test and the constructor, and should also add a successors function
class Problem:
def __init__(self, initial, goal=None):
self.initial = initial
self.goal = goal
def goalTest(self, state):
return state == self.goal
class TileProblem(Problem):
def __init__(self, initial, goal=None):
self.initial = initial
self.goal = goal
#used in priority queue
"(Uses Accumulated Manhattan Distance heuristic)"
def fAMD(self,element):
return self.g(element)+self.hAMD(element)
"(Uses Number of Misplaced Tiles heuristic)"
def fMT(self,element):
return self.g(element)+self.hMT(element)
#backward cost. #When this function is executed, the parent of the element has been already updated (by expand function in parent Node)
def g(self,element):
return element.pathCost
#forward cost
"(Accumulated Manhattan Distance heuristic)"
def hAMD(self,element):
goalState=self.goal
elementState=element.getState()
"We will calculate the accumulated Manhattan distance between goal state and element state"
accumulatedSum=0;
for i,tgoal in enumerate(goalState):
for j,vGoal in enumerate(tgoal):
for m, tElement in enumerate(elementState):
for n, vElement in enumerate(tElement):
if vGoal==vElement:
accumulatedSum=accumulatedSum + abs(i-m)+abs(j-n);#Manhattan distance
return accumulatedSum
#forward cost
"(Number of Misplaced Tiles heuristic)"
def hMT(self,element):
goalState=self.goal
elementState=element.getState()
"We will calculate the number of misplaced tiles between goal state and element state"
misplacedTileSum=0;
for m, tElement in enumerate(elementState):
for n, vElement in enumerate(tElement):
if(vElement!=goalState[m][n]):
misplacedTileSum=misplacedTileSum+1;
return misplacedTileSum;
The priority Queue Class as the following,
class PriorityQueue:
def __init__(self, f):
self.f = f ## save the function for later, when we apply it to incoming items (nodes)
self.q = []#defaultdict(list)
self.maximal_size=0
def push(self, item):
insort(self.q, (self.f(item), item))
self.updateMaximalSize();
def pop(self):
return self.q.pop(0)[1]
def empty(self):
return self.q==[];
def getMaximalSize(self):
return self.maximal_size
def updateMaximalSize(self):
if(len(self.q)>self.maximal_size):
self.maximal_size=len(self.q)
def getF(self):
return self.f;
# def printQ(self):
# for t in self.q:
# print("("+repr(t[0])+","+repr(t[1].getState())+")");
#returns the priority value of an element if it exists or None otherwise
#Needed in A* to judge whether to push an element to the queue or not by comparing its current priority to the updated one
def getPriorityValue(self,item):
for e in self.q:
if e[1].getState()==item.getState():
return e[0];
return None
#updates the priority value of an element in the queue. Needed in A* to decrease the priority value of an element
def updatePriorityValue(self,item):
self.q = [(v,i) if (i.getState() != item.getState()) else (self.f(item),item) for (v, i) in self.q]
self.updateMaximalSize();
my A* algorithm,
def aStarSearch(problem,frontier):
closed=[];#explored set
frontier.push(Node(problem.initial))
print("In A* Search, Popping The Following Nodes (States) in-order:")
while not frontier.empty():
node=frontier.pop();
print(node.getState()), #printing the history of popped nodes(states)
closed.append(node.getState()); # add it to closed state to prevent processing it again
if problem.goalTest(node.getState()): #if this is a goal node
return node; #just return it
children=node.expand(problem); #otherwise, lets expand it
for c in children: #looping over the children
if(c.getState() in closed): #already popped from the queue and closed
continue; #ignore it
priorityValue=frontier.getPriorityValue(c); #the child priority value in the queue
if(priorityValue==None): #not in the queue
frontier.push(c) #add it to the queue
elif (frontier.getF(c) < priorityValue): #in the queue but this is a better path
frontier.updatePriorityValue(c); #decrease the priority in the queue
#frontier.printQ();
return None;
Finally, in my main,
initial= ((7,2,4),(5,0,6),(8,3,1));#empty tile is marked with value 0
goal=((1,2,3),(4,5,6),(7,8,0));
"A* Graph Search using Accumulated Manhattan Distance heuristic"
print("A* Graph Search using Accumulated Manhattan Distance heuristic:")
tileAStarSearch = TileProblem(initial,goal);
frontier= PriorityQueue(tileAStarSearch.fAMD); #accumulated Manhattan distance heuristic
aStarResult=aStarSearch(tileAStarSearch, frontier);
When I run the code, I get the following error:
aStarResult=aStarSearch(tileAStarSearch, frontier);
File "C:\Users\zjalmahmoud\workspace\Tile_Problem\search.py", line 211, in aStarSearch
frontier.push(c) #add it to the queue
File "C:\Users\zjalmahmoud\workspace\Tile_Problem\utilsSimple.py", line 61, in push
insort(self.q, (self.f(item), item))
TypeError: unorderable types: Node() < Node()
I do not understand the cause of this problem. Why should the insort care about the items. I only expect it to push the item and sort by the "priority" value which is an integer in my case (the accumulated Manhattan Distance). How can I solve this problem? Thanks.
You have nodes with equal priority in the queue. Python then tries to order the (priority, node) tuple by comparing the nodes. That's because tuples are compared in lexicographical order, just like you'd sort names. With two names, if the first letters match, you compare the second letter, etc, until you have different letters and can order those, comparing tuples works the same way. If the priorities match, the nodes are compared.
Either make your nodes orderable (you can use the functools.total_ordering() decorator plus an __eq__ and a __lt__ method) or insert a counter value into the priority queue tuples to break the tie:
# at the top
from itertools import count
class PriorityQueue:
def __init__(self, f):
self.f = f ## save the function for later, when we apply it to incoming items (nodes)
self.q = []#defaultdict(list)
self.maximal_size=0
self._counter = count()
def push(self, item):
insort(self.q, (self.f(item), next(self._counter), item))
self.updateMaximalSize()
and update the rest of the PriorityQueue() class to look for the item at index 2 (or better still, at index -1). Update the list comprehension in the updatePriorityValue() method to unpack the queue items to (v, c, i) and to include those again in the left-hand-side expression.
The counter (produced by itertools.count()) inserts an ever increasing integer, so there will never be two tuples with (priority, counter, item) where both the priority and the counter are equal; thus the items are never compared.
This means that for equal priority, items that were inserted later win the tie. Use -next(self._counter) instead if you want items inserted earlier to win instead.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With