使用asyncio听按键

Question

有人可以提供一个代码示例，以非阻塞方式使用asynio监听按键，并在每次点击时将控制键放入控制台中？

这不是一个关于一些图形工具

Answer 1

问题所以安德烈Corbellini提供的链接是一个聪明的和彻底的解决问题的办法，但也相当复杂。如果你想要做的是提示用户输入一些输入（或模拟的raw_input），我更喜欢使用更简单的解决方案：

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

我写了一个名为aioconsole包的一部分，类似的东西。

它提供了一个名为get_standard_streams协程返回对应于stdin和。

下面是一个例子：

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

它还包括一个异步相当于input：

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

应该两个文件和非文件流工作。请参阅实施here。

Answer 3

使用队列将是使命令行的ASYN发生器，以及处理命令，因为他们进来，像这样的选择：

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