Thoughts on an asyncio interface to sounddevice

[multimeric]

python-sounddevice already has a nice callback interface. Do you think it would be possible to add a module within sounddevice that exposes an asyncio interface for it? This would let you do things like:

import sounddevice._async as sd

async def some_coroutine():
    async for chunk in sd.Stream(channels=2):
        # Process chunk, but let Python do things while waiting for the next chunk

I'd be happy to implement this if there's any interest. If not I'll write a separate async wrapper module that imports python-sounddevice. Let me know!

[mgeier]

Thanks for reaching out, that sounds really interesting!

For now I would like to keep compatibility with Python 2 (and probably pre-"async" Python 3). Therefore the code cannot live in the same module as the rest. However, it would of course be possible to turn the sounddevice module into a package and then have a sub-module with the async stuff.

I don't have much experience with asyncio and I cannot foresee if this would be feasible or not.

What about making a proof-of-concept implementation and then we can discuss if we should include it in the sounddevice module or if a separate wrapper makes more sense?

[multimeric]

My first attempt to solve this was by making a general async wrapper for callbacks (called Asynchronize), which might be generalisable beyond just sounddevice. Using this library I've written an example of how the async code looks with sounddevice: https://github.com/TMiguelT/Asynchronize/blob/master/example.py

I think it's a good start, but it's still a bit awkward. Because of the way that the sounddevice API is structured, nothing accepts a time limit. Instead, everything expects you to call stop() (or __leave__ the context manager) when you're done recording. This doesn't play well with the await paradigm, and forces the awkward code in my above example (with the separate function that calls sleep()).

I'd like to be able to do away with the simultaneous sleep command, and do this instead:

async for chunk in sd.Stream(channels=2, time=5)

What do you think of an API like this? How do you think it would be implemented? Should I make a separate submodule that provides an async class that inherits from Stream?

[multimeric]

Or possibly I could implement a timer utility in Asynchronize that helps in cases like this

[mgeier]

Thanks for the update! I'll have to read up on a few things and experiment a bit with your code until I can give any real feedback.

I'm not sure if this does anything fundamentally different, but did you see https://stackoverflow.com/a/53996189/?

I'm a bit concerned that you don't seem to be making a copy of the input buffer you get from the callback. The buffer may be re-used for one of the next calls to the callback, thereby overwriting the data that's already in the queue.

Did you also think about providing a way of sending audio data back to the callback?
Or is this impossible with the async approach?

If it's possible, how/where would signal processing be done, since I guess this should not block the asyncio thread?
Would you use something like a ThreadPoolExecutor for that?

What if you would like to do something in your callback function before putting data into the queue?
For example, select only a sub-set of channels to put into the queue?

[multimeric]

Did you see https://stackoverflow.com/a/53996189/?

Yes, that was me who asked the original question. I generalised the helpful answer into my Asynchronize library

I'm a bit concerned that you don't seem to be making a copy of the input buffer you get from the callback

You mean my library doesn't make a copy of the buffer when it adds it to the queue? Hmm that's a good point. But why then does it still print out a different mean value each time? Perhaps this proves that my generic library solution won't work, and sounddevice's idiosyncrasies (ie the buffer which is just a reference) need their own attention.

Did you also think about providing a way of sending audio data back to the callback

This might be a bit harder, but I think you could pass an async generator into the stream as an alternative to a callback. However the main use case for me is getting audio input, so this isn't a priority right now.

If it's possible, how/where would signal processing be done, since I guess this should not block the asyncio thread?

You're right about not blocking the event loop. Processing should probably be done with ProcessPoolExecutor and loop.run_in_executor() (ThreadPoolExecutor won't be any faster because of the GIL): https://stackoverflow.com/a/29147750/2148718

What if you would like to do something in your callback function before putting data into the queue?
For example, select only a sub-set of channels to put into the queue?

In my implementation, you can't. My library copies all of the callback arguments into the queue and it's only processed once it's yielded by the callback object. So I guess if the queue ever fills up, it will be using more memory that you might like. This might be another idiosyncrasy that my library doesn't handle especially well.

[mgeier]

