python - 强制 input() 从不同的线程读取换行符并向前执行

python - force input() to read newline character from a different thread and forward execution

我想找出一种方法来以编程方式避免内置 input() 方法停止并等待用户输入。 这是显示我正在尝试做的事情的片段:

import sys
from threading import Thread

def kill_input():
    sys.stdout.write('\n')
    sys.stdout.flush()   # just to make sure the output is really written to stdout and not bufferized

t = Thread(target=kill_input)
t.start()
foo = input('Press some key')
print('input() method has been bypassed')

预期行为:脚本执行并终止,无需等待按下回车键。

相反,发生的事情是程序停止等待用户输入一些内容。

在我看来,input() 应该读取另一个线程在 stdout 上打印的 换行符 ('\n'),并通过执行最终打印语句终止。该线程应该模拟用户按下回车键。不明白后面是怎么回事

也许另一种可能的方法是从非主线程关闭标准输入文件描述符并在主线程上捕获异常。

def kill_input():
    sys.stdin.close()

可能我想避免这个选项,而是想了解这个逻辑背后发生的事情,并找到一种方法来强制主线程从标准输入中读取一些模拟字符。

编辑 - 使用 子进程 模块

基于这些 related I've had a look to the subprocess 模块。我认为 Popen class 可以派上用场,所以我修改了我的脚本以利用管道

import sys
from subprocess import Popen, PIPE

def kill_input():
     proc = Popen(['python3', '-c', 'pass'], stdin=PIPE)
     proc.stdin.write('some text just to force parent proc to read'.encode())
     proc.stdin.flush()
     proc.stdin.close()
 
t = Thread(target=kill_input)
t.start()
sys.stdin.read()
print('input() method has been bypassed')

根据我的理解,这应该使用 Popen 创建一个进程(推荐 python3 -c 'pass' 就像一个占位符)其(应该?) stdin 是用父进程打开的 unix 管道。 我期望的是写入 子进程标准输入 的任何内容直接进入父进程 的 标准输入,以便被 sys.stdin.read()。所以程序不应该停止等待任何用户输入,它应该立即终止。不幸的是,它并没有发生,脚本仍在等待我按下 enter。我真的找不到解决方法。

[Python版本:3.8.5]

在您的第一段代码中,您正在写入 sys.stdout,默认情况下不会影响 sys.stdin 的内容。此外,默认情况下,您不能直接写入 sys.stdin,但您可以将其更改为不同的文件。为此,您可以使用 os.pipe(),它将 return 一个用于从新管道读取的文件描述符的元组,以及一个用于写入管道的文件描述符。

然后我们可以在这些文件描述符上使用 os.fdopen,并将 sys.stdin 分配给管道的读取端,而在另一个线程中我们写入管道的另一端。

import sys
import os
from threading import Thread

fake_stdin_read_fd, fake_stdin_write_fd = os.pipe()
fake_stdin_read = os.fdopen(fake_stdin_read_fd, 'r')
fake_stdin_write = os.fdopen(fake_stdin_write_fd, 'w')
sys.stdin = fake_stdin_read

def kill_input():
    fake_stdin_write.write('hello\n')
    fake_stdin_write.flush()

thread = Thread(target=kill_input)
thread.start()

input()
print('input() method has been bypassed!')