Tampon à anneau avec NUMPY / CTYPES

Je pense que vous devez prendre un recul de la pensée de style C ici. La mise à jour d'une ringbuffer pour chaque insertion ne sera jamais efficace. Un tampon à anneau est fondamentalement différent de l'interface de bloc de mémoire contiguë que la demande numpopie demande; Y compris la FFT que vous mentionnerez que vous voulez faire.

Une solution naturelle consiste à sacrifier un peu de mémoire pour des performances. Par exemple, si le nombre d'éléments que vous devez conserver dans votre tampon est N, allouez un tableau de N + 1024 (ou d'un nombre sensible). Ensuite, il vous suffit de déplacer n éléments autour de toutes les insertions de 1024, et vous avez toujours une vue contiguë de N éléments à agir directement disponible. P>

Edit: Voici un extrait de code qui implémente ce qui précède et devrait donner de bonnes performances. Notez cependant que vous seriez bien conseillé d'ajouter des morceaux, plutôt que par élément. Sinon, les avantages de performance de l'utilisation de NUMPY sont rapidement annulés, peu importe la manière dont vous implémentez votre Runbuffer. P>

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