[Python] Python multiprocessing queues have a delay that increases over time while plotting...

I have my script that opens 3 pyaudio streams, receiving data at 30, 40 and 50hz intervals (each stream uses a different chunk value based on the desired frequency)

import multiprocessing
import time
import numpy as np
import struct
import pyaudio as pa
import matplotlib.pyplot as plt


'''
lowest frequency(and interval between frequencies) f = rate / chunk

=> rate / f = chunk

Frequencies:
30hz => 60 hz => 90hz... chunk = 1470
40hz => 80 hz =>... chunk = 1102
50hz => 100hz =>... chunk = 882

'''

np.set_printoptions(threshold = np.inf)

FORMAT = pa.paInt16
CHANNELS = 1
RATE = 44100


p30 = pa.PyAudio()
p40 = pa.PyAudio()
p50 = pa.PyAudio()


# Function to record audio data at 30Hz intervals
def record_30hz_I(data_30hz_queue):

CHUNK_30hz = 1470

stream_30hz = p30.open(
format = FORMAT,
channels = CHANNELS,
rate = RATE,
input=True,
frames_per_buffer=CHUNK_30hz
)

while True:
data_30hz = stream_30hz.read(CHUNK_30hz)
data_30hz_Int = struct.unpack(str(CHUNK_30hz) + 'h', data_30hz)
fft_data_30hz = np.abs(np.fft.fft(data_30hz_Int))*2/(11000*CHUNK_30hz)
#print(fft_data_30hz)

if data_30hz_queue.empty:
data_30hz_queue.put(fft_data_30hz)


# Function to record audio data at 40Hz intervals
def record_40hz_I(data_40hz_queue):

CHUNK_40hz = 1102

stream_40hz = p40.open(
format = FORMAT,
channels = CHANNELS,
rate = RATE,
input=True,
frames_per_buffer=CHUNK_40hz
)

while True:
data_40hz = stream_40hz.read(CHUNK_40hz)
data_40hz_Int = struct.unpack(str(CHUNK_40hz) + 'h', data_40hz)
fft_data_40hz = np.abs(np.fft.fft(data_40hz_Int))*2/(11000*CHUNK_40hz)
#print(fft_data_40hz)

if data_40hz_queue.empty:
data_40hz_queue.put(fft_data_40hz)


# Function to record audio data at 50Hz intervals
def record_50hz_I(data_50hz_queue):

CHUNK_50hz = 882

stream_50hz = p50.open(
format = FORMAT,
channels = CHANNELS,
rate = RATE,
input=True,
frames_per_buffer=CHUNK_50hz
)


while True:
data_50hz = stream_50hz.read(CHUNK_50hz)
data_50hz_Int = struct.unpack(str(CHUNK_50hz) + 'h', data_50hz)
fft_data_50hz = np.abs(np.fft.fft(data_50hz_Int))*2/(11000*CHUNK_50hz)
#print(fft_data_50hz)

if data_50hz_queue.empty:
data_50hz_queue.put(fft_data_50hz)

# Function to plot FFT data
def plot(data_30hz_queue, data_40hz_queue, data_50hz_queue):

fig, (ax1) = plt.subplots(1)

x_fft1 = np.linspace(0, RATE, 1470)
x_fft2 = np.linspace(0, RATE, 1102)
x_fft3 = np.linspace(0, RATE, 882)

line_fft_30hz, = ax1.semilogx(x_fft1, np.random.rand(1470), 'b')
line_fft_40hz, = ax1.semilogx(x_fft2, np.random.rand(1102), 'g')
line_fft_50hz, = ax1.semilogx(x_fft3, np.random.rand(882), 'r')

ax1.set_xlim(20,RATE/2)
ax1.set_ylim(0,0.5)
fig.show()


while True:

line_fft_30hz.set_ydata(data_30hz_queue.get())
line_fft_40hz.set_ydata(data_40hz_queue.get())
line_fft_50hz.set_ydata(data_50hz_queue.get())

fig.canvas.draw()
fig.canvas.flush_events()



if __name__ == "__main__":

data_30hz_queue = multiprocessing.Queue()
data_40hz_queue = multiprocessing.Queue()
data_50hz_queue = multiprocessing.Queue()

p_record_30hz_I = multiprocessing.Process(target = record_30hz_I, args = (data_30hz_queue,))
p_record_40hz_I = multiprocessing.Process(target = record_40hz_I, args = (data_40hz_queue,))
p_record_50hz_I = multiprocessing.Process(target = record_50hz_I, args = (data_50hz_queue,))

p_record_50hz_I.start()
p_record_40hz_I.start()
p_record_30hz_I.start()


p_plot = multiprocessing.Process(target = plot, args = (data_30hz_queue, data_40hz_queue, data_50hz_queue, ))
p_plot.start()


So the blue line plotting the 30hz line works fine but the green (40hz) and red (50hz) line have a latency that increases over time and the red line has more delay than the green one.

I have tried printing the FFT data directly but it doesn't make a difference.

But if I remove the code to plot the blue 30hz line the green line works fine and the red one has less latency.

It's very possible I've just missed something about queues and how they work but I've searched and found nothing about this or someone with the same problem.

I would be very grateful for an explanation about what's happening here.

Continue reading...