プロンプトがどのように見えるかを知っている場合は、@Michael が提案したように、文字列操作を使用して目的の出力を取得できます。
import sys
from subprocess import Popen, PIPE
p = Popen([sys.executable, "names.py"], stdin=PIPE, stdout=PIPE,
universal_newlines=True)
output = p.communicate("Sue\n")[0]
prompt = "name? "
print(output.replace(prompt, prompt + "\n"))
出力
I am Joe.
What is your name?
Hi Sue !
プロンプトがどのように表示されるかわからない場合、子プロセスが非対話的に実行されるときにブロックバッファリングを使用すると、@Exp が提案するタイムアウトベースのソリューションが機能しない可能性があります。それは動作しますがnames.py
。select
出力を読み取るためにスレッドの代わりに使用するタイムアウトベースのソリューションを次に示します。
import os
import sys
from select import select
from subprocess import Popen, PIPE
timeout = 1 # seconds
with Popen([sys.executable, 'names.py'],
stdin=PIPE, stdout=PIPE, bufsize=0) as p:
while True:
ready = select([p.stdout], [], [], timeout)[0]
if ready: # there is something to read
data = os.read(p.stdout.fileno(), 512)
if not data: # EOF
break
sys.stdout.buffer.write(data) # echo subprocess output
elif p.poll() is None: # timeout, but subprocess is still running
print("") # print newline after the prompt
p.stdin.write(b"Sue\n") # answer the prompt
else: # subprocess exited
break
遅延の後、最初のコード例と同じ出力が生成されます。
一般にpexpect
、サブプロセスの対話モードをエミュレートするために使用できます。
プロンプトがどのように見えるかを知っている場合:
import sys
import pexpect
print(pexpect.run(sys.executable + " -mnames", events={r'name\? ': 'Sue\n'}))
# note: it echos our answer too (it can be avoided if necessary)
出力
I am Joe.
What is your name? Sue
Hi Sue !
応答のエコーバックを回避するタイムアウトベースのソリューションを次に示します。
import sys
import pexpect # pip install pexpect-u
child = pexpect.spawn(sys.executable + " -mnames", timeout=1)
child.logfile_read = sys.stdout # echo subprocess output
child.expect(pexpect.TIMEOUT)
print("") # print newline after the prompt
child.setecho(False) # don't echo our answer
child.sendline('Sue')
child.expect(pexpect.EOF)
child.close()
を使用して再現するにはsubprocess
、pty
モジュールを使用できます。
import os
import pty
import sys
from select import select
from subprocess import Popen, STDOUT
timeout = 1 # seconds
master_fd, slave_fd = pty.openpty()
with Popen([sys.executable, 'names.py'],
stdin=slave_fd, stdout=slave_fd, stderr=STDOUT,
bufsize=0) as p:
while True:
ready = select([master_fd], [], [], timeout)[0]
if ready: # there is something to read
data = os.read(master_fd, 512)
if not data: # EOF
break
sys.stdout.buffer.write(data) # echo subprocess output
elif p.poll() is None: # timeout, but subprocess is still running
# assume that child process waits for input after printing the prompt
answer = b"Sue\n"
os.write(master_fd, answer) # asnwer the prompt
os.read(master_fd, len(answer)) # don't echo our answer
else: # subprocess exited
break
os.close(slave_fd)
os.close(master_fd)