Benchmarks
Video speed test
"""
Tests the speed of image updates for an ImageItem and RawImageWidget.
The speed will generally depend on the type of data being shown, whether
it is being scaled and/or converted by lookup table, and whether OpenGL
is used by the view widget
"""
import argparse
import sys
from time import perf_counter
import numpy as np
import pyqtgraph as pg
from pyqtgraph.Qt import QT_LIB, QtCore, QtGui, QtWidgets
pg.setConfigOption('imageAxisOrder', 'row-major')
import importlib
ui_template = importlib.import_module(f'VideoTemplate_{QT_LIB.lower()}')
try:
import cupy as cp
pg.setConfigOption("useCupy", True)
_has_cupy = True
except ImportError:
cp = None
_has_cupy = False
try:
import numba
_has_numba = True
except ImportError:
numba = None
_has_numba = False
try:
from pyqtgraph.widgets.RawImageWidget import RawImageGLWidget
except ImportError:
RawImageGLWidget = None
parser = argparse.ArgumentParser(description="Benchmark for testing video performance")
parser.add_argument('--cuda', default=False, action='store_true', help="Use CUDA to process on the GPU", dest="cuda")
parser.add_argument('--dtype', default='uint8', choices=['uint8', 'uint16', 'float'], help="Image dtype (uint8, uint16, or float)")
parser.add_argument('--frames', default=3, type=int, help="Number of image frames to generate (default=3)")
parser.add_argument('--image-mode', default='mono', choices=['mono', 'rgb'], help="Image data mode (mono or rgb)", dest='image_mode')
parser.add_argument('--levels', default=None, type=lambda s: tuple([float(x) for x in s.split(',')]), help="min,max levels to scale monochromatic image dynamic range, or rmin,rmax,gmin,gmax,bmin,bmax to scale rgb")
parser.add_argument('--lut', default=False, action='store_true', help="Use color lookup table")
parser.add_argument('--lut-alpha', default=False, action='store_true', help="Use alpha color lookup table", dest='lut_alpha')
parser.add_argument('--size', default='512x512', type=lambda s: tuple([int(x) for x in s.split('x')]), help="WxH image dimensions default='512x512'")
args = parser.parse_args(sys.argv[1:])
if RawImageGLWidget is not None:
# don't limit frame rate to vsync
sfmt = QtGui.QSurfaceFormat()
sfmt.setSwapInterval(0)
QtGui.QSurfaceFormat.setDefaultFormat(sfmt)
app = pg.mkQApp("Video Speed Test Example")
win = QtWidgets.QMainWindow()
win.setWindowTitle('pyqtgraph example: VideoSpeedTest')
ui = ui_template.Ui_MainWindow()
ui.setupUi(win)
win.show()
if RawImageGLWidget is None:
ui.rawGLRadio.setEnabled(False)
ui.rawGLRadio.setText(ui.rawGLRadio.text() + " (OpenGL not available)")
else:
ui.rawGLImg = RawImageGLWidget()
ui.stack.addWidget(ui.rawGLImg)
# read in CLI args
ui.cudaCheck.setChecked(args.cuda and _has_cupy)
ui.cudaCheck.setEnabled(_has_cupy)
ui.numbaCheck.setChecked(_has_numba and pg.getConfigOption("useNumba"))
ui.numbaCheck.setEnabled(_has_numba)
ui.framesSpin.setValue(args.frames)
ui.widthSpin.setValue(args.size[0])
ui.heightSpin.setValue(args.size[1])
ui.dtypeCombo.setCurrentText(args.dtype)
ui.rgbCheck.setChecked(args.image_mode=='rgb')
ui.maxSpin1.setOpts(value=255, step=1)
ui.minSpin1.setOpts(value=0, step=1)
levelSpins = [ui.minSpin1, ui.maxSpin1, ui.minSpin2, ui.maxSpin2, ui.minSpin3, ui.maxSpin3]
if args.cuda and _has_cupy:
xp = cp
else:
xp = np
if args.levels is None:
ui.scaleCheck.setChecked(False)
ui.rgbLevelsCheck.setChecked(False)
else:
ui.scaleCheck.setChecked(True)
if len(args.levels) == 2:
ui.rgbLevelsCheck.setChecked(False)
ui.minSpin1.setValue(args.levels[0])
ui.maxSpin1.setValue(args.levels[1])
elif len(args.levels) == 6:
ui.rgbLevelsCheck.setChecked(True)
for spin,val in zip(levelSpins, args.levels):
spin.setValue(val)
else:
raise ValueError("levels argument must be 2 or 6 comma-separated values (got %r)" % (args.levels,))
ui.lutCheck.setChecked(args.lut)
ui.alphaCheck.setChecked(args.lut_alpha)
#ui.graphicsView.useOpenGL() ## buggy, but you can try it if you need extra speed.
vb = pg.ViewBox()
ui.graphicsView.setCentralItem(vb)
vb.setAspectLocked()
img = pg.ImageItem()
vb.addItem(img)
LUT = None
def updateLUT():
global LUT, ui
dtype = ui.dtypeCombo.currentText()
if dtype == 'uint8':
n = 256
else:
n = 4096
LUT = ui.gradient.getLookupTable(n, alpha=ui.alphaCheck.isChecked())
if _has_cupy and xp == cp:
LUT = cp.asarray(LUT)
ui.gradient.sigGradientChanged.connect(updateLUT)
updateLUT()
ui.alphaCheck.toggled.connect(updateLUT)
def updateScale():
global ui, levelSpins
if ui.rgbLevelsCheck.isChecked():
for s in levelSpins[2:]:
s.setEnabled(True)
else:
for s in levelSpins[2:]:
s.setEnabled(False)
updateScale()
ui.rgbLevelsCheck.toggled.connect(updateScale)
cache = {}
def mkData():
with pg.BusyCursor():
global data, cache, ui, xp
frames = ui.framesSpin.value()
width = ui.widthSpin.value()
height = ui.heightSpin.value()
cacheKey = (ui.dtypeCombo.currentText(), ui.rgbCheck.isChecked(), frames, width, height)
if cacheKey not in cache:
if cacheKey[0] == 'uint8':
dt = xp.uint8
loc = 128
scale = 64
mx = 255
elif cacheKey[0] == 'uint16':
dt = xp.uint16
loc = 4096
scale = 1024
mx = 2**16 - 1
elif cacheKey[0] == 'float':
dt = xp.float32
loc = 1.0
scale = 0.1
mx = 1.0
else:
raise ValueError(f"unable to handle dtype: {cacheKey[0]}")
chan_shape = (height, width)
if ui.rgbCheck.isChecked():
frame_shape = chan_shape + (3,)
else:
frame_shape = chan_shape
data = xp.empty((frames,) + frame_shape, dtype=dt)
view = data.reshape((-1,) + chan_shape)
for idx in range(view.shape[0]):
subdata = xp.random.normal(loc=loc, scale=scale, size=chan_shape)
# note: gaussian filtering has been removed as it slows down array
# creation greatly.
if cacheKey[0] != 'float':
xp.clip(subdata, 0, mx, out=subdata)
view[idx] = subdata
data[:, 10:50, 10] = mx
data[:, 48, 9:12] = mx
data[:, 47, 8:13] = mx
cache = {cacheKey: data} # clear to save memory (but keep one to prevent unnecessary regeneration)
data = cache[cacheKey]
updateLUT()
updateSize()
def updateSize():
global ui, vb
frames = ui.framesSpin.value()
width = ui.widthSpin.value()
height = ui.heightSpin.value()
dtype = xp.dtype(str(ui.dtypeCombo.currentText()))
rgb = 3 if ui.rgbCheck.isChecked() else 1
ui.sizeLabel.setText('%d MB' % (frames * width * height * rgb * dtype.itemsize / 1e6))
vb.setRange(QtCore.QRectF(0, 0, width, height))
def noticeCudaCheck():
global xp, cache
cache = {}
if ui.cudaCheck.isChecked():
if _has_cupy:
xp = cp
else:
xp = np
ui.cudaCheck.setChecked(False)
else:
xp = np
mkData()
def noticeNumbaCheck():
pg.setConfigOption('useNumba', _has_numba and ui.numbaCheck.isChecked())
mkData()
ui.dtypeCombo.currentIndexChanged.connect(mkData)
ui.rgbCheck.toggled.connect(mkData)
ui.widthSpin.editingFinished.connect(mkData)
ui.heightSpin.editingFinished.connect(mkData)
ui.framesSpin.editingFinished.connect(mkData)
ui.widthSpin.valueChanged.connect(updateSize)
ui.heightSpin.valueChanged.connect(updateSize)
ui.framesSpin.valueChanged.connect(updateSize)
ui.cudaCheck.toggled.connect(noticeCudaCheck)
ui.numbaCheck.toggled.connect(noticeNumbaCheck)
ptr = 0
lastTime = perf_counter()
fps = None
def update():
global ui, ptr, lastTime, fps, LUT, img
if ui.lutCheck.isChecked():
useLut = LUT
else:
useLut = None
downsample = ui.downsampleCheck.isChecked()
if ui.scaleCheck.isChecked():
if ui.rgbLevelsCheck.isChecked():
useScale = [
[ui.minSpin1.value(), ui.maxSpin1.value()],
[ui.minSpin2.value(), ui.maxSpin2.value()],
[ui.minSpin3.value(), ui.maxSpin3.value()]]
else:
useScale = [ui.minSpin1.value(), ui.maxSpin1.value()]
else:
useScale = None
if ui.rawRadio.isChecked():
ui.rawImg.setImage(data[ptr%data.shape[0]], lut=useLut, levels=useScale)
ui.stack.setCurrentIndex(1)
elif ui.rawGLRadio.isChecked():
ui.rawGLImg.setImage(data[ptr%data.shape[0]], lut=useLut, levels=useScale)
ui.stack.setCurrentIndex(2)
else:
img.setImage(data[ptr%data.shape[0]], autoLevels=False, levels=useScale, lut=useLut, autoDownsample=downsample)
ui.stack.setCurrentIndex(0)
#img.setImage(data[ptr%data.shape[0]], autoRange=False)
ptr += 1
now = perf_counter()
dt = now - lastTime
lastTime = now
if fps is None:
fps = 1.0/dt
else:
s = np.clip(dt*3., 0, 1)
fps = fps * (1-s) + (1.0/dt) * s
ui.fpsLabel.setText('%0.2f fps' % fps)
app.processEvents() ## force complete redraw for every plot
timer = QtCore.QTimer()
timer.timeout.connect(update)
timer.start(0)
if __name__ == '__main__':
pg.exec()
Line Plot update
#!/usr/bin/python
"""
Update a simple plot as rapidly as possible to measure speed.
"""
import argparse
from collections import deque
from time import perf_counter
import numpy as np
import pyqtgraph as pg
import pyqtgraph.functions as fn
import pyqtgraph.parametertree as ptree
from pyqtgraph.Qt import QtCore, QtGui, QtWidgets
# defaults here result in the same configuration as the original PlotSpeedTest
parser = argparse.ArgumentParser()
parser.add_argument('--noise', dest='noise', action='store_true')
parser.add_argument('--no-noise', dest='noise', action='store_false')
parser.set_defaults(noise=True)
parser.add_argument('--nsamples', default=5000, type=int)
parser.add_argument('--frames', default=50, type=int)
parser.add_argument('--fsample', default=1000, type=float)
parser.add_argument('--frequency', default=0, type=float)
parser.add_argument('--amplitude', default=5, type=float)
parser.add_argument('--opengl', dest='use_opengl', action='store_true')
parser.add_argument('--no-opengl', dest='use_opengl', action='store_false')
parser.set_defaults(use_opengl=None)
parser.add_argument('--allow-opengl-toggle', action='store_true',
help="""Allow on-the-fly change of OpenGL setting. This may cause unwanted side effects.
""")
args = parser.parse_args()
if args.use_opengl is not None:
pg.setConfigOption('useOpenGL', args.use_opengl)
pg.setConfigOption('enableExperimental', args.use_opengl)
# don't limit frame rate to vsync
sfmt = QtGui.QSurfaceFormat()
sfmt.setSwapInterval(0)
QtGui.QSurfaceFormat.setDefaultFormat(sfmt)
class MonkeyCurveItem(pg.PlotCurveItem):
def __init__(self, *args, **kwds):
super().__init__(*args, **kwds)
self.monkey_mode = ''
def setMethod(self, param, value):
self.monkey_mode = value
def paint(self, painter, opt, widget):
if self.monkey_mode not in ['drawPolyline']:
return super().paint(painter, opt, widget)
painter.setRenderHint(painter.RenderHint.Antialiasing, self.opts['antialias'])
painter.setPen(pg.mkPen(self.opts['pen']))
if self.monkey_mode == 'drawPolyline':
painter.drawPolyline(fn.arrayToQPolygonF(self.xData, self.yData))
app = pg.mkQApp("Plot Speed Test")
default_pen = pg.mkPen()
children = [
dict(name='sigopts', title='Signal Options', type='group', children=[
dict(name='noise', type='bool', value=args.noise),
dict(name='nsamples', type='int', limits=[0, None], value=args.nsamples),
dict(name='frames', type='int', limits=[1, None], value=args.frames),
dict(name='fsample', title='sample rate', type='float', value=args.fsample, units='Hz'),
dict(name='frequency', type='float', value=args.frequency, units='Hz'),
dict(name='amplitude', type='float', value=args.amplitude),
]),
dict(name='useOpenGL', type='bool', value=pg.getConfigOption('useOpenGL'),
readonly=not args.allow_opengl_toggle),
dict(name='enableExperimental', type='bool', value=pg.getConfigOption('enableExperimental')),
dict(name='pen', type='pen', value=default_pen),
dict(name='antialias', type='bool', value=pg.getConfigOption('antialias')),
dict(name='connect', type='list', limits=['all', 'pairs', 'finite', 'array'], value='all'),
dict(name='fill', type='bool', value=False),
dict(name='skipFiniteCheck', type='bool', value=False),
dict(name='plotMethod', title='Plot Method', type='list', limits=['pyqtgraph', 'drawPolyline'])
]
params = ptree.Parameter.create(name='Parameters', type='group', children=children)
pt = ptree.ParameterTree(showHeader=False)
pt.setParameters(params)
pw = pg.PlotWidget()
splitter = QtWidgets.QSplitter()
splitter.addWidget(pt)
splitter.addWidget(pw)
splitter.show()
pw.setWindowTitle('pyqtgraph example: PlotSpeedTest')
pw.setLabel('bottom', 'Index', units='B')
curve = MonkeyCurveItem(pen=default_pen, brush='b')
pw.addItem(curve)
rollingAverageSize = 1000
elapsed = deque(maxlen=rollingAverageSize)
def resetTimings(*args):
elapsed.clear()
def makeData(*args):
global data, connect_array, ptr
sigopts = params.child('sigopts')
nsamples = sigopts['nsamples']
frames = sigopts['frames']
Fs = sigopts['fsample']
A = sigopts['amplitude']
F = sigopts['frequency']
ttt = np.arange(frames * nsamples, dtype=np.float64) / Fs
data = A*np.sin(2*np.pi*F*ttt).reshape((frames, nsamples))
if sigopts['noise']:
data += np.random.normal(size=data.shape)
connect_array = np.ones(data.shape[-1], dtype=bool)
ptr = 0
pw.setRange(QtCore.QRectF(0, -10, nsamples, 20))
def onUseOpenGLChanged(param, enable):
pw.useOpenGL(enable)
def onEnableExperimentalChanged(param, enable):
pg.setConfigOption('enableExperimental', enable)
def onPenChanged(param, pen):
curve.setPen(pen)
def onFillChanged(param, enable):
curve.setFillLevel(0.0 if enable else None)
params.child('sigopts').sigTreeStateChanged.connect(makeData)
params.child('useOpenGL').sigValueChanged.connect(onUseOpenGLChanged)
params.child('enableExperimental').sigValueChanged.connect(onEnableExperimentalChanged)
params.child('pen').sigValueChanged.connect(onPenChanged)
params.child('fill').sigValueChanged.connect(onFillChanged)
params.child('plotMethod').sigValueChanged.connect(curve.setMethod)
params.sigTreeStateChanged.connect(resetTimings)
makeData()
fpsLastUpdate = perf_counter()
def update():
global curve, data, ptr, elapsed, fpsLastUpdate
options = ['antialias', 'connect', 'skipFiniteCheck']
kwds = { k : params[k] for k in options }
if kwds['connect'] == 'array':
kwds['connect'] = connect_array
# Measure
t_start = perf_counter()
curve.setData(data[ptr], **kwds)
app.processEvents(QtCore.QEventLoop.ProcessEventsFlag.AllEvents)
t_end = perf_counter()
elapsed.append(t_end - t_start)
ptr = (ptr + 1) % data.shape[0]
# update fps at most once every 0.2 secs
if t_end - fpsLastUpdate > 0.2:
fpsLastUpdate = t_end
average = np.mean(elapsed)
fps = 1 / average
pw.setTitle('%0.2f fps - %0.1f ms avg' % (fps, average * 1_000))
timer = QtCore.QTimer()
timer.timeout.connect(update)
timer.start(0)
if __name__ == '__main__':
pg.exec()
Scatter Plot update
#!/usr/bin/python
"""
For testing rapid updates of ScatterPlotItem under various conditions.
"""
from time import perf_counter
import numpy as np
import pyqtgraph as pg
import pyqtgraph.parametertree as ptree
from pyqtgraph.Qt import QtCore, QtWidgets
translate = QtCore.QCoreApplication.translate
app = pg.mkQApp()
param = ptree.Parameter.create(name=translate('ScatterPlot', 'Parameters'), type='group', children=[
dict(name='paused', title=translate('ScatterPlot', 'Paused: '), type='bool', value=False),
dict(name='count', title=translate('ScatterPlot', 'Count: '), type='int', limits=[1, None], value=500, step=100),
dict(name='size', title=translate('ScatterPlot', 'Size: '), type='int', limits=[1, None], value=10),
dict(name='randomize', title=translate('ScatterPlot', 'Randomize: '), type='bool', value=False),
dict(name='pxMode', title='pxMode: ', type='bool', value=True),
dict(name='useCache', title='useCache: ', type='bool', value=True),
dict(name='mode', title=translate('ScatterPlot', 'Mode: '), type='list', limits={translate('ScatterPlot', 'New Item'): 'newItem', translate('ScatterPlot', 'Reuse Item'): 'reuseItem', translate('ScatterPlot', 'Simulate Pan/Zoom'): 'panZoom', translate('ScatterPlot', 'Simulate Hover'): 'hover'}, value='reuseItem'),
])
for c in param.children():
c.setDefault(c.value())
pt = ptree.ParameterTree(showHeader=False)
pt.setParameters(param)
p = pg.PlotWidget()
splitter = QtWidgets.QSplitter()
splitter.addWidget(pt)
splitter.addWidget(p)
splitter.show()
data = {}
item = pg.ScatterPlotItem()
hoverBrush = pg.mkBrush('y')
ptr = 0
lastTime = perf_counter()
fps = None
timer = QtCore.QTimer()
def mkDataAndItem():
global data, fps
scale = 100
data = {
'pos': np.random.normal(size=(50, param['count']), scale=scale),
'pen': [pg.mkPen(x) for x in np.random.randint(0, 256, (param['count'], 3))],
'brush': [pg.mkBrush(x) for x in np.random.randint(0, 256, (param['count'], 3))],
'size': (np.random.random(param['count']) * param['size']).astype(int)
}
data['pen'][0] = pg.mkPen('w')
data['size'][0] = param['size']
data['brush'][0] = pg.mkBrush('b')
bound = 5 * scale
p.setRange(xRange=[-bound, bound], yRange=[-bound, bound])
mkItem()
def mkItem():
global item
item = pg.ScatterPlotItem(pxMode=param['pxMode'], **getData())
item.opts['useCache'] = param['useCache']
p.clear()
p.addItem(item)
def getData():
pos = data['pos']
pen = data['pen']
size = data['size']
brush = data['brush']
if not param['randomize']:
pen = pen[0]
size = size[0]
brush = brush[0]
return dict(x=pos[ptr % 50], y=pos[(ptr + 1) % 50], pen=pen, brush=brush, size=size)
def update():
global ptr, lastTime, fps
mode = param['mode']
if mode == 'newItem':
mkItem()
elif mode == 'reuseItem':
item.setData(**getData())
elif mode == 'panZoom':
item.viewTransformChanged()
item.update()
elif mode == 'hover':
pts = item.points()
old = pts[(ptr - 1) % len(pts)]
new = pts[ptr % len(pts)]
item.pointsAt(new.pos())
old.resetBrush() # reset old's brush before setting new's to better simulate hovering
new.setBrush(hoverBrush)
ptr += 1
now = perf_counter()
dt = now - lastTime
lastTime = now
if fps is None:
fps = 1.0 / dt
else:
s = np.clip(dt * 3., 0, 1)
fps = fps * (1 - s) + (1.0 / dt) * s
p.setTitle('%0.2f fps' % fps)
p.repaint()
# app.processEvents() # force complete redraw for every plot
mkDataAndItem()
for name in ['count', 'size']:
param.child(name).sigValueChanged.connect(mkDataAndItem)
for name in ['useCache', 'pxMode', 'randomize']:
param.child(name).sigValueChanged.connect(mkItem)
param.child('paused').sigValueChanged.connect(lambda _, v: timer.stop() if v else timer.start())
timer.timeout.connect(update)
timer.start(0)
if __name__ == '__main__':
pg.exec()
Multiple plots
#!/usr/bin/python
"""
Test the speed of rapidly updating multiple plot curves
"""
from time import perf_counter
import numpy as np
import pyqtgraph as pg
from pyqtgraph.Qt import QtCore
# pg.setConfigOptions(useOpenGL=True)
app = pg.mkQApp("MultiPlot Speed Test")
plot = pg.plot()
plot.setWindowTitle('pyqtgraph example: MultiPlotSpeedTest')
plot.setLabel('bottom', 'Index', units='B')
nPlots = 100
nSamples = 500
curves = []
for idx in range(nPlots):
curve = pg.PlotCurveItem(pen=({'color': (idx, nPlots*1.3), 'width': 1}), skipFiniteCheck=True)
plot.addItem(curve)
curve.setPos(0,idx*6)
curves.append(curve)
plot.setYRange(0, nPlots*6)
plot.setXRange(0, nSamples)
plot.resize(600,900)
rgn = pg.LinearRegionItem([nSamples/5.,nSamples/3.])
plot.addItem(rgn)
data = np.random.normal(size=(nPlots*23,nSamples))
ptr = 0
lastTime = perf_counter()
fps = None
count = 0
def update():
global curve, data, ptr, plot, lastTime, fps, nPlots, count
count += 1
for i in range(nPlots):
curves[i].setData(data[(ptr+i)%data.shape[0]])
ptr += nPlots
now = perf_counter()
dt = now - lastTime
lastTime = now
if fps is None:
fps = 1.0/dt
else:
s = np.clip(dt*3., 0, 1)
fps = fps * (1-s) + (1.0/dt) * s
plot.setTitle('%0.2f fps' % fps)
#app.processEvents() ## force complete redraw for every plot
timer = QtCore.QTimer()
timer.timeout.connect(update)
timer.start(0)
if __name__ == '__main__':
pg.exec()
コメント