Ciągłe drukowanie wyniku podprocesu podczas działania procesu

Aby uruchomić programy z moich skryptów Python, używam następującej metody:

def execute(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    output = process.communicate()[0]
    exitCode = process.returncode

    if (exitCode == 0):
        return output
    else:
        raise ProcessException(command, exitCode, output)

Więc kiedy uruchamiam taki proces Process.execute("mvn clean install"), mój program czeka, aż proces się zakończy, i dopiero wtedy otrzymuję pełne wyjście mojego programu. Jest to denerwujące, jeśli uruchamiam proces, który zajmuje trochę czasu.

Czy mogę pozwolić mojemu programowi zapisywać dane wyjściowe procesu wiersz po wierszu, odpytując dane wyjściowe procesu, zanim zakończy ono pętlę lub coś takiego?

** [EDYCJA] Przepraszam, nie szukałem zbyt dobrze przed opublikowaniem tego pytania. Wątek jest właściwie kluczem. Znaleziono tutaj przykład, który pokazuje, jak to zrobić: ** Python Subprocess.Popen z wątku

Można użyć ITER przetwarzać linie jak najszybciej wyjść sterujących nich lines = iter(fd.readline, ""). Oto pełny przykład pokazujący typowy przypadek użycia (podziękowania dla @jfs za pomoc):

from __future__ import print_function # Only Python 2.x
import subprocess

def execute(cmd):
    popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True)
    for stdout_line in iter(popen.stdout.readline, ""):
        yield stdout_line 
    popen.stdout.close()
    return_code = popen.wait()
    if return_code:
        raise subprocess.CalledProcessError(return_code, cmd)

# Example
for path in execute(["locate", "a"]):
    print(path, end="")

Ok, udało mi się go rozwiązać bez wątków (doceniamy wszelkie sugestie, dlaczego użycie wątków byłoby lepsze) za pomocą fragmentu tego pytania Przechwytywanie standardowego przebiegu podprocesu podczas jego działania

