-
-
Save rxa254/29144e03172c74fb3c5b2c12abe93015 to your computer and use it in GitHub Desktop.
pyqtgraph live running spectrogram from microphone
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Tested on Linux with python 3.7 | |
Must have portaudio installed (e.g. dnf install portaudio-devel) | |
pip install pyqtgraph pyaudio PyQt5 | |
""" | |
import numpy as np | |
import pyqtgraph as pg | |
import pyaudio | |
from PyQt5 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, exception_on_overflow=False) | |
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, int(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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment