CPU Usage Optimization on Sounddevice using Asyncio

Question

In order to decrease the CPU Usage on Sounddevice with Asyncio, I have tried to put time.sleep(0.1) on while loop, but this throws me a "_queue.Empty" error.

Is there any possible way I can fix this issue?

If it is not possible, is there any other way to decrease the cpu usage on my code?

Any hint or suggestion will be great

Thanks.

Code :

# Libraries
import sounddevice as sd
import numpy as np
import time
import queue
import asyncio

async def stream_generator(input_device_index, output_device_index):
    """Generator that yields blocks of input/output data as NumPy arrays.
    The output blocks are uninitialized and have to be filled with
    appropriate audio signals.
    """

    block_shift = 128

    q_in = asyncio.Queue()
    q_out = queue.Queue()
    loop = asyncio.get_event_loop()

    def callback(indata, outdata, frame_count, time_info, status):
        loop.call_soon_threadsafe(q_in.put_nowait, (indata.copy(), status))
        outdata[:] = q_out.get_nowait()

    # pre-fill output queue
    for _ in range(10):
        q_out.put(np.zeros((block_shift, 1), dtype=np.float32))

    stream = sd.Stream(device=(input_device_index, output_device_index), samplerate=16000,
            blocksize=block_shift, latency=0.2, callback=callback, dtype=np.float32, channels=1)
    with stream:
        while True:
            indata, status = await q_in.get()
            outdata = np.empty((block_shift, 1), dtype=np.float32)
            yield indata, outdata, status
            q_out.put_nowait(outdata)
            #time.sleep(0.1) # I wish to put sleep here for the optimizing cpu usage.

async def wire_coro(input_device_index, output_device_index):
    """Create a connection between audio inputs and outputs.
    Asynchronously iterates over a stream generator and for each block
    simply copies the input data into the output block.
    """
    async for indata, outdata, status in stream_generator(input_device_index, output_device_index):
        if status:
            print(status)
        outdata[:] = indata

async def main(input_device_index, output_device_index):
    audio_task = asyncio.create_task(wire_coro(input_device_index, output_device_index))
    for i in range(10, 0, -1):
        print(i)
        await asyncio.sleep(1)
    audio_task.cancel()


if __name__ == "__main__":
    print(sd.query_devices())
    try:
        asyncio.run(main(1, 4)) # please put input parameters accordingly what sd.query_devices says!
    except KeyboardInterrupt:
        sys.exit('\nInterrupted by user')

Error : error screenshot

Answer 1

I guess you based your code on https://github.com/spatialaudio/python-sounddevice/blob/master/examples/asyncio_generators.py (at least it looks very familiar), which I think would be worth mentioning.

How much CPU usage do you have? Why do you think adding a sleep() would help?

Even if it would help (which it doesn't), you should not use time.sleep() in async code, because this blocks the whole event loop. You should use asyncio.sleep() instead.

But again, why would you?

The code uses await q_in.get() in the loop, which waits (without wasting CPU) for the next available block of input data. This makes sure that the loop runs exactly once per audio block. If you artificially increase the runtime of the loop, it is no wonder that you get a queue.Empty error, because plain and simply the queue gets empty if you don't write enough to it.