在 Python 中重置生成器对象
- 2025-01-10 08:47:00
- admin 原创
- 95
问题描述：
我有一个由多个yield返回的生成器对象。调用此生成器的准备工作相当耗时。这就是为什么我想多次重用该生成器。
y = FunctionWithYield()
for x in y: print(x)
#here must be something to reset 'y'
for x in y: print(x)
当然，我考虑将内容复制到简单列表中。有没有办法重置我的生成器？
另请参阅： 如何在 Python 生成器中向前查看一个元素（peek）？
解决方案 1：
发电机无法倒转。您有以下选择：
再次运行生成器函数，重新开始生成：
y = FunctionWithYield()
for x in y: print(x)
y = FunctionWithYield()
for x in y: print(x)
将生成器结果存储在内存或磁盘上的数据结构中，您可以再次迭代：
y = list(FunctionWithYield())
for x in y: print(x)
# can iterate again:
for x in y: print(x)
选项1的缺点是它会再次计算值。如果这会占用大量 CPU，那么您最终会计算两次。另一方面，选项2的缺点是存储。整个值列表将存储在内存中。如果值太多，这可能不切实际。
因此，您有经典的内存与处理权衡。我无法想象一种无需存储值或重新计算值即可倒回生成器的方法。
您也可以tee按照其他答案的建议使用，但是在您的情况下，这仍然会将整个列表存储在内存中，因此结果和性能与选项 2 相同。
解决方案 2：
另一个选择是使用该itertools.tee()函数创建生成器的第二个版本：
import itertools
y = FunctionWithYield()
y, y_backup = itertools.tee(y)
for x in y:
print(x)
for x in y_backup:
print(x)
如果原始迭代可能无法处理所有项目，那么从内存使用的角度来看这可能是有益的。
解决方案 3：
>>> def gen():
... def init():
... return 0
... i = init()
... while True:
... val = (yield i)
... if val=='restart':
... i = init()
... else:
... i += 1
>>> g = gen()
>>> g.next()
0
>>> g.next()
1
>>> g.next()
2
>>> g.next()
3
>>> g.send('restart')
0
>>> g.next()
1
>>> g.next()
2
解决方案 4：
可能最简单的解决方案是将昂贵的部分包装在一个对象中并将其传递给生成器：
data = ExpensiveSetup()
for x in FunctionWithYield(data): pass
for x in FunctionWithYield(data): pass
这样，您可以缓存昂贵的计算。
如果您可以同时将所有结果保存在 RAM 中，则使用list()将生成器的结果具体化在一个简单的列表中并使用它。
解决方案 5：
我想为一个老问题提供不同的解决方案
class IterableAdapter:
def __init__(self, iterator_factory):
self.iterator_factory = iterator_factory
def __iter__(self):
return self.iterator_factory()
squares = IterableAdapter(lambda: (x * x for x in range(5)))
for x in squares: print(x)
for x in squares: print(x)
与类似方法相比，这种方法的优点list(iterator)是O(1)空间复杂度list(iterator)为O(n)。缺点是，如果您只能访问迭代器，而不能访问生成迭代器的函数，则无法使用此方法。例如，执行以下操作似乎合理，但实际上行不通。
g = (x * x for x in range(5))
squares = IterableAdapter(lambda: g)
for x in squares: print(x)
for x in squares: print(x)
解决方案 6：
使用包装函数来处理StopIteration
您可以为生成器生成函数编写一个简单的包装函数，用于跟踪生成器何时耗尽。它将使用StopIteration生成器在到达迭代结束时抛出的异常来实现这一点。
import types
def generator_wrapper(function=None, **kwargs):
assert function is not None, "Please supply a function"
def inner_func(function=function, **kwargs):
generator = function(**kwargs)
assert isinstance(generator, types.GeneratorType), "Invalid function"
try:
yield next(generator)
except StopIteration:
generator = function(**kwargs)
yield next(generator)
return inner_func
正如您上面所看到的，当我们的包装函数捕获到StopIteration异常时，它只是重新初始化生成器对象（使用函数调用的另一个实例）。
然后，假设您在某处定义生成器供应函数，如下所示，则可以使用 Python 函数装饰器语法来隐式包装它：
@generator_wrapper
def generator_generating_function(**kwargs):
for item in ["a value", "another value"]
yield item
解决方案 7：
如果 GrzegorzOledzki 的答案不够，你也许可以用它send()来实现你的目标。有关增强生成器和收益表达式的更多详细信息，请参阅PEP-0342 。
更新：另请参阅itertools.tee()。它涉及上面提到的内存与处理权衡，但与仅将生成器结果存储在中相比，它可能list节省一些内存；这取决于您如何使用生成器。
解决方案 8：
如果你的生成器是纯粹的，即它的输出只取决于传递的参数和步骤号，并且你希望生成的生成器可以重新启动，那么这里有一个可能有用的排序代码片段：
import copy
def generator(i):
yield from range(i)
g = generator(10)
print(list(g))
print(list(g))
class GeneratorRestartHandler(object):
def __init__(self, gen_func, argv, kwargv):
self.gen_func = gen_func
self.argv = copy.copy(argv)
self.kwargv = copy.copy(kwargv)
self.local_copy = iter(self)
def __iter__(self):
return self.gen_func(*self.argv, **self.kwargv)
def __next__(self):
return next(self.local_copy)
def restartable(g_func: callable) -> callable:
def tmp(*argv, **kwargv):
return GeneratorRestartHandler(g_func, argv, kwargv)
return tmp
@restartable
def generator2(i):
yield from range(i)
g = generator2(10)
print(next(g))
print(list(g))
print(list(g))
print(next(g))
输出：
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[]
0
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
1
解决方案 9：
来自tee 的官方文档：
一般来说，如果一个迭代器在另一个迭代器启动之前使用了大部分或全部数据，则使用 list() 比 tee() 更快。
list(iterable)因此，根据您的情况，最好使用。
解决方案 10：
您可以定义一个返回生成器的函数
def f():
def FunctionWithYield(generator_args):
code here...
return FunctionWithYield
现在你可以根据需要多次执行：
for x in f()(generator_args): print(x)
for x in f()(generator_args): print(x)
解决方案 11：
没有重置迭代器的选项。迭代器通常在迭代next()函数时弹出。唯一的方法是在迭代迭代器对象之前进行备份。请查看下文。
创建包含项目 0 至 9 的迭代器对象
i=iter(range(10))
迭代 next() 函数，弹出
print(next(i))
将迭代器对象转换为列表
L=list(i)
print(L)
output: [1, 2, 3, 4, 5, 6, 7, 8, 9]
因此项目 0 已弹出。此外，当我们将迭代器转换为列表时，所有项目都已弹出。
next(L)
Traceback (most recent call last):
File "<pyshell#129>", line 1, in <module>
next(L)
StopIteration
因此，在开始迭代之前，您需要将迭代器转换为列表以进行备份。可以使用以下方法将列表转换为迭代器iter(<list-object>)
解决方案 12：
您现在可以使用more_itertools.seekable（第三方工具）来重置迭代器。
通过安装> pip install more_itertools
import more_itertools as mit
y = mit.seekable(FunctionWithYield())
for x in y:
print(x)
y.seek(0) # reset iterator
for x in y:
print(x)
注意：迭代器前进时内存消耗会增加，因此要警惕较大的可迭代对象。
解决方案 13：
您可以使用itertools.cycle()来实现这一点，
您可以使用此方法创建一个迭代器，然后对迭代器执行 for 循环，迭代器将循环遍历其值。
例如：
def generator():
for j in cycle([i for i in range(5)]):
yield j
gen = generator()
for i in range(20):
print(next(gen))
将重复生成20个数字，0到4。
来自文档的说明：
Note, this member of the toolkit may require significant auxiliary storage (depending on the length of the iterable).
解决方案 14：
我不确定你说的昂贵的准备是什么意思，但我猜你确实有
data = ... # Expensive computation
y = FunctionWithYield(data)
for x in y: print(x)
#here must be something to reset 'y'
# this is expensive - data = ... # Expensive computation
# y = FunctionWithYield(data)
for x in y: print(x)
如果是这样，为什么不重复使用data？
解决方案 15：
好的，你说你想多次调用一个生成器，但是初始化的代价很高...那么这样的事情呢？
class InitializedFunctionWithYield(object):
def __init__(self):
# do expensive initialization
self.start = 5
def __call__(self, *args, **kwargs):
# do cheap iteration
for i in xrange(5):
yield self.start + i
y = InitializedFunctionWithYield()
for x in y():
print x
for x in y():
print x
或者，您可以创建自己的遵循迭代器协议并定义某种“重置”函数的类。
class MyIterator(object):
def __init__(self):
self.reset()
def reset(self):
self.i = 5
def __iter__(self):
return self
def next(self):
i = self.i
if i > 0:
self.i -= 1
return i
else:
raise StopIteration()
my_iterator = MyIterator()
for x in my_iterator:
print x
print 'resetting...'
my_iterator.reset()
for x in my_iterator:
print x
https://docs.python.org/2/library/stdtypes.html#iterator-types
http://anandology.com/python-practice-book/iterators.html
解决方案 16：
我的答案解决了略有不同的问题：如果生成器的初始化成本很高，并且每个生成的对象的生成成本也很高。但我们需要在多个函数中多次使用生成器。为了精确调用生成器和每个生成的对象一次，我们可以使用线程并在不同的线程中运行每个使用方法。由于 GIL，我们可能无法实现真正的并行，但我们将实现我们的目标。
这种方法在以下情况下表现良好：深度学习模型处理大量图像。结果是图像上许多对象的大量掩码。每个掩码都消耗内存。我们有大约 10 种方法可以生成不同的统计数据和指标，但它们会一次性获取所有图像。所有图像都无法装入内存。可以轻松重写这些方法以接受迭代器。
class GeneratorSplitter:
'''
Split a generator object into multiple generators which will be sincronised. Each call to each of the sub generators will cause only one call in the input generator. This way multiple methods on threads can iterate the input generator , and the generator will cycled only once.
'''
def __init__(self, gen):
self.gen = gen
self.consumers: List[GeneratorSplitter.InnerGen] = []
self.thread: threading.Thread = None
self.value = None
self.finished = False
self.exception = None
def GetConsumer(self):
# Returns a generator object.
cons = self.InnerGen(self)
self.consumers.append(cons)
return cons
def _Work(self):
try:
for d in self.gen:
for cons in self.consumers:
cons.consumed.wait()
cons.consumed.clear()
self.value = d
for cons in self.consumers:
cons.readyToRead.set()
for cons in self.consumers:
cons.consumed.wait()
self.finished = True
for cons in self.consumers:
cons.readyToRead.set()
except Exception as ex:
self.exception = ex
for cons in self.consumers:
cons.readyToRead.set()
def Start(self):
self.thread = threading.Thread(target=self._Work)
self.thread.start()
class InnerGen:
def __init__(self, parent: "GeneratorSplitter"):
self.parent: "GeneratorSplitter" = parent
self.readyToRead: threading.Event = threading.Event()
self.consumed: threading.Event = threading.Event()
self.consumed.set()
def __iter__(self):
return self
def __next__(self):
self.readyToRead.wait()
self.readyToRead.clear()
if self.parent.finished:
raise StopIteration()
if self.parent.exception:
raise self.parent.exception
val = self.parent.value
self.consumed.set()
return val
用法：
genSplitter = GeneratorSplitter(expensiveGenerator)
metrics={}
executor = ThreadPoolExecutor(max_workers=3)
f1 = executor.submit(mean,genSplitter.GetConsumer())
f2 = executor.submit(max,genSplitter.GetConsumer())
f3 = executor.submit(someFancyMetric,genSplitter.GetConsumer())
genSplitter.Start()
metrics.update(f1.result())
metrics.update(f2.result())
metrics.update(f3.result())
解决方案 17：
如果您想使用一组预定义的参数多次重用此生成器，则可以使用functools.partial。
from functools import partial
func_with_yield = partial(FunctionWithYield, arg0, arg1)
for i in range(100):
for x in func_with_yield():
print(x)
这会将生成器函数包装在另一个函数中，因此每次调用func_with_yield()它时都会创建相同的生成器函数。
解决方案 18：
可以通过代码对象来实现。以下是示例。
code_str="y=(a for a in [1,2,3,4])"
code1=compile(code_str,'<string>','single')
exec(code1)
for i in y: print i
1 2 3 4
for i in y: print i
exec(code1)
for i in y: print i
1 2 3 4
