朴素的解决方案
简单的算法是
- 对原始嵌套迭代执行元素复制。
- 制作
n 元素副本的副本。
- 获取与每个独立副本相关的坐标。
可以像这样实现
from itertools import tee
from operator import itemgetter
from typing import (Any,
Iterable,
Tuple,
TypeVar)
Domain = TypeVar('Domain')
def copy_nested_iterable(nested_iterable: Iterable[Iterable[Domain]],
*,
count: int = 2
) -> Tuple[Iterable[Iterable[Domain]], ...]:
def shallow_copy(iterable: Iterable[Domain]) -> Tuple[Iterable[Domain], ...]:
return tee(iterable, count)
copies = shallow_copy(map(shallow_copy, nested_iterable))
return tuple(map(itemgetter(index), iterables)
for index, iterables in enumerate(copies))
优点：
缺点：
- 如果我们想将我们的方法扩展到具有更高嵌套级别的可迭代对象（例如嵌套可迭代对象的可迭代对象等），这种方法看起来没有帮助。
我们可以做得更好。
改进的解决方案
如果我们查看 itertools.tee function documentation，它包含 Python 配方，在 functools.singledispatch decorator 的帮助下可以重写为
from collections import (abc,
deque)
from functools import singledispatch
from itertools import repeat
from typing import (Iterable,
Tuple,
TypeVar)
Domain = TypeVar('Domain')
@functools.singledispatch
def copy(object_: Domain,
*,
count: int) -> Iterable[Domain]:
raise TypeError('Unsupported object type: {type}.'
.format(type=type(object_)))
# handle general case
@copy.register(object)
# immutable strings represent a special kind of iterables
# that can be copied by simply repeating
@copy.register(bytes)
@copy.register(str)
# mappings cannot be copied as other iterables
# since they are iterable only by key
@copy.register(abc.Mapping)
def copy_object(object_: Domain,
*,
count: int) -> Iterable[Domain]:
return itertools.repeat(object_, count)
@copy.register(abc.Iterable)
def copy_iterable(object_: Iterable[Domain],
*,
count: int = 2) -> Tuple[Iterable[Domain], ...]:
iterator = iter(object_)
# we are using `itertools.repeat` instead of `range` here
# due to efficiency of the former
# more info at
# https://stackoverflow.com/questions/9059173/what-is-the-purpose-in-pythons-itertools-repeat/9098860#9098860
queues = [deque() for _ in repeat(None, count)]
def replica(queue: deque) -> Iterable[Domain]:
while True:
if not queue:
try:
element = next(iterator)
except StopIteration:
return
element_copies = copy(element,
count=count)
for sub_queue, element_copy in zip(queues, element_copies):
sub_queue.append(element_copy)
yield queue.popleft()
return tuple(replica(queue) for queue in queues)
优点：
- 处理更深层次的嵌套，甚至处理混合元素，如同一级别上的可迭代和不可迭代，
- 可以针对用户定义的结构进行扩展（例如，用于制作它们的独立深层副本）。
缺点：
测试
准备
让我们定义我们的嵌套迭代如下
nested_iterable = [range(10 ** index) for index in range(1, 7)]
由于迭代器的创建没有说明底层副本的性能，让我们定义迭代器耗尽的函数（描述为here）
exhaust_iterable = deque(maxlen=0).extend
时间
使用timeit 包
import timeit
def naive(): exhaust_iterable(copy_nested_iterable(nested_iterable))
def improved(): exhaust_iterable(copy_iterable(nested_iterable))
print('naive approach:', min(timeit.repeat(naive)))
print('improved approach:', min(timeit.repeat(improved)))
我的笔记本电脑上有 Python 3.5.4 中的 Windows 10 x64
naive approach: 5.1863865
improved approach: 3.5602296000000013
内存
使用memory_profiler package
Line # Mem usage Increment Line Contents
================================================
78 17.2 MiB 17.2 MiB @profile
79 def profile_memory(nested_iterable: Iterable[Iterable[Any]]) -> None:
80 68.6 MiB 51.4 MiB result = list(flatten(flatten(copy_nested_iterable(nested_iterable))))
对于“幼稚”的方法和
Line # Mem usage Increment Line Contents
================================================
78 17.2 MiB 17.2 MiB @profile
79 def profile_memory(nested_iterable: Iterable[Iterable[Any]]) -> None:
80 68.7 MiB 51.4 MiB result = list(flatten(flatten(copy_iterable(nested_iterable))))
对于“改进”的一个。
注意：我制作了不同的脚本运行，因为一次运行它们不会具有代表性，因为第二个语句将重用之前创建的底层 int 对象。
结论
我们可以看到这两个函数具有相似的性能，但最后一个支持更深层次的嵌套并且看起来相当可扩展。
广告
我已经从0.4.0 版本向lz package 添加了“改进的”解决方案，可以像这样使用
>>> from lz.replication import replicate
>>> iterable = iter(range(5))
>>> list(map(list, replicate(iterable,
count=3)))
[[0, 1, 2, 3, 4], [0, 1, 2, 3, 4], [0, 1, 2, 3, 4]]
它使用hypothesis framework 进行了基于属性的测试，因此我们可以确定它按预期工作。