Building a Snake game

Let's now build a simple Snake game. As usual, we will be making use of the Canvas widget to provide the platform for our Snake program.

We will use canvas.create_line to draw our snake, and canvas.create_rectangle to draw the snake-food.

Prepare for Lift Off

One of the primary objectives for this project is to introduce the Queue implementation in Python as we used it in conjunction with the threading module.

So far, we have built single-threaded applications. However, threading can be difficult to handle when there is more than one thread in an application, and these threads need to share attributes or resources among them. In this case, you cannot predict the thread execution order at all. OS does it very randomly and swiftly each time.

To handle this complexity, threading module provides some synchronization tools, such as locks, join, semaphores, events, and condition variables. However, it is—in most cases—safer and simpler to use queues. Simply put, a queue is a compound memory structure that is thread-safe; queues effectively channel access to a resource to multiple threads in a sequential order, and are a recommended design pattern that uses threads for most of the scenarios that require concurrency.

The Queue module provides a way to implement different kinds of queuing, such as FIFO (default implementation), LIFO queue, and Priority queue, and this module comes with a built-in implementation of all locking semantics required for running multithreaded programs.

Note

More information about the Queue module can be found in the following link:

http://docs.Python.org/2/library/queue.html

Here's a quick roundup of the basic usage of the Queue module:

myqueue = Queue() #create empty queue
myqueue.put(data)# put items into queue
task = myqueue.get () #get the next item in the queue
myqueue.task_done() # called when a queued task has completed
myqueue.join() # called when all tasks in queue get completed

Let's see a simple demonstration of using queue to implement a multithreaded application (refer to 7.02 threading with queue.py available in the code bundle):

