<p>要模拟<a href="http://wiki.bash-hackers.org/syntax/expansion/proc_subst" rel="nofollow noreferrer">bash process substitution</a>,请执行以下操作:</p>
<pre><code>#!/usr/bin/env python
from subprocess import check_call
check_call('someprogram <(someprocess) <(anotherprocess)',
shell=True, executable='/bin/bash')
</code></pre>
<p>在Python中,可以使用命名管道:</p>
^{pr2}$
<p>其中<code>named_pipes(n)</code>是:</p>
<pre><code>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)
</code></pre>
<hr/>
<p>实现bash进程替换的另一种更可取的方法(无需在磁盘上创建命名条目)是使用<code>/dev/fd/N</code>文件名(如果它们可用)作为<a href="https://stackoverflow.com/a/28844027/4279">suggested by @Dunes</a>。在FreeBSD上,<a href="https://www.freebsd.org/cgi/man.cgi?query=fdescfs&sektion=5" rel="nofollow noreferrer"><em>^{<cd3>}</em> (^{<cd4>}) creates entries for all file descriptors opened by the process</a>。要测试可用性,请运行:</p>
<pre><code>$ test -r /dev/fd/3 3</dev/null && echo /dev/fd is available
</code></pre>
<p>如果失败,请尝试将<code>/dev/fd</code>符号链接到<a href="http://man7.org/linux/man-pages/man5/proc.5.html" rel="nofollow noreferrer">^{<cd6>}</a>,就像在某些linux上那样:</p>
<pre><code>$ ln -s /proc/self/fd /dev/fd
</code></pre>
<p>下面是<code>/dev/fd</code>基于<code>someprogram <(someprocess) <(anotherprocess)</code>bash命令的实现:</p>
<pre><code>#!/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)
</code></pre>
<p>注意:在我的Ubuntu机器上,<code>subprocess</code>代码只在python3.4+中工作,尽管<code>pass_fds</code>在python3.2之后就可用了。在</p>