Слушайте нажатие клавиш с помощью asyncio

Question

Может ли кто-нибудь предоставить пример кода, который прослушивает нажатие клавиши неблокирующим способом с помощью asynio и помещает код клавиши в консоль при каждом нажатии?

Это не вопрос какого-то графического инструментария

Answer 1

Таким образом, ссылка, предоставленная Андреа Корбеллини, является умным и тщательным решением проблемы, но также довольно сложной. Если все, что вы хотите сделать, это попросить пользователя ввести какой-либо ввод (или имитировать 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 который возвращает два потока asyncio, соответствующих stdin а также stdout,

Вот пример:

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

Он должен работать как для файловых, так и для не файловых потоков. Смотрите реализацию здесь.

Answer 3

Линии чтения

Высокоуровневый чистый асинхронный способ сделать это выглядит следующим образом.

      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'но фактически пустая строка) означает конец файла. Обратите внимание, что это возможно для await reader.readline() возвращаться '' сразу после reader.at_eof()вернул False. См. Подробности в Python asyncio: StreamReader .

Здесь 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, что приведет к двум итерациям цикла. Но этого достаточно для символов ASCII, таких как буквы и ESC.

Если вам нужны фактические нажатия клавиш, такие как ALT, я полагаю, единственный способ - использовать подходящую библиотеку и заставить ее работать с asyncio, вызвав ее в отдельном потоке, как здесь . На самом деле подход библиотека + поток, вероятно, проще и в других случаях.

Под капотом

Если вам нужен более точный контроль, вы можете реализовать свой собственный протокол вместо StreamReaderProtocol: класс, реализующий любое количество функций <tcode id="4418396"></tcode>. Минимальный пример:

      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 возвращает кортеж (transport, protocol), куда protocol это пример MyReadProtocol (создается фабрикой протоколов, которая в нашем случае является тривиальной лямбдой), а transport это пример asyncio.ReadTransport (в частности, некоторая частная реализация, например _UnixReadPipeTransport в UNIX).

Но, в конце концов, все это шаблон, который в конечном итоге полагается на <tcode id="4418409"></tcode> (не имеет отношения к StreamReader).

Для Windows вам может потребоваться выбрать ProactorEventLoop(значение по умолчанию, начиная с Python 3.8), см. Python asyncio: Поддержка платформы .

Answer 4

Альтернативой использованию очередей было бы сделать командную строку асинхронным генератором и обрабатывать команды по мере их поступления, вот так:

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

Answer 5

Обновление Python 3.10 для решения, предоставленного user618895:

      class Prompt:
    def __init__(self):
        self.loop = asyncio.get_running_loop()
        self.q = asyncio.Queue()
        self.loop.add_reader(sys.stdin, self.got_input)

    def got_input(self):
        asyncio.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)
        # https://docs.python.org/3/library/asyncio-task.html#coroutine
        task = asyncio.create_task(self.q.get())
        return (await task).rstrip('\n')

Я протестировал его на клиенте веб-сокета внутри асинхронной функции, которая зависала в ожидании ввода, поэтому я заменил s = input("insert string")с s = await prompt("insert string")и теперь пинг-понг работает, даже когда программа ожидает ввода пользователя, соединение больше не останавливается, и проблема «время ожидания проверки активности понга истекло» решена.