Python3如何同时进行并行输入和输出？

Question

我需要设计一个脚本，它使用终端的顶部作为输出，其中在无限循环中每秒打印一些行，底部不断接受用户输入并在上面的部分中打印它们（在定期输出）。

换句话说，我需要设计一种外壳。

我尝试使用这样的幼稚方法进行多线程：

#!/usr/bin/python3

from math import acos
from threading import Thread
from random import choice
from time import sleep
from queue import Queue, Empty

commandQueue = Queue()

def outputThreadFunc():
    outputs = ["So this is another output","Yet another output","Is this even working"] # Just for demo
    while True:
        print(choice(outputs))
        try:
            inp = commandQueue.get(timeout=0.1)
            if inp == 'exit':
                return
            else:
                print(inp)
        except Empty:
            pass        
        sleep(1)

def inputThreadFunc():
    while True:
        command = input("> ") # The shell
        if command == 'exit':
            return
        commandQueue.put(command)

# MAIN CODE
outputThread = Thread(target=outputThreadFunc)
inputThread = Thread(target=inputThreadFunc)
outputThread.start()
inputThread.start()
outputThread.join()
inputThread.join()

print("Exit")

但正如预期的那样，随着用户不断输入，输出行与输入行合并。

有什么想法吗？

Answer 1

最简单的解决方案是使用两个脚本；一个是打印输出的服务器，另一个是将用户输入发送到服务器的客户端。然后您可以使用tmux 之类的标准解决方案在两个窗格中打开这两个脚本。

Answer 2

如 cmets 中所述，使用了 curses 库。

更新

使用两个subwin进行输入输出

#!/usr/bin/python3

import curses

from math import acos
from threading import Thread
from random import choice
from time import sleep
from queue import Queue, Empty


commandQueue = Queue()

stdscr = curses.initscr()
stdscr.keypad(True)

upperwin = stdscr.subwin(2, 80, 0, 0)
lowerwin = stdscr.subwin(2,0)

def outputThreadFunc():
    outputs = ["So this is another output","Yet another output","Is this even working"] # Just for demo
    while True:
        upperwin.clear()
        upperwin.addstr(f"{choice(outputs)}")
        try:
            inp = commandQueue.get(timeout=0.1)
            if inp == 'exit':
                return
            else:
                upperwin.addch('\n')
                upperwin.addstr(inp)
        except Empty:
            pass

        upperwin.refresh()
        sleep(1)
        


def inputThreadFunc():
    while True:
        global buffer

        lowerwin.addstr("->")

        command = lowerwin.getstr()

        if command:
            command = command.decode("utf-8")
            commandQueue.put(command)
            lowerwin.clear()

            lowerwin.refresh()
            if command == 'exit':
                return

            
        


# MAIN CODE
outputThread = Thread(target=outputThreadFunc)
inputThread = Thread(target=inputThreadFunc)
outputThread.start()
inputThread.start()
outputThread.join()
inputThread.join()

stdscr.keypad(False)
curses.endwin()
print("Exit")

旧解决方案

我已将您的示例编辑为使用 getch insted of input

#!/usr/bin/python3

import curses
import datetime

from math import acos
from threading import Thread
from random import choice
from time import sleep
from queue import Queue, Empty

INFO_REFRESH_SECONDS = 1

commandQueue = Queue()
buffer = list()  # stores your input buffer
stdscr = curses.initscr()
stdscr.keypad(True)

def outputThreadFunc():
    outputs = ["So this is another output","Yet another output","Is this even working"] # Just for demo
    info = choice(outputs), datetime.datetime.now()
    while True:

        if datetime.datetime.now() - info[1] > datetime.timedelta(seconds=INFO_REFRESH_SECONDS):
            # refresh info after certain period of time

            info = choice(outputs), datetime.datetime.now()  # timestamp which info was updated

        inp = ''
        buffer_text = ''.join(buffer)
        try:
            command = commandQueue.get(timeout=0.1)
            if command == 'exit':
                return
            inp = f"\n{command}"
        except Empty:
            pass 
        output_string = f"{info[0]}{inp}\n->{buffer_text}"
        stdscr.clear()
        stdscr.addstr(output_string)
        stdscr.refresh()
        if inp:
            # to make sure you see the command
            sleep(1)
        


def inputThreadFunc():
    while True:
        global buffer

        # get one character at a time
        key = stdscr.getch()
        curses.echo()

        if chr(key) == '\n':
            command = ''.join(buffer)
            commandQueue.put(command)
            if command == 'exit':
                return
            buffer = []
        elif key == curses.KEY_BACKSPACE:
            
            if buffer:
                buffer.pop()
        else:
            buffer.append(chr(key))

            
        


# MAIN CODE
outputThread = Thread(target=outputThreadFunc)
inputThread = Thread(target=inputThreadFunc)
outputThread.start()
inputThread.start()
outputThread.join()
inputThread.join()

stdscr.keypad(False)
curses.endwin()
print("Exit")

Answer 3

由于终端写入输出的方式，两者正在合并。它在缓冲区中收集输出，并在适当的时候立即输出所有内容。一个简单的解决方法是在每个实际语句之前使用'\n'，以便每个新输出位于单独的行上。

#!/usr/bin/python3

                    .
                    .
                    .

            if inp == 'exit':
                return
            else:
                print("\n", inp) # CHANGE OVER HERE

                    .
                    .
                    .

        command = input("\n> ") # CHANGE OVER HERE
        if command == 'exit':
            return

                    .
                    .
                    .

print("Exit")

请注意，由于两个线程并行运行，下一个输出将在您完成输入并按 Enter 输入之前打印（除非您的输入速度非常快并且反应速度非常快）。希望这能回答您的问题！