I tried running your example.py and it crashed the Python interpreter.
I also tried the example on SO (with a few additions, because it isn't runnable as is), and it also crashed the interpreter.

Probably I'm doing something very wrong here. As I said, I'm quite new to this asyncio stuff.

Do you happen to have an even more minimal (but fully working) code example?

Yes, that was me who asked the original question.

Oh, cool, I didn't realize that!

You mean my library doesn't make a copy of the buffer when it adds it to the queue?

Exactly.

But why then does it still print out a different mean value each time?

I guess there is not a single buffer that is reused but there are a few of them. So theoretically, if you make the queue a bit longer, you should see a repeating pattern in the mean values.
Or the interpreter crashes because you try to access memory you shouldn't access?

This very likely also depends on the operating system and host API that's used.

I tried it with Linux/ALSA.

(ie the buffer which is just a reference)

This is different in PyAudio, because IIRC it returns a copy of the data (in a bytes object).

I decided to return a raw buffer object without copying. On the one hand for efficiency, on the other hand to be symmetric to the handling of the output buffer.

ThreadPoolExecutor won't be any faster because of the GIL

I wouldn't be so sure about that. I guess it depends on what kind of processing you do. For example, many NumPy operations release the GIL and actually can run in parallel with the Python interpreter.

[multimeric]

Hmm, you should be able to just run example.py using Python 3.5+ after cloning the whole repo. What are the errors you're getting?

You can run native C code, and system-level IO in parallel with other Python code, but if you wanted to run multiple threads of Python concurrently (e.g. signal processing, and the event loop), you would need to use multiprocessing. numpy fits into that first category because it's implemented in native C.

[mgeier]

What are the errors you're getting?

Well that's the thing, the interpreter crashed. I didn't get any error message.

$ python3.7 example.py 
Mean: -0.0005440916866064072
Mean: -0.001177023397758603
Aborted

However, just now I tried it again using some different PortAudio devices.

I'm using the ALSA emulation of PulseAudio (on Linux) and it turns out that the crash happens when I use the default device, which happens to be some kind of "virtual" device.

When I choose an actual hardware device it doesn't crash.

So now I can start playing with it, but it would still be interesting how to avoid the crash ...

[mgeier]

So I played around a bit with this and I came up with #163.

What do you think about it?

I'm currently not concerned with how to generalize this, I just wanted to get it up and running.

Because of the way that the sounddevice API is structured, nothing accepts a time limit.

Would you suggest to add a time limit somewhere?

I could implement a timer utility

Do you really need a timer?

Can't you just exit from the async generator?
You probably have to call .close()?

[multimeric]

The reason I was concerned with a timer, is because often users will only want to record for a certain amount of time, not indefinitely like in your example. The timer aspect is what convoluted my example, so maybe if you can find an elegant way to do that in yours it would make it a better example. Yes you could exit from the generator, but when? You can't use sleep unless you do it in parallel with the actual iteration, like in my example. Or you could check what time it is every iteration, but that isn't very elegant.

Also, I like that your example shows you manipulating the output buffer, but can you explain why you needed a queue there? Can't you just zero the output buffer inside your callback each time?

[mgeier]

The reason I was concerned with a timer, is because often users will only want to record for a certain amount of time, not indefinitely like in your example.

That's a good point!
I think if you know in advance how long you want to record, you can use asyncio.wait_for() with a timeout argument.

I guess you'll have to wrap the async generator in a coroutine, e.g. like this (using the stream_generator() from my example):

async def wire_coro(**kwargs):
    async for indata, outdata, status in stream_generator(**kwargs):
        if status:
            print(status)
        outdata[:] = indata

Then you can run this coroutine with a timeout:

async def main(**kwargs):
    await asyncio.sleep(5)
    print('starting audio stuff')
    try:
        await asyncio.wait_for(wire_coro(**kwargs), timeout=5)
    except asyncio.TimeoutError:
        print('done with audio stuff')
    await asyncio.sleep(5)

if __name__ == "__main__":
    asyncio.run(main(blocksize=1024))

However, I think knowing the exact time in advance is not very common, or is it?

If there is some other asynchronous stuff that is supposed to happen during playback/recording, I think it makes sense to wrap the "audio stream generator" in an asyncio.Task and cancel() it whenever the time has come.

I've tried this in a new commit (117b8b8) in my example.

can you explain why you needed a queue there? Can't you just zero the output buffer inside your callback each time?

This is not about zeroing the output buffer!
Yes, if I only wanted to zero the output buffer, I could do that directly in the callback. But then I could just use an InputStream in the first place, right?

I'm actually sending audio data back in the output queue q_out. In my "wire" example it's only the unchanged input data, but I could do arbitrary signal processing there.

The interesting thing about this, which I wasn't aware of before, is that I have to use an asyncio.Queue for input and a queue.Queue for the output.
Or probably I should use call_soon_threadsafe() a second time ...?

[mgeier]

OK, I just tried it with a second call_soon_threadsafe() but this doesn't seem to be the way to go. I might be wrong though ...

[mgeier]

@TMiguelT I've added a few more examples to #163. Does that make sense?

In addition to a generator for input and output, I've also created an example of a generator for only input.

I guess an async generator for only output doesn't make sense, right?

Do you think I should add an example for async output (without input)?
If yes, do you have a concrete idea for an example?

Your original example from the very top of this page could now be written like this:

async def some_coroutine():
    async for chunk in async_generators.inputstream_generator(channels=2):
        # ...

I think I'll just keep my asyncio examples together with the rest of the examples, but if you have ideas how to generalize this, I would still be interested!

[mgeier]

@TMiguelT What do you think about my examples? Any further suggestions?

[mgeier]

I'm closing this. The asyncio examples are available in https://github.com/spatialaudio/python-sounddevice/tree/master/examples.

[multimeric]

Seems reasonable. Of course as a user I always find it nicer for code like this to be actually importable in the library, but this is certainly an improvement.

[mgeier]

Well, the examples themselves should actually be importable, but they are not pip-installed, so they have to be put into the Python path manually.

Anyway, I think the abstractions in those examples are not general enough to be included in the "official" sounddevice API.

Also, I wouldn't want to have a hard dependency on asyncio, since it isn't available on a few older Python versions. I wouldn't want to have a hard dependency on NumPy, either.

@TMiguelT I you want you can of course create an additional module that provides some importable async functionality! If you need some changes in the sounddevice API for that, let me know!