Asyncio and while loop

[PatLub85]

I would like to order a robot using asyncio.
For the tests I made a minimal program which does not do what I want!

import uasyncio as asyncio 
import time

async def forward():
    print ("forward")
    await asyncio.sleep_ms(10)

async def backwards():
    print ("backwards")
    await asyncio.sleep_ms(10)
  
async def hcsr():
    while True:
        print('Hcsr04')
        await asyncio.sleep_ms(100)

async def main(direction):

    if direction=="f":
        task=asyncio.create_task(hcsr())
        asyncio.create_task(forward())
        await asyncio.sleep(0.1)
    
    elif direction=="b":
        asyncio.create_task(backwards())
        await asyncio.sleep(0.1)

while True:
    direction=input("forward or backwards ? f or b  ")
    asyncio.run(main(direction))

To summarize, if I move forward, I would like the forward function to execute once and the hscr function to run in a loop, to test for the presence of an obstacle.
In my program, each function executes once.
And secondly when I order backwards, I would like the hscr function to be stopped. If the robot goes backwards and the probe tests forward, it's useless.
Do you have any idea on the feasibility of these two features?
Thanks in advance

[peterhinch]

You are right to use asyncio and what you want to do is feasible. I suggest you look at the tutorial to get more of a handle on how to write asyncio applications.

In particular asyncio.run() should not be called in a loop, because in typical firmware applications the function never returns. Your input routine should be another coroutine, running continuously. Note that tasks can be cancelled. Hopefully the tutorial will clarify all this.

[PatLub85]

Hi Peter,
I read your tutorial several times, and I understood the principle. As you advised me, asyncio.run() is no longer in a loop.
If f is chosen forward and hcsr are queued until the next await.
Then forward is executed, then hcsr for second, duration of await.
How can I make this hcsr loop continue until a key is pressed? Should we use an interuption? I'm a little in the fog!

import uasyncio as asyncio 
import time

async def forward():
    print ("forward")
    await asyncio.sleep_ms(10)
  
async def hcsr():
    while True:
        print('Hcsr04')
        await asyncio.sleep_ms(100)

async def main():
    while True:
        direction=input("forward or backwards ? f or b  ")
        if direction=="f":
            asyncio.create_task(forward())
            task=asyncio.create_task(hcsr())
            await asyncio.sleep(1)

asyncio.run(main())

[sosi-deadeye]

You can't. The input function is a synchronous blocking function. It will block the whole event loop in the same thread.
Code with explanation:

import uasyncio as asyncio


async def forward():
    print("forward")
    await asyncio.sleep_ms(1000)


async def backwards():
    print("backwards")
    await asyncio.sleep_ms(1000)


async def hcsr():
    # if the eventloop is blocked by a
    # sychronous function, then this while-true loop blocks
    # until the sychronous function is done

    while True:
        print("Hcsr04")
        await asyncio.sleep_ms(100)


async def main(direction):
    # not awaiting for hcsr
    # because it never finishes
    # and it will block if you await for it

    asyncio.create_task(hcsr())
    # now the task runs concurently in the eventloop

    while True:
        if direction == "f":
            # awaiting forward()
            await asyncio.create_task(forward())
            await asyncio.sleep(0.1)

        elif direction == "b":
            # awaiting backwards
            await asyncio.create_task(backwards())
            await asyncio.sleep(0.1)

        # input() here is a no no, because it blocks the whole eventloop
        # an async input function is required to do this right
        # but for testing it's ok

        # you'll see also, that hcsr is blocked, until
        # the user enters a direction
        direction = input("forward or backwards ? f or b  ")


if __name__ == "__main__":
    asyncio.run(main("f"))

If you find an async input function, you could use this. I don't know if someone made a library for it. There is an aiorepl, but it does not include an async input function.

If you're running this code on a Microcontroller, then just use two real buttons to test the functions. The access to gpios is very fast, and you won't recognize anything blocking, expect you program it to wait.

@peterhinch also has good async functions for buttons and other hardware related stuff, which should run async.

BTW: I was watching this video again: Asyncio in (Micro)Python
It's old, but gives good hints what to use.

[PatLub85]

Thank you for your feedback. I made this little program, thinking it was simpler to explain, but as you say the input function is a synchronous blocking function.
My real program is made with Microdot which is natively asynchronous and I will open another discussion
THANKS