Jak dołączyć do stdout dwóch podprocesów i rurę do standardowego wejścia nowych podproces w Pythonie

Question

Powiedzmy miałem następujące polecenie działa z powłoki

{ 
samtools view -HS header.sam;   # command1 
samtools view input.bam 1:1-50000000; # command2 
} | samtools view -bS - > output.bam # command3 

Dla tych z Was, którzy nie są zaznajomieni z widokiem samtools (Ponieważ jest to stackoverflow). W istocie chodzi o stworzenie nowego pliku bam, który ma nowy nagłówek. Pliki bam są zwykle dużymi skompresowanymi plikami, więc nawet przechodzenie przez ten plik w niektórych przypadkach może być czasochłonne. Jednym z alternatywnych podejść byłoby poddanie komendy2, a następnie użycie repozytorium samtools do przełączenia nagłówka. To przechodzi przez duży plik dwa razy. Powyższe polecenie przechodzi przez bam jednorazowo, co jest dobre dla większych plików bam (stają się większe niż 20 GB, nawet gdy są skompresowane - WGS).

Moje pytanie brzmi: jak zaimplementować polecenia tego typu w pythonie za pomocą podprocesu.

mam następujące:

fh_bam = open('output.bam', 'w') 
params_0 = [ "samtools", "view", "-HS", "header.sam" ] 
params_1 = [ "samtools", "view", "input.bam", "1:1-50000000"] 
params_2 = [ "samtools", "view", "-bS", "-" ] 
sub_0 = subprocess.Popen(params_0, stderr=subprocess.PIPE, stdout=subprocess.PIPE) 
sub_1 = subprocess.Popen(params_1, stderr=subprocess.PIPE, stdout=subprocess.PIPE) 
### SOMEHOW APPEND sub_1.stdout to sub_0.stdout 
sub_2 = subprocess.Popen(params_2, stdin=appended.stdout, stdout=fh_bam) 

Każda pomoc jest mile widziana. Dzięki.

Answer 1

Zakładam, że łączenie wyników z pierwszych dwóch podprocesów w pamięci nie jest możliwe ze względu na rozmiar plików. Sugerowałbym zawijanie wyników z dwóch pierwszych podprocesów w pliku takim jak. Wygląda na to, że będziesz potrzebować tylko metody read, ponieważ popen będzie czytał tylko z pliku stdin, nie szukając ani nie pisząc. Kod poniżej zakłada się, że powrót pusty łańcuch z odczytanych wystarczy podać strumień ma EOF

class concat(object): 
    def __init__(self, f1, f2): 
     self.f1 = f1 
     self.f2 = f2 

    def read(self, *args): 
     ret = self.f1.read(*args) 
     if ret == '': 
      ret = self.f2.read(*args) 
     return ret 

fh_bam = open('output.bam', 'w') 
params_0 = [ "samtools", "view", "-HS", "header.sam" ] 
params_1 = [ "samtools", "view", "input.bam", "1:1-50000000"] 
params_2 = [ "samtools", "view", "-bS", "-" ] 
sub_0 = subprocess.Popen(params_0, stderr=subprocess.PIPE, stdout=subprocess.PIPE) 
sub_1 = subprocess.Popen(params_1, stderr=subprocess.PIPE, stdout=subprocess.PIPE) 
### Somehow Append sub_1.stdout to sub_0.stdout 
sub_2 = subprocess.Popen(params_2, stdin=concat(sub_0.stdout, sub_1.stdout), stdout=fh_bam) 

celu wyjaśnienia, f1.read blokować i jedynie powrócić '', kiedy rura jest zamknięta/EOF'd. concat.read próbuje odczytywać tylko od f2, więc dane wyjściowe z f1 i f2 nie zostaną przeplatane. Istnieje oczywiście niewielki problem związany z wielokrotnym odczytywaniem końca f1, który można uniknąć, ustawiając zmienną flagi, aby wskazać plik do odczytania. Wątpię jednak, by to znacznie poprawiło wydajność.

Answer 2

