Keyboard input, async

Info

See the full example source code on GitHub.

Important

This example requires additional third-party packages. Install them with:

josephine@laptop:~$ pip install supriya[examples]

josephine@laptop:~$ git clone https://github.com/supriya-project/supriya.git
josephine@laptop:~$ cd supriya
josephine@laptop:~/supriya$ pip install -e .[examples]

Let’s revisit the previous keyboard input example, but this time we’ll refactor it to make it work with asyncio. Compared to the first keyboard input example, this is actually a walk in the park. Now we get to see why the previous example enforced strict decoupling between logical components.

Re-used components

We get to re-use many of the components from the non-async keyboard input example:

  • The NoteOn and NoteOff are just event placeholders, so don’t need any change for use with asyncio.

  • The PolyphonyManager acts against a generic write-only Context which makes it concurrency-agnostic.

  • And both the MidiHandler and QwertyHandler classes take callbacks for their listen() context manager, decoupling them from any concurrency-specific logic. While their callbacks can’t be async functions, they can still synchronously interact with code that other async logic relies on, like queues, futures, or even async tasks.

  • We even get to re-use the argument parsing logic, as nothing changes there.

Integration

The core difference between the sync and async examples is in the run() implementation. The async version sprinkles the async keyword into various locations, e.g. around booting, syncing, quitting, waiting on futures, etc. But it also has some deeper logical differences. Take a look, and then we’ll discuss:

async def run(input_handler: InputHandler) -> None:
    """
    Run the script.
    """

    async def on_boot(*args) -> None:  # run this during server.boot()
        server.add_synthdefs(polyphony.synthdef)  # add the polyphony's synthdef
        await server.sync()  # wait for the synthdef to load before moving on

    async def on_quitting(*args) -> None:  # run this during server.quit()
        polyphony.free_all()  # free all the synths
        await asyncio.sleep(0.5)  # wait for them to fade out before moving on

    def signal_handler(*args) -> None:
        exit_future.set_result(True)  # set the exit future flag

    def input_callback(event: NoteOn | NoteOff) -> None:
        # drop the event onto the queue
        loop.call_soon_threadsafe(queue.put_nowait, event)

    async def queue_consumer() -> None:
        while True:  # run forever
            # grab an event off the queue and play it
            polyphony.perform(await queue.get())

    # grab a reference to the current event loop
    loop = asyncio.get_event_loop()
    # create a future we can wait on to quit the script
    exit_future: asyncio.Future[bool] = loop.create_future()
    # create a server and polyphony manager
    server = supriya.AsyncServer()
    polyphony = PolyphonyManager(server=server)
    # setup lifecycle callbacks
    server.register_lifecycle_callback("BOOTED", on_boot)
    server.register_lifecycle_callback("QUITTING", on_quitting)
    # hook up Ctrl-C so we can gracefully shutdown the server
    loop.add_signal_handler(signal.SIGINT, signal_handler)
    # boot the server and let the user know we're ready to play
    await server.boot()
    print("Server online. Press Ctrl-C to exit.")
    # setup an event queue, turn on the input handler, and teach the input
    # handler to callback against the queue
    queue: asyncio.Queue[NoteOn | NoteOff] = asyncio.Queue()
    # setup a queue consumer task to run asynchronously
    queue_consumer_task = loop.create_task(queue_consumer())
    # turn on the input handler and teach it to callback against the queue
    with input_handler.listen(callback=input_callback):
        await exit_future  # wait for Ctrl-C
    queue_consumer_task.cancel()  # cancel the queue consumer task
    # stop the input handler and quit the server
    await server.quit()

The key difference is in how the input handling logic - which runs in a background thread - interacts with the async code running on the event loop in the main thread. In the previous example the input handlers were directly connected to the polyphony manager’s perform() method, but here we mediate the interaction via a Queue. The input handlers safely drop events onto the queue using call_soon_threadsafe(), and a separate coroutine (queue_consumer) runs forever in the event loop, waiting on events to pull off the queue and performing thowe against the polyphony manager.

We create the queue consumer as a task just before input handling starts and hold a reference to it so it doesn’t get garbage collected, and we cancel that task the after the exit future resolves, once we’ve shut down input handling.

Note

There are a lot of different ways of structuring async code (just like any code), and of integrating threaded code with async code. I’ve structured this async example to stay as close as possible to the sync version for the sake of pedagogy. It’s possible to implement this example without the queue as intermediary, but I like how it forces us to think explicitly about the boundaries between background threads and the event loop. As application complexity grows, navigating these thread boundaries becomes crucial, so let’s get used to it a little earlier.

Scripting

Like the run() function, the main() function looks very similar to the previous version. We’re merely wrapping the calls to run() in asyncio.run() to ensure the resulting coroutines are executed (and completed) in an event loop.

def main(args: list[str] | None = None) -> None:
    """
    The example entry-point function.
    """
    parsed_args = parse_args(args)
    if parsed_args.list_midi_inputs:
        # print out available MIDI input ports
        rtmidi.midiutil.list_input_ports()
    elif parsed_args.use_midi is not None:
        asyncio.run(run(MidiHandler(port=parsed_args.use_midi)))
    elif parsed_args.use_qwerty:
        asyncio.run(run(QwertyHandler()))

Invocation

You can invoke the script with …

josephine@laptop:~/supriya$ python -m examples.keyboard_input_async --help
usage: __main__.py [-h] (--list-midi-inputs | --use-midi PORT_NUMBER |
                   --use-qwerty)

Play notes via your QWERTY or MIDI keyboards

options:
  -h, --help            show this help message and exit
  --list-midi-inputs    list available MIDI inputs
  --use-midi PORT_NUMBER
                        play via MIDI keyboard
  --use-qwerty          play via QWERTY keyboard

… and you’ll see the options necessary to properly run it.