buffer ad anello con numpy/ctypes

Question

Sto sviluppando un client che riceverà i dati [EEG] su tcp e lo scriverà sul ring buffer. Ho pensato che possa essere molto comodo avere il buffer come un array ctypes o numpy perché è possibile creare una "vista" numpy in qualsiasi posizione di tale buffer e leggere/scrivere/elaborare i dati senza operazioni di copia. O è una cattiva idea in generale?

Tuttavia, non vedo come implementare un buffer circolare di dimensioni fisse in questo modo. Supponiamo di aver creato un oggetto buffer che è contiguo in memoria. Qual è il modo migliore per scrivere i dati quando viene raggiunta la fine del buffer?

Un modo possibile è iniziare a sovrascrivere i byte (già vecchi) dall'inizio quando il puntatore di scrittura raggiunge la fine dell'array di buffer. Vicino ai confini, tuttavia, la vista numpy di alcuni blocchi (per l'elaborazione) non può essere creata (o può?) In questo caso, perché alcuni di essi possono ancora essere posizionati alla fine del buffer array mentre un altro già in il suo inizio. Ho letto che è impossibile creare fette così circolari. Come risolvere questo?

UPD: Grazie a tutti per le risposte. Nel caso in cui qualcuno si trovi di fronte allo stesso problema, lo here è il codice finale che ho.

Answer 1

Se è necessaria una finestra di N byte, creare il buffer 2 * N byte e scrivere tutti gli input in due posizioni: i % N e , dove i è un contatore di byte. In questo modo hai sempre N byte consecutivi nel buffer.

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]) 

stampe

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

Answer 2

Un modo possibile è iniziare a sovrascrivere i byte (già vecchi) dall'inizio quando il puntatore di scrittura raggiunge la fine dell'array di buffer.

Questa è l'unica opzione in un buffer circolare di dimensioni fisse.

Ho letto che è impossibile creare tali segmenti circolari.

Ecco perché non lo farei con una vista Numpy. È possibile creare un wrapper class attorno a uno ndarray, tenendo il buffer/array, la capacità e un puntatore (indice) al punto di inserimento. Se si desidera ottenere il contenuto come un array di Numpy, si dovrà fare una copia in questo modo:

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

È comunque possibile impostare i valori di elementi sul posto se si implementa __setitem__ e __setslice__.

Answer 3

Una variante della risposta di @Janne Karila, per C ma non NumPy:
Se il buffer ad anello è molto ampia, come N x 1G, allora invece di raddoppiare tutto cosa, raddoppia un array di puntatori 2 * N alle sue righe. E.g. per N = 3, inizializzare

bufp = { buf[0], buf[1], buf[2], buf[0], buf[1], buf[2] }; 

Poi si scrivere i dati una sola volta, e anyfunc(bufp[j:j+3]) vede le righe in buf in ordine di tempo.

Answer 4

Penso che sia necessario fare un passo indietro dal pensiero stile C qui. L'aggiornamento di un ringbuffer per ogni singolo inserimento non sarà mai efficiente. Un ring-buffer è fondamentalmente diverso dall'interfaccia di blocco di memoria contigua che gli array di numpy richiedono; compreso il fft che dici di voler fare.

Una soluzione naturale è sacrificare un po 'di memoria a scopo di prestazioni. Ad esempio, se il numero di elementi da tenere nel buffer è N, allocare un array di N + 1024 (o un numero ragionevole). Quindi è necessario spostare solo N elementi attorno a ogni 1024 inserimenti e si ha sempre una vista contigua di N elementi su cui agire direttamente disponibili.

MODIFICA: questo è uno snippet di codice che implementa quanto sopra e dovrebbe fornire buone prestazioni. Nota, tuttavia, che ti consigliamo di aggiungere in blocchi, piuttosto che per elemento. Altrimenti, i vantaggi prestazionali dell'uso di numpy vengono rapidamente annullati, indipendentemente dal modo in cui implementate il vostro Ringbuffer.

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