¿Cómo uso subprocess.Popen para conectar múltiples procesos por conductos?

¿Cómo ejecuto el siguiente comando de shell usando Python? subprocess ¿módulo?

echo "input data" | awk -f script.awk | sort > outfile.txt

Los datos de entrada provendrán de una cadena, por lo que en realidad no necesito echo. He llegado hasta aquí, ¿alguien puede explicar cómo consigo que se canalice? sort ¿también?

p_awk = subprocess.Popen(["awk","-f","script.awk"],
                          stdin=subprocess.PIPE,
                          stdout=file("outfile.txt", "w"))
p_awk.communicate( "input data" )

ACTUALIZAR: Tenga en cuenta que si bien la respuesta aceptada a continuación no responde realmente a la pregunta formulada, creo que S. Lott tiene razón y es mejor evitar tener que resolver ese problema en primer lugar.

Estarías un poco más feliz con lo siguiente.

import subprocess

awk_sort = subprocess.Popen( "awk -f script.awk | sort > outfile.txt",
    stdin=subprocess.PIPE, shell=True )
awk_sort.communicate( b"input data\n" )

Delega parte del trabajo al shell. Deje que conecte dos procesos con una canalización.

Sería mucho más feliz reescribiendo ‘script.awk’ en Python, eliminando awk y la canalización.

Editar. Algunas de las razones para sugerir que awk no está ayudando.

[There are too many reasons to respond via comments.]

1. Awk está agregando un paso sin valor significativo. No hay nada único en el procesamiento de awk que Python no maneje.

2. La canalización de awk a sort, para grandes conjuntos de datos, puede mejorar el tiempo de procesamiento transcurrido. Para conjuntos cortos de datos, no tiene un beneficio significativo. Una medida rápida de awk >file ; sort file y awk | sort revelará ayudas de concurrencia. Con sort, rara vez ayuda porque sort no es un filtro de un solo paso.

3. La simplicidad del procesamiento “Python to sort” (en lugar de “Python to awk to sort”) evita que se hagan exactamente el tipo de preguntas aquí.

4. Python, aunque tiene más palabras que awk, también es explícito donde awk tiene ciertas reglas implícitas que son opacas para los novatos y confusas para los no especialistas.

5. Awk (como el script de shell en sí) agrega otro lenguaje de programación. Si todo esto se puede hacer en un lenguaje (Python), eliminar el shell y la programación awk elimina dos lenguajes de programación, lo que permite que alguien se concentre en las partes de la tarea que generan valor.

En pocas palabras: awk no puede agregar un valor significativo. En este caso, awk es un costo neto; añadía tanta complejidad que era necesario hacer esta pregunta. Eliminar awk será una ganancia neta.

barra lateral Por qué construir una tubería (a | b) Es tan difícil.

Cuando el caparazón se enfrenta a a | b tiene que hacer lo siguiente.

1. Fork un proceso secundario del shell original. Esto eventualmente se convertirá en b.

2. Construya una tubería OS. (no es un subproceso de Python. PIPE) pero llama os.pipe() que devuelve dos nuevos descriptores de archivo que están conectados a través de un búfer común. En este punto, el proceso tiene stdin, stdout, stderr de su padre, más un archivo que será “a’s stdout” y “b’s stdin”.

3. Tenedor de un niño. El hijo reemplaza su salida estándar con la salida estándar de la nueva a. ejecutivo el a proceso.

4. El hijo b cierra reemplaza su stdin con el nuevo stdin de b. ejecutivo el b proceso.

5. El hijo b espera a que a se complete.

6. El padre está esperando que b se complete.

Creo que lo anterior se puede usar recursivamente para generar a | b | cpero tiene que poner implícitamente entre paréntesis las canalizaciones largas, tratándolas como si fueran a | (b | c).

Dado que Python tiene os.pipe(), os.exec() y os.fork()y puedes reemplazar sys.stdin y sys.stdout, hay una manera de hacer lo anterior en Python puro. De hecho, es posible que pueda resolver algunos atajos usando os.pipe() y subprocess.Popen.

Sin embargo, es más fácil delegar esa operación al shell.

import subprocess

some_string = b'input_data'

sort_out = open('outfile.txt', 'wb', 0)
sort_in = subprocess.Popen('sort', stdin=subprocess.PIPE, stdout=sort_out).stdin
subprocess.Popen(['awk', '-f', 'script.awk'], stdout=sort_in, 
                 stdin=subprocess.PIPE).communicate(some_string)

Para emular una canalización de shell:

from subprocess import check_call

check_call('echo "input data" | a | b > outfile.txt', shell=True)

sin invocar el shell (ver 17.1.4.2. Reemplazo de tubería de revestimiento):

#!/usr/bin/env python
from subprocess import Popen, PIPE

a = Popen(["a"], stdin=PIPE, stdout=PIPE)
with a.stdin:
    with a.stdout, open("outfile.txt", "wb") as outfile:
        b = Popen(["b"], stdin=a.stdout, stdout=outfile)
    a.stdin.write(b"input data")
statuses = [a.wait(), b.wait()] # both a.stdin/stdout are closed already

plumbum proporciona algo de azúcar de sintaxis:

#!/usr/bin/env python
from plumbum.cmd import a, b # magic

(a << "input data" | b > "outfile.txt")()

El análogo de:

#!/bin/sh
echo "input data" | awk -f script.awk | sort > outfile.txt

es:

#!/usr/bin/env python
from plumbum.cmd import awk, sort

(awk["-f", "script.awk"] << "input data" | sort > "outfile.txt")()

La respuesta aceptada es eludir el problema. aquí hay un fragmento que encadena la salida de múltiples procesos: tenga en cuenta que también imprime el comando de shell (algo) equivalente para que pueda ejecutarlo y asegurarse de que la salida sea correcta.

#!/usr/bin/env python3

from subprocess import Popen, PIPE

# cmd1 : dd if=/dev/zero bs=1m count=100
# cmd2 : gzip
# cmd3 : wc -c
cmd1 = ['dd', 'if=/dev/zero', 'bs=1M', 'count=100']
cmd2 = ['tee']
cmd3 = ['wc', '-c']
print(f"Shell style : {' '.join(cmd1)} | {' '.join(cmd2)} | {' '.join(cmd3)}")

p1 = Popen(cmd1, stdout=PIPE, stderr=PIPE) # stderr=PIPE optional, dd is chatty
p2 = Popen(cmd2, stdin=p1.stdout, stdout=PIPE)
p3 = Popen(cmd3, stdin=p2.stdout, stdout=PIPE)

print("Output from last process : " + (p3.communicate()[0]).decode())

# thoretically p1 and p2 may still be running, this ensures we are collecting their return codes
p1.wait()
p2.wait()
print("p1 return: ", p1.returncode)
print("p2 return: ", p2.returncode)
print("p3 return: ", p3.returncode)

http://www.python.org/doc/2.5.2/lib/node535.html cubrió esto bastante bien. ¿Hay alguna parte de esto que no entendiste?

Su programa sería bastante similar, pero el segundo Popen tendría stdout= a un archivo, y no necesitaría la salida de su .communicate().

Inspirado en la respuesta de @Cristian. Me encontré con el mismo problema, pero con un comando diferente. Así que estoy poniendo mi ejemplo probado, que creo que podría ser útil:

grep_proc = subprocess.Popen(["grep", "rabbitmq"],
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE)
subprocess.Popen(["ps", "aux"], stdout=grep_proc.stdin)
out, err = grep_proc.communicate()

Esto está probado.

Lo que se ha hecho

• Declarado perezoso grep ejecución con stdin desde pipe. Este comando se ejecutará en el ps ejecución del comando cuando la tubería se llenará con la salida estándar de ps.
• Llamado el comando principal ps con stdout dirigido a la tubería utilizada por el grep mando.
• Grep se comunicó para obtener la salida estándar de la tubería.

Me gusta de esta manera porque es una concepción de tubería natural envuelta suavemente con subprocess interfaces

Las respuestas anteriores perdieron un punto importante. Reemplazo de tubería de revestimiento es básicamente correcto, como lo señala geocar. Está casi suficiente para correr communicate en el último elemento de la tubería.

El problema restante es pasar el datos de entrada a la tubería. Con múltiples subprocesos, un simple communicate(input_data) en el último elemento no funciona, se cuelga para siempre. Debe crear una canalización y un niño manualmente como este:

import os
import subprocess

input = """\
input data
more input
""" * 10

rd, wr = os.pipe()
if os.fork() != 0: # parent
    os.close(wr)
else:              # child
    os.close(rd)
    os.write(wr, input)
    os.close(wr)
    exit()

p_awk = subprocess.Popen(["awk", "{ print $2; }"],
                         stdin=rd,
                         stdout=subprocess.PIPE)
p_sort = subprocess.Popen(["sort"], 
                          stdin=p_awk.stdout,
                          stdout=subprocess.PIPE)
p_awk.stdout.close()
out, err = p_sort.communicate()
print (out.rstrip())

Ahora el hijo proporciona la entrada a través de la canalización y el padre llama a communicar(), que funciona como se esperaba. Con este enfoque, puede crear canalizaciones largas arbitrarias sin recurrir a “delegar parte del trabajo al shell”. Desafortunadamente, el documentación de subprocesos no menciona esto.

Hay formas de lograr el mismo efecto sin tuberías:

from tempfile import TemporaryFile
tf = TemporaryFile()
tf.write(input)
tf.seek(0, 0)

Ahora usa stdin=tf por p_awk. Es cuestión de gustos lo que prefieras.

Lo anterior todavía no es 100% equivalente a las canalizaciones bash porque el manejo de la señal es diferente. Puede ver esto si agrega otro elemento de tubería que trunca la salida de sortp.ej head -n 10. Con el código de arriba, sort imprimirá un mensaje de error “Tubo roto” para stderr. No verá este mensaje cuando ejecute la misma canalización en el shell. (Sin embargo, esa es la única diferencia, el resultado en stdout es el mismo). La razón parece ser que Python Popen conjuntos SIG_IGN por SIGPIPEmientras que la cáscara lo deja en SIG_DFLy sortEl manejo de la señal de es diferente en estos dos casos.