Exercism v3 launches on Sept 1st 2021. Learn more! πŸš€πŸš€πŸš€
Avatar of rootulp

rootulp's solution

to Circular Buffer in the Python Track

Published at Jul 13 2018 · 0 comments
Instructions
Test suite
Solution

Note:

This solution was written on an old version of Exercism. The tests below might not correspond to the solution code, and the exercise may have changed since this code was written.

A circular buffer, cyclic buffer or ring buffer is a data structure that uses a single, fixed-size buffer as if it were connected end-to-end.

A circular buffer first starts empty and of some predefined length. For example, this is a 7-element buffer:

[ ][ ][ ][ ][ ][ ][ ]

Assume that a 1 is written into the middle of the buffer (exact starting location does not matter in a circular buffer):

[ ][ ][ ][1][ ][ ][ ]

Then assume that two more elements are added β€” 2 & 3 β€” which get appended after the 1:

[ ][ ][ ][1][2][3][ ]

If two elements are then removed from the buffer, the oldest values inside the buffer are removed. The two elements removed, in this case, are 1 & 2, leaving the buffer with just a 3:

[ ][ ][ ][ ][ ][3][ ]

If the buffer has 7 elements then it is completely full:

[6][7][8][9][3][4][5]

When the buffer is full an error will be raised, alerting the client that further writes are blocked until a slot becomes free.

When the buffer is full, the client can opt to overwrite the oldest data with a forced write. In this case, two more elements β€” A & B β€” are added and they overwrite the 3 & 4:

[6][7][8][9][A][B][5]

3 & 4 have been replaced by A & B making 5 now the oldest data in the buffer. Finally, if two elements are removed then what would be returned is 5 & 6 yielding the buffer:

[ ][7][8][9][A][B][ ]

Because there is space available, if the client again uses overwrite to store C & D then the space where 5 & 6 were stored previously will be used not the location of 7 & 8. 7 is still the oldest element and the buffer is once again full.

[D][7][8][9][A][B][C]

Exception messages

Sometimes it is necessary to raise an exception. When you do this, you should include a meaningful error message to indicate what the source of the error is. This makes your code more readable and helps significantly with debugging. Not every exercise will require you to raise an exception, but for those that do, the tests will only pass if you include a message.

To raise a message with an exception, just write it as an argument to the exception type. For example, instead of raise Exception, you should write:

raise Exception("Meaningful message indicating the source of the error")

Running the tests

To run the tests, run the appropriate command below (why they are different):

  • Python 2.7: py.test circular_buffer_test.py
  • Python 3.4+: pytest circular_buffer_test.py

Alternatively, you can tell Python to run the pytest module (allowing the same command to be used regardless of Python version): python -m pytest circular_buffer_test.py

Common pytest options

  • -v : enable verbose output
  • -x : stop running tests on first failure
  • --ff : run failures from previous test before running other test cases

For other options, see python -m pytest -h

Submitting Exercises

Note that, when trying to submit an exercise, make sure the solution is in the $EXERCISM_WORKSPACE/python/circular-buffer directory.

You can find your Exercism workspace by running exercism debug and looking for the line that starts with Workspace.

For more detailed information about running tests, code style and linting, please see the help page.

Source

Wikipedia http://en.wikipedia.org/wiki/Circular_buffer

Submitting Incomplete Solutions

It's possible to submit an incomplete solution so you can see how others have completed the exercise.

circular_buffer_test.py

import unittest

from circular_buffer import (
    CircularBuffer,
    BufferFullException,
    BufferEmptyException
)


# Tests adapted from `problem-specifications//canonical-data.json` @ v1.1.0

