Печать нажатых клавиш каждую секунду с асинхронным программированием - PullRequest
Новая
       18

Печать нажатых клавиш каждую секунду с асинхронным программированием

0 голосов
/ 26 августа 2018

Поскольку примеры асинхронного программирования все еще довольно редки, я пытаюсь придумать свое собственное использование нового async def. Я подумал, что самым простым примером, который я мог бы сделать, был бы цикл обработки событий, который прослушивает вводимые пользователем данные и выводит их обратно.

Я хочу цикл обработки событий, который прослушивает клавиши, которые нажимает пользователь, и печатает их каждую секунду. Я пытаюсь сделать это, имея одну задачу, которая печатает из очереди клавиш, и вторую задачу, которая прослушивает нажатые клавиши и добавляет их в очередь.

Мне не хватает хорошего способа асинхронного прослушивания нажатий клавиш. Вот что у меня сейчас. 

import asyncio

KEY_QUEUE = []

async def printer():
    while True:
        await asyncio.sleep(1)
        print('In the last second you pressed:', *KEY_QUEUE)
        KEY_QUEUE.clear()

async def listener():
    while True:
        ... # await a key to be pressed and add it to KEY_QUEUE

loop = asyncio.get_event_loop()

loop.create_task(printer())
loop.create_task(listener())

loop.run_forever()

Ожидаемый результат будет выглядеть так 

In the last second you pressed: h e l l o
In the last second you pressed: w o r
In the last second you pressed: l d 
In the last second you pressed:
In the last second you pressed: i t    w o
In the last second you pressed: r k s

Я сомневаюсь, что есть приемлемая сопрограмма, такая как asyncio.await_pressed_key, как бы мы ее создали?

Любой другой подход приветствуется, так как моя цель - не заставить этот конкретный пример работать, а скорее генерировать значимые асинхронные примеры программирования.

1 Ответ

0 голосов
/ 26 августа 2018

Предполагая, что вы хотите читать символы из TTY (терминальное устройство в Unix-подобной системе), для доступа к символам по мере их ввода требуется:

  • TTY, чтобы быть в «raw»mode ", чтобы система возвращала символы без ожидания ввода новой строки пользователем;

  • стандартный ввод, который является блокирующим, для преобразования в асинхронное неблокированиеstream.

Первый предоставляется модулем tty, который поставляется со стандартной библиотекой Python, а второй - aioconsole сторонняя библиотека.

С этими двумя ваш пример может выглядеть следующим образом: 

import asyncio, aioconsole, sys, tty

async def main():
    loop = asyncio.get_event_loop()
    typed = []
    p = loop.create_task(printer(typed))
    await listener(typed)
    p.cancel()

async def listener(typed):
    tty.setraw(sys.stdin.fileno())
    stdin, _ = await aioconsole.stream.get_standard_streams()
    while True:
        ch = await stdin.read(1)
        if ch == b'\x03':  # ctrl-c
            break
        typed.append(ch)
    tty.setcbreak(sys.stdin.fileno())

async def printer(typed):
    while True:
        await asyncio.sleep(1)
        print('In the last second you pressed', typed, end='\r\n')
        del typed[:]

asyncio.get_event_loop().run_until_complete(main())

Этот пример будет работать только на Unix-подобных системах.

...