Hello, world!, context-agnostic¶

Info

See the full example source code on GitHub.

Let’s revisit the first “hello, world!” example and apply the concept of contexts to create the same C-major chord in non-realtime, “sync” realtime and “async” realtime contexts.

We’ll demonstrate how some code can be “context-agnostic”: it doesn’t care if you’re running it live or using it to create a pre-recorded audio file offline.

This might seem a little abstract, but it’s helpful for code re-use, and is foundational for more complex audio applications. Remember how commercial DAWs let you both play your song live and also render it to disk as quickly as possible? Context-agnostic performance logic is part of unlocking that.

Performance logic¶

Let’s extract some performance logic out of the first “hello, world!” example.

We can pull out the logic for starting the chord into its own function, passing the context in as an argument, and returning the list of synths when we’re done:

def play_synths(context: supriya.Context) -> list[supriya.Synth]:
    """
    Play a C-major chord on ``context``.
    """
    # Define a C-major chord in Hertz
    frequencies = [261.63, 329.63, 392.00]
    # Create an empty list to store synths in:
    synths: list[supriya.Synth] = []
    # Add the default synthdef to the server and open a "completion" context
    # manager to group further commands for when the synthdef finishes loading:
    with context.add_synthdefs(supriya.default):
        # Loop over the frequencies:
        for frequency in frequencies:
            # Create a synth using the default synthdef and the frequency
            # and add it to the list of synths:
            synths.append(
                context.add_synth(synthdef=supriya.default, frequency=frequency)
            )
    return synths

Note

Look closely at the context usage when starting the chord: we don’t see any usage of at() to specify when the chord plays. Realtime and non-realtime contexts have subtly different semantics around timestamps both in terms of whether timestamps are mandatory and what the exact time in the timestamp should be. Because of those different semantics, we’ll treat setting up the Moment as a context-specific implementation detail external to this function.

And we can pull out the logic for stopping the chord into its own function, passing a list of synths in as an argument, and return nothing because there’s nothing else to be done:

def stop_synths(synths: list[supriya.Synth]) -> None:
    """
    Stop ``synths``.
    """
    # Loop over the synths and free them
    for synth in synths:
        synth.free()

Nice. We’ve decoupled playing the chord from stopping the chord, and we’ve removed any notion of a concrete context type. These functions can work with both servers and scores interchangeably.

Context management¶

While having context-agnostic logic is great, we still need specific contexts to run our context-agnostic code with.

Let’s create separate functions for playing our C-major chord: two realtime functions using a Server and AsyncServer respectively, and one non-realtime function using a Score.

Threaded realtime¶

Performing with a threaded server is simple, and looks very much like an abridged version of the main() function from the very first example:

def run_threaded() -> None:
    """
    Run the example on a realtime threaded
    :py:class:`~supriya.contexts.realtime.Server`.
    """
    # Create a server and boot it:
    server = supriya.Server().boot()
    # Start an OSC bundle to run immediately:
    with server.at():
        # Start playing the synths
        synths = play_synths(context=server)
    # Let the notes play for 4 seconds:
    time.sleep(4)
    # Loop over the synths and free them:
    stop_synths(synths)
    # Wait a second for the notes to fade out:
    time.sleep(1)
    # Quit the server:
    server.quit()

Async realtime¶

Performing with an async server is virtually identical to performing with a threaded server. The only differences are the server’s class name, and a sprinkling of async and await keywords (and yes, await asyncio.sleep() instead of time.sleep()):

async def run_async() -> None:
    """
    Run the example on an realtime async
    :py:class:`~supriya.contexts.realtime.AsyncServer`.
    """
    # Create an async server and boot it:
    server = await supriya.AsyncServer().boot()
    # Start an OSC bundle to run immediately:
    with server.at():
        # Start playing the synths:
        synths = play_synths(context=server)
    # Let the notes play for 4 seconds:
    await asyncio.sleep(4)
    # Loop over the synths and free them:
    stop_synths(synths)
    # Wait a second for the notes to fade out:
    await asyncio.sleep(1)
    # Quit the async server:
    await server.quit()

