español - Python lee desde el subproceso stdout y stderr por separado mientras conserva el orden

subprocess python 3 example

Question

Tengo un subproceso de Python del que estoy tratando de leer la salida y las secuencias de error. Actualmente lo tengo funcionando, pero solo puedo leer desde stderr después de haber terminado de leer desde stdout . Así es como se ve:

process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout_iterator = iter(process.stdout.readline, b"")
stderr_iterator = iter(process.stderr.readline, b"")
for line in stdout_iterator:
# Do stuff with line
print line
for line in stderr_iterator:
# Do stuff with line
print line

Como puede ver, el bucle stderr for no puede comenzar hasta que se complete el bucle stdout . ¿Cómo puedo modificar esto para poder leer de ambos en el orden correcto en que aparecen las líneas?

Para aclarar: todavía necesito saber si una línea vino de stdout o stderr porque se tratarán de manera diferente en mi código.

Answer 1

Aquí hay una solución basada en selectors , pero que conserva el orden y transmite caracteres de longitud variable (incluso caracteres únicos).

El truco es usar read1() , en lugar de read() .

import selectors
import subprocess
import sys
p = subprocess.Popen(
"python random_out.py", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
sel = selectors.DefaultSelector()
sel.register(p.stdout, selectors.EVENT_READ)
sel.register(p.stderr, selectors.EVENT_READ)
while True:
for key, _ in sel.select():
data = key.fileobj.read1().decode()
if not data:
exit()
if key.fileobj is p.stdout:
print(data, end="")
else:
print(data, end="", file=sys.stderr)

Si desea un programa de prueba, use esto.

import sys
from time import sleep
for i in range(10):
print(f" x{i} ", file=sys.stderr, end="")
sleep(0.1)
print(f" y{i} ", end="")
sleep(0.1)

Answer 2

El código en su pregunta puede llegar a un punto muerto si el proceso hijo produce suficiente salida en stderr (~ 100 KB en mi máquina Linux).

Hay un método de communicate() que permite leer tanto desde stdout como desde stderr por separado:

from subprocess import Popen, PIPE
process = Popen(command, stdout=PIPE, stderr=PIPE)
output, err = process.communicate()

Si necesita leer las transmisiones mientras el proceso secundario aún se está ejecutando, la solución portátil es usar hilos (no probados):

from subprocess import Popen, PIPE
from threading import Thread
from Queue import Queue # Python 2
def reader(pipe, queue):
try:
with pipe:
for line in iter(pipe.readline, b''''):
queue.put((pipe, line))
finally:
queue.put(None)
process = Popen(command, stdout=PIPE, stderr=PIPE, bufsize=1)
q = Queue()
Thread(target=reader, args=[process.stdout, q]).start()
Thread(target=reader, args=[process.stderr, q]).start()
for _ in range(2):
for source, line in iter(q.get, None):
print "%s: %s" % (source, line),

Ver:

• Python: lea la entrada de transmisión de subprocess.communicate ()
• Lectura sin bloqueo en un subproceso.PIPE en python
• ¿El subproceso de Python obtiene la salida de los niños al archivo y al terminal?

Answer 3

El orden en que un proceso escribe datos en diferentes tuberías se pierde después de la escritura.

No hay forma de saber si stdout se ha escrito antes de stderr.

Puede intentar leer los datos simultáneamente de varios descriptores de archivos de manera no bloqueadora tan pronto como los datos estén disponibles, pero esto solo minimizaría la probabilidad de que el pedido sea incorrecto.

Este programa debería demostrar esto:

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import os
import select
import subprocess
testapps={
''slow'': ''''''
import os
import time
os.write(1, ''aaa'')
time.sleep(0.01)
os.write(2, ''bbb'')
time.sleep(0.01)
os.write(1, ''ccc'')
'''''',
''fast'': ''''''
import os
os.write(1, ''aaa'')
os.write(2, ''bbb'')
os.write(1, ''ccc'')
'''''',
''fast2'': ''''''
import os
os.write(1, ''aaa'')
os.write(2, ''bbbbbbbbbbbbbbb'')
os.write(1, ''ccc'')
''''''
}
def readfds(fds, maxread):
while True:
fdsin, _, _ = select.select(fds,[],[])
for fd in fdsin:
s = os.read(fd, maxread)
if len(s) == 0:
fds.remove(fd)
continue
yield fd, s
if fds == []:
break
def readfromapp(app, rounds=10, maxread=1024):
f=open(''testapp.py'', ''w'')
f.write(testapps[app])
f.close()
results={}
for i in range(0, rounds):
p = subprocess.Popen([''python'', ''testapp.py''], stdout=subprocess.PIPE
, stderr=subprocess.PIPE)
data=''''
for (fd, s) in readfds([p.stdout.fileno(), p.stderr.fileno()], maxread):
data = data + s
results[data] = results[data] + 1 if data in results else 1
print ''running %i rounds %s with maxread=%i'' % (rounds, app, maxread)
results = sorted(results.items(), key=lambda (k,v): k, reverse=False)
for data, count in results:
print ''%03i x %s'' % (count, data)
print
print "=> if output is produced slowly this should work as whished"
print " and should return: aaabbbccc"
readfromapp(''slow'', rounds=100, maxread=1024)
print
print "=> now mostly aaacccbbb is returnd, not as it should be"
readfromapp(''fast'', rounds=100, maxread=1024)
print
print "=> you could try to read data one by one, and return"
print " e.g. a whole line only when LF is read"
print " (b''s should be finished before c''s)"
readfromapp(''fast'', rounds=100, maxread=1)
print
print "=> but even this won''t work ..."
readfromapp(''fast2'', rounds=100, maxread=1)

y genera algo como esto:

=> if output is produced slowly this should work as whished
and should return: aaabbbccc
running 100 rounds slow with maxread=1024
100 x aaabbbccc
=> now mostly aaacccbbb is returnd, not as it should be
running 100 rounds fast with maxread=1024
006 x aaabbbccc
094 x aaacccbbb
=> you could try to read data one by one, and return
e.g. a whole line only when LF is read
(b''s should be finished before c''s)
running 100 rounds fast with maxread=1
003 x aaabbbccc
003 x aababcbcc
094 x abababccc
=> but even this won''t work ...
running 100 rounds fast2 with maxread=1
003 x aaabbbbbbbbbbbbbbbccc
001 x aaacbcbcbbbbbbbbbbbbb
008 x aababcbcbcbbbbbbbbbbb
088 x abababcbcbcbbbbbbbbbb

Answer 4

Escribí algo para hacer esto hace mucho tiempo . Todavía no lo he portado a Python 3, pero no debería ser demasiado difícil (¡se aceptan parches!)

Si lo ejecuta de forma independiente, verá muchas de las diferentes opciones. En cualquier caso, le permite distinguir stdout de stderr.

Answer 5

Esto funciona para mí (en Windows): https://github.com/waszil/subpiper

from subpiper import subpiper
def my_stdout_callback(line: str):
print(f''STDOUT: {line}'')
def my_stderr_callback(line: str):
print(f''STDERR: {line}'')
my_additional_path_list = [r''c:/important_location'']
retcode = subpiper(cmd=''echo magic'',
stdout_callback=my_stdout_callback,
stderr_callback=my_stderr_callback,
add_path_list=my_additional_path_list)

Answer 6

Sé que esta pregunta es muy antigua, pero esta respuesta puede ayudar a otros que se topan con esta página para buscar una solución para una situación similar, así que la estoy publicando de todos modos.

He creado un fragmento de python simple que fusionará cualquier cantidad de tuberías en una sola. Por supuesto, como se indicó anteriormente, el orden no se puede garantizar, pero esto es lo más cercano que creo que puede obtener en Python.

Genera un hilo para cada una de las tuberías, las lee línea por línea y las coloca en una Cola (que es FIFO). El hilo principal recorre la cola, produciendo cada línea.

import threading, queue
def merge_pipes(**named_pipes):
r''''''
Merges multiple pipes from subprocess.Popen (maybe other sources as well).
The keyword argument keys will be used in the output to identify the source
of the line.
Example:
p = subprocess.Popen([''some'', ''call''],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
outputs = {''out'': log.info, ''err'': log.warn}
for name, line in merge_pipes(out=p.stdout, err=p.stderr):
outputs[name](line)
This will output stdout to the info logger, and stderr to the warning logger
''''''
# Constants. Could also be placed outside of the method. I just put them here
# so the method is fully self-contained
PIPE_OPENED=1
PIPE_OUTPUT=2
PIPE_CLOSED=3
# Create a queue where the pipes will be read into
output = queue.Queue()
# This method is the run body for the threads that are instatiated below
# This could be easily rewritten to be outside of the merge_pipes method,
# but to make it fully self-contained I put it here
def pipe_reader(name, pipe):
r"""
reads a single pipe into the queue
"""
output.put( ( PIPE_OPENED, name, ) )
try:
for line in iter(pipe.readline,''''):
output.put( ( PIPE_OUTPUT, name, line.rstrip(), ) )
finally:
output.put( ( PIPE_CLOSED, name, ) )
# Start a reader for each pipe
for name, pipe in named_pipes.items():
t=threading.Thread(target=pipe_reader, args=(name, pipe, ))
t.daemon = True
t.start()
# Use a counter to determine how many pipes are left open.
# If all are closed, we can return
pipe_count = 0
# Read the queue in order, blocking if there''s no data
for data in iter(output.get,''''):
code=data[0]
if code == PIPE_OPENED:
pipe_count += 1
elif code == PIPE_CLOSED:
pipe_count -= 1
elif code == PIPE_OUTPUT:
yield data[1:]
if pipe_count == 0:
return

Answer 7

Según el documento de Python

Popen.stdout Si el argumento stdout era PIPE, este atributo es un objeto de archivo que proporciona resultados del proceso secundario. De lo contrario, es Ninguno.

Popen.stderr Si el argumento stderr era PIPE, este atributo es un objeto de archivo que proporciona una salida de error del proceso secundario. De lo contrario, es Ninguno.

Debajo de la muestra puedes hacer lo que quieras

prueba.py

print "I''m stdout"
raise Exception("I''m Error")

impresora.py

import subprocess
p = subprocess.Popen([''python'', ''test.py''], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print "Normal"
std_lines = p.stdout.readlines()
for line in std_lines:
print line.rstrip()
print "Error"
stderr_lines = p.stderr.readlines()
for line in stderr_lines:
print line.rstrip()

Salida:

Normal
I''m stdout
Error
Traceback (most recent call last):
File "test.py", line 3, in <module>
raise Exception("I''m Error")
Exception: I''m Error