python - 窥探 Python 中的 Popen 管道流

Question

背景:
Linux 上的 Python 2.6.6。 DNA 序列分析流程的第一部分。
我想从挂载的远程存储(LAN)中读取一个可能压缩过的文件，如果它是压缩过的；将它压缩到流中(即使用 gunzip FILENAME -c)，如果流(文件)的第一个字符是“@”，则将整个流路由到一个过滤程序中，该程序接受标准输入的输入，否则直接将其通过管道传输到本地磁盘上的文件。我想尽量减少从远程存储中读取/查找文件的次数(只通过一次文件应该是不可能的？)。

示例输入文件的内容，前四行对应于 FASTQ 格式的一条记录:

@I328_1_FC30MD2AAXX:8:1:1719:1113/1                                        
GTTATTATTATAATTTTTTACCGCATTTATCATTTCTTCTTTATTTTCATATTGATAATAAATATATGCAATTCG
+I328_1_FC30MD2AAXX:8:1:1719:1113/1                                        
hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhahhhhhhfShhhYhhQhh]hhhhffhU\UhYWc

不应通过管道传输到过滤程序的文件包含如下所示的记录(前两行对应 FASTA 格式的一条记录):

>I328_1_FC30MD2AAXX:8:1:1719:1113/1
GTTATTATTATAATTTTTTACCGCATTTATCATTTCTTCTTTATTTTCATATTGATAATAAATATATGCAATTCG

有些人编写了半伪代码来可视化我想做的事情(我知道这不可能按照我写的方式)。我希望它有意义:

if gzipped:
    gunzip = Popen(["gunzip", "-c", "remotestorage/file.gz"], stdout=PIPE)
    if gunzip.stdout.peek(1) == "@": # This isn't possible
        fastq = True
    else:
        fastq = False
if fastq:
    filter = Popen(["filter", "localstorage/outputfile.fastq"], stdin=gunzip.stdout).communicate()
else:
    # Send the gunzipped stream to another file

请忽略代码不会像我在这里编写的那样运行以及我没有错误处理等事实，所有这些都已经在我的其他代码中了。我只需要帮助查看流或找到解决方法。如果你能 gunzip.stdout.peek(1) 我会很棒，但我知道这是不可能的。

到目前为止我尝试了什么:
我认为 subprocess.Popen 可能会帮助我实现这一目标，并且我尝试了很多不同的想法，其中包括尝试使用某种 io.BufferedRandom() 对象将流写入但我无法弄清楚如何会工作。我知道流是不可搜索的，但也许解决方法是读取 gunzip-stream 的第一个字符，然后创建一个新流，您首先在其中输入“@”或“>”，具体取决于文件内容，然后填充其余部分gunzip.stdout-stream 到新流中。然后这个新流将被送入过滤器的 Popen 标准输入。

请注意，文件大小可能比可用内存大几倍。我不想从远程存储中多次读取源文件，也不想进行不必要的文件访问。

欢迎任何想法!如果我说得不够清楚，请问我问题，以便我澄清。

Answer 1

这是您的首先根据文件内容输入“@”或“>”，然后将 gunzip.stdout-stream 的其余部分填充到新流中的实现提案。我只测试了测试的本地文件分支，但它应该足以证明这个概念。

if gzipped:
    source = Popen(["gunzip", "-c", "remotestorage/file.gz"], stdout=PIPE)
else:
    source = Popen(["cat", "remotestorage/file"], stdout=PIPE)
firstchar = source.stdout.read(1)
# "unread" the char we've just read
source = Popen([r"(printf '\x%02x' && cat)" % ord(firstchar)],
               shell=True, stdin=source.stdout, stdout=PIPE)

# Now feed the output to a filter or to a local file.
flocal = None
try:
    if firstchar == "@":
        filter = Popen(["filter", "localstorage/outputfile.fastq"],
                       stdin=source.stdout)
    else:
        flocal = open('localstorage/outputfile.stream', 'w')
        filter = Popen(["cat"], stdin=source.stdout, stdout=flocal)
    filter.communicate()
finally:
    if flocal is not None:
        flocal.close()

想法是从 source 命令的输出中读取单个字符，然后使用 (printf '\xhh' && cat) 重新创建原始输出，从而有效地实现 peek。替换流将 shell=True 指定为 Popen，将其留给 shell 和 cat 来完成繁重的工作。数据始终保留在管道中，永远不会完全读入内存。请注意，shell 的服务仅针对实现取消读取已查看字节的 Popen 的单个调用请求，而不是涉及用户提供的文件名的调用。即使在那个时候，字节也会转义为十六进制，以确保 shell 在调用 printf 时不会破坏它。

可以进一步清理代码以实现名为 peek 的实际函数，该函数返回已查看的内容和替换 new_source。