Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
pyqtgraph live running spectrogram from microphone
import numpy as np
import pyqtgraph as pg
import pyaudio
from PyQt4 import QtCore, QtGui
FS = 44100 #Hz
CHUNKSZ = 1024 #samples
class MicrophoneRecorder():
def __init__(self, signal):
self.signal = signal
self.p = pyaudio.PyAudio()
self.stream = self.p.open(format=pyaudio.paInt16,
channels=1,
rate=FS,
input=True,
frames_per_buffer=CHUNKSZ)
def read(self):
data = self.stream.read(CHUNKSZ)
y = np.fromstring(data, 'int16')
self.signal.emit(y)
def close(self):
self.stream.stop_stream()
self.stream.close()
self.p.terminate()
class SpectrogramWidget(pg.PlotWidget):
read_collected = QtCore.pyqtSignal(np.ndarray)
def __init__(self):
super(SpectrogramWidget, self).__init__()
self.img = pg.ImageItem()
self.addItem(self.img)
self.img_array = np.zeros((1000, CHUNKSZ/2+1))
# bipolar colormap
pos = np.array([0., 1., 0.5, 0.25, 0.75])
color = np.array([[0,255,255,255], [255,255,0,255], [0,0,0,255], (0, 0, 255, 255), (255, 0, 0, 255)], dtype=np.ubyte)
cmap = pg.ColorMap(pos, color)
lut = cmap.getLookupTable(0.0, 1.0, 256)
# set colormap
self.img.setLookupTable(lut)
self.img.setLevels([-50,40])
# setup the correct scaling for y-axis
freq = np.arange((CHUNKSZ/2)+1)/(float(CHUNKSZ)/FS)
yscale = 1.0/(self.img_array.shape[1]/freq[-1])
self.img.scale((1./FS)*CHUNKSZ, yscale)
self.setLabel('left', 'Frequency', units='Hz')
# prepare window for later use
self.win = np.hanning(CHUNKSZ)
self.show()
def update(self, chunk):
# normalized, windowed frequencies in data chunk
spec = np.fft.rfft(chunk*self.win) / CHUNKSZ
# get magnitude
psd = abs(spec)
# convert to dB scale
psd = 20 * np.log10(psd)
# roll down one and replace leading edge with new data
self.img_array = np.roll(self.img_array, -1, 0)
self.img_array[-1:] = psd
self.img.setImage(self.img_array, autoLevels=False)
if __name__ == '__main__':
app = QtGui.QApplication([])
w = SpectrogramWidget()
w.read_collected.connect(w.update)
mic = MicrophoneRecorder(w.read_collected)
# time (seconds) between reads
interval = FS/CHUNKSZ
t = QtCore.QTimer()
t.timeout.connect(mic.read)
t.start(1000/interval) #QTimer takes ms
app.exec_()
mic.close()
rxa254 commented Mar 5, 2017

I get:
IOError: [Errno -9981] Input overflowed

and then a bunch of
IOError: [Errno -9988] Stream closed

when trying this. I've tried playing around with the chunk size, but no luck. Do you find this still works with modern pyqt ?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment