8

我尝试定义一个mycount()可以使用生成器函数重置的生成器函数send(0)，如下例所示。一切正常，除非我send(0)在尚未启动的新生成器对象上使用。在这种情况下，它给出了一个TypeError. 是否有任何功能可以检查生成器是否已启动，或者在这种情况下我是否必须捕获TypeError并创建一个新的生成器对象mycount(0)？

def mycount(value):
    while True:
        v = yield value
        if v == None:
            value = value + 1
        else:
            value = v

g = mycount(3)
print(next(g))    # prints 3
print(next(g))    # prints 4
print(g.send(0))  # prints 0
print(next(g))    # prints 1
print(next(g))    # prints 2

g2 = mycount(3)
g2.send(0)
# TypeError: can't send non-None value to a just-started generator
4

4 回答 4

9

为避免向刚启动的生成器发送非None值，您需要先调用nextor send(None)。我同意其他人的观点，即 David Beazley 的协程装饰器（在 python 3.x 中，您需要调用__next__()函数而不是next()）是一个不错的选择。尽管那个特殊的装饰器很简单，但我也成功地使用了copipes库，它很好地实现了 Beazley 演示文稿中的许多实用程序，包括协程。

关于是否可以检查生成器是否已启动 - 在 Python 3 中，您可以使用inspect.getgeneratorstate。这在 Python 2 中不可用，但CPython 实现是纯 python 并且不依赖于 Python 3 的任何新内容，因此您可以以相同的方式检查自己：

if generator.gi_running:
    return GEN_RUNNING
if generator.gi_frame is None:
    return GEN_CLOSED
if generator.gi_frame.f_lasti == -1:
    return GEN_CREATED
return GEN_SUSPENDED

具体来说，g2如果inspect.getgeneratorstate(g2) != inspect.GEN_CREATED.

于 2013-07-18T01:05:10.507 回答
1

由于您的错误意味着send必须None在刚刚启动的生成器 (docs-link)上调用该函数。

你可以从那里赶上TypeError并滚动：

    #...
    try:
        g2.send(0)
    except TypeError:
        #Now you know it hasn't started, etc.
        g2.send(None)

无论哪种方式，它都不能用于“重置”生成器，它只需要重新制作。

这里对生成器概念和语法进行了很好的概述，涵盖了生成器的链接和其他高级主题。

于 2013-07-17T15:38:03.837 回答
1

特别是，您可能会找到一种方法来使用第consumer4 页中描述的装饰器。David Beazley 的“发电机技巧”的 I-131，J. Gwyn 提供了一个链接：

def consumer(func):
    def start(*args,**kwargs):
        c = func(*args,**kwargs)
        c.next()
        return c
    return start

我在我的代码中使用了类似的东西。

请注意，if v is None优先于if v == None.

于 2013-07-17T17:55:55.613 回答
0

这是一个完整的实现 Python2 兼容例程，getgeneratorstate(gtor)，带有测试代码。

import unittest
import enum

class GtorState(enum.Enum):
    GEN_RUNNING ='GEN_RUNNING'
    GEN_CLOSED ='GEN_CLOSED'
    GEN_CREATED ='GEN_CREATED'
    GEN_SUSPENDED ='GEN_SUSPENDED'

    @staticmethod
    def getgeneratorstate(gtor):
        if gtor.gi_running:
            return GtorState.GEN_RUNNING

        if gtor.gi_frame is None:
            return GtorState.GEN_CLOSED

        if gtor.gi_frame.f_lasti == -1:
            return GtorState.GEN_CREATED

        return GtorState.GEN_SUSPENDED
    #end-def

def coro000():
    """ a coroutine that does little 

    """ 
    print('-> coroutine started')
    x =yield
    print('-> coroutine received ', x)


class Test_Coro(unittest.TestCase):
    def test_coro000(self):

        my_coro000 =coro000()
        self.assertEqual( GtorState.getgeneratorstate(my_coro000), GtorState.GEN_CREATED)

        next(my_coro000)  # prints '-> coroutine started'
        self.assertEqual( GtorState.getgeneratorstate(my_coro000), GtorState.GEN_SUSPENDED)

        try:
            my_coro000.send(42)  # prints '-> coroutine received 42 
            self.assertEqual( GtorState.getgeneratorstate(my_coro000), GtorState.GEN_SUSPENDED)

            self.fail('should have raised StopIteration ')

        except StopIteration:
            self.assertTrue(True, 'On exit a coroutine will throw StopIteration')
            self.assertEqual( GtorState.getgeneratorstate(my_coro000), GtorState.GEN_CLOSED)
于 2017-10-25T10:15:06.897 回答