Асинхронный ввод в консоли без помех от вторичного потока

Question

Недавно я писал CLI серверную программу на python, моя программа работает асинхронно, я столкнулся с такой проблемой:

Скажу сразу, в интернете был такой вопрос, только на англ. stack-overflow, но он уже не актуален (ему 14 лет), да и тот метод уже не работает

В мою консоль одновременно приходит сообщения от сервера, и в тоже время я хочу реализовать асинхронный ввод команд, в интернете я нашел такой пример так называемого "row_input()", (извиняюсь если что-то перепутал):

import asyncio
import sys
import websockets


async def ainput(string: str) -> str:
    await asyncio.to_thread(sys.stdout.write, f'{string} ')
    return (await asyncio.to_thread(sys.stdin.readline)).rstrip('\n')

Эта функция позволяет асинхронно вводить текст игнорируя приходящею информацию, однако при получении информации, недописанный текст сбрасывается, сообщение от сервера дополняет строку и каретка переводится на следующую строку, мне же надо, что б текст до ввода, например "input command:" оставался снизу, а так же поле ввода сохранялось при получении информации. По идее, при получении нового сообщения, консоль должна считать незаконченный ввод, убрать строку, написать сообщение, и заново начать ввод, но уже с использованием недописанного текста. Также для доп. информации, я работаю на Windows 10

Answer 1

Сначала адаптирую первый пример из этого ответа под Python 3:

import readline
import threading
import time


def noisy_thread():
    while True:
        time.sleep(3)
        print('\r'+' '*(len(readline.get_line_buffer())+2)+'\r', end="")
        print('Interrupting text!')
        print('> ' + readline.get_line_buffer(), end="", flush=True)


threading.Thread(target=noisy_thread, daemon=True).start()
while True:
    s = input('> ')

Изменения минимальные:

Функция print вместо оператора
input вместо raw_input
Изменился способ запуска потока
Ну и sys.stdout.write заменил на print с параметром end="", и sys.stdout.flush() заменил на параметр flush=True у print. Это на работоспособность кода не влияет, но так код выглядит гораздо чище.

Тестировал на Linux, но по идее должно работать и на Windows.

Суть кода вкратце: перед каждым выводом текста с экрана удаляется промпт и то, что пользователь успел ввести (затирается пробелами), выводится текст, потом опять выводится промпт и то, что пользователь ввел. Магия в том, что input продолжает работать как ни в чем не бывало, в том числе работает удаление ранее введенного текста. Очень интересное решение.

Дальше, адаптирую под asyncio:

import asyncio
import readline


_prompt = ""


def uninterrupting_print(*args, **kwargs):
    print("\r" + " " * (len(_prompt) + len(readline.get_line_buffer())) + "\r", end="")
    print(*args, **kwargs)
    print(_prompt + readline.get_line_buffer(), end="", flush=True)


async def async_input(prompt: str):
    global _prompt
    _prompt = prompt
    return await asyncio.to_thread(input, prompt)


async def spammer_loop():
    while True:
        await asyncio.sleep(1)
        uninterrupting_print("Interrupting text!")


async def input_loop():
    while True:
        user_input = await async_input("input command: ")
        print("User input:", user_input)
        if user_input == "exit":
            break


async def main():
    spammer_task = asyncio.create_task(spammer_loop())
    input_task = asyncio.create_task(input_loop())
    _done, pending = await asyncio.wait([spammer_task, input_task], return_when=asyncio.FIRST_COMPLETED)
    for task in pending:
        task.cancel()


asyncio.run(main())

В вашем случае обработку данных с сервера нужно поместить в spammer_loop, и получаемые данные выводить через uninterrupting_print. Обработку команд от пользователя, соответственно, делаете в input_loop.

БЛОГ НА HUSL

Асинхронный ввод в консоли без помех от вторичного потока

Ответы (1 шт):