Proper Asyncio programming construct

[neliel123]

I am trying to understand and make use of Asyncio in my MicroPython project and I have been reading the awesome tutorial by Peter Hinch about this subject to the best of my ability. I have not used Asyncio so I am just learning through various available resources.

So I have an RFID reader with an LCD and some other buttons. This is my first attempt at making my own program that will run in a loop.

from mfrc522 import MFRC522
import utime
import asyncio
from lcd_i2c import LCD


class Display:
    def __init__(self, i2c, lcd):
        self.i2c = i2c
        self.lcd = lcd
        self.lcd.begin()
        
    def display(self, message):
        print(message)
        
class RFIDReader:
    def __init__(self, reader, lcd_display):
        self.reader = reader
        self.task = asyncio.create_task(self.scan_cards())
        self.lcd_display = lcd_display
        
        
    async def scan_cards(self):
        while True:
            print("Waiting for cards")
            await asyncio.sleep_ms(100)
            self.lcd_display("Sample Message")

async def main():
    # Configure OLED first
    I2C_ADDR = 0x27     # DEC 39, HEX 0x27
    NUM_ROWS = 2
    NUM_COLS = 16
    i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=10000)
    lcd = LCD(addr=I2C_ADDR, cols=NUM_COLS, rows=NUM_ROWS, i2c=i2c)
    
    lcd_display = Display(i2c, lcd)
    
    rfid_card_reader = MFRC522(spi_id=0,sck=2,miso=4,mosi=3,cs=1,rst=0)
    rfid = RFIDReader(rfid_card_reader, lcd_display)
    
    
    await rfid.task



try:
    asyncio.run(main())
except KeyboardInterrupt:
    print('Interrupted')
finally:
    asyncio.new_event_loop()

My doubts are:

1. Is my current setup correct for an asyncio program that will run continuously?
2. When displaying something on LCD or OLED, do I still need to make my function asynchronous and add an await asyncio.sleep(1)?
3. If I add another sensor where I will continually poll it, does it mean that another await task for it in my main() function?

I am a newbie so please enlighten me, especially on the asyncio part. I am learning things as I do it so I would like some guidance from experts.
Thanks in advance.

[peterhinch]

The program structure looks OK. I think the line

self.lcd_display("Sample Message")

should read:

self.lcd_display.display("Sample Message")

Re 2. it depends on what the display function does. If it rapidly outputs a string on I2C there is no need for the function to be asynchronous because it will only block the scheduler for a short period (<1ms). There is only a purpose in making display asynchronous if it is going to take a long time. For example a display with a large frame buffer might take hundreds of ms to update: blocking for that period would upset other tasks so you might write an async function that looked like this:

async def display(self):
    for n in range(10):
        self.write(self.buf[1000*n: 1000*(n + 1)])  # Write out 1/10 of the buffer then
        asyncio.sleep_ms(0)   # Let other tasks run

Whether to make a function synchronous or asynchronous depends on what the code must do in the context of the application as a whole. As a rule of thumb if something blocks for < 1ms it's best being synchronous. If it blocks for > 10ms then it's worth considering the effect on other tasks. The reason is that the overhead in task switching is about 300μs on most platforms.

As for 3. it's entirely reasonable to start multiple asynchronous polling tasks in main().

Lastly you don't need to trap the keyboard interrupt. This is in some tutorial samples to show how this can be done should you wish.

[neliel123]

Hi Peter, thank you very much for helping out beginners like me. I especially like this nugget wisdom of yours below as I struggle to think when should I make it a synchronous or an asynchronous call

Whether to make a function synchronous or asynchronous depends on what the code must do in the context of the application as a whole. As a rule of thumb if something blocks for < 1ms it's best being synchronous. If it blocks for > 10ms then it's worth considering the effect on other tasks. The reason is that the overhead in task switching is about 300μs on most platforms.

I hope you will be patient even more as I have some additional doubts.

When I scanned something on my RFID reader and it detected a card I was hoping to validate it on my API server.

class RFIDReader:
    async def get_card_info(card_id):
        url = f"http://my-api-host:3000/cards/{card_id}"
        r = urequests.get(url)
        # MORE CODE

    async def scan_cards(self):
        while True:
            print("Waiting for cards")
            ## CARD DETECTED -> VALIDATE WITH API
            if stat == self.reader.OK:
                (stat, uid) = self.reader.SelectTagSN()
                card_details = await get_card_info(uid)  <<--------- IS THIS PROPER?

            await asyncio.sleep_ms(100)

My initial work is to add an async code that will use urequests library and gather the returned information.

I am second-guessing myself if this is the most python-asyncio way of doing this.
My doubts are:

1. Is my current code okay? I could not find a CPython aiohttp library in MicroPython so I have used urequests.
2. Or should I have created a new asyncio.task just for calling the API and calling it here? (Not sure if this is possible though)
3. The code for RFIDReader might get bloated as I have several other API calls within. The class RFIDReader is not just interfacing with my RFID but handles the API call as well. Not sure if there is a much better Python Asyncio way of doing these.

Hoping for your guidance. Thanks!

[peterhinch]

I have no experience of RFID so I'd appreciate some picture of the physical design of the system. Is the RFID reader a physical device attached to your MicroPython host? If so, how is it connected? What is the resource which you are accessing via HTTP? Is it on a LAN or on the internet? Please give me some picture of what the system aims to achieve.

Re your code, if card_info is supposed to be a class member it should read as follows:

class RFIDReader:
    async def get_card_info(self, card_id):
        url = f"http://my-api-host:3000/cards/{card_id}"
        r = urequests.get(url)
        # MORE CODE
        return card_details   # Need this line

    async def scan_cards(self):
        while True:
            print("Waiting for cards")
            ## CARD DETECTED -> VALIDATE WITH API
            if stat == self.reader.OK:
                (stat, uid) = self.reader.SelectTagSN()
                card_details = await self.get_card_info(uid)

            await asyncio.sleep_ms(100)

I'm not sure what is your query about the method call, but the method of retrieving card_detalls is fine.

Until I have some grasp of the physical design it's hard to comment in detail but r = urequests.get(url) is a blocking synchronous call which will cause other tasks to pause while it is in progress.

[neliel123]

Hi Peter,
So this is a school research work that I am working on wherein I have these setup

• A Raspberry Pi Pico W
• An RC522 RFID with Mifare cards
• OLED
• and a keypad (passcode)

The goal is to design a security system, written entirely in MicroPython and using Asyncio.

• Each student is assigned a unique Mifare card and he/she should swipe his/her RFID card.
• Each RFID card has written information that I could cross-check with the database running on my laptop.
• Using REST API, I could validate each RFID tag
• If successful then it should proceed to the next validation which is a unique passcode per each student by inputting on my keypad.
• I would also call a REST API to validate the inputted passcode,

Until I have some grasp of the physical design it's hard to comment in detail but r = urequests.get(url) is a blocking synchronous call which will cause other tasks to pause while it is in progress.

Yes, this is a blocking call and I could not find a library with an asynchronous HTTP call in MicroPython. I think aiohttp is not present on the MicroPython. Is there an alternative available for MicroPython?

Once I have figured this out (RFID REST API Call) I have to work on the Keypad thing and how to introduce that in this project.
I saw in your documentation that you have an asynchronous module for a switch and pushbutton. I am not sure if I could use that in this setup. Do you think that it is feasible?

Thank you for your interest in this research project of mine and looking forward to your reply.

[peterhinch]

Thank you for clarifying the design.

I don't know of an async HTTP library for MicroPython, but this is not something I've investigated in detail. However the blocking of get will be brief as the database is local. By the time you read the DB, the RFID read is complete. So, in the context of the application, brief blocking at that point in time may not matter.

Re keypads they normally are wired as a crosspoint array. To read them you assert each column line in turn and read out any rows that are active. It would be feasible to write an async driver, and it's something I've considered occasionally, but as far as I'm aware nobody has actually done it.

My switch and pushbutton drivers assume discrete switches wired either to Vcc or to Gnd. They would not work with a crosspoint array.

[neliel123]

Thanks, Peter for this tip. I actually have been thinking that calling this blocking call may not be that time-consuming so I will let it be as it is.

Re keypads they normally are wired as a crosspoint array. To read them you assert each column line in turn and read out any rows that are active. It would be feasible to write an async driver, and it's something I've considered occasionally, but as far as I'm aware nobody has actually done it.

Yes, I have not seen anything that is using asyncio for this device. I am only seeing the usual non-async way of reading this device.
Regarding this one, could I just wrap this checking of keypad button click in an async function iterate through all the buttons, and then just add await asyncio.sleep(0) so that this would not block everything?

I am thinking of doing something like this?

async def read_keypad():
    # Code to read keypad synchronously
    await asyncio.sleep(0)

Hoping we could have an async library to deal with these devices.

[peterhinch]

You could, but it wouldn't be optimal. A single scan of the keypad could be done synchronously because it would be fast. Pseudocode:

for each column:
    assert the column line
    current_rows[column] = read the rows

However the driver needs to do quite a bit more than that. It needs to compare the present current_rows array to the previous state. If an individual cell has been newly asserted (a keypress) a callback is run. It then updates the previous state and waits out any contact bounce before repeating. In that wait period it would yield to the scheduler.

Writing such a driver would be straightforward.

That is a very crude driver which might suffice in your application, but it does assume keys are pressed one at a time. Real keyboard drivers have n-key rollover which adds substantial complexity.

[peterhinch]

As a further comment, you can buy keypads with a decoder chip and an I2C interface. Such a unit would be easy to use if you don't want to write a driver for a keypad array.

[andrewleech]

There's an async http client library here: https://github.com/micropython/micropython-lib/tree/master/micropython/uaiohttpclient

[neliel123]

Hi,
Thanks for this tip.
I do have a question as I am having some trouble calling my API with this library

import uaiohttpclient as aiohttp
			

async def process_data(self, data):			
        response = None
        try:
#             response = urequests.get(config.BASE_URL)
            print(f"config.BASE_URL :: {config.BASE_URL}")
            response = await aiohttp.request("GET", config.BASE_URL)
            print("HTTP Response:", response.text)
        except Exception as e:
            print("An error occurred during HTTP request:", str(e))
        finally:
            if response:
                response.close()

I am getting An error occurred during HTTP request: -2 when using this library but switching it to urequests has no issue, I have looked at the example.py of the MicroPython lib though,
Not sure where to look at the "-2" error code is.

[peterhinch]

@neliel123 A wet afternoon got the better of me so I wrote this keypad driver. It handles key rollover (where a second key is pressed before the first is released). In principle it can handle any sized keypad array (because of Python's arbitrary precision arithmetic).

# keys.py
import uasyncio as asyncio
class keypad:
    def __init__(self, rowpins, colpins, callback):  # Tuples of pins. Rows are OUT, cols are IN
        self.rowpins = rowpins
        self.colpins = colpins
        self.nbuttons = len(rowpins) * len(colpins)
        self.callback = callback
        self.cur = 0
        self.prev = 0
        for opin in self.rowpins:  # Initialise output pins
            opin(0)
        asyncio.create_task(self.scan())

    async def scan(self):
        while True:
            await asyncio.sleep_ms(0)
            self.cur = 0
            for opin in self.rowpins:
                opin(1)  # Assert output
                for ipin in self.colpins:
                    self.cur <<= 1
                    self.cur |= ipin()
                opin(0)
            if self.cur == self.prev:  # No change
                continue  # No change
            pressed = self.cur & ~self.prev
            self.prev = self.cur
            if not pressed:  # Button release
                continue
            # A key was pressed
            for v in range(self.nbuttons):  # Find button index
                if pressed & 1:
                    break
                pressed >>= 1
            self.callback(v)
            await asyncio.sleep_ms(50)  # Wait out bounce

# test script. Pins based on Rpi Pico.
from machine import Pin
rowpins = [Pin(p, Pin.OUT) for p in range(10, 14)]
colpins = [Pin(p, Pin.IN, Pin.PULL_DOWN) for p in range(16, 20)]
def cb(v):
    print(v)  # Print scan code.

async def main():
    kp = keypad(rowpins, colpins, cb)
    await asyncio.sleep(60)

asyncio.run(main())

It is rather crude, running a callback on a keypress but you might want to experiment. I plan to improve the API replacing the callback with a design based on an async iterator, which I'll post in my async library.

Comments welcome.

[EDIT]
Updated keyboard driver with code simplified and an improved API here

[neliel123]

Wow! Thanks, Peter! I was looking at the code and I have read this RingbufQueue somewhere in your documentation but I am not sure when and how to use it. I will be taking this for a test drive.

[peterhinch]

For the benefit of anyone wanting to use this driver the following demonstrates its use. Pin definitions are for a Pico/Pico W:

import asyncio
from primitives import Keyboard
from machine import Pin
rowpins = [Pin(p, Pin.OPEN_DRAIN) for p in range(10, 14)]
colpins = [Pin(p, Pin.IN, Pin.PULL_UP) for p in range(16, 20)]

async def main():
    kp = Keyboard(rowpins, colpins)
    async for scan_code in kp:
        print(scan_code)
        if not scan_code:
            break  # Quit on key with code 0

asyncio.run(main())