buffer melingkar yang efisien?

Saya ingin membuat buffer melingkar yang efisien dengan python (dengan tujuan mengambil rata-rata nilai integer di buffer).

Apakah ini cara yang efisien untuk menggunakan daftar untuk mengumpulkan nilai?

def add_to_buffer( self, num ):
    self.mylist.pop( 0 )
    self.mylist.append( num )

Apa yang lebih efisien (dan mengapa)?

Saya akan menggunakan collections.dequedengan maxlenarg

>>> import collections
>>> d = collections.deque(maxlen=10)
>>> d
deque([], maxlen=10)
>>> for i in xrange(20):
...     d.append(i)
... 
>>> d
deque([10, 11, 12, 13, 14, 15, 16, 17, 18, 19], maxlen=10)

Ada resepnya di dokumen dequeyang serupa dengan yang Anda inginkan. Penegasan saya bahwa ini yang paling efisien sepenuhnya didasarkan pada kenyataan bahwa itu diterapkan di C oleh kru yang sangat terampil yang memiliki kebiasaan untuk membuat kode kedudukan tertinggi.

muncul dari kepala daftar menyebabkan seluruh daftar disalin, jadi tidak efisien

Anda sebaiknya menggunakan daftar / larik dengan ukuran tetap dan indeks yang bergerak melalui buffer saat Anda menambah / menghapus item

Berdasarkan jawaban MoonCactus , berikut adalah sebuah circularlistkelas. Perbedaannya dengan versinya adalah bahwa di sini c[0]akan selalu memberikan elemen terlama, elemen tambahan c[-1]terakhir, c[-2]kedua dari belakang ... Ini lebih natural untuk aplikasi.

c = circularlist(4)
c.append(1); print c, c[0], c[-1]    #[1]              1, 1
c.append(2); print c, c[0], c[-1]    #[1, 2]           1, 2
c.append(3); print c, c[0], c[-1]    #[1, 2, 3]        1, 3
c.append(8); print c, c[0], c[-1]    #[1, 2, 3, 8]     1, 8
c.append(10); print c, c[0], c[-1]   #[10, 2, 3, 8]    2, 10
c.append(11); print c, c[0], c[-1]   #[10, 11, 3, 8]   3, 11

Kelas:

class circularlist(object):
    def __init__(self, size, data = []):
        """Initialization"""
        self.index = 0
        self.size = size
        self._data = list(data)[-size:]

    def append(self, value):
        """Append an element"""
        if len(self._data) == self.size:
            self._data[self.index] = value
        else:
            self._data.append(value)
        self.index = (self.index + 1) % self.size

    def __getitem__(self, key):
        """Get element by index, relative to the current index"""
        if len(self._data) == self.size:
            return(self._data[(key + self.index) % self.size])
        else:
            return(self._data[key])

    def __repr__(self):
        """Return string representation"""
        return self._data.__repr__() + ' (' + str(len(self._data))+' items)'

[Diedit]: Menambahkan dataparameter opsional untuk memungkinkan inisialisasi dari daftar yang ada, misalnya:

circularlist(4, [1, 2, 3, 4, 5])      #  [2, 3, 4, 5] (4 items)
circularlist(4, set([1, 2, 3, 4, 5])) #  [2, 3, 4, 5] (4 items)
circularlist(4, (1, 2, 3, 4, 5))      #  [2, 3, 4, 5] (4 items)

Deque Python lambat. Anda juga bisa menggunakan numpy.roll Bagaimana cara memutar angka dalam array numpy bentuk (n,) atau (n, 1)?

Dalam tolok ukur ini, deque adalah 448ms. Numpy.roll adalah 29ms http://scimusing.wordpress.com/2013/10/25/ring-buffers-in-pythonnumpy/

ok dengan penggunaan kelas deque, tetapi untuk permintaan pertanyaan (rata-rata) ini adalah solusi saya:

>>> from collections import deque
>>> class CircularBuffer(deque):
...     def __init__(self, size=0):
...             super(CircularBuffer, self).__init__(maxlen=size)
...     @property
...     def average(self):  # TODO: Make type check for integer or floats
...             return sum(self)/len(self)
...
>>>
>>> cb = CircularBuffer(size=10)
>>> for i in range(20):
...     cb.append(i)
...     print "@%s, Average: %s" % (cb, cb.average)
...
@deque([0], maxlen=10), Average: 0
@deque([0, 1], maxlen=10), Average: 0
@deque([0, 1, 2], maxlen=10), Average: 1
@deque([0, 1, 2, 3], maxlen=10), Average: 1
@deque([0, 1, 2, 3, 4], maxlen=10), Average: 2
@deque([0, 1, 2, 3, 4, 5], maxlen=10), Average: 2
@deque([0, 1, 2, 3, 4, 5, 6], maxlen=10), Average: 3
@deque([0, 1, 2, 3, 4, 5, 6, 7], maxlen=10), Average: 3
@deque([0, 1, 2, 3, 4, 5, 6, 7, 8], maxlen=10), Average: 4
@deque([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], maxlen=10), Average: 4
@deque([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], maxlen=10), Average: 5
@deque([2, 3, 4, 5, 6, 7, 8, 9, 10, 11], maxlen=10), Average: 6
@deque([3, 4, 5, 6, 7, 8, 9, 10, 11, 12], maxlen=10), Average: 7
@deque([4, 5, 6, 7, 8, 9, 10, 11, 12, 13], maxlen=10), Average: 8
@deque([5, 6, 7, 8, 9, 10, 11, 12, 13, 14], maxlen=10), Average: 9
@deque([6, 7, 8, 9, 10, 11, 12, 13, 14, 15], maxlen=10), Average: 10
@deque([7, 8, 9, 10, 11, 12, 13, 14, 15, 16], maxlen=10), Average: 11
@deque([8, 9, 10, 11, 12, 13, 14, 15, 16, 17], maxlen=10), Average: 12
@deque([9, 10, 11, 12, 13, 14, 15, 16, 17, 18], maxlen=10), Average: 13
@deque([10, 11, 12, 13, 14, 15, 16, 17, 18, 19], maxlen=10), Average: 14

Meskipun sudah ada banyak jawaban bagus di sini, saya tidak dapat menemukan perbandingan langsung waktu untuk opsi yang disebutkan. Oleh karena itu, silakan temukan upaya saya yang sederhana dengan perbandingan di bawah ini.

Untuk tujuan pengujian saja, kelas dapat beralih antara listbuffer berbasis-a, buffer berbasis-a collections.deque, dan Numpy.rollbuffer berbasis- a .

Perhatikan bahwa file update metode hanya menambahkan satu nilai pada satu waktu, agar tetap sederhana.

import numpy
import timeit
import collections


class CircularBuffer(object):
    buffer_methods = ('list', 'deque', 'roll')

    def __init__(self, buffer_size, buffer_method):
        self.content = None
        self.size = buffer_size
        self.method = buffer_method

    def update(self, scalar):
        if self.method == self.buffer_methods[0]:
            # Use list
            try:
                self.content.append(scalar)
                self.content.pop(0)
            except AttributeError:
                self.content = [0.] * self.size
        elif self.method == self.buffer_methods[1]:
            # Use collections.deque
            try:
                self.content.append(scalar)
            except AttributeError:
                self.content = collections.deque([0.] * self.size,
                                                 maxlen=self.size)
        elif self.method == self.buffer_methods[2]:
            # Use Numpy.roll
            try:
                self.content = numpy.roll(self.content, -1)
                self.content[-1] = scalar
            except IndexError:
                self.content = numpy.zeros(self.size, dtype=float)

# Testing and Timing
circular_buffer_size = 100
circular_buffers = [CircularBuffer(buffer_size=circular_buffer_size,
                                   buffer_method=method)
                    for method in CircularBuffer.buffer_methods]
timeit_iterations = 1e4
timeit_setup = 'from __main__ import circular_buffers'
timeit_results = []
for i, cb in enumerate(circular_buffers):
    # We add a convenient number of convenient values (see equality test below)
    code = '[circular_buffers[{}].update(float(j)) for j in range({})]'.format(
        i, circular_buffer_size)
    # Testing
    eval(code)
    buffer_content = [item for item in cb.content]
    assert buffer_content == range(circular_buffer_size)
    # Timing
    timeit_results.append(
        timeit.timeit(code, setup=timeit_setup, number=int(timeit_iterations)))
    print '{}: total {:.2f}s ({:.2f}ms per iteration)'.format(
        cb.method, timeit_results[-1],
        timeit_results[-1] / timeit_iterations * 1e3)