def execute(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

    # Poll process for new output until finished
    while True:
        nextline = process.stdout.readline()
        if nextline == '' and process.poll() is not None:
            break
        sys.stdout.write(nextline)
        sys.stdout.flush()

    output = process.communicate()[0]
    exitCode = process.returncode

    if (exitCode == 0):
        return output
    else:
        raise ProcessException(command, exitCode, output)

Aby wydrukować wynik podprocesu wiersz po wierszu, jak tylko bufor stdout zostanie opróżniony w Pythonie 3:

from subprocess import Popen, PIPE, CalledProcessError

with Popen(cmd, stdout=PIPE, bufsize=1, universal_newlines=True) as p:
    for line in p.stdout:
        print(line, end='') # process line here

if p.returncode != 0:
    raise CalledProcessError(p.returncode, p.args)

Uwaga: nie potrzebujesz p.poll()- pętla kończy się po osiągnięciu eof. I nie potrzebujesz iter(p.stdout.readline, '')- błąd odczytu z wyprzedzeniem został naprawiony w Pythonie 3.

Zobacz także Python: odczyt danych strumieniowych z subprocess.communicate () .

Jest naprawdę bardzo prosty sposób, aby to zrobić, gdy chcesz po prostu wydrukować wynik:

import subprocess
import sys

def execute(command):
    subprocess.check_call(command, stdout=sys.stdout, stderr=subprocess.STDOUT)

Tutaj po prostu wskazujemy podproces na nasze standardowe wyjście i wykorzystujemy istniejące API powodzenia lub wyjątku.

@tokland

wypróbowałem kod i poprawiłem go dla 3.4, a Windows dir.cmd to proste polecenie dir, zapisane jako plik cmd

import subprocess
c = "dir.cmd"

def execute(command):
    popen = subprocess.Popen(command, stdout=subprocess.PIPE,bufsize=1)
    lines_iterator = iter(popen.stdout.readline, b"")
    while popen.poll() is None:
        for line in lines_iterator:
            nline = line.rstrip()
            print(nline.decode("latin"), end = "\r\n",flush =True) # yield line

execute(c)

Jeśli ktoś chce czytać z obu stdouti stderrjednocześnie używać wątków, oto co wymyśliłem:

import threading
import subprocess
import Queue

class AsyncLineReader(threading.Thread):
    def __init__(self, fd, outputQueue):
        threading.Thread.__init__(self)

        assert isinstance(outputQueue, Queue.Queue)
        assert callable(fd.readline)

        self.fd = fd
        self.outputQueue = outputQueue

    def run(self):
        map(self.outputQueue.put, iter(self.fd.readline, ''))

    def eof(self):
        return not self.is_alive() and self.outputQueue.empty()

    @classmethod
    def getForFd(cls, fd, start=True):
        queue = Queue.Queue()
        reader = cls(fd, queue)

        if start:
            reader.start()

        return reader, queue


process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(stdoutReader, stdoutQueue) = AsyncLineReader.getForFd(process.stdout)
(stderrReader, stderrQueue) = AsyncLineReader.getForFd(process.stderr)

# Keep checking queues until there is no more output.
while not stdoutReader.eof() or not stderrReader.eof():
   # Process all available lines from the stdout Queue.
   while not stdoutQueue.empty():
       line = stdoutQueue.get()
       print 'Received stdout: ' + repr(line)

       # Do stuff with stdout line.

   # Process all available lines from the stderr Queue.
   while not stderrQueue.empty():
       line = stderrQueue.get()
       print 'Received stderr: ' + repr(line)

       # Do stuff with stderr line.

   # Sleep for a short time to avoid excessive CPU use while waiting for data.
   sleep(0.05)

print "Waiting for async readers to finish..."
stdoutReader.join()
stderrReader.join()

# Close subprocess' file descriptors.
process.stdout.close()
process.stderr.close()

print "Waiting for process to exit..."
returnCode = process.wait()

if returnCode != 0:
   raise subprocess.CalledProcessError(returnCode, command)

Po prostu chciałem się tym podzielić, ponieważ skończyłem na tym pytaniu, próbując zrobić coś podobnego, ale żadna z odpowiedzi nie rozwiązała mojego problemu. Mam nadzieję, że to komuś pomaga!

Pamiętaj, że w moim przypadku proces zewnętrzny zabija proces, który my Popen().

Dla każdego, kto próbuje odpowiedzieć na to pytanie, aby uzyskać stdout ze skryptu Python, zauważ, że Python buforuje swoje standardowe wyjście, dlatego może minąć trochę czasu, zanim zobaczysz standardowe wyjście.

Można to naprawić, dodając następujące elementy po każdym zapisie standardowym w skrypcie docelowym:

sys.stdout.flush()

W Pythonie> = 3.5 używanie subprocess.rundziała dla mnie:

import subprocess

cmd = 'echo foo; sleep 1; echo foo; sleep 2; echo foo'
subprocess.run(cmd, shell=True)

(pobieranie danych wyjściowych podczas wykonywania również działa bez shell=True) https://docs.python.org/3/library/subprocess.html#subprocess.run

Aby odpowiedzieć na pierwotne pytanie, najlepszym sposobem IMO jest po prostu przekierowanie podprocesu stdoutbezpośrednio do twojego programu stdout(opcjonalnie to samo można zrobić dla stderr, jak w przykładzie poniżej)

p = Popen(cmd, stdout=sys.stdout, stderr=sys.stderr)
p.communicate()

Ten PoC stale odczytuje dane wyjściowe z procesu i można uzyskać do nich dostęp w razie potrzeby. Zachowywany jest tylko ostatni wynik, wszystkie inne dane wyjściowe są odrzucane, co zapobiega wzrostowi pamięci PIPE:

import subprocess
import time
import threading
import Queue


class FlushPipe(object):
    def __init__(self):
        self.command = ['python', './print_date.py']
        self.process = None
        self.process_output = Queue.LifoQueue(0)
        self.capture_output = threading.Thread(target=self.output_reader)

    def output_reader(self):
        for line in iter(self.process.stdout.readline, b''):
            self.process_output.put_nowait(line)

    def start_process(self):
        self.process = subprocess.Popen(self.command,
                                        stdout=subprocess.PIPE)
        self.capture_output.start()

    def get_output_for_processing(self):
        line = self.process_output.get()
        print ">>>" + line


if __name__ == "__main__":
    flush_pipe = FlushPipe()
    flush_pipe.start_process()

    now = time.time()
    while time.time() - now < 10:
        flush_pipe.get_output_for_processing()
        time.sleep(2.5)

    flush_pipe.capture_output.join(timeout=0.001)
    flush_pipe.process.kill()

print_date.py

#!/usr/bin/env python
import time

if __name__ == "__main__":
    while True:
        print str(time.time())
        time.sleep(0.01)

Dane wyjściowe: Możesz wyraźnie zobaczyć, że dane wyjściowe pochodzą tylko z ~ 2,5 s odstępu między nimi.

>>>1520535158.51
>>>1520535161.01
>>>1520535163.51
>>>1520535166.01

Działa to przynajmniej w Python3.4

import subprocess

process = subprocess.Popen(cmd_list, stdout=subprocess.PIPE)
for line in process.stdout:
    print(line.decode().strip())

Żadna z odpowiedzi tutaj nie zaspokoiła wszystkich moich potrzeb.

1. Brak wątków dla standardowego wyjścia (bez kolejek itp.)
2. Nie blokuje, ponieważ muszę sprawdzić, czy dzieją się inne rzeczy
3. Użyj PIPE, ponieważ potrzebowałem zrobić wiele rzeczy, np. Strumień wyjściowy, zapisz do pliku dziennika i zwróć kopię ciągu danych wyjściowych.

Małe tło: używam ThreadPoolExecutor do zarządzania pulą wątków, z których każdy uruchamia podproces i uruchamia je jednocześnie. (W Python2.7, ale powinno to również działać w nowszych wersjach 3.x). Nie chcę używać wątków tylko do zbierania danych wyjściowych, ponieważ chcę jak najwięcej dostępnych dla innych rzeczy (pula 20 procesów użyłaby 40 wątków tylko do uruchomienia; 1 dla wątku procesu i 1 dla standardowego wyjścia ... i więcej, jeśli chcesz stderr, tak myślę)

Usuwam tutaj wiele wyjątków i takich, więc jest to oparte na kodzie, który działa w środowisku produkcyjnym. Mam nadzieję, że nie zrujnowałem go w kopii i wkleiłem. Również opinie bardzo mile widziane!

import time
import fcntl
import subprocess
import time

proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

# Make stdout non-blocking when using read/readline
proc_stdout = proc.stdout
fl = fcntl.fcntl(proc_stdout, fcntl.F_GETFL)
fcntl.fcntl(proc_stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)

def handle_stdout(proc_stream, my_buffer, echo_streams=True, log_file=None):
    """A little inline function to handle the stdout business. """
    # fcntl makes readline non-blocking so it raises an IOError when empty
    try:
        for s in iter(proc_stream.readline, ''):   # replace '' with b'' for Python 3
            my_buffer.append(s)

            if echo_streams:
                sys.stdout.write(s)

            if log_file:
                log_file.write(s)
    except IOError:
        pass

# The main loop while subprocess is running
stdout_parts = []
while proc.poll() is None:
    handle_stdout(proc_stdout, stdout_parts)

    # ...Check for other things here...
    # For example, check a multiprocessor.Value('b') to proc.kill()

    time.sleep(0.01)

# Not sure if this is needed, but run it again just to be sure we got it all?
handle_stdout(proc_stdout, stdout_parts)

stdout_str = "".join(stdout_parts)  # Just to demo

Jestem pewien, że dodano tutaj koszty ogólne, ale w moim przypadku nie stanowi to problemu. Funkcjonalnie robi to, czego potrzebuję. Jedyną rzeczą, której nie rozwiązałem, jest to, dlaczego działa to idealnie w przypadku komunikatów w dzienniku, ale widzę, że niektóre printwiadomości pojawiają się później i jednocześnie.

W Pythonie 3.6 użyłem tego:

import subprocess

cmd = "command"
output = subprocess.call(cmd, shell=True)
print(process)