Python 中的异步倒计时

Question

使用：python 3.8 on windows 10

我正在编写一个在 while 循环中运行的脚本。在循环结束时，我希望它等待 5 秒让用户输入然后重新启动，或者如果他们不输入则退出。

我以前从未真正使用过它，但我认为 asyncio 会很有用，因为它允许定义可等待对象。然而事实证明，将事情整合在一起比我预期的要困难。

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())

就目前而言，如果我等待五秒钟，该过程不会结束。它也没有明显倒计时（我最好的，也是最荒谬的，猜测是它可能在 main 中循环太快了？尝试按住“输入”键- 有和没有打印 running)。此外，当我切换到 time.sleep 时，显示效果很好，但 countdown 功能似乎从来没有 returns.

以前我曾尝试使用 input 语句而不是 keyboard.record；那块。我也尝试过使用 asyncio.wait_for;但是超时从未到来，尽管再次注册了“输入”键（也就是说，即使它有效也不会打印倒计时）。我也试过 asyncio.as_completed 但我无法从可迭代对象中解析出任何有用的东西。不过很高兴在那里错了！

此外，如果您可以将倒计时概括为非整数时间跨度 <3

，则可加分

wait_for 方法：

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')

此外，如果您可以将倒计时概括为非整数时间，则可加分：P

干杯！

Answer 1

如果您查看 docs for keyboard.record，它会显示：

Note: this is a blocking function.

这就是您的进程在 5 秒后没有结束的原因。它在 keyboard.record('enter', True) 阻塞。如果你打算坚持使用键盘模块，你需要做的是在 'enter' 键上创建一个钩子。我将一个快速演示与您的代码放在一起：

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())

已创建回调 notify_enter，它会在触发时设置 asyncio.Event。 get() 任务在退出之前等待此事件触发。由于我不知道您的其他代码在做什么，因此直到您等待这两个任务之前，我们才将挂钩绑定到回车键的 key_down 事件，并且我们会在其中一个任务完成后立即解除绑定。我将所有内容都包装在 class 中，因此事件可以在回调中访问，因为没有办法传递参数。

Python 中的异步倒计时

Asynchronous Countdowns in Python

python

terminal

python-asyncio

countdown

干杯！