Di sistem saya, ini menghasilkan:

list:  total 1.06s (0.11ms per iteration)
deque: total 0.87s (0.09ms per iteration)
roll:  total 6.27s (0.63ms per iteration)

Bagaimana dengan solusi dari Python Cookbook , termasuk klasifikasi ulang dari instance buffer cincin ketika sudah penuh?

class RingBuffer:
    """ class that implements a not-yet-full buffer """
    def __init__(self,size_max):
        self.max = size_max
        self.data = []

    class __Full:
        """ class that implements a full buffer """
        def append(self, x):
            """ Append an element overwriting the oldest one. """
            self.data[self.cur] = x
            self.cur = (self.cur+1) % self.max
        def get(self):
            """ return list of elements in correct order """
            return self.data[self.cur:]+self.data[:self.cur]

    def append(self,x):
        """append an element at the end of the buffer"""
        self.data.append(x)
        if len(self.data) == self.max:
            self.cur = 0
            # Permanently change self's class from non-full to full
            self.__class__ = self.__Full

    def get(self):
        """ Return a list of elements from the oldest to the newest. """
        return self.data

# sample usage
if __name__=='__main__':
    x=RingBuffer(5)
    x.append(1); x.append(2); x.append(3); x.append(4)
    print(x.__class__, x.get())
    x.append(5)
    print(x.__class__, x.get())
    x.append(6)
    print(x.data, x.get())
    x.append(7); x.append(8); x.append(9); x.append(10)
    print(x.data, x.get())

Pilihan desain penting dalam penerapannya adalah, karena objek ini menjalani transisi status yang tidak dapat diubah di beberapa titik dalam masa pakainya — dari buffer tidak penuh ke buffer penuh (dan perubahan perilaku pada saat itu) —saya memodelkannya dengan mengubahnya self.__class__ . Ini berfungsi bahkan di Python 2.2, selama kedua kelas memiliki slot yang sama (misalnya, ini berfungsi dengan baik untuk dua kelas klasik, seperti RingBuffer dan__Full dalam resep ini).

Mengubah kelas instance mungkin aneh dalam banyak bahasa, tetapi ini adalah alternatif Pythonic untuk cara lain mewakili perubahan keadaan sesekali, masif, tidak dapat diubah, dan diskrit yang sangat mempengaruhi perilaku, seperti dalam resep ini. Untung Python mendukungnya untuk semua jenis kelas.

Kredit: Sébastien Keim

Anda juga bisa melihat resep Python yang cukup lama ini .

Ini adalah versi saya sendiri dengan NumPy array:

#!/usr/bin/env python

import numpy as np

class RingBuffer(object):
    def __init__(self, size_max, default_value=0.0, dtype=float):
        """initialization"""
        self.size_max = size_max

        self._data = np.empty(size_max, dtype=dtype)
        self._data.fill(default_value)

        self.size = 0

    def append(self, value):
        """append an element"""
        self._data = np.roll(self._data, 1)
        self._data[0] = value 

        self.size += 1

        if self.size == self.size_max:
            self.__class__  = RingBufferFull

    def get_all(self):
        """return a list of elements from the oldest to the newest"""
        return(self._data)

    def get_partial(self):
        return(self.get_all()[0:self.size])

    def __getitem__(self, key):
        """get element"""
        return(self._data[key])

    def __repr__(self):
        """return string representation"""
        s = self._data.__repr__()
        s = s + '\t' + str(self.size)
        s = s + '\t' + self.get_all()[::-1].__repr__()
        s = s + '\t' + self.get_partial()[::-1].__repr__()
        return(s)

class RingBufferFull(RingBuffer):
    def append(self, value):
        """append an element when buffer is full"""
        self._data = np.roll(self._data, 1)
        self._data[0] = value

Yang ini tidak membutuhkan perpustakaan apa pun. Itu menumbuhkan daftar dan kemudian berputar di dalam menurut indeks.

