如何连续编写子进程的 stdin？

Question

我有一个脚本行为类似于下面的subscript.py，我想从另一个脚本运行它，因为它是用Python 2.7编写的。我尝试了一些子进程函数，但它们都没有提供与终端本机运行的下标相同的功能。

subscript.py

import threading
from queue import Queue
import argparse

class FooBar():
def __init__(self) -> None:
self.q = Queue()
foo = threading.Thread(target=self.foo)
bar = threading.Thread(target=self.bar)
foo.start()
bar.start()

def foo(self):
while True:
text = input()
print(f"{text} is sent by foo()")
self.q.put(text)

def bar(self):
while True:
if not self.q.empty():
text = self.q.get_nowait()
print(f"{text} is sent by bar()")
if __name__ == "__main__":
c = FooBar()

main1.py

import subprocess
subprocess.run("py subscript.py", text=True, input="foon")

main2.py这是一个阻塞类型。因此，不可能用stdin提供下标。

import subprocess
output = subprocess.call("py subscript.py")

Answer 1

第一个选项在字符串写入后立即关闭stdin，第二个选项不给您任何写入stdin的机会，因此它挂起。您可以使用较低级别的Popen调用来获取到stdin的管道并自己编写它。

>>> import sys
>>> import subprocess
>>> proc = subprocess.Popen([sys.executable, "subscript.py"], stdin=subprocess.PIPE)
>>> proc.stdin.write(b"hellon")
6
>>> proc.stdin.flush()
>>> hello is sent by foo()
hello is sent by bar()

该管道需要字节对象。但它可以放在一个文本包装器中，该包装器为您进行字符串到字节的编码(如果您愿意，还可以使用print)。

>>> import io
>>> txt_stdin = io.TextIOWrapper(proc.stdin)
>>> print("hello again", file=txt_stdin, flush=True)
>>> hello again is sent by foo()
hello again is sent by bar()

但是仍然有很多缺失。如果程序也写东西，您可能还需要为标准输出和标准错误使用管道。如果它期待一个终端，如果您也在类似unix的系统上，则可能需要使用伪tty或pexpect。