Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion numpy_ringbuffer/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,11 +162,17 @@ def __len__(self):
return self._right_index - self._left_index

def __getitem__(self, item):
if len(self) == 0:
raise IndexError("Buffer is empty")
# handle simple (b[1]) and basic (b[np.array([1, 2, 3])]) fancy indexing specially
if not isinstance(item, tuple):
item_arr = np.asarray(item)
if issubclass(item_arr.dtype.type, np.integer):
item_arr = (item_arr + self._left_index) % self._capacity
if self.is_full:
item_arr = (item_arr + self._left_index) % self._capacity
else:
item_arr = ((item_arr + self._left_index)
% self._right_index)
Comment on lines +174 to +175
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This looks wrong to me for cases when left_index != 0

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

test_append causes the buffer to wrap, which makes self._left_index be greater than 0, and it passes

return self._arr[item_arr]

# for everything else, get it right at the expense of efficiency
Expand Down
17 changes: 17 additions & 0 deletions tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,23 @@ def test_degenerate(self):
except IndexError:
self.fail()

def test_raises_error_indexing_empty_buffer(self):
r = RingBuffer(10)
with self.assertRaises(IndexError,
msg="Fails to raise an IndexError when "
"trying to index an empty buffer."):
r[0]

def test_negative_indices_give_recent_data_with_unfull_buffer(self):
r = RingBuffer(10)
for i in range(5):
r.append(i)
for i in reversed(range(-len(r), 0)):
self.assertAlmostEqual(r[i], r[i+5],
msg="Fails to index from the back of "
"the filled portion of the buffer "
"given a negative index.")

if not hasattr(TestAll, 'assertRaisesRegex'):
TestAll.assertRaisesRegex = TestAll.assertRaisesRegexp

Expand Down