旨在与人类交互的程序不同于旨在与其他程序交互的程序。您的square.py
脚本更接近前一类。可能的问题:
- 提示在一行中间结束,该行强制父脚本一次读取一个字节,而不是整行或已知块以避免阻塞
- 孩子的答案没有明确刷新。这意味着可以在非交互模式下启用块缓冲,
subprocess
例如,当它通过pty
以下是使用subprocess
当前形式的模块与它交互的方法:
#!/usr/bin/env python
from __future__ import print_function
import sys
from itertools import cycle
from subprocess import Popen, PIPE
from textwrap import dedent
# start child process
p = Popen([sys.executable or 'python', '-u', '-c', dedent("""
for i in range(10):
x = int(input('Enter x-dimension: '))
print(x*x)
""")], stdin=PIPE, stdout=PIPE, universal_newlines=True, bufsize=1)
for n in cycle([3, 1, 4, 15, 926]): # infinite loop
while p.poll() is None: # while the subprocess is running
# send input to the child
print(n, file=p.stdin)
# read & parse answer
data = p.stdout.readline().rpartition(' ')[2]
if not data: # EOF
answer = None
break # exit inner loop
answer = int(data)
if answer == 1: # show example when input depends on output
n += 1
else: # done with given `n`
break # exit inner loop
else: # subprocess ended
break # exit outer loop
if answer is not None:
print("Input %3d Output %6d" % (n, answer))
p.communicate() # close pipes, wait for the child to terminate
这是同样的事情,但使用pexpect
(用于比较):
#!/usr/bin/env python
import sys
from itertools import cycle
from textwrap import dedent
import pexpect
child = pexpect.spawnu(sys.executable or 'python', ['-c', dedent("""
for i in range(10):
x = int(input('Enter x-dimension: '))
print(x*x)
""")])
for n in cycle([3, 1, 4, 15, 926]):
while True:
i = child.expect([pexpect.EOF, u'x-dimension:'])
if i == 0: # EOF
answer = None
child.close()
sys.exit()
elif i == 1: # child waits for input
child.sendline(str(n))
child.expect(u'\\n\\d+\\s')
answer = int(child.after)
if answer == 1:
n += 1
else:
break
else:
assert 0
else: # child terminated
break
if answer is not None:
print("Input %3d Output %6d" % (n, answer))
这两个脚本都是为了支持同一来源的 Python 2 和 Python 3 而编写的。
注意:基于 - 的脚本中有一个-u
参数,subprocess
即使在非交互模式下,它们也允许在它们可用时立即读取它们。pexpect
-based 脚本无需此类开关即可工作。stdio
基于 - 的程序可以使用stdbuf
,unbuffer
实用程序或通过提供 pty来无缓冲/进行行缓冲。
您可以看到,即使是最简单的子脚本 ( square.py
) 也需要克服几个问题才能正常工作。
当子程序期望从另一个程序运行同时保持人类可读(可调试)时,一切都会变得更简单。在这种情况下,square.py
可能如下所示:
#!/usr/bin/env python
import sys
import time
for line in iter(sys.stdin.readline, ''): # get line as soon as it is available
print(int(line)**2) # find square
sys.stdout.flush() # make the answer available immediately
time.sleep(.5) # a delay to show that the answer is available immediately
它可以subprocess
在“一次全部”模式下从基于 - 的模块中使用:
import sys
from subprocess import Popen, PIPE
L = [2, 7, 1] # numbers to be squared
p = Popen([sys.executable or 'python', 'square.py'], stdin=PIPE, stdout=PIPE,
universal_newlines=True, bufsize=-1)
answers = map(int, p.communicate("\n".join(map(str, L)))[0].splitlines())
或者一次一个数字:
#!/usr/bin/env python
import sys
from subprocess import Popen, PIPE
answers = []
p = Popen([sys.executable or 'python', 'square.py'], stdin=PIPE, stdout=PIPE,
bufsize=1)
for c in [b'2', b'7', b'1']:
p.stdin.write(c + b'\n')
p.stdin.flush()
answers.append(int(p.stdout.readline()))
print(answers)
p.communicate() # close pipes, wait for child to finish
print(answers)
从 C 程序中获取数组;你可以json
模块:
import json
from subprocess import Popen, PIPE
p = Popen(['./c-program', 'other', 'args'], stdin=PIPE, stdout=PIPE, bufsize=1)
p.stdin.write(json.dumps({'parameter': 8}).encode() + b'\n') # send input
p.stdin.flush()
result = json.loads(p.stdout.readline().decode()) # e.g., {"result": [0, 0, 7]}
# ...
p.communicate() # close pipes, wait for child to finish