4

我正在使用 Python 解析一个大文件。我想做的是

If condition =True
   append to list A
else 
   append to list B

我想为此使用生成器表达式 - 以节省内存。我正在输入实际代码。

def is_low_qual(read):
    lowqual_bp=(bq for bq in phred_quals(read) if bq < qual_threshold)  
    if iter_length(lowqual_bp) >  num_allowed:
        return True
    else:
        return False  

lowqual=(read for read in SeqIO.parse(r_file,"fastq") if is_low_qual(read)==True)
highqual=(read for read in SeqIO.parse(r_file,"fastq") if is_low_qual(read)==False)


SeqIO.write(highqual,flt_out_handle,"fastq")
SeqIO.write(lowqual,junk_out_handle,"fastq")

def iter_length(the_gen):
    return sum(1 for i in the_gen)
4

4 回答 4

6

您可以与anditertools.tee结合使用：itertools.ifilteritertools.ifilterfalse

import itertools
def is_condition_true(x):
    ...

gen1, gen2 = itertools.tee(sequences)
low = itertools.ifilter(is_condition_true, gen1)
high = itertools.ifilterfalse(is_condition_true, gen2)

tee即使序列本身是生成器，使用也可确保函数正常工作。

但是请注意，如果和以不同的速率消耗（例如，如果在使用之前耗尽），它tee本身可能会使用相当多的内存（最多为 size 的列表）。len(sequences)lowhighlowhigh

于 2012-08-24T16:30:50.080 回答
1

我认为您正在努力避免两次迭代您的集合。如果是这样，这种方法有效：

high, low = [], []
_Nones = [high.append(x) if is_condition_true() else low.append(x) for x in sequences]

这可能不是建议的，因为它使用列表理解来解决副作用。这通常是反pythonic。

于 2012-08-24T16:31:07.267 回答
1

只是添加一个更一般的答案：如果您主要关心的是内存，您应该使用一个循环遍历整个文件的生成器，并处理每个项目的低或高。就像是：

for r in sequences:
    if condition_true(r):
        handle_low(r)
    else:
        handle_high(r)

如果您需要在使用之前收集所有高/低元素，那么您无法防范潜在的内存命中。原因是您在阅读它们之前无法知道哪些元素是高/低的。如果你必须先处理低，而事实证明所有元素实际上都是高的，你别无选择，只能将它们存储在一个列表中，这将占用内存。使用一个循环执行此操作可以让您一次处理一个元素，但您必须平衡这一点与其他问题（即，以这种方式执行此操作有多麻烦，这将取决于您正在尝试做什么与数据）。

于 2012-08-24T16:46:59.830 回答
0

这很难优雅地做到。这是可行的：

from itertools import tee, ifilter, ifilterfalse
low, high = [f(condition, g) for f, g in zip((ifilter, ifilterfalse), tee(seq))]

请注意，当您从一个生成的迭代器（例如low）中消费项目时， tee 中的内部双端队列必须扩展以包含您尚未消费的任何项目high（包括，不幸的是，那些ifilterfalse将拒绝的项目）。因此，这可能不会像您希望的那样节省内存。

这是一个使用尽可能少的额外内存的实现：

def filtertee(func, iterable, codomain=(False, True)):
    it = iter(iterable)
    deques = dict((r, deque()) for r in codomain)
    def gen(mydeque):
        while True:
            while not mydeque:          # as long as the local deque is empty
                newval = next(it)       # fetch a new value,
                result = func(newval)   # find its image under `func`,
                try:
                    d = deques[result]  # find the appropriate deque, and
                except KeyError:
                    raise ValueError("func returned value outside codomain")
                d.append(newval)        # add it.
            yield mydeque.popleft()
    return dict((r, gen(d)) for r, d in deques.items())

这会将 adict从函数的 codomain 返回到生成器，该生成器提供在 下采用该值的项目func：

gen = filtertee(condition, seq)
low, high = gen[True], gen[False]

请注意，您有责任确保condition仅返回codomain.

于 2012-08-24T16:31:58.840 回答