Using BFS to solve 8-puzzle game using Python 2

Question

I'm trying to solve the 8-puzzle game using BFS, DFS and A* algorithms implemented using Python 2.7. For now, I have managed to solve a couple of test cases using BFS and I want to know how I can improve the implementation of the algorithm as well as the structure of my program.

The program currently is divided into 4 files:

class Board:
 """ The Board class represents the low-level physical configuration of the 
 8-puzzle game. """
 # The 8-puzzle board can be represented as a list of length 8
 def __init__(self, initial_values=[]):
 self.value = initial_values
 def __eq__(self, other): 
 return self.value == other.value
 def __str__(self):
 return str(self.value)
 def __hash__(self):
 return hash(str(self))
 # If 0 is in the top most block, then up is invalid
 def up(self):
 pos = self.value.index(0)
 if pos in (0, 1, 2):
 return None
 else:
 new_val = list(self.value)
 new_val[pos], new_val[pos-3] = new_val[pos-3], new_val[pos]
 return new_val
 # If 0 is in the bottom most block, then up is invalid
 def down(self):
 pos = self.value.index(0)
 if pos in (6, 7, 8):
 return None
 else:
 new_val = list(self.value)
 new_val[pos], new_val[pos+3] = new_val[pos+3], new_val[pos]
 return new_val
 # If 0 is in the left most block, then up is invalid
 def left(self):
 pos = self.value.index(0)
 if pos in (0, 3, 6):
 return None
 else:
 new_val = list(self.value)
 new_val[pos], new_val[pos-1] = new_val[pos-1], new_val[pos]
 return new_val
 # If 0 is in the right most block, then up is invalid
 def right(self):
 pos = self.value.index(0)
 if pos in (2, 5, 8):
 return None
 else:
 new_val = list(self.value)
 new_val[pos], new_val[pos+1] = new_val[pos+1], new_val[pos]
 return new_val

Then we have state.py to represent high-level moves for the game:

from board import Board
class State:
 """ Handles the state of the game """
 def __init__(self, initial_state=[]):
 self.current = Board(initial_state)
 def __eq__(self, other): 
 return self.current == other.current
 def __str__(self):
 return str(self.current)
 def __hash__(self):
 return hash(str(self))
 def up(self):
 up = self.current.up()
 if up is not None:
 return State(up)
 else:
 return self
 def down(self):
 down = self.current.down()
 if down is not None:
 return State(down)
 else:
 return self
 def left(self):
 left = self.current.left()
 if left is not None:
 return State(left)
 else:
 return self
 def right(self):
 right = self.current.right()
 if right is not None:
 return State(right)
 else:
 return self
 def successors(self):
 succ = []
 up = self.current.up()
 if up != None:
 succ.append(State(up))
 down = self.current.down()
 if down != None:
 succ.append(State(down))
 left = self.current.left()
 if left != None:
 succ.append(State(left))
 right = self.current.right()
 if right != None:
 succ.append(State(right))
 return succ

Then the search.py module contains the algorithms (only BFS for now):

from collections import namedtuple
def goal_test(state):
 return str(state) == str(range(0, 9))
# BFS Search
def bfs(start):
 """ 
 Performs breadth-first search starting with the 'start' as the beginning
 node. Returns a namedtuple 'Success' which contains namedtuple 'position'
 (includes: node, cost, depth, prev), 'max_depth' and 'nodes_expanded'
 if a node that passes the goal test has been found.
 """
 # SearchPos used for bookeeping and finding the path:
 SearchPos = namedtuple('SearchPos', 'node, cost, depth, prev')
 # Initial position does not have a predecessor
 position = SearchPos(start, 0, 0, None)
 # frontier contains unexpanded positions
 frontier = [position]
 explored = set()
 while len(frontier) > 0:
 # current position is the first position in the frontier
 position = frontier.pop(0)
 node = position.node
 # goal test: return success if True
 if goal_test(node):
 max_depth = max([pos.depth for pos in frontier])
 Success = namedtuple('Success', 
 'position, max_depth, nodes_expanded')
 success = Success(position, max_depth, len(explored))
 return success
 # expanded nodes are added to explored set
 explored.add(node)
 # All reachable positions from current postion is added to frontier
 for neighbor in node.successors():
 new_position = SearchPos(neighbor, position.cost + 1,
 position.depth + 1, position)
 frontier_check = neighbor in [pos.node for pos in frontier]
 if neighbor not in explored and not frontier_check:
 frontier.append(new_position)
 # the goal could not be reached.
 return None

