Pobieranie danych wyjściowych w czasie rzeczywistym przy użyciu podprocesu

Próbuję napisać skrypt opakowujący dla programu wiersza poleceń (weryfikacja svnadmin), który będzie wyświetlał ładny wskaźnik postępu operacji. Wymaga to ode mnie możliwości zobaczenia każdego wiersza wyjścia z opakowanego programu, gdy tylko zostanie on wyprowadzony.

Pomyślałem, że po prostu uruchomię program za pomocą subprocess.Popen, użyję stdout=PIPE, a następnie przeczytam każdą linię, gdy się pojawi i odpowiednio z nią postąpię. Jednak gdy uruchomiłem następujący kod, dane wyjściowe wydawały się być gdzieś buforowane, powodując, że pojawiały się w dwóch fragmentach, wierszach od 1 do 332, a następnie od 333 do 439 (ostatnia linia danych wyjściowych)

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', '')

Po przyjrzeniu się nieco dokumentacji podprocesu odkryłem, że bufsizeparametr to Popen, więc próbowałem ustawić rozmiar bufora na 1 (bufor każdej linii) i 0 (brak bufora), ale żadna z wartości nie wydawała się zmieniać sposobu dostarczania linii.

W tym momencie zacząłem sięgać po słomki, więc napisałem następującą pętlę wyjściową:

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

ale uzyskał ten sam wynik.

Czy możliwe jest uzyskanie danych wyjściowych programu w czasie rzeczywistym programu wykonanego przy użyciu podprocesu? Czy jest jakaś inna opcja w Pythonie, która jest kompatybilna z przyszłością (nie exec*)?

Próbowałem tego iz jakiegoś powodu podczas kodu

for line in p.stdout:
  ...

bufory agresywnie, wariant

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

nie. Najwyraźniej jest to znany błąd: http://bugs.python.org/issue3907 (problem jest teraz „zamknięty” od 29 sierpnia 2018 r.)

p = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=1)
for line in iter(p.stdout.readline, b''):
    print line,
p.stdout.close()
p.wait()

Dane wyjściowe podprocesu można skierować bezpośrednio do strumieni. Uproszczony przykład:

subprocess.run(['ls'], stderr=sys.stderr, stdout=sys.stdout)

Możesz spróbować tego:

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

Jeśli użyjesz readline zamiast read, w niektórych przypadkach komunikat wejściowy nie zostanie wydrukowany. Wypróbuj to za pomocą polecenia, które wymaga wejścia inline i przekonaj się sam.

Żywo podproces stdin i stdout z asyncio w Pythonie blogu przez Kevin McCarthy pokazuje jak zrobić to z asyncio:

import asyncio
from asyncio.subprocess import PIPE
from asyncio import create_subprocess_exec


async def _read_stream(stream, callback):
    while True:
        line = await stream.readline()
        if line:
            callback(line)
        else:
            break


async def run(command):
    process = await create_subprocess_exec(
        *command, stdout=PIPE, stderr=PIPE
    )

    await asyncio.wait(
        [
            _read_stream(
                process.stdout,
                lambda x: print(
                    "STDOUT: {}".format(x.decode("UTF8"))
                ),
            ),
            _read_stream(
                process.stderr,
                lambda x: print(
                    "STDERR: {}".format(x.decode("UTF8"))
                ),
            ),
        ]
    )

    await process.wait()


async def main():
    await run("docker build -t my-docker-image:latest .")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

W Pythonie 3.x proces może się zawiesić, ponieważ dane wyjściowe są tablicą bajtów zamiast ciągu. Upewnij się, że dekodujesz go na ciąg.

Począwszy od Pythona 3.6 można to zrobić za pomocą parametru encodingw Popen Constructor . Kompletny przykład:

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)

Uwaga, że ten kod przekierowuje stderr do stdouti obsługuje błędy wyjściowych .

Rozwiązany problem z danymi wyjściowymi w czasie rzeczywistym: napotkałem podobny problem w Pythonie podczas przechwytywania danych wyjściowych w czasie rzeczywistym z programu C. Dodałem fflush(stdout);w moim kodzie C. U mnie to zadziałało. Oto kod.

Program C:

