具有numpy/ctype的环形缓冲区

Question

我正在开发一个客户端，它将通过tcp接收EEG数据并将其写入环形缓冲区。我认为将缓冲区作为ctype或numpy数组非常方便，因为可以对缓冲区的任何位置创建numpy“视图”，并在不进行任何复制操作的情况下读取/写入/处理数据。还是总体来说是个坏主意？

但是，我不知道如何以这种方式实现大小固定的循环缓冲区。假设我已经创建了一个缓冲区对象，它在内存中是连续的。当到达缓冲区的末尾时，写数据的最好方法是什么？

一种可能的方法是从写入指针到达缓冲区数组末尾时开始覆盖(已经旧的)字节。然而，在边界附近，无法创建某个块(用于处理)的numpy视图(或者可以吗？)在这种情况下，因为其中一些仍然可以定位在缓冲区数组的末尾，而另一些已经处于初始状态。我读过这样的圆形切片是不可能的。如何解决这个问题？

UPD:感谢大家的回答。如果有人也面临同样的问题，这里是我得到的最后代码。

Answer 1

如果需要N个字节的窗口，那么将缓冲区设置为2*N字节，并将所有输入写入两个位置：i % N和i % N + N，其中i是字节计数器。这样，缓冲区中总是有N个连续字节。

data = 'Data to buffer'
N = 4
buf = 2*N*['\00']

for i,c in enumerate(data):
    j = i % N
    buf[j] = c
    buf[j+N] = c
    if i >= N-1:
        print ''.join(buf[j+1:j+N+1]) 

版画

Data
ata 
ta t
a to
 to 
to b
o bu
 buf
buff
uffe
ffer

Answer 2

我认为你应该从C风格的思维中后退一步。为每一个插入更新一个循环缓冲区永远都不会是有效的。环形缓冲区从根本上不同于numpy数组所要求的连续内存块接口；包括您提到的要做的fft。

一个自然的解决方案是为了性能而牺牲一点点内存。例如，如果需要在缓冲区中保存的元素数为N，则分配一个N+1024数组(或一些合理的数字)。然后，您只需要在每1024个插入时移动N个元素，并且您总是有一个连续的N元素视图来对直接可用的元素进行操作。

编辑:下面是实现上述功能的代码片段，应该具有良好的性能。不过，请注意，建议您添加块，而不是每个元素。否则，使用numpy的性能优势会很快失效，而不管您是如何实现循环缓冲区的。

import numpy as np

class RingBuffer(object):
    def __init__(self, size, padding=None):
        self.size = size
        self.padding = size if padding is None else padding
        self.buffer = np.zeros(self.size+self.padding)
        self.counter = 0

    def append(self, data):
        """this is an O(n) operation"""
        data = data[-self.padding:]
        n = len(data)
        if self.remaining < n: self.compact()
        self.buffer[self.counter+self.size:][:n] = data
        self.counter += n

    @property
    def remaining(self):
        return self.padding-self.counter
    @property
    def view(self):
        """this is always an O(1) operation"""
        return self.buffer[self.counter:][:self.size]
    def compact(self):
        """
        note: only when this function is called, is an O(size) performance hit incurred,
        and this cost is amortized over the whole padding space
        """
        print 'compacting'
        self.buffer[:self.size] = self.view
        self.counter = 0

rb = RingBuffer(10)
for i in range(4):
    rb.append([1,2,3])
    print rb.view

rb.append(np.arange(15))
print rb.view  #test overflow

Answer 3

一种可能的方法是从写入指针到达缓冲区数组末尾时开始覆盖(已经旧的)字节。

这是固定大小的环形缓冲器中唯一的选择。

我读过这样的圆形切片是不可能的。

这就是为什么我不会用Numpy的观点来做这件事。您可以在class周围创建一个ndarray包装器，保存缓冲区/数组、容量和指向插入点的指针(索引)。如果要以Numpy数组的形式获取内容，则必须复制如下所示的内容：

buf = np.array([1,2,3,4])
indices = [3,0,1,2]
contents = buf[indices]    # copy

如果您实现了__setitem__和__setslice__，仍然可以在适当的地方设置元素的值。