Asynchronous Countdowns in Python

Question

using: python 3.8 on windows 10

I am working on a script that runs in a while loop. At the end of the loop I want it to wait for 5 seconds for the user to give input and then restart, or exit if they do not give input.

I've never actually used it before but I assumed asyncio would be useful because it allows for the definition of awaitables. However bringing things together is proving more difficult than I'd anticipated.

import keyboard, asyncio, time as t



async def get():
    keyboard.record('enter', True)
    return True

async def countdown(time):
    while time > 0:
        print(f'This program will exit in {int(time)} seconds. Press enter to start over.', end='\r')
        await asyncio.sleep(1)
        # t.sleep(1)
        time -= 1
    print(f'This program will exit in 0 seconds. Press enter to start over.', end='\r')
    return False


async def main():
    running = True
    while running:
        # other code
        
        
        clock = asyncio.create_task(countdown(5))
        check = asyncio.create_task(get())
        
        done, pending = await asyncio.wait({clock, check}, return_when=asyncio.FIRST_COMPLETED)
        running = next(iter(done)).result()
        print(running, end='\r')
    print('bye')
        

asyncio.run(main())

As it stands, the process doesn't end if I wait for five seconds. It also doesn't visibly count down (my best, and most ridiculous, guess is that it may be looping too fast in main? Try holding down the "enter" key - with and without printing running). Also, when I switch to time.sleep, the display works fine, but it doesn't seem as though the countdown function ever returns.

Previously I'd tried using an input statement instead of keyboard.record; that blocks. I had also tried using asyncio.wait_for; but the timeout never came, though again, the "enter" key was registered (that said, it wouldn't have printed the countdown even if it had worked). I also tried asyncio.as_completed but I was unable to parse anything useful from the iterable. Happy to be wrong there though!

Also, bonus points if you can generalize the countdown to non-integer time spans <3

---

The wait_for approach:

async def main():
    running = True
    while running:
        
        try:
            running = await asyncio.wait_for(get(), 5)
        except asyncio.TimeoutError:
            running = False
        print(running)
    print('bye')


asyncio.run(main())

---

as_completed

async def main():
    running = True
    while running:
        ## other code
        clock = asyncio.create_task(countdown(5))
        check = asyncio.create_task(get())
        for f in asyncio.as_completed({check, clock}):
            # print(f.cr_frame)
            # print(f.cr_await, f.cr_running, f.cr_origin, f.cr_frame, f.cr_code)
            print(f.cr_await, f.cr_running, f.cr_origin)
            
    print('bye')

---

Also, bonus points if you can generalize the countdown to non-integer time :P

cheers!

Answer 1

If you look in the docs for keyboard.record it says:

Note: this is a blocking function.

This is why your process didn't end after 5 seconds. It was blocking at keyboard.record('enter', True). If you are going to stick with the keyboard module, what you need to do is create a hook on the 'enter' key. I put a quick demo together with your code:

import asyncio
import keyboard


class Program:
    def __init__(self):
        self.enter_pressed = asyncio.Event()
        self._loop = asyncio.get_event_loop()

    async def get(self):
        await self.enter_pressed.wait()
        return True

    @staticmethod
    async def countdown(time):
        while time > 0:
            print(f'This program will exit in {int(time)} seconds. Press enter to start over.')
            await asyncio.sleep(1)
            # t.sleep(1)
            time -= 1
        print(f'This program will exit in 0 seconds. Press enter to start over.')
        return False

    def notify_enter(self, *args, **kwargs):
        self._loop.call_soon_threadsafe(self.enter_pressed.set)


    async def main(self):
        running = True
        while running:
            # other code

            keyboard.on_press_key('enter', self.notify_enter)
            self.enter_pressed.clear()
            clock = asyncio.create_task(self.countdown(5))
            check = asyncio.create_task(self.get())

            done, pending = await asyncio.wait({clock, check}, return_when=asyncio.FIRST_COMPLETED)
            keyboard.unhook('enter')
            for task in pending:
                task.cancel()

            running = next(iter(done)).result()
            print(running)

        print('bye')


async def main():
    program = Program()
    await program.main()

asyncio.run(main())

There is callback created notify_enter, that sets an asyncio.Event whenever it's fired. The get() task waits for this event to trigger before it exits. Since I didn't know what your other code is doing, we don't bind a hook to the enter key's key_down event until right before you await both tasks and we unbind it right after one of the tasks completes. I wrapped everything up in a class so the event is accessible in the callback, since there isn't a way to pass parameters in.