Bash подстановка стиля в стиле Python Popen

В Bash вы можете легко перенаправить вывод процесса во временный дескриптор файла, и все это автоматически обрабатывается Bash следующим образом:

$ mydaemon --config-file <(echo "autostart: True \n daemonize: True")

или вот так:

$ wc -l <(ls)
15 /dev/fd/63

посмотреть, как это не перенаправление stdin:

$ vim <(echo "Hello World") 
vim opens a text file containing "Hello world"
$ echo  "Hello World" | vim
Vim: Warning: Input is not from a terminal

Во втором примере вы можете увидеть, как Bash автоматически создает файловый дескриптор и позволяет передавать вывод программы в другую программу.

Теперь на мой вопрос: как я могу сделать то же самое с Python, используя Popen в модуле подпроцесса?

Я использую обычный файл клмеров и просто его читаю, но моя программа теперь генерирует определенный список клмеров во время выполнения на основе пользовательских параметров. Я хотел бы избежать записи во временный файл вручную, потому что разрешение разрешений на файлы может вызвать проблемы для моих примитивных пользователей.

Вот мой код для запуска моей программы и записи stdout с фактическим файлом "kmer_file"

input_file = Popen(["pram_axdnull", str(kmer), input_file, kmer_file], stdout=PIPE)

Я создал функцию с именем generate_kmers, которая возвращает строку, которая может быть записана в файл легко (включая новые строки) или в StringIO. У меня также есть python script, который является автономным, чтобы делать то же самое

Итак, теперь я хочу передать его в качестве третьего параметра:

Это не работает:

kmer_file = stringIO(generate_kmers(3))
input_file = Popen(["pram_axdnull", str(kmer), input_file, kmer_file], stdout=PIPE)

И это не делает:

kmer_file = Popen(["generate_kmers", str(kmer)], stdout=PIPE)
input_file = Popen(["pram_axdnull", str(kmer), input_file, kmer_file.stdout], stdout=PIPE)

Итак, у меня нет идей.

Кто-нибудь знает о хорошем способе разрешить это? Я думал об использовании опции shell = True и использовании фактического башизма <(), но я этого не понял.

Спасибо!

Ответ 1

Если pram_axdnull понимает соглашение "-" как означающее: "read from stdin", вы могли бы:

p = Popen(["pram_axdnull", str(kmer), input_filename, "-"],
          stdin=PIPE, stdout=PIPE)
output = p.communicate(generate_kmers(3))[0]

Если вход генерируется внешним процессом:

kmer_proc = Popen(["generate_kmers", str(kmer)], stdout=PIPE)
p = Popen(["pram_axdnull", str(kmer), input_filename, "-"],
          stdin=kmer_proc.stdout, stdout=PIPE)
kmer_proc.stdout.close()
output = p.communicate()[0]

Если pram_axdnull не понимает соглашение "-":

import os
import tempfile
from subprocess import check_output

with tempfile.NamedTemporaryFile() as file:
    file.write(generate_kmers(3))
    file.delete = False

try:
    p = Popen(["pram_axdnull", str(kmer), input_filename, file.name],
              stdout=PIPE)
    output = p.communicate()[0]
    # or
    # output = check_output(["pram_axdnull", str(kmer), input_filename, 
                             file.name])
finally:
    os.remove(file.name)

Чтобы создать временный файл с помощью внешнего процесса:

from subprocess import check_call

with tempfile.NamedTemporaryFile() as file:
    check_call(["generate_kmers", str(kmer)], stdout=file)
    file.delete = False

Чтобы избежать ожидания создания всех копиров, то есть для одновременного чтения и чтения копий, вы можете использовать os.mkfifo() в Unix (предложенный @cdarke):

import os
import shutil
import tempfile
from contextlib import contextmanager
from subprocess import Popen, PIPE

@contextmanager
def named_pipe():
    dirname = tempfile.mkdtemp()
    try:
        path = os.path.join(dirname, 'named_pipe')
        os.mkfifo(path)
        yield path
    finally:
        shutil.rmtree(dirname)

with named_pipe() as path:
    p = Popen(["pram_axdnull", str(kmer), input_filename, path],
              stdout=PIPE) # read from path
    with open(path, 'wb') as wpipe:
        kmer_proc = Popen(["generate_kmers", str(kmer)],
                          stdout=wpipe) # write to path
    output = p.communicate()[0]
    kmer_proc.wait()

Ответ 1

Ответ 2