Jejaknya sangat kecil (tidak ada perpustakaan), dan setidaknya berjalan dua kali lebih cepat dari dequeue. Ini memang bagus untuk menghitung rata-rata bergerak, tetapi perlu diketahui bahwa item tidak disimpan berdasarkan usia seperti di atas.

class CircularBuffer(object):
    def __init__(self, size):
        """initialization"""
        self.index= 0
        self.size= size
        self._data = []

    def record(self, value):
        """append an element"""
        if len(self._data) == self.size:
            self._data[self.index]= value
        else:
            self._data.append(value)
        self.index= (self.index + 1) % self.size

    def __getitem__(self, key):
        """get element by index like a regular array"""
        return(self._data[key])

    def __repr__(self):
        """return string representation"""
        return self._data.__repr__() + ' (' + str(len(self._data))+' items)'

    def get_all(self):
        """return a list of all the elements"""
        return(self._data)

Untuk mendapatkan nilai rata-rata, misal:

q= CircularBuffer(1000000);
for i in range(40000):
    q.record(i);
print "capacity=", q.size
print "stored=", len(q.get_all())
print "average=", sum(q.get_all()) / len(q.get_all())

Hasil dalam:

capacity= 1000000
stored= 40000
average= 19999

real 0m0.024s
user 0m0.020s
sys  0m0.000s

Ini sekitar 1/3 waktu ekuivalen dengan dequeue.

Saya pernah mengalami masalah ini sebelum melakukan pemrograman serial. Pada waktu lebih dari setahun yang lalu, saya juga tidak dapat menemukan implementasi yang efisien, jadi saya akhirnya menulis satu sebagai ekstensi C dan itu juga tersedia di pypi di bawah lisensi MIT. Ini sangat mendasar, hanya menangani buffer dari karakter bertanda 8-bit, tetapi memiliki panjang yang fleksibel, sehingga Anda dapat menggunakan Struct atau sesuatu di atasnya jika Anda membutuhkan sesuatu selain karakter. Saya melihat sekarang dengan pencarian google bahwa ada beberapa opsi hari ini, jadi Anda mungkin ingin melihatnya juga.

Jawaban Anda tidak benar. Circular buffer main memiliki dua prinsip （https://en.wikipedia.org/wiki/Circular_buffer )

1. Lenth buffer ditetapkan;
2. Pertama masuk pertama keluar;
3. Saat Anda menambah atau menghapus item, item lainnya tidak boleh berpindah posisinya

kode Anda di bawah ini:

def add_to_buffer( self, num ):
    self.mylist.pop( 0 )
    self.mylist.append( num )

Mari pertimbangkan situasi di mana daftar itu penuh, dengan menggunakan kode Anda:

self.mylist = [1, 2, 3, 4, 5]

sekarang kami menambahkan 6, daftar diubah menjadi

self.mylist = [2, 3, 4, 5, 6]

item yang diharapkan 1 dalam daftar telah berubah posisinya

kode Anda adalah antrian, bukan buffer lingkaran.

Jawaban Basj menurut saya paling manjur.

Ngomong-ngomong, penyangga lingkaran dapat meningkatkan kinerja operasi untuk menambahkan item.

Dari Github:

class CircularBuffer:

    def __init__(self, size):
        """Store buffer in given storage."""
        self.buffer = [None]*size
        self.low = 0
        self.high = 0
        self.size = size
        self.count = 0

    def isEmpty(self):
        """Determines if buffer is empty."""
        return self.count == 0

    def isFull(self):
        """Determines if buffer is full."""
        return self.count == self.size

    def __len__(self):
        """Returns number of elements in buffer."""
        return self.count

    def add(self, value):
        """Adds value to buffer, overwrite as needed."""
        if self.isFull():
            self.low = (self.low+1) % self.size
        else:
            self.count += 1
        self.buffer[self.high] = value
        self.high = (self.high + 1) % self.size

    def remove(self):
        """Removes oldest value from non-empty buffer."""
        if self.count == 0:
            raise Exception ("Circular Buffer is empty");
        value = self.buffer[self.low]
        self.low = (self.low + 1) % self.size
        self.count -= 1
        return value

    def __iter__(self):
        """Return elements in the circular buffer in order using iterator."""
        idx = self.low
        num = self.count
        while num > 0:
            yield self.buffer[idx]
            idx = (idx + 1) % self.size
            num -= 1

    def __repr__(self):
        """String representation of circular buffer."""
        if self.isEmpty():
            return 'cb:[]'

        return 'cb:[' + ','.join(map(str,self)) + ']'

