使用asyncio收听按键

Question

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

这不是一些图形工具包的问题

Answer 1

读行

执行此操作的高级纯异步方法如下。

import asyncio
import sys

async def main():
        # Create a StreamReader with the default buffer limit of 64 KiB.
        reader = asyncio.StreamReader()
        pipe = sys.stdin
        loop = asyncio.get_event_loop()
        await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), pipe)

        async for line in reader:
                print(f'Got: {line.decode()!r}')

asyncio.run(main())

可以更明确地编写循环async for line in reader，例如，如果您想在循环内打印提示或捕获异常：

 while True:
            print('Prompt: ', end='', flush=True)
            try:
                line = await reader.readline()
                if not line:
                    break
            except ValueError:
                print('Line length went over StreamReader buffer limit.')
            else:
                print(f'Got: {line.decode()!r}')

空line（不是'\n'但实际上是空字符串''）意味着文件结束。请注意，返回 False后也可以立即await reader.readline()返回。有关详细信息，请参阅Python asyncio：StreamReader。''reader.at_eof()

这readline()是异步收集一行输入。也就是说，事件循环可以在读者等待更多字符时运行。相反，在其他答案中，事件循环可能会阻塞：它可以检测到某些输入可用，输入调用函数sys.stdin.readline()，然后阻塞它，直到结束行可用（阻止任何其他任务进入循环）。当然，在大多数情况下这不是问题，因为结束行与（在行缓冲的情况下，这是默认的）一起可用，或者在（在其他情况下，假设行相当短）任何初始字符之后不久可用线。

一个字符一个字符地读取

await reader.readexactly(1)从管道读取时，您还可以使用逐字节读取单个字节。从终端读取按键时，需要正确设置，请参阅Python中的关键侦听器？了解更多。在 UNIX 上：

import asyncio
import contextlib
import sys
import termios

@contextlib.contextmanager
def raw_mode(file):
    old_attrs = termios.tcgetattr(file.fileno())
    new_attrs = old_attrs[:]
    new_attrs[3] = new_attrs[3] & ~(termios.ECHO | termios.ICANON)
    try:
        termios.tcsetattr(file.fileno(), termios.TCSADRAIN, new_attrs)
        yield
    finally:
        termios.tcsetattr(file.fileno(), termios.TCSADRAIN, old_attrs)

async def main():
    with raw_mode(sys.stdin):
        reader = asyncio.StreamReader()
        loop = asyncio.get_event_loop()
        await loop.connect_read_pipe(lambda: asyncio.StreamReaderProtocol(reader), sys.stdin)

        while not reader.at_eof():
            ch = await reader.read(1)
            # '' means EOF, chr(4) means EOT (sent by CTRL+D on UNIX terminals)
            if not ch or ord(ch) <= 4:
                break
            print(f'Got: {ch!r}')

asyncio.run(main())

请注意，这实际上并不是一次一个字符或一个键：如果用户按下一个提供多字节字符的组合键（例如 ALT+E），那么按下 ALT 时不会发生任何事情，终端将发送两个字节按 E，这将导致循环迭代两次。但对于字母和 ESC 等 ASCII 字符来说已经足够了。

如果您需要像 ALT 这样的实际按键，我想唯一的方法是使用合适的库并通过在单独的线程中调用它来使其与 asyncio 一起工作，就像这里一样。事实上，在其他情况下，库+线程方法也可能更简单。

在引擎盖下

如果您想要更好的控制，您可以实现自己的协议来代替StreamReaderProtocol：实现任意数量的asyncio.Protocol. 最小的例子：

class MyReadProtocol(asyncio.Protocol):
    def __init__(self, reader: asyncio.StreamReader):
        self.reader = reader

    def connection_made(self, pipe_transport):
        self.reader.set_transport(pipe_transport)

    def data_received(self, data: bytes):
        self.reader.feed_data(data)

    def connection_lost(self, exc):
        if exc is None:
            self.reader.feed_eof()
        else:
            self.reader.set_exception(exc)

您可以用自己的缓冲机制替换 StreamReader。在调用 后connect_read_pipe(lambda: MyReadProtocol(reader), pipe)，将恰好有一次对 的调用connection_made，然后是任意多次调用data_received（数据取决于终端和 python 缓冲选项），最后恰好有一次对 的调用connection_lost（在文件结尾或出错时）。如果您需要它们，connect_read_pipe则返回一个 tuple (transport, protocol)，其中protocol是 的实例MyReadProtocol（由协议工厂创建，在我们的例子中是一个简单的 lambda），而transport是 的实例asyncio.ReadTransport（特别是一些私有实现，例如_UnixReadPipeTransport在 UNIX 上）。

但最终这都是最终依赖的所有样板loop.add_reader（与 无关StreamReader）。

对于 Windows，您可能需要选择ProactorEventLoop（自 Python 3.8 起的默认设置），请参阅Python asyncio：平台支持。

Answer 2

因此，安德里亚·科贝里尼（Andrea 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 3

我在名为aioconsole的软件包中编写了类似的内容。

它提供了一个称为的协程get_standard_streams，该协程返回对应于和的两个异步流。stdinstdout

这是一个例子：

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: ') 

它应适用于文件流和非文件流。在此处查看实现。