UPDATE: see **Michel Pelletier ** comment below, for using numpy.roll.

Recently I’ve been developing a scrolling oscilloscope for real-time data. An important first step is to store the data in a non-real-time buffer, which will be input to the plotting functions.

### Requirements

For my purposes, the length is around 10,000 to 100,000 single precision floats, with blocks of 1,000 to 10,000 being added 40-20 times a second. The buffer will be read about 10 times a second as a numpy array.

### Deque

The first place I looked for a ring buffer was in the standard library. Within `collections`

there’s a promising data structure called `deque`

which by default is a first-in-first-out queue, with an optional max length argument. Though a queue isn’t a ring buffer, it can have the same behavior, making my code base simpler and smaller.

from collections import deque import numpy as np def ringbuff_deque_test(): ringlen = 100000 ringbuff = deque(np.zeros(ringlen, dtype='f'), ringlen) for i in range(40): ringbuff.extend(np.zeros(10e3, dtype='f')) # write np.array(ringbuff) #read

Is it fast enough? Using Ipython’s %timeit:

In [32]: timeit ringbuff_deque_test() 1 loops, best of 3: 19.7 s per loop

Sadly, no. Looking a little deeper, it turns out that reading a deque object into a numpy array is painfully slow. Converting the deque to a list first

np.array(list(ringbuff))

reduces the time to 448 milliseconds, still too slow. Removing the read command entirely brings our test to only 30 milliseconds! I’m not going to re-invent the plotting library, so no `deque`

for this project. However, this does look like a viable option for non-numpy projects.

### Writing a numpy class

The next step is to implement a ring buffer in numpy. Because I’m always adding arrays of length greater than 1, I only wrote an extend method.

class RingBuffer(): "A 1D ring buffer using numpy arrays" def __init__(self, length): self.data = np.zeros(length, dtype='f') self.index = 0 def extend(self, x): "adds array x to ring buffer" x_index = (self.index + np.arange(x.size)) % self.data.size self.data[x_index] = x self.index = x_index[-1] + 1 def get(self): "Returns the first-in-first-out data in the ring buffer" idx = (self.index + np.arange(self.data.size)) %self.data.size return self.data[idx] def ringbuff_numpy_test(): ringlen = 100000 ringbuff = RingBuffer(ringlen) for i in range(40): ringbuff.extend(np.zeros(10000, dtype='f')) # write ringbuff.get() #read

Is it fast enough?

In [33]: timeit ringbuff_numpy_test() 100 loops, best of 3: 105 ms per loop

Yes. 105 milliseconds of computation over a typical second in the application.

Not fast enough for your project? Look into cython or numba to implement a `for`

loop instead of creating index arrays.

Next step: efficient scrolling plots in python (pyqtgraph?, galry?).