主页 / 问题 / 1595065
Accepted
TosiX
Asked:2024-09-28 00:51:06 +0000 UTC2024-09-28 00:51:06 +0000 UTC

异步输入到控制台，不受辅助线程干扰

  • 772

最近我在用python写CLI一个服务器程序，我的程序异步运行，我遇到了以下问题：

我马上说网上有这样的问题，只是英文的。堆栈溢出，但它不再相关（它已有 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')

该功能允许您异步输入игнорируя带有传入信息的文本，但是，当接收信息时，未完成的文本сбрасывается，来自服务器的消息补充了该行，并且каретка переводится на следующую строку，我需要输入之前的文本，例如“输入命令：”保留在底部，接收信息时也会保存输入字段。理论上，当收到新消息时，控制台应该读取未完成的输入，删除该行，写入消息，然后再次开始输入，但使用未完成的文本。还可以用于额外的信息，我工作的目的Windows 10

python

1 个回答

  1. Best Answer

    首先，我将把这个答案中的第一个示例改编为 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=Truey 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。

    • 1

相关问题