https://github.com/heineman/python-data-structures/blob/master/2.%20Ubiquitous%20Lists/circBuffer.py

Pertanyaan aslinya adalah: buffer melingkar " efisien ". Menurut efisiensi yang diminta ini, jawaban dari aaronasterling tampaknya pasti benar. Menggunakan kelas khusus yang diprogram dengan Python dan membandingkan waktu pemrosesan dengan collections.deque menunjukkan akselerasi x5.2 kali dengan deque! Berikut kode yang sangat sederhana untuk mengujinya:

class cb:
    def __init__(self, size):
        self.b = [0]*size
        self.i = 0
        self.sz = size
    def append(self, v):
        self.b[self.i] = v
        self.i = (self.i + 1) % self.sz

b = cb(1000)
for i in range(10000):
    b.append(i)
# called 200 times, this lasts 1.097 second on my laptop

from collections import deque
b = deque( [], 1000 )
for i in range(10000):
    b.append(i)
# called 200 times, this lasts 0.211 second on my laptop

Untuk mengubah deque menjadi daftar, cukup gunakan:

my_list = [v for v in my_deque]

Anda kemudian akan mendapatkan O (1) akses acak ke item deque. Tentu saja, ini hanya berguna jika Anda perlu melakukan banyak akses acak ke deque setelah menyetelnya sekali.

Ini menerapkan prinsip yang sama ke beberapa buffer yang dimaksudkan untuk menampung pesan teks terbaru.

import time
import datetime
import sys, getopt

class textbffr(object):
    def __init__(self, size_max):
        #initialization
        self.posn_max = size_max-1
        self._data = [""]*(size_max)
        self.posn = self.posn_max

    def append(self, value):
        #append an element
        if self.posn == self.posn_max:
            self.posn = 0
            self._data[self.posn] = value   
        else:
            self.posn += 1
            self._data[self.posn] = value

    def __getitem__(self, key):
        #return stored element
        if (key + self.posn+1) > self.posn_max:
            return(self._data[key - (self.posn_max-self.posn)])
        else:
            return(self._data[key + self.posn+1])


def print_bffr(bffr,bffer_max): 
    for ind in range(0,bffer_max):
        stored = bffr[ind]
        if stored != "":
            print(stored)
    print ( '\n' )

def make_time_text(time_value):
    return(str(time_value.month).zfill(2) + str(time_value.day).zfill(2)
      + str(time_value.hour).zfill(2) +  str(time_value.minute).zfill(2)
      + str(time_value.second).zfill(2))


def main(argv):
    #Set things up 
    starttime = datetime.datetime.now()
    log_max = 5
    status_max = 7
    log_bffr = textbffr(log_max)
    status_bffr = textbffr(status_max)
    scan_count = 1

    #Main Loop
    # every 10 secounds write a line with the time and the scan count.
    while True: 

        time_text = make_time_text(datetime.datetime.now())
        #create next messages and store in buffers
        status_bffr.append(str(scan_count).zfill(6) + " :  Status is just fine at : " + time_text)
        log_bffr.append(str(scan_count).zfill(6) + " : " + time_text + " : Logging Text ")

        #print whole buffers so far
        print_bffr(log_bffr,log_max)
        print_bffr(status_bffr,status_max)

        time.sleep(2)
        scan_count += 1 

if __name__ == '__main__':
    main(sys.argv[1:])  

Anda dapat memeriksa buffer melingkar ini berdasarkan larik numpy ukuran yang telah ditentukan sebelumnya. Idenya adalah Anda membuat buffer (mengalokasikan memori untuk array numpy) dan kemudian menambahkannya. Penyisipan data dan pengambilan sangat cepat. Saya telah membuat modul ini untuk tujuan yang sama seperti yang Anda butuhkan. Dalam kasus saya, saya memiliki perangkat yang menghasilkan data integer. Saya membaca data dan meletakkannya di buffer melingkar untuk analisis dan pemrosesan di masa mendatang.