Printing pressed keys every second with asynchronous programming

Question

Since examples of asynchronous programming are still fairly sparse, I am trying to come up with my own using the new async def. I figured the most basic example I could make would be an event loop which listens to the user inputs and prints them back.

I want an event loop which listens to the keys the user presses and prints them every second. I'm trying to do so by having one task that prints from a queue of keys and a second task that listens to pressed keys and adds them to the queue.

I am missing a good way to asynchronously listen to key strokes. Here is what I have for now.

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

Expected output would look like this

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

I doubt there is an awaitable coroutine such as asyncio.await_pressed_key, how would we proceed to create one?

Any other approach is welcome as well as my goal is not to make this specific example work, but rather to generate meaningful asynchronous programmign examples.

Answer 1

Assuming that you want to read characters from a TTY (terminal device on Unix-like system), accessing characters as they are typed requires:

• the TTY to be in "raw mode", so that the system returns characters without waiting for a newline to be typed by the user;

• the standard input, which are blocking, to be wrapped into an asyncio non-blocking stream.

The former is provided by the tty module that comes with the Python standard library, and the latter by the aioconsole third-party library.

With those two, your example could look like this:

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

This example will only work on Unix-like systems.