Несколько каналов в подпроцессе

Я пытаюсь использовать Sailfish, который принимает несколько файлов fastq в качестве аргументов, в конвейере ruffus. Я выполняю Sailfish, используя модуль подпроцесса в python, но <() в вызове подпроцесса не работает, даже когда я устанавливаю shell=True.

Это команда, которую я хочу выполнить с помощью python: 

sailfish quant [options] -1 <(cat sample1a.fastq sample1b.fastq) -2 <(cat sample2a.fastq sample2b.fastq) -o [output_file]

Или (предпочтительно): 

sailfish quant [options] -1 <(gunzip sample1a.fastq.gz sample1b.fastq.gz) -2 <(gunzip sample2a.fastq.gz sample2b.fastq.gz) -o [output_file]

Обобщение: 

someprogram <(someprocess) <(someprocess)

Как бы я сделал это в python? Подпроцесс-это правильный подход?

2 6
python subprocess named-pipes

2 ответа:

Для эмуляции процесса подстановки bash : 

#!/usr/bin/env python
from subprocess import check_call

check_call('someprogram <(someprocess) <(anotherprocess)',
           shell=True, executable='/bin/bash')

В Python можно использовать именованные каналы: 

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

with named_pipes(n=2) as paths:
    someprogram = Popen(['someprogram'] + paths)
    processes = []
    for path, command in zip(paths, ['someprocess', 'anotherprocess']):
        with open(path, 'wb', 0) as pipe:
            processes.append(Popen(command, stdout=pipe, close_fds=True))
    for p in [someprogram] + processes:
        p.wait()

Где named_pipes(n) - это: 

import os
import shutil
import tempfile
from contextlib import contextmanager

@contextmanager
def named_pipes(n=1):
    dirname = tempfile.mkdtemp()
    try:
        paths = [os.path.join(dirname, 'named_pipe' + str(i)) for i in range(n)]
        for path in paths:
            os.mkfifo(path)
        yield paths
    finally:
        shutil.rmtree(dirname)
Другой и более предпочтительный способ (нет необходимости создавать именованную запись на диске) для реализации подстановки процесса bash-использовать имена файлов /dev/fd/N (если они доступны), как , предложенный @Dunes. На FreeBSD, fdescfs(5) (/dev/fd/#) создает записи для всех файловых дескрипторов, открытых процесс . Чтобы проверить доступность, выполните: 
$ test -r /dev/fd/3 3</dev/null && echo /dev/fd is available

Если это не удается; попробуйте создать символическую ссылку /dev/fd на proc(5) как это делается на некоторых Линиксах: 

$ ln -s /proc/self/fd /dev/fd

Вот /dev/fd-основанная реализация команды someprogram <(someprocess) <(anotherprocess) bash: 

#!/usr/bin/env python3
from contextlib import ExitStack
from subprocess import CalledProcessError, Popen, PIPE

def kill(process):
    if process.poll() is None: # still running
        process.kill()

with ExitStack() as stack: # for proper cleanup
    processes = []
    for command in [['someprocess'], ['anotherprocess']]:  # start child processes
        processes.append(stack.enter_context(Popen(command, stdout=PIPE)))
        stack.callback(kill, processes[-1]) # kill on someprogram exit

    fds = [p.stdout.fileno() for p in processes]
    someprogram = stack.enter_context(
        Popen(['someprogram'] + ['/dev/fd/%d' % fd for fd in fds], pass_fds=fds))
    for p in processes: # close pipes in the parent
        p.stdout.close()
# exit stack: wait for processes
if someprogram.returncode != 0: # errors shouldn't go unnoticed
   raise CalledProcessError(someprogram.returncode, someprogram.args)

Примечание: На моей машине Ubuntu код subprocess работает только в Python 3.4+, несмотря на то, что pass_fds доступен начиная с Python 3.2.

8

В то время как Дж.Ф. Себастьян дал ответ, используя именованные каналы, это можно сделать с анонимными каналами.

import shlex
from subprocess import Popen, PIPE

inputcmd0 = "zcat hello.gz" # gzipped file containing "hello"
inputcmd1 = "zcat world.gz" # gzipped file containing "world"

def get_filename(file_):
    return "/dev/fd/{}".format(file_.fileno())

def get_stdout_fds(*processes):
    return tuple(p.stdout.fileno() for p in processes)

# setup producer processes
inputproc0 = Popen(shlex.split(inputcmd0), stdout=PIPE)
inputproc1 = Popen(shlex.split(inputcmd1), stdout=PIPE)

# setup consumer process
# pass input processes pipes by "filename" eg. /dev/fd/5
cmd = "cat {file0} {file1}".format(file0=get_filename(inputproc0.stdout), 
    file1=get_filename(inputproc1.stdout))
print("command is:", cmd)
# pass_fds argument tells Popen to let the child process inherit the pipe's fds
someprogram = Popen(shlex.split(cmd), stdout=PIPE, 
    pass_fds=get_stdout_fds(inputproc0, inputproc1))

output, error = someprogram.communicate()

for p in [inputproc0, inputproc1, someprogram]:
    p.wait()

assert output == b"hello\nworld\n"
2