Booting, quitting and sleeping are all async operations in this paradigm, but sending messages to the context doesn’t change at all. “Writes” or “mutations” against the server are never an asynchronous operation. We just fire them off.

Non-realtime¶

Performing with a non-realtime score is a little different. While the score doesn’t need to be booted or quit, we do have to be absolutely explicit about when everything happens. Non-realtime contexts have no concept of “now”, so every moment needs to have an explicit timestamp.

We open a moment at 0 seconds and start playing the synths. We don’t need to sleep since nothing is happening live, so instead we open a second moment at 4 seconds to stop playing the synths. And to handle the same 1 second fade out that the realtime functions achieve via a final sleep(1), we just open a third moment at 5 seconds and… do nothing. The final no-op operation against the context just signals to non-realtime rendering that we should keep processing audio until that timestamp.

def run_nonrealtime() -> None:
    """
    Run the example on a non-realtime
    :py:class:`~supriya.contexts.nonrealtime.Score`.
    """
    # Create a score with stereo outputs:
    score = supriya.Score(output_bus_channel_count=2)
    # Start an OSC bundle to run at 0 seconds:
    with score.at(0):
        # Start playing the synths:
        synths = play_synths(context=score)
    # Start an OSC bundle to run at 4 seconds:
    with score.at(4):
        # Loop over the synths and free them:
        stop_synths(synths)
    # Start an OSC bundle to run at 5 seconds:
    with score.at(5):
        # A no-op message to tell the score there's nothing left to do:
        score.do_nothing()
    # Render the score to disk and open the soundfile:
    supriya.play(score)

Note

The timestamps used with scores are all absolute, starting from 0 seconds.

You don’t need to open moments in any particular order: the moment at 5 seconds could’ve been used before the moments at 4 or 0. But stopping the synths had to be applied after staring them, because we still needed Synth objects with IDs to act against.

Timestamps can be visited multiple times too, allowing you to make multiple passes against the same score. Additional commands run at the same timestamp will be appended to the sequence of commands to be run at that timestamp.

The call at the end to play() renders the score to disk and opens the result in your default audio player.

Scripting¶

Now that we have three different ways of creating contexts to use our context-agnostic performance logic, we need a way to yoke them into a script.

Our main() function parses some command-line arguments, and then calls the appropriate run_...() function depending on what we passed to the CLI on invocation:

def main(args: list[str] | None = None) -> None:
    """
    The example entry-point function.
    """
    parsed_args = parse_args(args)
    if parsed_args.realtime_threaded:
        run_threaded()
    elif parsed_args.realtime_async:
        asyncio.run(run_async())
    elif parsed_args.nonrealtime:
        run_nonrealtime()

And, for simplicity, we’ll wrap up the CLI argument parsing into its own function, using Python’s argparse module to create an argument parser, create a mandatory mutually-exclusive group of flags, and then parse whatever CLI arguments were passed when we run the script:

def parse_args(args: list[str] | None = None) -> argparse.Namespace:
    """
    Parse CLI arguments.
    """
    parser = argparse.ArgumentParser(
        description="Play a C-major chord via different kinds of contexts"
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "--realtime-threaded",
        action="store_true",
        help="use a realtime threaded Server",
    )
    group.add_argument(
        "--realtime-async",
        action="store_true",
        help="use a realtime asyncio-enabled AsyncServer",
    )
    group.add_argument(
        "--nonrealtime", action="store_true", help="use a non-realtime Score"
    )
    return parser.parse_args(args)

Invocation¶

You can invoke the script with …

josephine@laptop:~/supriya$ python -m examples.hello_world_contexts --help
usage: __main__.py [-h] (--realtime-threaded | --realtime-async |
                   --nonrealtime)

Play a C-major chord via different kinds of contexts

options:
  -h, --help           show this help message and exit
  --realtime-threaded  use a realtime threaded Server
  --realtime-async     use a realtime asyncio-enabled AsyncServer
  --nonrealtime        use a non-realtime Score

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

For example:

josephine@laptop:~/supriya$ python -m examples.hello_world_contexts --realtime-threaded