Finally, solver.py is used to carry out the search:

from state import State
import search
import time
import resource
def trace_path(last_pos):
 pos = last_pos.prev
 next_pos = last_pos
 path = []
 while pos != None:
 if pos.node.up() == next_pos.node:
 path.append("Up")
 elif pos.node.down() == next_pos.node:
 path.append("Down")
 elif pos.node.left() == next_pos.node:
 path.append("Left")
 elif pos.node.right() == next_pos.node:
 path.append("Right")
 pos = pos.prev
 next_pos = next_pos.prev
 return path[::-1]
start_time = time.time()
config = [1,2,5,3,4,0,6,7,8]
game = State(config)
result = search.bfs(game)
final_pos = result.position
max_depth = result.max_depth
nodes_expanded = result.nodes_expanded
print "path_to_goal:", trace_path(final_pos)
print "cost_of_path:", final_pos.cost
print "nodes_expanded:", nodes_expanded
print "search_depth:", final_pos.depth
print "max_search_depth:", max_depth
print "running_time:", time.time() - start_time
print "max_ram_usage", resource.getrusage(1)[2]

• πάντα ῥεῖ: "For now, I have managed to solve a couple of test cases ..." Does that mean your code doesn't work for all testcases?
• Ramiz Rahman: Works on the ones that I have come up with, but haven't tested it against the unseen test cases. The trouble is my DFS implementation is extremely slow for some reason, even though the above code is quite fast on the same test cases. I wanted to figure out if there is any problem here and make corrections so that I can improve and tackle the more difficult problem without directly taking help on that one.
• jschnei: If you want help on your DFS implementation, you should include it here! Incidentally, it is not surprising that DFS would take much longer than BFS/A* in this case; DFS might explore up to the entire graph before finding the answer, while BFS only explores up to however many moves it needs to.
• Ramiz Rahman: @jschnei Thanks, I will be posting help for the DFS implementation. However, I do want help with the overall structure of my code and want to know if my BFS code is well written or not and how I can improve upon it. As for DFS being slow on this particular problem, I'm well aware of it and know that almost every possible node needs to be visited before finding the solution; even so, it shouldn't take more than 60 seconds yet after 30 minutes the code does not stop execution and there is no infinite loop.

Answer 1

State

This class does very little useful work, and can be integrated in the Board class

Board

Here a docstring with an explanation about the internal representation of the board can be helpful. I managed to deduce that you structured your board like this:

+---+---+---+
| 0 | 1 | 2 |
+---+---+---+
| 3 | 4 | 5 |
+---+---+---+
| 6 | 7 | 8 |
+---+---+---+

You use 0 as the placeholder for the free position, and the rest is just a flattened list of this board.

It will help you next time you go visit this, that you need to make it clear to anyone wanting to do maintenance on this Board. Your users don't need to be aware of this representation.

The drawback of this, is the inflexibility. Imagine you want to use another board size, then you'll need to make huge changes to this class. This can be made easier like this:

class Board(object):
 def __init__(self, initial_values=(), dimensions=(3,3)):
 rows, columns = dimensions
 assert len(initial_values) == rows * columns
 self._values = tuple(initial_values)
 self._rows = rows
 self._columns = columns

Here I use a tuple instead of a list to make the board immutable. This way you don't need the conversion to str in __hash__.

Instead of hardcoding which coordinates can not do certain moves, I would provide 4 little helper methods

 def _top_row(self):
 return range(self._rows)
 def _bottom_row(self):
 bottom_left = (self._rows - 1) * self._columns
 return range(bottom_left, self._rows * self._columns, self._columns)
 def _left_row(self):
 return range(0, self._rows * self._columns, self._columns)
 def _right_row(self):
 return range(self._columns - 1, self._columns * self._rows, self._columns)

