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.
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.
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 : 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
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 : 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.