make a metronome with python asyncio
Apr 4, 2018
5 minute read

What we will build

We are going to make a metronome using Python’s asyncio module. Our metronome is going tick at a steady rate, executing a collection of subroutines at each tick. I’ve used this module to control a light & music show – but you can use this code for running any operation at regular intervals. The code is relatively straight-forward – I just think it’s a neat introdoction to the power of async IO!

Intro to async IO

Python’s asyncio module is an everything-and-the-kitchen-sink package of parallel & concurrent programming tools. We’re just going to look at a very simple use case for this library: running an infinite loop with a sleep in each iteration.

In synchronous single-threaded programs, calling sleep() does just that – it puts the thread to sleep, and the thread does no work until it wakes up. With Python’s asyncio there is a kind of ‘sleep’ where the thread does not stop working, but instead gets a free pass for a time to do work elsewhere. With calling asyncio.sleep(), the thread refrains from working at the current routine, and makes itself available to do work elswhere. The single thread does not go to sleep – it rushes off to do as much work as it can before being yanked back ‘awake’ to the place it briefly adjourned.

Without this asyncio.sleep(), if we wanted our program to keep working during a sleep, we would need more than 1 thread! As anyone who has tried it knows: multithreaded programming is neither easy nor simple. But, as we’ll see, getting work out your program while putting a thread ‘to sleep’ is a cinch with asyncio.

The upshot: we’re going to write code with infinite loops, 1 thread, and high-performance.

Metronome spec time

We are going to make a metronome. A metronome is used typically in music to synchronize a performance to a regular tempo. Traditionally, a tempo is defined in terms of beats-per-minute. It terms of code, our metronome is going to beat at regular intervals, and at each beat it will execute a collection of callbacks.

All our metronome needs is (1) the beats-per-minute tempo, and (2) a list of callbacks. We can take a first pass at the code

import asyncio
from typing import Callable, Any, List

CallBack = Callable[None, Any]
CallBacks = List[Callback]

class Metronome:
  def __init__(tempo: int, callbacks: CallBacks):
    self.tempo = tempo
    self.callbacks = callbacks

  async def start(self):
    while True:
      # tempo is beats-per-minute.  60bpm is a beat every 1 second
      sleep_time = 60 / self.tempo
      asyncio.sleep(sleep_time)

      for callback in self.callbacks:
        callback()

This is a good start! It’s a minimum viable metronome, for sure. We instantiate the class with 2 arguments: a tempo and a list of callbacks. The tempo is used for determining the sleep time, and at each beat in the loop, our metronome executes all the callbacks.

How can we improve it? Let’s make 2 changes. The first will improve performance, and the second will improve the regularity of our sleep interval, so that the metronome does not drift much.

1. Async everywhere

I want to use my metronome for sending commands to things in the outside world. Maybe my code will send MIDI messages, or control an array of LEDs. This IO-bound work takes time – time which can block our main thread. This sort of IO-focused processing is what Python’s asyncio library is made for! If we ensure that our callbacks are all async subroutines, then we can mitigate the impact of IO on our code’s performance. This is just a simple change to our metronome’s API & implementation.

Now with all async callbacks:

import asyncio
from typing import Callable, Any, List, Awaitable

CallBack = Callable[None, Awaitable[Any]]
CallBacks = List[Callback]

class Metronome:
  def __init__(tempo: int, callbacks: CallBacks):
    self.tempo = tempo
    self.callbacks = callbacks

  async def start(self):
    while True:
      # tempo is beats-per-minute.  60bpm is a beat every 1 second
      sleep_time = 60 / self.tempo
      asyncio.sleep(sleep_time)

      for callback in self.callbacks:
        await callback()

It’s a simple change with big consequences. It ensures that our callbacks will not block our thread while waiting on IO to complete.

2. Prevent drift over repeated intervals

Making our callbacks asynchronous does not make them run faster. The IO operations will still take time. This time will create a drift in our metronome intervals, where one beat might take longer than another owing to how long it needed to wait for its callbacks to complete. We need to keep track of the time it takes to run a beat’s worth of callbacks to completion. This is a simple enough feature to implement:

import asyncio
import time
from typing import Callable, Any, List, Awaitable

CallBack = Callable[None, Awaitable[Any]]
CallBacks = List[Callback]

class Metronome:
  def __init__(tempo: int, callbacks: CallBacks):
    self.tempo = tempo
    self.callbacks = callbacks

  async def start(self):
    offset = 0

    while True:
      sleep_time = max(0, (60 / self.bpm - offset))
      asyncio.sleep(sleep_time)

      # start timer
      t0 = time.time()

      for callback in self.callbacks:
        await callback()

      t1 = time.time()

      # calculate offset
      offset = t1 - t0

Now we are timing our callbacks, and reduce our sleep time accordingly. This won’t perfectly eliminate drift, but the minor variations between beats should be well within what a human ear can detect.

Example code

Under the hood, our metronome is an infinite loop. We can run this infinite loop in the midst of a single-threaded program because we utilize asyncio.sleep() in our metronome’s loop, which frees up our thread to do other work while it is wating for the next beat.

The .start() method of the metronome returns a coroutine. This coroutine needs to be passed off to an asyncio event loop before it will run. Here is a simple example usage:

import asyncio
from metronome import Metronome

# define a callback
async def do_beat():
  print('BEAT')

# collect all the metronome's callbacks
cbs = [do_beat]

# set tempo to 120 beats-per-minute
tempo = 120

# create the metronome
metronome = Metronome(tempo=tempo, callbacks=cbs)

# produce the coroutine
metronome_coro = metronome.start()

# asyncio boilerplate to run the coroutine
loop = asyncio.get_event_loop()
loop.run_until_complete(metronome_coro)

comments powered by Disqus