156

コマンド ライン プログラム (svnadmin verify) のラッパー スクリプトを作成しようとしています。これには、ラップされたプログラムが出力されるとすぐに、出力の各行を確認できる必要があります。

subprocess.Popen、 useを使用してプログラムを実行し、stdout=PIPE各行が入ってくるたびに読み取り、それに応じて行動するだけだと考えました。ただし、次のコードを実行すると、出力がどこかにバッファリングされているように見え、1 行目から 332 行目、333 行目から 439 行目 (出力の最後の行) の 2 つのチャンクに出力が表示されました。

from subprocess import Popen, PIPE, STDOUT

p = Popen('svnadmin verify /var/svn/repos/config', stdout = PIPE, 
        stderr = STDOUT, shell = True)
for line in p.stdout:
    print line.replace('\n', '')

subprocess のドキュメントを少し見たところ、bufsizeパラメータがPopenにあることがわかりました。そのため、bufsize を 1 (行ごとにバッファリングする) と 0 (バッファなし) に設定してみましたが、どちらの値でも行の配信方法が変わるようには見えませんでした。

この時点で、私はストローをつかみ始めていたので、次の出力ループを書きました。

while True:
    try:
        print p.stdout.next().replace('\n', '')
    except StopIteration:
        break

しかし、同じ結果が得られました。

サブプロセスを使用して実行されたプログラムの「リアルタイム」プログラム出力を取得することは可能ですか? Python に前方互換性がある (ないexec*) 他のオプションはありますか?

4

20 に答える 20

87

私はこれを試しましたが、何らかの理由でコードが

for line in p.stdout:
  ...

積極的にバッファリングし、バリアント

while True:
  line = p.stdout.readline()
  if not line: break
  ...

ではない。どうやらこれは既知のバグです: http://bugs.python.org/issue3907 (この問題は 2018 年 8 月 29 日に「クローズ」されました)

于 2009-04-29T17:26:53.640 に答える
39

バッファ サイズを 1 に設定すると、基本的にプロセスが出力をバッファリングしないようになります。

p = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=1)
for line in iter(p.stdout.readline, b''):
    print line,
p.stdout.close()
p.wait()
于 2011-06-20T16:13:51.240 に答える
25

これを試すことができます：

import subprocess
import sys

process = subprocess.Popen(
    cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)

while True:
    out = process.stdout.read(1)
    if out == '' and process.poll() != None:
        break
    if out != '':
        sys.stdout.write(out)
        sys.stdout.flush()

read の代わりに readline を使用すると、入力メッセージが出力されない場合があります。インライン入力が必要なコマンドで試してみてください。

于 2009-04-29T17:21:10.203 に答える
11

Python 3.x では、出力が文字列ではなくバイト配列であるため、プロセスがハングする場合があります。必ず文字列にデコードしてください。

Python 3.6 以降では、 Popen Constructor のパラメーターを使用して実行encodingできます。完全な例:

process = subprocess.Popen(
    'my_command',
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    shell=True,
    encoding='utf-8',
    errors='replace'
)

while True:
    realtime_output = process.stdout.readline()

    if realtime_output == '' and process.poll() is not None:
        break

    if realtime_output:
        print(realtime_output.strip(), flush=True)

このコードは、出力エラーにリダイレクト stderrしstdoutて処理することに注意してください。

于 2019-09-17T08:49:38.963 に答える
3

私はしばらく前に同じ問題に遭遇しました。read私の解決策は、サブプロセスの実行が完了していなくてもすぐに返されるメソッドの繰り返しを捨てることでした。

于 2009-04-29T17:24:09.200 に答える
2

この「プラグアンドプレイ」機能はこちらで見つかりました。魔法のように働きました！

import subprocess

def myrun(cmd):
    """from
    http://blog.kagesenshi.org/2008/02/teeing-python-subprocesspopen-output.html
    """
    p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT)
    stdout = []
    while True:
        line = p.stdout.readline()
        stdout.append(line)
        print line,
        if line == '' and p.poll() != None:
            break
    return ''.join(stdout)
于 2016-08-03T13:28:47.707 に答える
2

これは、私がこれに常に使用する基本的なスケルトンです。これにより、タイムアウトの実装が容易になり、避けられないハング プロセスに対処できます。

import subprocess
import threading
import Queue

def t_read_stdout(process, queue):
    """Read from stdout"""

    for output in iter(process.stdout.readline, b''):
        queue.put(output)

    return

process = subprocess.Popen(['dir'],
                           stdout=subprocess.PIPE,
                           stderr=subprocess.STDOUT,
                           bufsize=1,
                           cwd='C:\\',
                           shell=True)

queue = Queue.Queue()
t_stdout = threading.Thread(target=t_read_stdout, args=(process, queue))
t_stdout.daemon = True
t_stdout.start()

while process.poll() is None or not queue.empty():
    try:
        output = queue.get(timeout=.5)

    except Queue.Empty:
        continue

    if not output:
        continue

    print(output),

t_stdout.join()
于 2017-12-27T14:11:04.513 に答える
1

完全なソリューション:

import contextlib
import subprocess

# Unix, Windows and old Macintosh end-of-line
newlines = ['\n', '\r\n', '\r']
def unbuffered(proc, stream='stdout'):
    stream = getattr(proc, stream)
    with contextlib.closing(stream):
        while True:
            out = []
            last = stream.read(1)
            # Don't loop forever
            if last == '' and proc.poll() is not None:
                break
            while last not in newlines:
                # Don't loop forever
                if last == '' and proc.poll() is not None:
                    break
                out.append(last)
                last = stream.read(1)
            out = ''.join(out)
            yield out

def example():
    cmd = ['ls', '-l', '/']
    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        # Make all end-of-lines '\n'
        universal_newlines=True,
    )
    for line in unbuffered(proc):
        print line

example()
于 2013-03-08T07:33:30.493 に答える
1

ログをリアルタイムでコンソールに転送するだけの場合

以下のコードは両方で機能します

 p = subprocess.Popen(cmd,
                         shell=True,
                         cwd=work_dir,
                         bufsize=1,
                         stdin=subprocess.PIPE,
                         stderr=sys.stderr,
                         stdout=sys.stdout)
于 2020-08-09T04:21:35.467 に答える
1

ノンブロッキング readlines でpexpectを使用すると、この問題が解決します。これは、パイプがバッファリングされているため、アプリの出力がパイプによってバッファリングされているため、バッファがいっぱいになるかプロセスが停止するまで、その出力に到達できません。

于 2010-07-06T14:09:22.727 に答える