对生成器使用前瞻

Question

我在 Python 中实现了一个基于生成器的扫描器，它将字符串标记为 （令牌类型，令牌值）形式的元组：

for token in scan("a(b)"):
    print token

会打印

("literal", "a")
("l_paren", "(")
...

下一个任务意味着解析令牌流，为此，我需要能够在不向前移动指针的情况下从当前项向前查看一项。事实上，迭代器和生成器不会一次提供完整的项目序列，而是根据需要提供每个项目，这使得与列表相比，前瞻有点棘手，因为除非调用 __next__()，否则下一个项目是未知的。

基于生成器的前瞻的简单实现是什么样的？目前我正在使用一种解决方法，这意味着从生成器中列出一个列表：

token_list = [token for token in scan(string)]

然后，可以通过以下方式轻松实现前瞻：

try:
    next_token = token_list[index + 1]
except: IndexError:
    next_token = None

当然，这很好用。但是考虑到这一点，我的第二个问题出现了：首先让scan() 成为生成器真的有什么意义吗？

Answer 1

您可以使用lazysequence，这是一个不可变的序列，它包装了一个可迭代对象并将消耗的项目缓存在一个内部缓冲区中。您可以像使用任何列表或元组一样使用它，但迭代器的高级程度仅为给定操作所需的程度。

这是您的示例使用惰性序列时的样子：

from lazysequence import lazysequence

token_list = lazysequence(token for token in scan(string))

try:
    next_token = token_list[index + 1]
except IndexError:
    next_token = None

以下是您自己实现惰性序列的方法：

from collections.abc import Sequence


class lazysequence(Sequence):
    def __init__(self, iterable):
        self._iter = iter(iterable)
        self._cache = []

    def __iter__(self):
        yield from self._cache
        for item in self._iter:
            self._cache.append(item)
            yield item

    def __len__(self):
        return sum(1 for _ in self)

    def __getitem__(self, index):
        for position, item in enumerate(self):
            if index == position:
                return item

        raise IndexError("lazysequence index out of range")

这是一个幼稚的实现。这里缺少的一些东西：

• 惰性序列最终会将所有项目存储在内存中。无法获得不再缓存项目的普通迭代器。
• 在布尔上下文 (if s) 中，将评估整个序列，而不仅仅是第一项。
• len(s) 和 s[i] 需要遍历序列，即使项目已经存储在内部缓存中。
• 不支持负索引 (s[-1]) 和切片 (s[:2])。

PyPI 包解决了这些问题以及更多问题。最后一个警告适用于上述实现和包：

• 显式优于隐式。客户端可能会更好地传递一个迭代器并处理它的限制。例如，客户可能不希望 len(s) 承担将迭代器消耗到其结束的成本。

披露：我是lazysequence的作者。

Answer 2

使用itertools.tee 构造一个简单的前瞻包装器：

from itertools import tee, islice

class LookAhead:
    'Wrap an iterator with lookahead indexing'
    def __init__(self, iterator):
        self.t = tee(iterator, 1)[0]
    def __iter__(self):
        return self
    def next(self):
        return next(self.t)
    def __getitem__(self, i):
        for value in islice(self.t.__copy__(), i, None):
            return value
        raise IndexError(i)

使用该类来包装现有的可迭代对象或迭代器。然后，您可以使用 next 进行正常迭代，也可以使用索引查找进行前瞻。

>>> it = LookAhead([10, 20, 30, 40, 50])
>>> next(it)
10
>>> it[0]
20
>>> next(it)
20
>>> it[0]
30
>>> list(it)
[30, 40, 50]

要在 Python 3 下运行此代码，只需将 next 方法更改为 __next__。

Answer 3

如果我只需要 1 个元素的前瞻性，我将如何简洁地编写它：

SEQUENCE_END = object()

def lookahead(iterable):
    iter = iter(iterable)
    current = next(iter)
    for ahead in iter:
        yield current,ahead
        current = ahead
    yield current,SEQUENCE_END

