用 Python 的 Popen 替换 Bash 风格的进程

2022-01-18 00:00:00 python subprocess popen bash

问题描述

在 Bash 中，您可以轻松地将进程的输出重定向到临时文件描述符，并且所有这些都由 bash 自动处理，如下所示:

In Bash you can easily redirect the output of a process to a temporary file descriptor and it is all automagically handled by bash like this:

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

或者像这样:

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

看看它怎么不是标准输入重定向:

see how it is not stdin redirection:

$ 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 如何自动创建文件描述符并允许您将程序的输出传递给另一个程序.

You can see in the second example how bash automatically creates a file descriptor and allows you to pass the output of a program to another program.

现在我的问题是:我怎样才能用 Python 做同样的事情，在 subprocess 模块中使用 Popen?

Now onto my question: How can I do the same thing with Python, using Popen in the subprocess module?

我一直在使用普通的 kmers 文件并将其读入，但我的程序现在根据用户参数在运行时生成一个特定的 kmers 列表.我想避免手动写入临时文件，因为处理文件权限可能会给我的原始用户带来问题.

I have been using a normal file of kmers and just reading it in, but my program now generates a specific list of kmers at runtime based on user parameters. I'd like to avoid writing to a temporary file manually because dealing with file permissions could cause problems for my primitive users.

这是我运行程序并使用实际文件kmer_file"捕获标准输出的代码

Here is my code to run my program and capture the stdout with an actual file "kmer_file"

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

我创建了一个名为 generate_kmers 的函数，它返回一个可以轻松写入文件(包括换行符)或 StringIO 的字符串.我也有一个独立的 python 脚本来做同样的事情

I created a function called generate_kmers which returns a string that can be written out to a file easily (includes newlines) or to a StringIO. I also have a python script that is standalone to do the same thing

所以现在我想将它作为我的第三个参数传入:

So now I want to pass it in as my 3rd parameter:

这不起作用:

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 选项并使用 <() 的实际bashism，但我还没有弄清楚.

Does anyone know of a good way to resolve this? I was thinking using the shell=True option and using the actual bashism of <() but I haven't figured that out.

谢谢！

解决方案

如果 pram_axdnull 将 "-" 约定理解为:从标准输入读取"，那么您可以:

If pram_axdnull understands "-" convention to mean: "read from stdin" then you could:

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

如果输入是由外部进程生成的:

If the input is generated by external process:

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 不理解 "-" 约定:

If pram_axdnull doesn't understand "-" convention:

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)

使用外部进程生成临时文件:

To generate temporary file using external process:

from subprocess import check_call

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

为避免等待所有 kmers 生成，即同时写入/读取 kmers，您可以在 Unix 上使用 os.mkfifo()(@cdarke 建议):

To avoid waiting for all kmers to be generated i.e., to write/read kmers simultaneously, you could use os.mkfifo() on Unix (suggested by @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()

相关文章