Listen to keypress with asyncio

Question

Can somebody provide a sample of code which listen to keypress in nonblocking manner with asynio and put the keycode in console on every click?

It's not a question about some graphical toolkit

Answer 1

So the link provided by Andrea Corbellini is a clever and thorough solution to the problem, but also quite complicated. If all you want to do is prompt your user to enter some input (or simulate raw_input), I prefer to use the much simpler solution:

import sys
import functools
import asyncio as aio

class Prompt:
    def __init__(self, loop=None):
        self.loop = loop or aio.get_event_loop()
        self.q = aio.Queue(loop=self.loop)
        self.loop.add_reader(sys.stdin, self.got_input)

    def got_input(self):
        aio.ensure_future(self.q.put(sys.stdin.readline()), loop=self.loop)

    async def __call__(self, msg, end='\n', flush=False):
        print(msg, end=end, flush=flush)
        return (await self.q.get()).rstrip('\n')

prompt = Prompt()
raw_input = functools.partial(prompt, end='', flush=True)

async def main():
    # wait for user to press enter
    await prompt("press enter to continue")

    # simulate raw_input
    print(await raw_input('enter something:'))

loop = aio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Answer 2

I wrote something similar as part of a package called aioconsole.

It provides a coroutine called get_standard_streams that returns two asyncio streams corresponding to stdin and stdout.

Here's an example:

import asyncio
import aioconsole

async def echo():
    stdin, stdout = await aioconsole.get_standard_streams()
    async for line in stdin:
        stdout.write(line)

loop = asyncio.get_event_loop()
loop.run_until_complete(echo())

It also includes an asynchronous equivalent to input:

something = await aioconsole.ainput('Entrer something: ') 

It should work for both file and non-file streams. See the implementation here.

Answer 3

An alternative to using queues would be to make the command line an asyn generator, and process the commands as they come in, like so:

import asyncio
import sys

class UserInterface(object):

def __init__(self, task, loop):
    self.task = task
    self.loop = loop

    def get_ui(self):
        return asyncio.ensure_future(self._ui_task())

    async def _ui_cmd(self):
        while True:
            cmd = sys.stdin.readline()
            cmd = cmd.strip()
            if cmd == 'exit':
                self.loop.stop()
                return
            yield cmd

    async def _ui_task(self):
        async for cmd in self._ui_cmd():
            if cmd == 'stop_t':
                self.task.stop()
            elif cmd == 'start_t':
                self.task.start()