Я хочу создать эффективный круглый буфер в python (с целью усреднения целочисленных значений в буфере).
Является ли это эффективным способом использования списка для сбора значений?
def add_to_buffer( self, num ):
self.mylist.pop( 0 )
self.mylist.append( num )
Что было бы более эффективным (и почему)?
Я бы использовал collections.deque с maxlen arg
>>> 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)
В документах для deque есть recipe, который похож на то, что вы хотите. Мое утверждение о том, что оно наиболее эффективно, полностью зависит от того, что он реализован на C невероятно опытным экипажем, который привык проворачивать верхний код надреза.
выскакивание из заголовка списка приводит к копированию всего списка, поэтому неэффективно
Вместо этого вы должны использовать список/массив фиксированного размера и индекс, который перемещается по буфере при добавлении/удалении элементов
Python deque медленный. Вместо этого вы также можете использовать numpy.roll. Как вы вращаете числа в массиве numpy формы (n,) или (n, 1)?
В этом тесте deque составляет 448 мс. Numpy.roll составляет 29 мс http://scimusing.wordpress.com/2013/10/25/ring-buffers-in-pythonnumpy/
Основываясь на ответе MoonCactus, вот класс circularlist. Разница с его версией заключается в том, что здесь c[0] всегда будет давать самый старый добавленный элемент, c[-1] последний добавляемый элемент, c[-2] предпоследний... Это более естественно для приложений.
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
Класс:
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)'
[Отредактировано]: добавлен необязательный параметр data, чтобы разрешить инициализацию из существующих списков, например:
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)
ok с использованием класса deque, но для запросов (средний) это мое решение:
>>> 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
Хотя здесь уже есть множество отличных ответов, я не смог найти прямого сравнения времени для упомянутых вариантов. Поэтому, пожалуйста, найдите мою скромную попытку сравнения ниже.
Только для целей тестирования класс может переключаться между буфером list -based, буфер collections.deque -based и буфером Numpy.roll -based.
Обратите внимание, что метод update добавляет только одно значение за раз, чтобы сделать его простым.
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)
В моей системе это дает:
list: total 1.06s (0.11ms per iteration)
deque: total 0.87s (0.09ms per iteration)
roll: total 6.27s (0.63ms per iteration)
Как насчет решения из Python Cookbook, включая переклассификацию экземпляра кольцевого буфера, когда он заполнится?
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 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())
Примечательный выбор дизайна в реализации состоит в том, что, поскольку эти объекты претерпевают необратимый переход состояния в некоторый момент их времени жизни - от незаполненного буфера до полного буфера (и поведение изменяется в этой точке) - я смоделировал это, изменив
self.__class__. Это работает даже в Python 2.2, если оба класса имеют одинаковые слоты (например, он отлично работает для двух классических классов, таких как RingBuffer и__Fullв этом рецепте).Изменение класса экземпляра может быть странным во многих языках, но это Pythonic альтернатива другим способам представления случайных, массовых, необратимых и дискретных изменений состояния, которые сильно влияют на поведение, как в этом рецепте. Хорошо, что Python поддерживает его для всех видов классов.
Кредит: Себастьян Кейм
Вы также можете увидеть этот довольно старый рецепт Python.
Вот моя собственная версия с массивом NumPy:
#!/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
Для этого не требуется никакой библиотеки. Он увеличивает список, а затем циклически меняет индекс.
Отпечаток очень маленький (без библиотеки), и он работает в два раза быстрее, чем минимум. Это полезно для вычисления скользящих средних, но помните, что элементы не сортируются по возрасту, как указано выше.
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)
Чтобы получить среднее значение, например:
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())
Результаты в:
capacity= 1000000
stored= 40000
average= 19999
real 0m0.024s
user 0m0.020s
sys 0m0.000s
Это примерно 1/3 времени эквивалента с dequeue.
У меня была эта проблема перед выполнением последовательного программирования. В то время, чуть более года назад, я тоже не мог найти никаких эффективных реализаций, поэтому я закончил тем, что писал один как расширение C и он также доступен на pypi под лицензией MIT. Он супер базовый, обрабатывает только буферы 8-разрядных символов, но имеет гибкую длину, поэтому вы можете использовать Struct или что-то поверх него, если вам нужно что-то другое, кроме символов. Я вижу теперь с поиском google, что есть несколько вариантов в наши дни, поэтому вы можете также посмотреть на них.
Вы отвечаете неправильно. Циркулярный буферный буфер имеет два параметра (https://en.wikipedia.org/wiki/Circular_buffer)
ваш код ниже:
def add_to_buffer( self, num ):
self.mylist.pop( 0 )
self.mylist.append( num )
Рассмотрим ситуацию, когда список заполнен, используя ваш код:
self.mylist = [1, 2, 3, 4, 5]
теперь добавим 6, список изменится на
self.mylist = [2, 3, 4, 5, 6]
элементы, ожидающие 1 в списке, изменили свою позицию
ваш код представляет собой очередь, а не буфер окружности.
Ответ Basj, я думаю, наиболее эффективный.
Кстати, буфер окружности может обеспечить производительность операции для добавления элемента.
Из 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
Первоначальный вопрос: круговой буфер эффективный. В соответствии с такой эффективностью, ответ от aaronasterling кажется окончательно правильным. Используя выделенный класс, запрограммированный в Python, и сравнивая обработку времени с коллекциями .deque показывает ускорение x5.2 раз с deque! Вот очень простой код для проверки этого:
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
Чтобы преобразовать deque в список, просто используйте:
my_list = [v for v in my_deque]
Затем вы получите O (1) случайный доступ к элементам deque. Конечно, это полезно только в том случае, если вам нужно сделать много произвольных обращений к deque после того, как установили его один раз.
Это применяет тот же принцип к некоторым буферам, предназначенным для хранения самых последних текстовых сообщений.
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:])