Nhiều đường ống trong quy trình con

Question

Tôi đang cố gắng sử dụng Sailfish, trong đó có nhiều tệp fastq làm đối số, trong một đường ống ruffus. Tôi thực hiện Sailfish sử dụng mô-đun subprocess trong python, nhưng <() trong cuộc gọi subprocess không hoạt động ngay cả khi tôi đặt shell=True.

Đây là lệnh tôi muốn thực hiện sử dụng python:

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

hoặc (tốt nhất):

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

Một khái quát:

someprogram <(someprocess) <(someprocess) 

Làm thế nào tôi sẽ đi về làm điều này trong python? Là subprocess các phương pháp tiếp cận đúng?

Answer 1

Để cạnh tranh với bash process substitution:

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

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

Trong Python, bạn có thể sử dụng tên Ống:

#!/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() 

nơi named_pipes(n) là:

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) 

---

Một và nhiều hơn nữa trước cách có thể lên men (không cần phải tạo một mục có tên trên đĩa) để thực hiện thay thế tiến trình bash là sử dụng tên tệp /dev/fd/N (nếu chúng có sẵn) là suggested by @Dunes. Trên FreeBSD, fdescfs(5) (/dev/fd/#) creates entries for all file descriptors opened by the process. Để kiểm tra tính khả dụng, hãy chạy:

$ test -r /dev/fd/3 3</dev/null && echo /dev/fd is available 

Nếu không thành công; cố gắng liên kết mềm /dev/fd để proc(5) vì nó được thực hiện trên một số Linuxes:

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

Dưới đây là /dev/fd thực hiện dựa trên các lệnh 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) 

Lưu ý: trên máy tính Ubuntu của tôi, mã subprocess chỉ hoạt động trong Python 3.4+, mặc dù pass_fds khả dụng kể từ Python 3.2.

Answer 2

Trong khi J.F. Sebastian đã cung cấp câu trả lời bằng cách sử dụng các đường ống có tên, có thể thực hiện điều này bằng các đường ống ẩn danh.

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"