Circular Queue Structure ( array-backed)

Question

I need some help in writing a python program that will implement a circular queue data structure (array-backed). I've completed a few of the methods already but I'm a bit stumped when it comes to adding and taking things away from the queue, as well as checking the values within it. I believe this is of the first-in-first-out structure. Here's what I have for the body so far

class Queue:
    ''' Constructor '''
    def __init__(self, limit):
        self.limit = limit
        self.data = [None] * limit
        self.queue = []
        self.head = -1
        self.tail = -1
        self.count = 0


    def dequeue(self):
        if self.count == 0:
            raise RuntimeError
        else:
            self.head = 0
            x = self.queue.pop(0)
            if self.head == self.tail:
                self.head = -1
                self.tail = -1
            else:
                self.tail -= 1
            self.count -= 1
            #self.head += 1
            return x

    def enqueue(self, item):
        if self.count == self.limit:
            raise RuntimeError
        else:
            self.count += 1
            self.queue.append(item)
            self.tail += 1

    def __str__(self):
        return " ".join([str(v) for v in self.queue])


    def resize(self, new_limit):
        new_q = [None]*self.limit
        old_q = self.queue
        for i in range(len(old_q)):
            new_q[i] = old_q[i]
        self.limit = new_limit
        self.queue = new_q


    def empty(self):
        return 0 == self.count

    def iter(self):
        listt = []
        for v in self.queue:
            listt.append(v)
        return listt

What I 've written thus far makes the most sense to me but if I were to test this with the following code block I'd get an error saying 10 != 4. This code will fail the 9th line of the test, tc.assertEqual(q.data.count(None), 4) I'm unsure why my code is producing the value 10 at this time. What would allow for this class to pass the given test?

from unittest import TestCase
tc = TestCase()

q = Queue(10)

for i in range(6):
    q.enqueue(i)

tc.assertEqual(q.data.count(None), 4)

for i in range(5):
    q.dequeue()

tc.assertFalse(q.empty())
tc.assertEqual(q.data.count(None), 9)
tc.assertEqual(q.head, q.tail)
tc.assertEqual(q.head, 5)

for i in range(9):
    q.enqueue(i)

with tc.assertRaises(RuntimeError):
    q.enqueue(10)

for x, y in zip(q, [5] + list(range(9))):
    tc.assertEqual(x, y)

Why not using an existing container such as collections.deque ? — Julien Goupy, Jun 18 '16 at 17:10
... adding to this: Sample SO Q&A for illustration: [efficient circular buffer?](http://stackoverflow.com/questions/4151320/efficient-circular-buffer) ... but if you need to write one yourself, I suggest the answer of @Blckknght ;-) — Dilettant, Jun 18 '16 at 17:20

Blckknght · Answer 1 · 2016-06-18T17:36:50.190

I'm pretty sure that all the code using self.queue is wrong. That attribute isn't needed at all. The whole point of the data attribute is to use it to store the values. Use the indexes head and tail to figure out where in data to put things (and where to take them from):

class Queue:
    ''' Constructor '''
    def __init__(self, limit):
        self.limit = limit
        self.data = [None] * limit
        self.head = 0
        self.tail = -1
        self.count = 0

    def dequeue(self):
        if self.count == 0:
            raise RuntimeError
        else:
            x = self.data[self.head]
            self.head = (self.head + 1) % self.limit
            self.count -= 1
            return x

    def enqueue(self, item):
        if self.count == self.limit:
            raise RuntimeError
        else:
            self.count += 1
            self.tail = (self.tail + 1) % self.limit
            self.data[self.tail] = item

    def __str__(self):
        return " ".join([str(v) for v in self]) # use __iter__

    def resize(self, new_limit):
        if new_limit < self.count:
            raise RuntimeError
        new_data = [None]*new_limit
        for i, item in enumerate(self):
            new_data[i] = item
        self.data = new_data
        self.head = 0
        self.tail = self.count - 1

    def empty(self):
        return 0 == self.count

    def __bool__(self): # this is better than empty()
        return self.count != 0

    def __iter__(self): # renamed from iter so you can use it in a for loop
        for i in range(self.count):
            return self.data[(self.head + i) % self.limit]

You should probably also have a __len__ method.

Okay this is starting to make more sense, I just had the head and tail initialized to -1 because the list was empty and that's how I'm being told to represent these attributes when the queue is emptied — Devon, Jun 18 '16 at 17:30
Oh, actually I may be getting things wrong on empty queues. I was thinking it would just work like normal, but you may need `self.tail` to be less than self.head` for it to work properly. — Blckknght, Jun 18 '16 at 17:36

score 1 · Answer 2 · answered Jun 18 '16 at 17:01

1

I'd get an error stating that the Queue class doesn't have a data attribute

I don't have the error you mention when running your test on your code.

answered Jun 18 '16 at 17:01

Faibbus

1,115
10
18

I ran it again and that went away, I've updated the actual error I'm getting – Devon Jun 18 '16 at 17:20

score 0 · Answer 3 · answered Jan 29 '19 at 14:19

If for some reasons you don't want to use built-in collections.deque module, here is an example of how you can implement your own circular buffer:

"""
Example of circular buffer using regular list
"""


class CircularBuffer:
    def __init__(self, size):
        self.buffer = [None] * size
        self.size = size
        self.count = 0
        self.tail = 0
        self.head = 0

    @property
    def is_empty(self):
        return self.count == 0

    @property
    def is_full(self):
        return self.count == self.size

    def __len__(self):
        return self.count

    def add(self, value):
        # if buffer is full overwrite the value
        if self.is_full:
            self.tail = (self.tail + 1) % self.size
        else:
            self.count += 1
        self.buffer[self.head] = value
        self.head = (self.head + 1) % self.size

    def remove(self):
        if self.count == 0:
            raise Exception("Circular Buffer is empty")
        value = self.buffer[self.tail]
        self.tail = (self.tail + 1) % self.size
        self.count -= 1
        return value

    def __iter__(self):
        index = self.tail
        counter = self.count
        while counter > 0:
            yield self.buffer[index]
            index = (index + 1) % self.size
            counter -= 1

    def __repr__(self):
        return "[]" if self.is_empty else "[" + ",".join(str(i) for i in self) + "]"

Circular Queue Structure ( array-backed)

3 Answers3