and another little helper for the current position.

 @property
 def _current(self):
 return self.values.index(0)

For the moves on the board, I would raise an Exception to signify an illegal move, instead of returning None.

 def up(self):
 """
 if `up` is a valid move, return a new board position
 with the empty piece shifted up. 
 If `up` is invalid, raise an `IllegalMove`
 """
 pos = self._current
 if pos in self._top_row():
 raise IllegalMove
 new_pos = pos - self._columns
 new_board = list(self._values)
 new_board[pos], new_board[new_pos] = new_board[new_pos], new_board[pos]
 return Board(new_board, dimensions=(self._rows, self._columns))

To define the valid successors, I would use a generator, instead of returning a list. This makes this very simple

 def valid_successors(self):
 moves = self.up, self.down, self.left, self.right
 for move in moves:
 try:
 yield move()
 except IllegalMove:
 pass

The goal_test can also be part of this Board:

 def goal_test(self):
 return self._values == sorted(self._values)

Answer 2

I have noticed only one clue in your code,

When you check if a neighbour is in the frontier

With this code :

 frontier_check = neighbor in [pos.node for pos in frontier]

Membership testing in a list is slower than membership testing in a set. consider this question for more information.

I found useful tips in a discussion on edx platform about this assignment here is the link.

Here is what he said :

When I was just starting, I used a normal python list for keeping track of explored nodes. It ran extremely slow when the depth of the solution was bigger than 6 or 7. I started investigating a little bit and found out that 99% of the cpu time was used for membership testing. So I changed the list for a set which improved performance, but the program was still taking too much to find the solution. This is because my frontier class used a deque which also has exponential membership testing times. So I made an implementation where I kept track of states in a dictionary, which has linear time complexity for membership testing, and the queue was just being used for the order in which nodes are expanded, and it now calculates the solution in less than 10 seconds for any problem. In conclusion, don't use 'node in list' when 'list' has exponential membership testing. Data structures that don't have exponential membership testing: non ordered lists like Sets, or Hash Maps/Dictionaries

here is a working implementation for me :

with a frontier and frontier_set where I keep track of explored node for future testing . here is a correction of your code :

just add 2 lines of code and remove 1 and it speed up the whole process.

from collections import deque
def bfs(start):
 """ 
 Performs breadth-first search starting with the 'start' as the beginning
 node. Returns a namedtuple 'Success' which contains namedtuple 'position'
 (includes: node, cost, depth, prev), 'max_depth' and 'nodes_expanded'
 if a node that passes the goal test has been found.
 """
 # SearchPos used for bookeeping and finding the path:
 SearchPos = namedtuple('SearchPos', 'node, cost, depth, prev')
 # Initial position does not have a predecessor
 position = SearchPos(start, 0, 0, None)
 # frontier contains unexpanded positions
 frontier = deque([position])
 frontier_set = {position}
 explored = set()
 while frontier:
 # current position is the first position in the frontier
 position = frontier.popleft()
 node = position.node
 # goal test: return success if True
 if goal_test(node):
 max_depth = max(pos.depth for pos in frontier)
 Success = namedtuple('Success', 
 'position, max_depth, nodes_expanded')
 success = Success(position, max_depth, len(explored))
 return success
 # expanded nodes are added to explored set
 explored.add(node)
 # All reachable positions from current postion is added to frontier
 for neighbor in node.successors():
 new_position = SearchPos(neighbor, position.cost + 1,
 position.depth + 1, position)
 if neighbor not in explored and new_position not in frontier_set:
 frontier.append(new_position)
 frontier_set.add(new_position)

• Maarten Fabré: isn't a collections.deque a better container for frontier than a list. And your while len(frontier) > 0: can be simplified to while frontier:
• Maarten Fabré: there is no need for the [] in max([pos.depth for pos in frontier])
• Espoir Murhabazi: Thankd for those comment !! Suggest an edit i will approve them