Multitasking: “Fucking up multiple things at the same time!” – MicroPython uasyncio howto

Most code for simple IoT devices is a script with an endless loop with the code for the connected hardware like reading a sensor or polling something from the network followed by a delay. Keeping that pattern in mind helps to write new or convert your existing code for asynchronous execution.

If you want to combine multiple tasks your device should do but independent of each other (e.g. different delays) the most efficient way is to use uasyncio. There’s also code example on the linked page, but for the purpose in this case I am not using tasks but gather and console printed instead of blinking.

import uasyncio

async def print_async(task_id, period_ms):
    while True:
        print(task_id)
        await uasyncio.sleep_ms(period_ms)

async def main(task_id_1, task_id_2):
    await uasyncio.gather(print_async(task_id_1, 700), print_async(task_id_2, 400))

uasyncio.run(main('one', 'two'))

This defines the function print_async which console prints a string and waits for the given milliseconds. The uasyncio.sleep funcions do not block like the time.sleep, but signal asyncio that in the meantime something different can be done. Perfect for the given problem!

If you don’t include a call to uasyncio’s sleep method in your while-true-loop, it will be stuck in it and nothing else will be executed. If you want to include a loop without delay call asyncio.sleep(0) and asyncio will do its best to execute it continiously without delay while still performing the other given tasks.

Functions with the keyword async always return a coroutine. A coroutine is a task which can be executed asynchronously (seemingly parallel) with other tasks. You can wait for a coroutine execution to finish by putting await in front of it. In our case we want to run them independantly at the same time but they mustn’t finish. The loops in print_async already runs forever and to run them both simultaneously we use uasyncio.gather.
It takes the coroutines of the function calls and waits for them to finish. This is mostly used to “gather” different requests which can be done simultaneously as they do not depend on each other. In our case the coroutines will run forever – just what we want.

The only thing you have to do is replace print_async with the real code for the IoT device similar to the code below.

import uasyncio
import dht
from machine import Pin

sensor = dht.DHT22(Pin(5))
measure_delay_s = 30
led = Pin(2, Pin.OUT)
blink_alert = False

async def measure(delay_s:int) -> None:
    while True:
        try:
            sensor.measure()
            temp = sensor.temperature()
            blink_alert = 19 <= temp < 27
            print(f'Temp: {temp} Hum: {sensor.humidity()}')
        except Exception as e:
            print(f"An error occurred - continuing. \n{e}")
        await uasyncio.sleep(delay_s)

async def blink_alert() -> None:
    while True:
        try:
            if blink_alert:
                led.on()
                uasyncio.sleep_ms(500)
                led.off()
        except Exception as e:
            print(f"An error occurred - continuing. \n{e}")
        await uasyncio.sleep(10)

async def main(delay_s:int) -> None:
    await uasyncio.gather(measure(30), blink_alert())

uasyncio.run(main(measure_delay_s))

You can do much more with uasyncio like handling async tasks and synchronizing with events, but that’s not goal of this post. I hope this was helpful for understanding the first steps you need to do when using asyncio and enabling your IoT device to do more than one loop.

Official asyncio MicroPython doc

Great Youtube videos about this topic: ArjanCodes about asyncio and TechWithTim about asyncio
Keep in mind they are about Python, not MicroPython, which is a subset. Most of it is transferable.