потоковая передача данных в команду с подпроцессом.Popen

Question

Мне часто приходится сортировать коллекцию файлов, содержащих заголовки. Поскольку сортировка зависит от содержимого заголовка, этот пример использования более сложный, чем аналогичные вопросы (например, Is there a way to ignore header lines in a UNIX sort?).

Я надеялся использовать Python для чтения файлов, вывода заголовка первого файла, а затем передать хвосты в сортировку. Я попытался это как доказательство концепции:

#!/usr/bin/env python 

import io 
import subprocess 
import sys 

header_printed = False 

sorter = subprocess.Popen(['sort'], stdin=subprocess.PIPE) 

for f in sys.argv[1:]: 
    fd = io.open(f,'r') 
    line = fd.readline() 
    if not header_printed: 
     print(line) 
     header_printed = True 
    sorter.communicate(line) 

Когда называется header-sort fileA fileB с FiLea и FILEB, содержащие линии, как

c float int 
Y 0.557946  413 
F 0.501935  852 
F 0.768102  709 

я получаю:

# sort file 1 
Traceback (most recent call last): 
    File "./archive/bin/pipetest", line 17, in <module> 
    sorter.communicate(line) 
    File "/usr/lib/python2.7/subprocess.py", line 785, in communicate 
    self.stdin.write(input) 
ValueError: I/O operation on closed file 

Проблема заключается в том связь берет строку, и труба закрывается после записи. Это означает, что содержимое должно быть полностью прочитано в памяти. связь не принимает генератор (я пробовал).

Еще более простой демонстрацией этого является:

>>> import subprocess 
>>> p = subprocess.Popen(['tr', 'a-z', 'A-Z'], stdin=subprocess.PIPE) 
>>> p.communicate('hello') 
HELLO(None, None) 
>>> p.communicate('world') 
Traceback (most recent call last): 
    File "<ipython-input-14-d6873fd0f66a>", line 1, in <module> 
    p.communicate('world') 
    File "/usr/lib/python2.7/subprocess.py", line 785, in communicate 
    self.stdin.write(input) 
ValueError: I/O operation on closed file 

Итак, вопрос в том, что правильный путь (с Popen или иным образом) для потоковой передачи данных в трубу в Python?

Answer 1

Просто напишите в трубу непосредственно:

#!/usr/bin/env python2 
import fileinput 
import subprocess 

process = subprocess.Popen(['sort'], stdin=subprocess.PIPE) 
with process.stdin as pipe, fileinput.FileInput() as file: 
    for line in file: 
     if file.isfirstline(): # print header 
      print line, 
     else: # pipe tails 
      pipe.write(line) 
process.wait() 

Answer 2

Вы можете использовать запись/чтение от stdin и stdout, однако в зависимости от вашего подпроцесса вам нужен «промывочный механизм» для подпроцесса для обработки вашего ввода. Нижеприведенный код работает для первой части, но поскольку он закрывает stdin, он также убивает подпроцесс. Если вы измените его на flush(), или если вы можете добавить некоторые завершающие символы для подталкивания вашего подпроцесса, вы можете его использовать. Еще бы, я бы рекомендовал взглянуть на Multithreading in Python, особенно pipes.

p=subprocess.Popen(['tr','a-z','A-Z'],stdin=subprocess.PIPE,stdout=subprocess.PIPE) 
p.stdin.write("hello\n") 
p.stdin.close() 
p.stdout.readline() 
'HELLO\n' 

Answer 3

Для вашего конкретного случая, если вы только прошли subprocess.PIPE для одной стандартной ручки (в вашем случае, stdin), то в вашем примере, вы можете смело называть sorter.stdin.write(line) снова и снова. Когда вы закончите запись вывода, вызовите sorter.stdin.close(), так что sort знает, что вход завершен, и он может выполнять фактические операции сортировки и вывода (sorter.communicate() без аргументов, вероятно, тоже будет работать, в противном случае после закрытия stdin вы, вероятно, захотите позвонить sorter.wait(), чтобы закончить).

Если вам нужно иметь дело с более чем одной стандартной рукояткой, подходящей по порядку, то есть правильный путь: threading с выделенной нитью для каждой трубы, которая должна обрабатываться за пределами первой (относительно простая в концепции, но в супертяжелом и вводит все головные боли потоковой передачи) или с использованием модуля select (или в Python 3.4+, selectors), что довольно сложно получить, но может (при некоторых обстоятельствах) быть более эффективным. Наконец, есть creating temporary files for output, поэтому вы можете напрямую писать в процесс stdin, пока процесс записывается в файл (и поэтому не будет блокироваться); вы можете прочитать файл в свободное время (обратите внимание, что подпроцесс не обязательно будет очищать свои собственные выходные буферы до тех пор, пока он не выйдет, поэтому выход может не прибыть незамедлительно в ответ на ваш вход до тех пор, пока дополнительные входы и выходы не будут заполнены и не будут сброшены буфер).

subprocess.Popen «s .communicate() метода использует либо тему или select сам модуль примитивов (в зависимости от поддержки ОС, реализация находится под various _communicate methods here) всякий раз, когда вы передаете subprocess.PIPE более чем одной из стандартных ручек; это то, как вы должны это делать.