[python] 서브 프로세스 stdout을 한 줄씩 읽으십시오

내 파이썬 스크립트는 하위 프로세스를 사용하여 매우 시끄러운 리눅스 유틸리티를 호출합니다. 모든 출력을 로그 파일에 저장하고 일부를 사용자에게 보여주고 싶습니다. 나는 다음과 같이 작동 할 것이라고 생각했지만 유틸리티가 상당한 양의 출력을 생성 할 때까지 응용 프로그램에 출력이 표시되지 않습니다.

#fake_utility.py, just generates lots of output over time
import time
i = 0
while True:
   print hex(i)*512
   i += 1
   time.sleep(0.5)

#filters output
import subprocess
proc = subprocess.Popen(['python','fake_utility.py'],stdout=subprocess.PIPE)
for line in proc.stdout:
   #the real code does filtering here
   print "test:", line.rstrip()

필자가 실제로 원하는 동작은 필터 스크립트가 서브 프로세스로부터 수신 될 때 각 라인을 인쇄하는 것입니다. tee파이썬 코드와 비슷 하지만 파이썬 코드와 비슷 합니다.

내가 무엇을 놓치고 있습니까? 이것도 가능합니까?

최신 정보:

a sys.stdout.flush()가 fake_utility.py에 추가되면 코드는 python 3.1에서 원하는 동작을 갖습니다. 파이썬 2.6을 사용하고 있습니다. 사용 proc.stdout.xreadlines()은 py3k와 동일하게 작동 한다고 생각 하지만 그렇지 않습니다.

업데이트 2 :

다음은 최소 작업 코드입니다.

#fake_utility.py, just generates lots of output over time
import sys, time
for i in range(10):
   print i
   sys.stdout.flush()
   time.sleep(0.5)

#display out put line by line
import subprocess
proc = subprocess.Popen(['python','fake_utility.py'],stdout=subprocess.PIPE)
#works in python 3.0+
#for line in proc.stdout:
for line in iter(proc.stdout.readline,''):
   print line.rstrip()

답변

파이썬으로 마지막으로 작업 한 지 오랜 시간이 지났지 만 문제는 문 for line in proc.stdout을 반복하기 전에 전체 입력을 읽는 문에 있다고 생각 합니다. 해결책은 readline()대신 사용하는 것입니다.

#filters output
import subprocess
proc = subprocess.Popen(['python','fake_utility.py'],stdout=subprocess.PIPE)
while True:
  line = proc.stdout.readline()
  if not line:
    break
  #the real code does filtering here
  print "test:", line.rstrip()

물론 하위 프로세스의 버퍼링을 처리해야합니다.

참고 : 문서에 따르면 반복 미리보기가있는 솔루션은 readline()미리 읽기 버퍼를 제외하고 를 사용하는 것과 동일해야 하지만 제안 된 변경으로 인해 다른 결과가 나왔습니다 (Windows XP의 Python 2.5).

답변

파티에 늦었지만 여기에서 가장 간단한 해결책이 무엇인지 생각하지 않는 것에 놀랐습니다.

import io
import subprocess

proc = subprocess.Popen(["prog", "arg"], stdout=subprocess.PIPE)
for line in io.TextIOWrapper(proc.stdout, encoding="utf-8"):  # or another encoding
    # do something with line

(파이썬 3이 필요합니다.)

답변

실제로 반복자를 정렬하면 버퍼링이 문제가 될 수 있습니다. 하위 프로세스의 파이썬에게 출력을 버퍼링하지 않도록 지시 할 수 있습니다.

proc = subprocess.Popen(['python','fake_utility.py'],stdout=subprocess.PIPE)

된다

proc = subprocess.Popen(['python','-u', 'fake_utility.py'],stdout=subprocess.PIPE)

파이썬 내에서 파이썬을 호출 할 때 이것이 필요했습니다.

답변

이러한 추가 매개 변수를 다음으로 전달하려고합니다 subprocess.Popen.

bufsize=1, universal_newlines=True

그런 다음 예제와 같이 반복 할 수 있습니다. (파이썬 3.5에서 테스트)

답변

둘다 순회를 가능하게하는 기능 stdout과 stderr라인으로 라인을 동시에, 실시간

경우에 당신은 모두 출력 스트림을 얻을 필요 stdout하고 stderr동시에 다음과 같은 기능을 사용할 수 있습니다.

이 함수는 큐를 사용하여 두 Popen 파이프를 단일 반복기로 병합합니다.

여기에 함수를 만듭니다 read_popen_pipes():

from queue import Queue, Empty
from concurrent.futures import ThreadPoolExecutor


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


def read_popen_pipes(p):

    with ThreadPoolExecutor(2) as pool:
        q_stdout, q_stderr = Queue(), Queue()

        pool.submit(enqueue_output, p.stdout, q_stdout)
        pool.submit(enqueue_output, p.stderr, q_stderr)

        while True:

            if p.poll() is not None and q_stdout.empty() and q_stderr.empty():
                break

            out_line = err_line = ''

            try:
                out_line = q_stdout.get_nowait()
            except Empty:
                pass
            try:
                err_line = q_stderr.get_nowait()
            except Empty:
                pass

            yield (out_line, err_line)

read_popen_pipes() 사용:

import subprocess as sp


with sp.Popen(my_cmd, stdout=sp.PIPE, stderr=sp.PIPE, text=True) as p:

    for out_line, err_line in read_popen_pipes(p):

        # Do stuff with each line, e.g.:
        print(out_line, end='')
        print(err_line, end='')

    return p.poll() # return status-code

답변

루프없이 줄을 읽을 수도 있습니다. python3.6에서 작동합니다.

import os
import subprocess

process = subprocess.Popen(command, stdout=subprocess.PIPE)
list_of_byte_strings = process.stdout.readlines()

답변

내가 python3에 이것을 시도하고, 작업 소스

def output_reader(proc):
    for line in iter(proc.stdout.readline, b''):
        print('got line: {0}'.format(line.decode('utf-8')), end='')


def main():
    proc = subprocess.Popen(['python', 'fake_utility.py'],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.STDOUT)

    t = threading.Thread(target=output_reader, args=(proc,))
    t.start()

    try:
        time.sleep(0.2)
        import time
        i = 0

        while True:
        print (hex(i)*512)
        i += 1
        time.sleep(0.5)
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=0.2)
            print('== subprocess exited with rc =', proc.returncode)
        except subprocess.TimeoutExpired:
            print('subprocess did not terminate in time')
    t.join()