forked from adnanaziz/EPIJudge
-
Notifications
You must be signed in to change notification settings - Fork 0
/
circular_queue.py
68 lines (51 loc) · 2.02 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
61
62
63
64
65
66
67
68
from test_framework import generic_test
from test_framework.test_failure import TestFailure
class Queue:
SCALE_FACTOR = 2
def __init__(self, capacity):
self._entries = [None] * capacity
self._head = self._tail = self._num_queue_elements = 0
def enqueue(self, x):
if self._num_queue_elements == len(self._entries): # Needs to resize.
# Makes the queue elements appear consecutively.
self._entries = (
self._entries[self._head:] + self._entries[:self._head])
# Resets head and tail.
self._head, self._tail = 0, self._num_queue_elements
self._entries += [None] * (
len(self._entries) * Queue.SCALE_FACTOR - len(self._entries))
self._entries[self._tail] = x
self._tail = (self._tail + 1) % len(self._entries)
self._num_queue_elements += 1
def dequeue(self):
if not self._num_queue_elements:
raise IndexError('empty queue')
self._num_queue_elements -= 1
result = self._entries[self._head]
self._head = (self._head + 1) % len(self._entries)
return result
def size(self):
return self._num_queue_elements
def queue_tester(ops):
q = Queue(1)
for (op, arg) in ops:
if op == 'Queue':
q = Queue(arg)
elif op == 'enqueue':
q.enqueue(arg)
elif op == 'dequeue':
result = q.dequeue()
if result != arg:
raise TestFailure(
"Dequeue: expected " + str(arg) + ", got " + str(result))
elif op == 'size':
result = q.size()
if result != arg:
raise TestFailure(
"Size: expected " + str(arg) + ", got " + str(result))
else:
raise RuntimeError("Unsupported queue operation: " + op)
if __name__ == '__main__':
exit(
generic_test.generic_test_main("circular_queue.py",
'circular_queue.tsv', queue_tester))