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]

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)