Ultimo: an interface framework based on async iterators

[corranwebster]

Over the last few weeks I've put together Ultimo a Micropython framework for uasyncio-based interface code built around tasks that consume async iterators.

For example, you can write code like the following to print temperature and time asynchronously on a Raspberry Pi Pico as follows:

import asyncio
from machine import ADC

from ultimo.pipelines import Dedup
from ultimo_machine.gpio import PollADC
from ultimo_machine.time import PollRTC

async def temperature():
    async for value in PollADC(ADC.CORE_TEMP, 10.0):
        t = 27 - (3.3 * value / 0xFFFF - 0.706) / 0.001721
        print(t)

async def clock():
    async for current_time in Dedup(PollRTC(0.1)):
        print(current_time)

async def main():
    temperature_task = asyncio.create_task(temperature())
    clock_task = asyncio.create_task(clock())
    await asyncio.gather(temperature_task, clock_task)

if __name__ == '__main__':
    asyncio.run(main())

Ultimo also provides convenience decorators and a syntax for building data-flow pipelines from basic building blocks using the bitwise-or (or "pipe" operator):

@pipe
def denoise(value):
    """Denoise uint16 values to 6 significant bits."""
    return value & 0xfc00

async def main():
    led_brightness = PollADC(26, 0.1) | denoise() | Dedup() | PWMSink(25)
    await asyncio.gather(led_brightness.create_task())

Because Ultimo is built on top of the standard asyncio.Task infrastructure it should interoperate well with standard asyncio code.

Hopefully other people will find this useful.

Github: https://github.com/unital/ultimo
Documentation: https://unital.github.io/ultimo/
Tutorial: https://unital.github.io/ultimo/user_guide/tutorial.html
Examples: https://unital.github.io/ultimo/user_guide/examples.html

Disclaimers:

• I have a personal project that I am planning to use Ultimo for, I haven't built anything really big with it yet.
• I'm a long-time Pythonista, but this is my first substantial Micropython work, so I'd appreciate any feedback on best-practices, how to make things more efficient, cross-device support, etc.

[bidrohini]

Certain features in MicroPython's uasyncio may not fully align with CPython's asyncio. For example, versions of MicroPython earlier than 1.17 have some differences in how uasyncio behaves (e.g., memory allocation in tasks), so it would be worth testing Ultimo on different MicroPython versions

[corranwebster]

Thanks - I know there are differences because I really wanted to have access to PEP 525-style async generators and I know that you don't need to be as paranoid about Tasks being garbage collected out from under you - but yes, the version of Micropython has to support at least PEP 492-style async for.

Probably the best way to handle this is to add some weekly tests against older versions of Micropython.

[peterhinch]

Is this line of code (above) correct?

    await asyncio.gather(led_brightness.create_task())

shouldn't it read

    await asyncio.gather(asyncio.create_task(led_brightness))

?

Re old firmware versions I think it's valid to state "This library requires firmware version or later". Maybe check the version in code and throw if necessary,

[corranwebster]

No, the code is correct: there is a convenience method create_task() which is equivalent to asyncio.create_task(led_brightness.run()). It would be really slick if the result of that expression was an awaitable but a simple convenience method seems to work well.

I've now got automated regression tests for 1.20 onwards, which seems reasonable backwards compatibility. I'm using https://github.com/BrianPugh/install-micropython in my GitHub Actions and it is unhappy installing versions earlier than that.

BTW, thanks for your async on Micropython docs-they were helpful for working out what the differences were between regular Python asyncio and Micropython.