하위 프로세스를 한 줄씩 읽다.

하위 프로세스를 한 줄씩 읽다.

내 python 스크립트는 매우 시끄러운 Linux 유틸리티를 호출하기 위해 하위 프로세스를 사용한다.모든 출력을 로그 파일에 저장하고 사용자에게 일부를 보여주고 싶다.나는 다음과 같은 것이 효과가 있을 것이라고 생각했지만, 유틸리티가 상당한 양의 출력을 생산하기 전에는 출력이 내 어플리케이션에 나타나지 않는다.

#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()

내가 정말 원하는 동작은 필터 스크립트가 하위 프로세스에서 수신되는 대로 각 줄을 인쇄하는 것이다.python 코드로만 하는 것 처럼요.

제가 무엇을 빠뜨리고 있나요?이게 가능할까?

---

업데이트:

만약sys.stdout.flush()3fake_pyte에서 .py에 추가된 코드는 python 3.1에서 원하는 동작을 가진다.파이톤 2.6을 쓰고 있어.당신은 그것을 사용하는 것이proc.stdout.xreadlines()파이3k와 동일하게 작동하지만, 그렇지 않다.

---

업데이트 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() ( 때문에)는 나에게 (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

(이러려면 Python 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

그런 다음 예시와 같이 반복할 수 있다. (Python 3.5로 테스트됨)

두 에 대해 할 수 있는 은 기이다.stdout그리고stderr동시에, 실시간으로, 한 줄씩

두 가지 모두에 대한 출력 스트림을 가져와야 하는 경우stdout그리고stderr동시에 다음과 같은 기능을 사용할 수 있다.

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

여기서 함수를 생성한다.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()

파이톤3로 해봤는데 효과가 있었어, 출처

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()

다음과 같은 Rmulmulo의 대답 수정은 Python 2와 3(2.7.12와 3.6.1)에서 나에게 효과가 있다.

import os
import subprocess

process = subprocess.Popen(command, stdout=subprocess.PIPE)
while True:
  line = process.stdout.readline()
  if line != '':
    os.write(1, line)
  else:
    break

피몬트 3.5는 그 방법들을 추가했다.run()그리고call()에게subprocess모듈(둘 다 a를 반환함)CompletedProcess이의를 제기하다이것으로 너는 괜찮다.proc.stdout.splitlines():

proc = subprocess.run( comman, shell=True, capture_output=True, text=True, check=True )
for line in proc.stdout.splitlines():
   print "stdout:", line

하위 프로세스 실행 방법을 사용하여 Python에서 셸 명령을 실행하는 방법을 참조하십시오.

참조URL: https://stackoverflow.com/questions/2804543/read-subprocess-stdout-line-by-line