python - 深度复制嵌套迭代(或改进的 itertools.tee 用于迭代的迭代)

Question

前言

我有一个测试，其中我正在使用嵌套的可迭代对象(nested iterable 我的意思是只有可迭代对象作为元素的可迭代对象)。

作为测试级联考虑

from itertools import tee
from typing import (Any,
                    Iterable)


def foo(nested_iterable: Iterable[Iterable[Any]]) -> Any:
    ...


def test_foo(nested_iterable: Iterable[Iterable[Any]]) -> None:
    original, target = tee(nested_iterable)  # this doesn't copy iterators elements

    result = foo(target)

    assert is_contract_satisfied(result, original)


def is_contract_satisfied(result: Any,
                          original: Iterable[Iterable[Any]]) -> bool:
    ...

例如foo 可能是简单的身份函数

def foo(nested_iterable: Iterable[Iterable[Any]]) -> Iterable[Iterable[Any]]:
    return nested_iterable

contract 只是检查扁平化的可迭代对象是否具有相同的元素

from itertools import (chain,
                       starmap,
                       zip_longest)
from operator import eq
...
flatten = chain.from_iterable


def is_contract_satisfied(result: Iterable[Iterable[Any]],
                          original: Iterable[Iterable[Any]]) -> bool:
    return all(starmap(eq,
                       zip_longest(flatten(result), flatten(original),
                                   # we're assuming that ``object()``
                                   # will create some unique object
                                   # not presented in any of arguments
                                   fillvalue=object())))

但是如果一些 nested_iterable 元素是一个迭代器，它可能会被耗尽，因为 tee 正在制作浅拷贝，而不是深拷贝，即对于给定的 foo 和 is_contract_satisfied 下一条语句

>>> test_foo([iter(range(10))])

导致可预测

Traceback (most recent call last):
  ...
    test_foo([iter(range(10))])
  File "...", line 19, in test_foo
    assert is_contract_satisfied(result, original)
AssertionError

问题

如何深度复制任意嵌套的iterable？

注意事项

我知道 copy.deepcopy function , 但它不适用于文件对象。

Answer 1

天真的解决方案

简单的算法是

1. 执行原始嵌套 iterable 的逐元素复制。
2. 创建 n 个元素副本。
3. 获取与每个独立副本相关的坐标。

可以这样实现

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)

优点:

• 处理更深层次的嵌套，甚至处理混合元素，例如同一层次上的可迭代对象和不可迭代对象，
• 可以针对用户定义的结构进行扩展(例如，制作它们的独立深拷贝)。

缺点:

• 可读性较差(但据我们所知 "practicality beats purity" )，
• 提供一些与调度相关的开销(但这没关系，因为它基于具有 O(1) 复杂性的字典查找)。

---

测试

准备

让我们定义我们的嵌套迭代如下

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对象。

---

结论

正如我们所见，这两个函数具有相似的性能，但最后一个函数支持更深层次的嵌套并且看起来非常可扩展。

广告

我已经为 lz package 添加了“改进的”解决方案从 0.4.0 版本可以像这样使用

>>> 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 进行了基于属性的测试, 所以我们可以确定它按预期工作。