例子：

>>> for x,ahead in lookahead(range(3)):
>>>     print(x,ahead)
0, 1
1, 2
2, <object SEQUENCE_END>

Answer 4

它并不漂亮，但这可能会满足你的需求：

def paired_iter(it):
    token = it.next()
    for lookahead in it:
        yield (token, lookahead)
        token = lookahead
    yield (token, None)

def scan(s):
    for c in s:
        yield c

for this_token, next_token in paired_iter(scan("ABCDEF")):
    print "this:%s next:%s" % (this_token, next_token)

打印：

this:A next:B
this:B next:C
this:C next:D
this:D next:E
this:E next:F
this:F next:None

Answer 5

既然你说你正在标记一个字符串而不是一个通用的可迭代对象，我建议最简单的解决方案就是扩展你的标记器以返回一个 3 元组： (token_type, token_value, token_index)，其中token_index 是字符串中标记的索引。然后，您可以向前、向后或字符串中的任何其他位置查看。只是不要越过终点。我认为最简单、最灵活的解决方案。

此外，您无需使用列表推导式从生成器创建列表。只需调用它的 list() 构造函数：

 token_list = list(scan(string))

Answer 6

这是一个允许将单个项目发送回生成器的示例

def gen():
    for i in range(100):
        v=yield i           # when you call next(), v will be set to None
        if v:
            yield None      # this yields None to send() call
            v=yield v       # so this yield is for the first next() after send()

g=gen()

x=g.next()
print 0,x

x=g.next()
print 1,x

x=g.next()
print 2,x # oops push it back

x=g.send(x)

x=g.next()
print 3,x # x should be 2 again

x=g.next()
print 4,x

Answer 7

那里的答案非常好，但我最喜欢的方法是使用itertools.tee——给定一个迭代器，它会返回两个（或更多，如果需要）可以独立推进。它根据需要在内存中缓冲（即，如果迭代器彼此之间没有非常“失步”，则不会太多）。例如：

import itertools
import collections

class IteratorWithLookahead(collections.Iterator):
  def __init__(self, it):
    self.it, self.nextit = itertools.tee(iter(it))
    self._advance()
  def _advance(self):
    self.lookahead = next(self.nextit, None)
  def __next__(self):
    self._advance()
    return next(self.it)

你可以用这个类包装任何迭代器，然后使用包装器的.lookahead 属性来知道未来要返回的下一个项目是什么。我喜欢将所有真正的逻辑留给 itertools.tee，只提供这种薄胶水！-)

Answer 8

您可以编写一个包装器来缓冲来自生成器的一些项目，并提供一个lookahead() 函数来查看这些缓冲项目：

class Lookahead:
    def __init__(self, iter):
        self.iter = iter
        self.buffer = []

    def __iter__(self):
        return self

    def next(self):
        if self.buffer:
            return self.buffer.pop(0)
        else:
            return self.iter.next()

    def lookahead(self, n):
        """Return an item n entries ahead in the iteration."""
        while n >= len(self.buffer):
            try:
                self.buffer.append(self.iter.next())
            except StopIteration:
                return None
        return self.buffer[n]

Answer 9

Paul's 是一个很好的答案。具有任意前瞻的基于类的方法可能类似于：

class lookahead(object):
    def __init__(self, generator, lookahead_count=1):
        self.gen = iter(generator)
        self.look_count = lookahead_count

    def __iter__(self):
        self.lookahead = []
        self.stopped = False
        try:
            for i in range(self.look_count):
                self.lookahead.append(self.gen.next())
        except StopIteration:
            self.stopped = True
        return self

    def next(self):
        if not self.stopped:
            try:
                self.lookahead.append(self.gen.next())
            except StopIteration:
                self.stopped = True
        if self.lookahead != []:
            return self.lookahead.pop(0)
        else:
            raise StopIteration

x = lookahead("abcdef", 3)
for i in x:
    print i, x.lookahead