-
Notifications
You must be signed in to change notification settings - Fork 0
/
circular-queue.py
60 lines (50 loc) · 1.87 KB
/
circular-queue.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class CircularQueue:
"""
A circlular queue: a first-in-first-out data structure with a fixed buffer size.
"""
def __init__(self, size):
if type(size) is not int:
raise TypeError("Queue size must be a postive integer.")
if size <= 0:
raise ValueError("Queue size must be a postive integer.")
self.queue = []
self.read_pos = 0
self.write_pos = 0
self.size = size
self.queue = [None for i in range(size)]
def enqueue(self, element):
"""
Adds an element to the buffer if the buffer is not already full.
:param element: The element you wish to add.
:returns: The element itself if it was added, or `None` if the buffer was full.
"""
if self.queue[self.write_pos] is None:
self.queue[self.write_pos] = element
self.write_pos = (self.write_pos + 1) % self.size
return element
else:
# Buffer is full
return None
def dequeue(self):
"""
Removes an element from the buffer if the buffer is not already empty.
:returns: The element removed (or `None` if the buffer was empty).
"""
if self.queue[self.read_pos] is None:
# Buffer is empty
return None
else:
item = self.queue[self.read_pos]
self.queue[self.read_pos] = None
self.read_pos = (self.read_pos + 1) % self.size
return item
def clear(self):
"""Clears the contents of the queue."""
self.queue = [None for i in range(self.size)]
self.read_pos = self.write_pos = 0
def print(self):
"""
Prints the queue to the console as a list, starting with the element that will
be read next.
"""
return self.queue[self.read_pos :] + self.queue[: self.read_pos]