Мне нужно построить непрерывный ввод с помощью pyqtgraph, поэтому я использую круговой буфер для хранения данных. Я использую deque с maxlen для выполнения этой работы. (Python 2.7, numpy 1.9.2, pyqtgraph 0.9.10)
from collections import deque
def create_cbuffer(self):
buffer_len = self.BUFFER_LEN*self.number_of_points
data = [0]*buffer_len # buffer_len = 160k
self.cbuffer[0] = deque(data, maxlen=buffer_len)
buffer_len = self.BUFFER_LEN
data = [0]*buffer_len
self.cbuffer[1] = deque(data, maxlen=buffer_len)
После этого я использую его так:
import time
def update_cbuffer(self):
data_points, data = data_feeds() # data get every 16ms as lists
start_t = time.time()
self.cbuffer[0].extend(data_points) # Thanks to @PadraicCunningham
# for k in xrange(0, self.number_of_points):
# self.cbuffer[0].append(data_points[k])
self.cbuffer[1].append(data)
fin_t = time.time() - start_t
настроить сюжет как:
self.curve[0] = self.plots[0].plot(self.X_AXIS,
[0]*self.BUFFER_LEN*self.number_of_points,
pen=pg.intColor(color_idx_0),name='plot1')
self.curve[1] = self.plots[1].plot(self.X_AXIS_2, [0]*self.BUFFER_LEN,
pen=pg.intColor(color_idx_1),name='plot2')
обновить сюжет как:
def update_plots(self):
self.curve[0].setData(self.X_AXIS, self.cbuffer[0])
self.curve[0].setPos(self.ptr, 0)
self.curve[1].setData(self.X_AXIS_2, self.cbuffer[1])
self.curve[1].setPos(self.ptr, 0)
self.ptr += 0.016
Затем я вызываю его с помощью QTimer:
self.timer = QtCore.QTimer()
self.timer.timeout.connect(self.update_cbuffer)
self.timer.timeout.connect(self.update_plots)
self.timer.start(16)
Вопрос:
1. Когда я рисую, кажется, что это намного медленнее, чем 16 мс. Есть идеи по ускорению?
2. Когда я определяю время update_plots() с помощью time.time() и вычисляю его среднее время выполнения (total_time/number_of_runs), оно постепенно увеличивается, я пытаюсь понять причина этого.
Есть предложения? Я новичок в Python, я могу сделать некоторые ошибки в коде, пожалуйста, не стесняйтесь указывать на это. Спасибо за вашу помощь заранее.
p.s. Я пробовал разные циклические буферы, как это предлагается в эффективном циклическом буфере?
class Circular_Buffer():
def __init__(self, buffer_len, data_type='float'):
if data_type == 'int':
self.__buffer = np.zeros(buffer_len, dtype=int)
else:
self.__buffer = np.zeros(buffer_len)
self.__counter = 0
def append(self, data):
self.__buffer = np.roll(self.__buffer, -1)
self.__buffer[-1] = data
def get(self):
return self.__buffer
Но в мой случай.
Я также пробовал это:
class CB_list():
def __init__(self, buffer_len):
self.__buffer = [0]*buffer_len
def append(self, data):
self.__buffer = self.__buffer[1:]
self.__buffer.append(data)
def get(self):
return self.__buffer
Он работает так же, как deque, поэтому я придерживаюсь deque.
РЕДАКТИРОВАТЬ: Извините, я сделал ошибку вчера. Я уже исправил это в коде.
data = [0]*buffer_len # buffer_len = 16k <--- Should be 160k instead
data
, просто используйтеself.cbuffer[0].extend(data)
. Если вы добавляете фрагмент, используйтеself.cbuffer[0].extend(itertools.islice(data,None,self.number_of_points))
- person Padraic Cunningham   schedule 15.04.2015data_feeds
? - person sebastian   schedule 15.04.2015