class CircularBufferTest(unittest.TestCase):
    def test_read_empty_buffer(self):
        buf = CircularBuffer(1)
        with self.assertRaisesWithMessage(BufferEmptyException):
            buf.read()

    def test_read_just_written_item(self):
        buf = CircularBuffer(1)
        buf.write('1')
        self.assertEqual(buf.read(), '1')

    def test_write_and_read_back_one_item(self):
        buf = CircularBuffer(1)
        buf.write('1')
        self.assertEqual(buf.read(), '1')
        with self.assertRaisesWithMessage(BufferEmptyException):
            buf.read()

    def test_write_and_read_back_multiple_items_ordered(self):
        buf = CircularBuffer(2)
        buf.write('1')
        buf.write('2')
        self.assertEqual(buf.read(), '1')
        self.assertEqual(buf.read(), '2')

    def test_cant_write_to_full_buffer(self):
        buf = CircularBuffer(1)
        buf.write('1')
        with self.assertRaisesWithMessage(BufferFullException):
            buf.write('2')

    def test_alternate_write_and_read(self):
        buf = CircularBuffer(1)
        buf.write('1')
        self.assertEqual(buf.read(), '1')
        buf.write('2')
        self.assertEqual(buf.read(), '2')

    def test_read_back_oldest_item(self):
        buf = CircularBuffer(3)
        buf.write('1')
        buf.write('2')
        self.assertEqual(buf.read(), '1')
        buf.write('3')
        self.assertEqual(buf.read(), '2')
        self.assertEqual(buf.read(), '3')

    def test_clearing_buffer(self):
        buf = CircularBuffer(1)
        buf.write('1')
        buf.clear()
        with self.assertRaisesWithMessage(BufferEmptyException):
            buf.read()

    def test_clear_free_buffer_for_write(self):
        buf = CircularBuffer(1)
        buf.write('1')
        buf.clear()
        buf.write('2')
        self.assertEqual(buf.read(), '2')

    def test_clear_does_nothing_on_empty_buffer(self):
        buf = CircularBuffer(1)
        buf.clear()
        buf.write('1')
        self.assertEqual(buf.read(), '1')

    def test_overwrite_non_full_buffer(self):
        buf = CircularBuffer(2)
        buf.write('1')
        buf.overwrite('2')
        self.assertEqual(buf.read(), '1')
        self.assertEqual(buf.read(), '2')

    def test_overwrite_replaces_oldest_item(self):
        buf = CircularBuffer(2)
        buf.write('1')
        buf.write('2')
        buf.overwrite('3')
        self.assertEqual(buf.read(), '2')
        self.assertEqual(buf.read(), '3')

    def test_overwrite_replaces_oldest_remaining_item(self):
        buf = CircularBuffer(3)
        buf.write('1')
        buf.write('2')
        buf.write('3')
        self.assertEqual(buf.read(), '1')
        buf.write('4')
        buf.overwrite('5')
        self.assertEqual(buf.read(), '3')
        self.assertEqual(buf.read(), '4')
        self.assertEqual(buf.read(), '5')

    # Utility functions
    def setUp(self):
        try:
            self.assertRaisesRegex
        except AttributeError:
            self.assertRaisesRegex = self.assertRaisesRegexp

    def assertRaisesWithMessage(self, exception):
        return self.assertRaisesRegex(exception, r".+")


if __name__ == '__main__':
    unittest.main()
from collections import deque


class CircularBuffer:

    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = deque([], capacity)

    def read(self):
        if self.empty():
            raise BufferEmptyException
        return self.buffer.popleft()

    def write(self, data):
        if self.full():
            raise BufferFullException
        self.buffer.append(data)

    def overwrite(self, data):
        if self.full():
            self.buffer.popleft()
        self.buffer.append(data)

    def clear(self):
        self.buffer.clear()

    def empty(self):
        return len(self.buffer) == 0

    def full(self):
        return len(self.buffer) == self.capacity


class BufferFullException(Exception):
    pass


class BufferEmptyException(Exception):
    pass

Community comments

Find this solution interesting? Ask the author a question to learn more.

What can you learn from this solution?

A huge amount can be learned from reading other people’s code. This is why we wanted to give exercism users the option of making their solutions public.

Here are some questions to help you reflect on this solution and learn the most from it.

  • What compromises have been made?
  • Are there new concepts here that you could read more about to improve your understanding?