在 Python 中写入生成的进程的重定向输入

Question

对不起，我真的不是python人，但情况让我很受打击，这对你们python人来说是很明显的事情。

我正在尝试与衍生的控制台进程进行交互通信（在我们的用例中，它是一个控制台程序，它与旧的但仍然很好的 HP 仪器进行通信，用于测量实验中的一些物理变量）。
我在这里找到了一些有见地的例子：
Running interactive program from within python

但是当试图获得灵感时（我不想要那个计时器），我开始从头开始编写这个，使用通常存在于 windows 盒子上的 ftp.exe 程序进行测试，具有以下内容：

#!/usr/bin/python
import subprocess, sys
mystdout=sys.stdout
p = subprocess.Popen("ftp.exe", stdin=subprocess.PIPE, stdout=subprocess.PIPE) #, shell=True)
sys.stdout=p.stdin
print('bye\n'.encode())
sys.stdout=mystdout
print('done')

但是print('bye\n'.encode()) 的结果是：

print('bye\n'.encode())
TypeError: a bytes-like object is required, not 'str'

我不能使用 subprocess 的 communicate() 方法，因为它似乎不是很有交互性（仅限一次性）。

请你给我一个提示，我的笨蛋在哪里？这台机器在 Windows 上运行 Python 3.6.1，但更友好的 3.7.3 的 Linux 机器给出了相同的问候。

Answer 1

有趣，我刚刚发现添加

universal_newlines=True

Popen() 的最后一个参数解决了我的问题。我猜这个流不喜欢print()，因为它是以二进制模式打开的。添加此标志以文本模式打开流。我将发布一个完整的示例，包括交互式通信。

编辑：

哦，现在我找到了更多与此问题相关的资源。

• Non blocking reading from a subprocess output stream in Python
• Error: No module named 'fcntl'
• A non-blocking read on a subprocess.PIPE in Python
• Interactive input/output using Python
• Running an interactive command from within Python

在 Linux 上似乎更容易，因为有 fcntl 包允许 使用非阻塞 read() 设置管道。不幸的是，它似乎在 Windows 上不可用:( 我希望使用 peek() 调用，但令人惊讶的是，它也是 阻塞管道！ :(

所以我们要么被线程卡住，要么被现代难以阅读的 async/await 风格卡住。

基于上面链接的灵感，我已经达到了以下简单的工作 ftp 示例：

#!/usr/bin/python

import time, sys, subprocess, threading, queue

def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()    

def getOutput(outQueue):
    outStr = ''
    try:
        while True: 
            outStr+=outQueue.get_nowait()

    except queue.Empty:
        return outStr           

p = subprocess.Popen("ftp.exe", stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) 

outQueue = queue.Queue()
errQueue = queue.Queue()

outThread = threading.Thread(target=enqueue_output, args=(p.stdout, outQueue))
errThread = threading.Thread(target=enqueue_output, args=(p.stderr, errQueue))

outThread.daemon = True
errThread.daemon = True

outThread.start()
errThread.start()

p.stdin.write('status\n')
p.stdin.flush()
time.sleep(0.2)

errors = getOutput(errQueue)
output = getOutput(outQueue)

print("err:" + errors)
print("out:" + output)

time.sleep(2)

p.stdin.write('help\n')
p.stdin.flush()
time.sleep(0.2)

errors = getOutput(errQueue)
output = getOutput(outQueue)

print("err:" + errors)
print("out:" + output)

time.sleep(2)

p.stdin.write('bye\n')
p.stdin.flush()
time.sleep(0.2)

errors = getOutput(errQueue)
output = getOutput(outQueue)

print("err:" + errors)
print("out:" + output)

time.sleep(2)

print('done')

最后一点：当使用它与我自己的 c++ 程序通信时，我必须在每次输出后发出fflush(stdout)。甚至换行符也没有帮助这些东西在不冲洗的情况下落入管道。