(nie mogę wypowiedzieć się smutno, ale to „odpowiedź” jest komentarz do odpowiedzi cmidi jest, jeśli ktoś może go przenieść byłoby mile widziane - PS: To odpowiedź została już usunięta ...)

Marco wyraźnie powiedział, że polecenia generują dużo wydruków, około 20 GB. Jeśli użyjesz funkcji komunikacji(), będzie oczekiwał na zakończenie procesu, co oznacza, że ​​deskryptor "fd" będzie musiał przechowywać dużą ilość danych. W praktyce system operacyjny będzie w międzyczasie przepłukiwał dane na dysk, chyba że na komputerze jest więcej niż 20 GB wolnej pamięci RAM. Kończysz więc pisanie pośrednich danych na dysk, których chciał uniknąć oryginalny autor. +1 za odpowiedź sirlarka!

Answer 3

Jeśli masz już polecenia powłoki w ciągu następnie można po prostu go uruchomić jak:

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

check_call(r""" 
{ 
samtools view -HS header.sam;   # command1 
samtools view input.bam 1:1-50000000; # command2 
} | samtools view -bS - > output.bam # command3 
""", shell=True) 

do emulowania rurociągu w Pythonie:

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

# start command3 to get stdin pipe, redirect output to the file 
with open('output.bam', 'wb', 0) as output_file: 
    command3 = Popen("samtools view -bS -".split(), 
        stdin=PIPE, stdout=output_file) 
# start command1 with its stdout redirected to command3 stdin 
command1 = Popen('samtools view -HS header.sam'.split(), 
       stdout=command3.stdin) 
rc_command1 = command1.wait() #NOTE: command3.stdin is not closed, no SIGPIPE or a write error if command3 dies 
# start command2 after command1 finishes 
command2 = Popen('samtools view input.bam 1:1-50000000'.split(), 
       stdout=command3.stdin) 
command3.stdin.close() # inform command2 if command3 dies (SIGPIPE or a write error) 
rc_command2 = command2.wait() 
rc_command3 = command3.wait() 

Answer 4

Podczas Popen akceptuje plikopodobnym obiekty, w rzeczywistości używa bazowych uchwytów plików/deskryptorów, a nie metod odczytu i zapisu obiektów plików do komunikacji, jak @JF Sebastian słusznie zwraca uwagę. Lepszym sposobem na to jest użycie rury (os.pipe()), która nie używa dysku.Pozwala to na podłączenie strumienia wyjściowego bezpośrednio do strumienia wejściowego innego procesu, który jest dokładnie tym, czego potrzebujesz. Problemem jest wtedy tylko kwestia serializacji, aby upewnić się, że dwa strumienie źródłowe nie przeplatają się.

import os 
import subprocess 

r, w = os.pipe() 

fh_bam = open('output.bam', 'w') 
params_0 = [ "samtools", "view", "-HS", "header.sam" ] 
params_1 = [ "samtools", "view", "input.bam", "1:1-50000000"] 
params_2 = [ "samtools", "view", "-bS", "-" ] 
sub_sink = subprocess.Popen(params_2, stdin=r, stdout=fh_bam, bufsize=4096) 
sub_src1 = subprocess.Popen(params_0, stderr=subprocess.PIPE, stdout=w, bufsize=4096) 
sub_src1.communicate() 
sub_src2 = subprocess.Popen(params_1, stderr=subprocess.PIPE, stdout=w, bufsize=4096) 
sub_src2.communicate() 

otwieramy umywalka (Czytelnik rurki), a następnie communicate z procesami źródła tylko w celu uniknięcia potencjalnego zablokowania Jak wspomniano @Ariel. To również zmusza pierwszy proces źródłowy do ukończenia i przepłukania jego wyjścia przez rurę, zanim drugi proces źródłowy otrzyma szansę na zapisanie do rury, zapobiegając przeplataniu/zrzucaniu danych wyjściowych. Możesz grać z wartością bufsize, aby poprawić wydajność.

To jest dokładnie to, co robi polecenie powłoki.