#include <stdio.h>
void main()
{
    int count = 1;
    while (1)
    {
        printf(" Count  %d\n", count++);
        fflush(stdout);
        sleep(1);
    }
}

Program w Pythonie:

#!/usr/bin/python

import os, sys
import subprocess


procExe = subprocess.Popen(".//count", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)

while procExe.poll() is None:
    line = procExe.stdout.readline()
    print("Print:" + line)

Wynik:

Print: Count  1
Print: Count  2
Print: Count  3

Jakiś czas temu napotkałem ten sam problem. Moim rozwiązaniem było porzucenie iteracji dla readmetody, która wróci natychmiast, nawet jeśli podproces nie zostanie zakończony itp.

W zależności od przypadku użycia możesz również chcieć wyłączyć buforowanie w samym podprocesie.

Jeśli podproces będzie procesem w Pythonie, możesz to zrobić przed wywołaniem:

os.environ["PYTHONUNBUFFERED"] = "1"

Lub alternatywnie przekaż to w envargumencie doPopen .

W przeciwnym razie, jeśli korzystasz z systemu Linux / Unix, możesz użyć stdbufnarzędzia. Np. Jak:

cmd = ["stdbuf", "-oL"] + cmd

Zobacz także tutaj ostdbuf lub inne opcje.

(Zobacz także tutaj, aby uzyskać tę samą odpowiedź).

Użyłem tego rozwiązania, aby uzyskać dane wyjściowe w czasie rzeczywistym w podprocesie. Ta pętla zostanie zatrzymana, gdy tylko proces się zakończy, pomijając potrzebę instrukcji break lub możliwej nieskończonej pętli.

sub_process = subprocess.Popen(my_command, close_fds=True, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

while sub_process.poll() is None:
    out = sub_process.stdout.read(1)
    sys.stdout.write(out)
    sys.stdout.flush()

Znalazłem tę funkcję „plug-and-play” tutaj . Działał jak urok!

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)

Możesz użyć iteratora dla każdego bajtu na wyjściu podprocesu. Pozwala to na aktualizację inline (linie kończące się '\ r' nadpisują poprzednią linię wyjściową) z podprocesu:

from subprocess import PIPE, Popen

command = ["my_command", "-my_arg"]

# Open pipe to subprocess
subprocess = Popen(command, stdout=PIPE, stderr=PIPE)


# read each byte of subprocess
while subprocess.poll() is None:
    for c in iter(lambda: subprocess.stdout.read(1) if subprocess.poll() is None else {}, b''):
        c = c.decode('ascii')
        sys.stdout.write(c)
sys.stdout.flush()

if subprocess.returncode != 0:
    raise Exception("The subprocess did not terminate correctly.")

Kompletne rozwiązanie:

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

To jest podstawowy szkielet, którego zawsze używam do tego. Ułatwia wdrażanie limitów czasu i jest w stanie poradzić sobie z nieuniknionymi zawieszonymi procesami.

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

Użycie pexpect z nieblokującymi readlines rozwiąże ten problem. Wynika to z faktu, że potoki są buforowane, a więc wyjście aplikacji jest buforowane przez potok, dlatego nie możesz dostać się do tego wyjścia, dopóki bufor się nie zapełni lub proces nie umrze.

(To rozwiązanie zostało przetestowane w Pythonie 2.7.15)
Wystarczy sys.stdout.flush () po każdym czytaniu / zapisie linii:

while proc.poll() is None:
    line = proc.stdout.readline()
    sys.stdout.write(line)
    # or print(line.strip()), you still need to force the flush.
    sys.stdout.flush()

Kilka odpowiedzi sugerujących python 3.x lub pthon 2.x, poniższy kod będzie działał w obu przypadkach.

 p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,)
    stdout = []
    while True:
        line = p.stdout.readline()
        if not isinstance(line, (str)):
            line = line.decode('utf-8')
        stdout.append(line)
        print (line)
        if (line == '' and p.poll() != None):
            break

jeśli chcesz tylko przekazać dziennik do konsoli w czasie rzeczywistym

Poniższy kod będzie działał dla obu

 p = subprocess.Popen(cmd,
                         shell=True,
                         cwd=work_dir,
                         bufsize=1,
                         stdin=subprocess.PIPE,
                         stderr=sys.stderr,
                         stdout=sys.stdout)