Escuchar keypress con asyncio

¿Alguien puede proporcionar una muestra de código que escuche las pulsaciones de teclas sin locking con asynio y coloque el código clave en la consola con cada clic?

No es una pregunta sobre algún conjunto de herramientas gráficas.

Así que el enlace proporcionado por Andrea Corbellini es una solución inteligente y completa para el problema, pero también es bastante complicado. Si todo lo que desea hacer es pedirle a su usuario que ingrese algo de entrada (o simular raw_input), prefiero usar la solución mucho más simple:

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

Una alternativa al uso de colas sería hacer de la línea de comandos un generador asínico y procesar los comandos a medida que ingresan, de esta manera:

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

Escribí algo similar como parte de un paquete llamado aioconsole .

Proporciona una línea llamada get_standard_streams que devuelve dos secuencias asyncio correspondientes a stdin y stdout .

Aquí hay un ejemplo:

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

También incluye un equivalente asíncrono a la input :

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

Debería funcionar tanto para archivos como para archivos que no son archivos. Vea la implementación aquí .