import Queue
import threading
class Worker(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            task = self.queue.get()
            self.taskHandler(task)
    
    def taskHandler(self, job):
        print'doing task %s'%job
        self.queue.task_done()
    def main(tasks):
        queue = Queue.Queue()
        for task in tasks:
            queue.put(task)
    # create some threads and assign them queue
        for i in range(6):
            mythread = Worker(queue)
            mythread.setDaemon(True)
            mythread.start()
        queue.join()
        print'all tasks completed'
    
    if __name__ == "__main__":
            tasks = 'A B C D E F'.split()
            main(tasks)

The description of the code is as follows:

  • We first create a Worker class, which inherits from the threading module of Python. The __init__ method takes in a queue as its argument.
  • We then override the run method of the threading module to get each item from the queue using queue.get(), which is then passed on to the taskHandler method, which actually executes the task specified in the current queue item. In our example, it does nothing useful but printing the name of the task.
  • After the work is done on a particular thread by our taskHandler method, it sends a signal to the queue telling that the task has been completed using the queue.task_done() method.
  • Outside our Worker class, we create an empty queue in our main() method. This queue is populated with a list of tasks using queue.put(task).
  • We then create six different threads and pass this populated queue as its argument. Now that the tasks are handled by the queue, all threads automatically ensure that the tasks are completed in a sequence in which they are encountered by the threads, without causing any deadlocks or two different threads trying to work on the same queued task.
  • At the time of creating each thread, we also create a pool of daemon threads using the mythread.setDaemon(True) method. Doing this passes control to our main program once all threads have completed execution. If you comment out the line, the program would still run, but would fail to exit after all threads have completed executing the tasks in the queue. Without the daemon threads, you'd have to keep track of all the threads and tell them to exit before your program could completely quit.
  • Finally, the queue.join() method ensures that the program flow waits there until the queue is empty.

Now that we know how to use queues to handle multithreaded applications effectively, let's build our Snake game. In its final form, the game would be like the one shown in the following screenshot (refer to the 7.03 game of snake.py Python file available in the code bundle):

Prepare for Lift Off

Engage Thrusters

  1. Let's start coding our game, by first creating a basic GUI class.
    class GUI(Tk):
    def __init__(self, queue):
        Tk.__init__(self)
        self.queue = queue
        self.is_game_over = False
        self.canvas = Canvas(self, width=495, height=305, bg='#FF75A0')
        self.canvas.pack()
        self.snake = self.canvas.create_line((0, 0), (0,0), fill='#FFCC4C', width=10)
        self.food = self.canvas.create_rectangle(0, 0, 0, 0, fill='#FFCC4C', outline='#FFCC4C')
        self.points_earned = self.canvas.create_text(455, 15, fill='white', text='Score: 0')
        self.queue_handler()

    The description of the code is as follows:

    • This code should be mostly familiar to you by now, because we have created similar GUI classes several times in the past.
    • However, rather than passing the root instance as an argument to its __init__ method, our GUI class now inherits from the Tk class. The line Tk.__init__(self)ensures that the root window is available to all methods of this class. This way we can avoid writing root attribute on every line by referencing self.root simply as self.
    • We then initialize the canvas, line (snake), rectangle (food) and text (to display score).
    • We then call the function queueHandler(). This yet to be defined method would be similar to main method defined in the previous queue example. This would be the central method which will process all tasks in the queue. We will come back to define this method once we have added some tasks to the queue.
  2. Now, we will create the Food class, as shown in the following code snippet:
    class Food():
    def __init__(self, queue):
        self.queue = queue
        self.generate_food()

    def generate_food(self):
           x = random.randrange(5, 480, 10)
            y = random.randrange(5, 295, 10)
        self.position = x, y
        self.exppos = x - 5, y - 5, x + 5, y + 5
        self.queue.put({'food': self.exppos})

    The description of the code is as follows:

    • Because we want to process all data centrally from within a queue, we pass the queue as an argument to the __init__ method of the Food class. We choose to run this from the main program thread to demonstrate how a code which is being executed in the main thread can communicate with attributes and methods from other threads.
    • The __init__ method calls another method called generate_food(), which is responsible for generating the snake-food at random positions on the canvas.
    • The generate_food method generates a random (x, y) position on the canvas. However, because the place where the coordinates coincide is just a small point on the canvas, it would be barely visible. We therefore generate an expanded coordinate (self.exppos) ranging from five values less than the (x,y) coordinate up to five values higher than the same coordinate. Using this range, we can create a small rectangle on the canvas which would be easily visible and would represent our food.

    Note

    However, we do not create the rectangle here. Instead, we pass the coordinates for the food (rectangle) into our queue using queue.put. Because this queue is to be made available to all our classes, we will have a centralized worker named queue_handler(), which will process this queue to generate the rectangle from our GUI class later. This is the central idea behind a Queue implementation.

  3. Let's now create the Snake class. We have already passed a task to generate our food to the central queue. However, no thread was involved in the task. We could also generate our Snake class without using threads. However, because we are talking about ways to implement multithreaded applications, let's implement our Snake class to work from a separate thread (refer to 7.03 game of snake.py):
    class Snake(threading.Thread):
    def __init__(self, gui, queue):
        threading.Thread.__init__(self)
        self.gui = gui
        self.queue = queue
        self.daemon = True
        self.points_earned = 0
        self.snake_points = [(495, 55), (485, 55), (475, 55), (465, 55), (455, 55)]
        self.food = Food(queue)
        self.direction = 'Left'
        self.start()

def run(self):
    while not self.gui.is_game_over:
        self.queue.put({'move':self.snake_points})
        time.sleep(0.1)
        self.move()

    The description of the code is as follows:

    • First, we create a class named Snake to run from a separate thread. This class takes the GUI and queue as its input arguments.
    • We initialize the points earned by the player from zero and set the initial location of the snake using the attribute self.snake_points.
    • Finally, we start the thread and create an infinite loop to call the move() method at small intervals. During every run of the loop, the method populates the queue with a dictionary having the key as 'move' and the value equal to the updated position of the snake through the self.snake_points attribute.
  4. In this step, we will be making the snake move.

    The thread initialized above calls the Snake class move() method to move the snake around on the canvas. However, before we can move the snake, we need to know the direction in which the snake should move. This obviously depends on the particular key pressed by the user (Left/Right/Top/Down key).

    Accordingly, we need to bind these four events to the Canvas widget. We will do the actual binding later. However, we can now create a method named called key_pressed, which takes the key_press event itself as its argument and sets the direction value according to the key that is pressed.

    def key_pressed(self, e):
	self.direction = e.keysym

    Now that we have the directions, let's code the move method:

    def move(self):
    new_snake_point = self.calculate_new_coordinates()
    if self.food.position == new_snake_point:
        self.points_earned += 1
        self.queue.put({'points_earned':self.points_earned })
        self.food.generate_food()
    else:
        self.snake_points.pop(0)
        self.check_game_over(new_snake_point)
        self.snake_points.append(new_snake_point)

def calculate_new_coordinates(self):
    last_x, last_y = self.snake_points[-1]
    if self.direction == 'Up':
        new_snake_point = last_x, (last_y - 10)
    elif self.direction == 'Down':
        new_snake_point = last_x, (last_y + 10)
    elif self.direction == 'Left':
        new_snake_point = (last_x - 10), last_y
    elif self.direction == 'Right':
        new_snake_point = (last_x + 10), last_y
    return new_snake_point

def check_game_over(self, snake_point):
    x,y = snake_point[0], snake_point[1]
    if not -5 < x < 505 or not -5 < y < 315 or snake_point in self.snake_points:
            self.queue.put({'game_over':True})

    The description for the code is as follows:

    • First, the move method obtains the latest coordinates for the snake depending on the keyboard event. It uses a separate method called calculate_new_coordinates to get the latest coordinates.
    • It then checks if the location of the new coordinates coincide with the location of the food. If they match, it increases the score of the player by one and calls the Food class generate_food method to generate a new food at a new location.
    • If the current point does not coincide with the food coordinates, it deletes the last item from the snake coordinates using self.snake_points.pop(0).
    • Then, it calls another method named check_game_over to check if the snake collides against the wall or against itself. If the snake does collide, it appends a new dictionary item in the queue with the value 'game_over':True.
    • Finally, if the game is not over, it appends the new position of the snake to the list self.snake_points. This is automatically added to the queue, because we have defined self.queue.put({'move':self.snake_points}) in the Snake class's run() method to update every 0.1 seconds as long as the game is not over.
  5. Now, let's create the Queue handler.

    We now have a Food class feeding the centralized queue from the main program thread. We also have the Snake class adding data to the queue from one thread and a GUI class running the queue_handler method from another thread. So, the queue is the central point of interaction between these three threads.

    Now, it is time to handle these data to update the content on the canvas. We accordingly define the queue_handler() method in our GUI class to work on items in the queue.

    def queue_handler(self):
    try:
        while True:
            task = self.queue.get(block=False)
            if task.has_key('game_over'):
                self.game_over()
            elif task.has_key('move'):
                points = [x for point in task['move'] for x in point]
                self.canvas.coords(self.snake, *points)
            elif task.has_key('food'):
                self.canvas.coords(self.food, *task['food'])
            elif task.has_key('points_earned'):
             self.canvas.itemconfigure(self.points_earned, text='Score:{}'.format(task['points_earned']))
           self.queue.task_done()
    except Queue.Empty:
        if not self.is_game_over:
            self.canvas.after(100, self.queue_handler)

    The description for the code is as follows:

    • The queue_handler method gets into an infinite loop looking for tasks in the queue using task = self.queue.get(block=False). If the queue becomes empty, the loop is restarted using canvas.after.
    • Once a task is fetched from the queue, the method checks its key.
    • If the key is 'game_over', it calls another method named game_over() that we defined next.
    • If the key of task is 'move', it uses canvas.coords to move the line to its new position.
    • If the key is 'points_earned', it updates the score on the canvas.
    • When execution of a task completes, it signals the thread with the task_done() method.

    Note

    queue.get can take both block=True (default) and block=False as its argument.

    When the block is set to False, it removes and returns an item from the queue, if available. If the queue is empty, it raises Queue.Empty. When the block is set to True, queue.get fetches an item from the queue by suspending the calling thread, if required, until an item is available.

  6. In this step, we will code the method to handle the game_over feature for the game.

    The queue_handler method calls the game_over method in case of a matching queue key:

    def game_over(self):
    self.is_game_over = True
    self.canvas.create_text(200, 150, fill='white', text='Game Over') 
    quitbtn = Button(self, text='Quit', command =self.destroy)
    self.canvas.create_window(200, 180, anchor='nw', window=quitbtn)

    The description for the code is as follows:

    • We first set the game_over attribute to True. This helps us exit out of the infinite loop of queue_handler. Then, we add a text on the canvas displaying the content Game Over.
    • We also add a Quit button inside the canvas, which has a command callback attached to quit the root window.

      Tip

      Take a note of how to attach other widgets inside the canvas widget.

  7. Let's Run the game. The game is now ready. To run the game, we create a function outside all other classes named main():
    def main():
    queue = Queue.Queue()
    gui     = GUI(queue)
    snake = Snake(gui, queue)
    gui.bind('<Key-Left>', snake.key_pressed)
    gui.bind('<Key-Right>', snake.key_pressed)
    gui.bind('<Key-Up>', snake.key_pressed)
    gui.bind('<Key-Down>', snake.key_pressed)
    gui.mainloop()

if __name__ == '__main__':
    main()

    We create an empty queue, and pass it as an argument to all three of our classes so that they can feed tasks into the queue. We also bind the four directional keys to the key_pressed method, which is defined earlier within our Snake class.

Objective Complete – Mini Debriefing

Our game is now functional. Go try your hands at controlling the snake, while keeping its stomach filled.

To summarize, we created three classes such as Food, Snake, and GUI. These three classes feed information about the task related to their class to a centralized queue which is passed as an argument to all the classes.

Then, we create a centralized method named queue_handler, which handle tasks from the queue by polling tasks one at a time and completing it in a non-blocking manner.

The game could have been implemented without threads and queues, but it would have been slower, longer, and more complex. By using queues to manage data from multiple threads effectively, we have been able to contain the program to less than 150 lines of code.

Hopefully, you should now be able to implement queues for managing other programs that you design at your work.

..................Content has been hidden....................

You can't read the all page of ebook, please click here login for view all page.
Reset
18.216.77.153