Блог

использование: python 3.8 в 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 ? Попробуйте удерживать нажатой клавишу «enter» — с печатью и без печати running ).
Кроме того, когда я переключаюсь time.sleep , отображение работает нормально, но не похоже, что countdown функция когда-либо возвращается.

Ранее я пытался использовать input оператор вместо keyboard.record ; который блокирует. Я также пытался использовать asyncio.wait_for ; но тайм-аут так и не наступил, хотя, опять же, была зарегистрирована клавиша «enter» (тем не менее, он не напечатал бы обратный отсчет, даже если бы он работал). Я также пытался 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

приветствия!

Если вы посмотрите в документации для keyboard.record, там написано:

Примечание: это блокирующая функция.

Вот почему ваш процесс не завершился через 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 клавиши enter до тех пор, пока вы не дождетесь выполнения обеих задач, и мы не отвяжем его сразу после завершения одной из задач. Я завернул все в класс, чтобы событие было доступно при обратном вызове, поскольку нет способа передать параметры.