时间戳

首页 > 解决方案 > 在 Python 中写入生成的进程的重定向输入

python-3.x - 在 Python 中写入生成的进程的重定向输入

问题描述

对不起，我真的不是蟒蛇人，但情况打击了我，这对你们蟒蛇人来说是非常明显的。

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

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

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

标签： python-3.x

解决方案

有趣的是，我刚刚发现添加

universal_newlines=True

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

编辑：

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

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

所以我们要么被线程困住，要么被现代难以阅读的异步/等待风格困住。

基于上面链接的灵感，我已经达到了以下简单的工作 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)在每次输出后发出。甚至换行符也没有帮助这些东西在不冲洗的情况下落入管道。

推荐阅读