CircuitPython_async

About

Pure Python cooperative multitasking implementation for the async/await language syntax.

Loosely modeled after CPython's standard asyncio; focused on CircuitPython.

Typically, when you need to wait around for something you have to choose between just doing time.sleep() and having a hitch in your app OR manually interleaving tasks and tracking their state & timers.

asynccp interleaves your tasks at await points in the same general way as asyncio does on regular python. Instead of blocking with time.sleep() you'll await asynccp.delay() to let the microcontroller work on other things.

The async and await keywords are supported in Circuitpython 6.0. They may be unavailable on your m0 microcontroller because of flash space.

Examples

Plain synchronous loop task with async support

importasynccpasyncdefread_some_sensor(self):
passasyncdefcheck_button(self):
passasyncdefupdate_display(self):
passasyncdefloop():
awaitread_some_sensor()
awaitcheck_button()
awaitupdate_display()
defrun():
asynccp.add_task(loop())
asynccp.run()
if__name__=='__main__':
run()

Scheduled app instead of loop

importasynccpimportasynccp.time.DurationasDurationclassApp:
def__init__(self):
self.button_state=0self.sensor_state=0asyncdefread_some_sensor(self):
passasyncdefcheck_button(self):
passasyncdefupdate_display(self):
passdefrun():
app=App()
asynccp.schedule(frequency=Duration.of_seconds(5), coroutine_function=app.read_some_sensor)
asynccp.schedule(frequency=80, coroutine_function=app.check_button)
asynccp.schedule(frequency=15, coroutine_function=app.update_display)
asynccp.run()
if__name__=='__main__':
run()

Multiplex SPI bus without manual coordination

Using asynccp.managed_resource.ManagedResource you can share an SPI bus between concurrent tasks without explicit coordination.

defsetup_spi():
fromasynccp.managed_resourceimportManagedResourceimportdigitalioimportboard# Configure the hardwarespi=board.SPI()
sensor_cs=digitalio.DigitalInOut(board.D4)
sensor_cs.direction=digitalio.Direction.OUTPUTsdcard_cs=digitalio.DigitalInOut(board.D5)
sdcard_cs.direction=digitalio.Direction.OUTPUT# Set up acquire/release workflow for the SPI busdefset_active(pin):
pin.value=Truedefset_inactive(pin):
pin.value=False# Configure the physical spi as a managed resource with callbacks that manage the CS pinmanaged_spi=ManagedResource(spi, on_acquire=set_active, on_release=set_inactive)
# Get awaitable handles for each CS using this SPI bussensor_handle=managed_spi.handle(pin=sensor_cs)
sdcard_handle=managed_spi.handle(pin=sdcard_cs)
returnsensor_handle, sdcard_handle

And with these configured resource handles you can use them without checking whether anything is busy. Things will efficiently wait when they have to, and charge right on through when there's nothing using the bus currently.

asyncdefread_sensor(sensor_handle):
asyncwithsensor_handleasbus:
awaitsend_read_request_to_sensor(bus)
# Consider a BME680 which needs a delay before reading the requested result.# Let's let something else use the bus while it's waitingawaitasynccp.delay(seconds=0.1)
asyncwithsensor_handleasbus:
returnawaitread_result_from_sensor(bus)
asyncdeflog_to_sdcard(sdcard_handle):
asyncwithsdcard_handleasbus:
bytes_written=awaitwrite_to_sdcard(bus)
sensor_handle, sdcard_handle=setup_spi()
asynccp.schedule(Duration.of_milliseconds(123), read_sensor, sensor_handle)
sd_log_scheduled_task=asynccp.schedule(Duration.of_seconds(1.5), log_to_sdcard, sdcard_handle)
asynccp.run()

Some toy example code

Uses this library for the rotary button

importasynccpfromcpy_rotaryimportRotaryButton# Some state. Global state is not super cool but whatevsreading_sensor=False# Define the top-level workflows (you would have to write this stuff no matter what)asyncdefread_sensor():
globalreading_sensorreading_sensor=Truetry:
i2c.send(payload)
awaitasynccp.delay(1) # Don't block your loading beach ball while the sensor is sensing.i2c.read(payload) # if you have some buffered i2c thingfinally:
reading_sensor=Falseasyncdefanimate_beach_ball():
globalreading_sensorifreading_sensor:
set_animation_state() # hopefully this is quick - if not, maybe there's something inside to `await`asyncdefread_from_3d_printer():
passrotary=RotaryButton()
# ---------- asynccp wiring begins here ---------- ## Schedule the workflows at whatever frequency makes senseasynccp.schedule(Duration.of_milliseconds(100), coroutine_function=read_sensor)
asynccp.schedule(Duration.of_milliseconds(100), coroutine_function=animate_beach_ball)
asynccp.schedule(Duration.of_milliseconds(200), corouting_function=read_from_3d_printer)
asynccp.schedule(Duration.of_milliseconds(10), coroutine_function=rotary.loop)
# And let asynccp do while Trueasynccp.run()
# ---------- asynccp wiring ends here ---------- #