从 Haskell 翻译的 Python 中，Count 懒惰地运行

Posted 2023-03-28

【中文标题】从 Haskell 翻译的 Python 中，Count 懒惰地运行

【英文标题】：Count runs lazily in Python translate from Haskell

【发布时间】：2021-12-07 19:26:03

【问题描述】：

我正在尝试编写一个生成器函数（或实现等效函数），它在 Python 中采用可迭代的 xs 并计算“运行”。 （这是 Bird 的 Thinking Functionally with Haskell 中的一个问题，我想使用 Python 的惰性特性将其翻译成 Python。）所以

list(iter(count_runs(['a', 'a', 'b', 'c', 'a', 'd', 'd'])))
# => [(2, 'a'), (1, 'b'), (1, c'), (1, 'a'), (2, 'd')]

在 Haskell 中是

countRuns :: [a] -> [(Int, a)]
countRuns [] = []
countRuns x:xs = (1 + length us, x):countRuns vs
               where us, vs = span (==x) xs 

在 Python 中，我想写一些类似的东西

from itertools import takewhile, dropwhile

def count_runs(xs):
  # get first element x of xs, if it exists
  us, vs = (takewhile(lambda y: y==x, xs), 
            dropwhile(lambda y: y==x, xs))
  yield (1 + len(list(us)), x)
  yield from count_runs(vs)

但问题是vs 已经是一个迭代器，所以如果我在下一次递归中调用takewhile 和dropwhile 会遇到麻烦。 （当我在下一次递归中调用list(takewhile(..., xs)) 时，它也会删除dropwhile(..., xs) 的第一个元素，因为它们都在查看同一个迭代器。

如何解决这个问题，获取第二行第一个元素的正确方法是什么？

【问题讨论】：

itertools.tee() 也许？

您并不需要takewhile 和dropwhile，因为Python 迭代器是可变的。但是，takewhile 也没有使用模式匹配，因此它将消耗来自xs 的非x 值以停止产生值。

itertools.groupby 已经完成了您需要的大部分工作。

我想我有点困惑。如果问题是执行takewhile 将导致dropwhile 从您希望下一个takewhile 开始的位置开始，那么就......根本不要打电话给dropwhile？ us = takewhile(lambda y: y==x, xs); yield (1+len(list(us)), x); yield from count_runs(xs)（不把这个放在答案中，因为我不知道 Python，所以可能有一个微妙的理由不这样做。）

【参考方案1】：

span 和takewhile 之间的一个显着区别是takewhile 使用第一个非x 值以确定何时停止产生值。结果，您将丢失输入中的任何单例项；特别是，takewhile 在生成as 的领先集时失去了第一个b。迭代器协议无法查看迭代器的下一个元素，也无法放回它所使用的元素。

相反，您需要两个独立的迭代器：一个用于takewhile 以生成所需的前缀，另一个用于删除递归调用的前缀。

def count_runs(xs):
    try:
        x = next(xs)
    except StopIteration:
        return

    t1, t2 = tee(xs)

    us = list(takewhile(lambda y: y == x, t1))
    yield (1 + len(us), x)
    yield from count_runs(dropwhile(lambda y: y == x, t2))

（请注意，itertools 文档在其recipe section 中将类似于span 的东西作为before_and_after 函数实现。它不使用tee，但我建议您参考实际实现以了解详细信息） .

def before_and_after(xs):
    ...

def count_runs(xs):
    try:
        x = next(xs)
    except StopIteration:
        return

    first, second = before_and_after(lambda y: y == x, xs)
    yield (1 + len(list(first)), x)
    yield from count_runs(second)

)

不过，大部分工作已经由itertools.groupby 为您完成。

def count_runs(xs):
    yield from ((len(list(v)), k) for k, v in groupby(xs))

【讨论】：

如果你想避免创建列表只是为了获取它们的长度，你可以用sum(1 for _ in v)替换那个位