Coroutines and argument tupling

[Marshall T. Vandegrift]

Hi,

I'm trying to write a decorator which allows one to produce simple
coroutines by just writing a function as a generator expression which
re-receives it's arguments as a tuple from each yield. For example:

@coroutine
def nextn(n=1):
values = []
for i in itertools.count():
while n <= 0:
(n,) = (yield values)
values = []
values.append(i)
n = n - 1

print nextn() # => [0]
print nextn(3) # => [1, 2, 3]
print nextn(n=2) # => [4, 5]
print nextn() # => [6]

I've got this working, but have two questions. First, is there a better
way to generically transform function arguments into a tuple than I'm
doing now? That way being with this pretty hideous chunk of code:

class ArgPacker(object):
def __init__(self, function):
args, varargs, varkw, defaults = inspect.getargspec(function)
self.args = args or []
self.varargs = (varargs is not None) and 1 or 0
self.varkw = (varkw is not None) and 1 or 0
self.nargs = len(self.args) + self.varargs + self.varkw
defaults = defaults or []
defargs = self.args[len(self.args) - len(defaults):]
self.defaults = dict([(k, v) for k, v in izip(defargs, defaults)])

def pack(self, *args, **kwargs):
args = list(args)
result = [None] * self.nargs
for i, arg in izip(xrange(len(self.args)), self.args):
if args:
result = args.pop(0)
elif arg in kwargs:
result = kwargs[arg]
del kwargs[arg]
elif arg in self.defaults:
result = self.defaults[arg]
else:
return None
if self.varargs:
result[len(self.args)] = args
elif args:
return None
if self.varkw:
result[-1] = kwargs
elif kwargs:
return None
return tuple(result)

I also tried a version using exec, which was much tighter, but used
exec.

Second, am I trying to hammer a nail with a glass bottle here? The
ugliness of the ArgPacker class makes me suspect that I should perhaps
just manually create and track a generator when I need a function with
generator-like properties.

Thanks!

-Marshall

 

[Bjoern Schliessmann]

I'm trying to write a decorator which allows one to produce simple
coroutines by just writing a function as a generator expression
which re-receives it's arguments as a tuple from each yield.

May I ask why? Passing it the same arguments over and over is no
use; and there is the send method.

The ugliness of the ArgPacker class makes me suspect that I should
perhaps just manually create and track a generator when I need a
function with generator-like properties.

What do you mean? I don't quite understand why you'd have to "track"
a generator for getting generator-like properties.

Regards,


Björn

 

[Marshall T. Vandegrift]

May I ask why? Passing it the same arguments over and over is no
use; and there is the send method.

That's what I meant. The wrapper produced by the decorator passes the
arguments back into the generator as a tuple via the `send' method.

What do you mean? I don't quite understand why you'd have to "track"
a generator for getting generator-like properties.

Using the trivial `nextn' example from my original post with my
decorator lets you do just:

print nextn(2) # => [0, 1]
print nextn(3) # => [2, 3, 4]
print nextn() # => [5]

Without the decorator that becomes:

gen = nextn(2)
print gen.next() # => [0, 1]
print gen.send(3) # => [2, 3, 4]
print gen.send(1) # => [5]

The former is just that smidgen nicer, and allows you to continue to
make use of argument defaults and varadic arguments if so desired.

-Marshall

 

[attn.steven.kuo]

May I ask why? Passing it the same arguments over and over is no
use; and there is the send method.

That's what I meant. The wrapper produced by the decorator passes the
arguments back into the generator as a tuple via the `send' method.

What do you mean? I don't quite understand why you'd have to "track"
a generator for getting generator-like properties.

Using the trivial `nextn' example from my original post with my
decorator lets you do just:

print nextn(2) # => [0, 1]
print nextn(3) # => [2, 3, 4]
print nextn() # => [5]

Without the decorator that becomes:

gen = nextn(2)
print gen.next() # => [0, 1]
print gen.send(3) # => [2, 3, 4]
print gen.send(1) # => [5]

The former is just that smidgen nicer, and allows you to continue to
make use of argument defaults and varadic arguments if so desired.

Do you really need a generator or co-routine to do this? Maybe
you can just use a closure:

import itertools
class Foo(object):
it = itertools.count(0)

def __call__(self):
return self.y

def __init__(self, n=1):
self.y = [ Foo.it.next() for each in xrange(n) ]

def nextn(n=1):
return Foo(n)()

print nextn()
print nextn(3)
print nextn(n=2)
print nextn()

 

[Marshall T. Vandegrift]

Do you really need a generator or co-routine to do this? Maybe
you can just use a closure:

For my trivial example, sure -- there are lots of ways to do it. Here's
a slightly better example: the `read' method of a file-like object which
sequentially joins the contents of a sequence of other file-like
objects:

@coroutine
def read(self, size=-1):
data = ''
for file in self.files:
this = file.read(size - len(data))
data = ''.join([data, this])
while this:
if len(data) == size:
_, size = (yield data)
data = ''
this = file.read(size - len(data))
data = ''.join([data, this])
yield data
while True:
yield ''

-Marshall

 

[Bjoern Schliessmann]

Without the decorator that becomes:

gen = nextn(2)
print gen.next() # => [0, 1]
print gen.send(3) # => [2, 3, 4]
print gen.send(1) # => [5]

The former is just that smidgen nicer, and allows you to continue
to make use of argument defaults and varadic arguments if so
desired.

The solution I'd use is a decorator that calls next automatically
one time after instantiation. Then you can use send normally, and
don't have to care about any initial parameters, which makes the
code clearer (initial parameters should be used for setup purposes,
but not for the first iteration, IMHO). It'd look like this (from
PEP 342, http://www.python.org/dev/peps/pep-0342/):

def consumer(func):
def wrapper(*args,**kw):
gen = func(*args, **kw)
gen.next()
return gen
wrapper.__name__ = func.__name__
wrapper.__dict__ = func.__dict__
wrapper.__doc__ = func.__doc__
return wrapper

@consumer
def nextn():
...

gen = nextn()
print gen.send(2) # => [0, 1]
print gen.send(3) # => [2, 3, 4]
print gen.send(1) # => [5]

Regards,


Björn

 

[Marshall T. Vandegrift]

The solution I'd use is a decorator that calls next automatically one
time after instantiation. Then you can use send normally, and don't
have to care about any initial parameters, which makes the code
clearer (initial parameters should be used for setup purposes, but not
for the first iteration, IMHO). It'd look like this (from PEP 342,
http://www.python.org/dev/peps/pep-0342/):

I'd seen the consumer decorator, and it certainly is cleaner than just
using a generator. I don't like how it hides the parameter signature in
the middle of the consumer function though, and it also doesn't provide
for argument default values. It's the difference between:

...
def __init__(self, ...):
...
self.consumer = self._function(value)
...

def function(self, first, second=3, *args, **kwargs):
self.consumer.send((first, second, args, kwargs))

@consumer
def _function(self, setup):
...
first, second, args, kwargs = yield # initial 'next'
while condition:
...
first, second, args, kwargs = yield retval

Versus just:

@coroutine
def function(self, first, second=3, *args, **kwargs):
...
while condition:
...
first, second, args, kwargs = yield retval

Thanks in any case for the replies! Since I've apparently decided my
ArgPacker is worth it, the complete code for my coroutine decorator
follows.

-Marshall


import inspect
import types
import functools
from itertools import izip

__all__ = [ 'coroutine' ]

class ArgPacker(object):
def __init__(self, function):
args, varargs, varkw, defaults = inspect.getargspec(function)
self.args = args or []
self.varargs = (varargs is not None) and 1 or 0
self.varkw = (varkw is not None) and 1 or 0
self.nargs = len(self.args) + self.varargs + self.varkw
defaults = defaults or []
defargs = self.args[len(self.args) - len(defaults):]
self.defaults = dict([(k, v) for k, v in izip(defargs, defaults)])

def pack(self, *args, **kwargs):
args = list(args)
result = [None] * self.nargs
for i, arg in izip(xrange(len(self.args)), self.args):
if args:
result = args.pop(0)
elif arg in kwargs:
result = kwargs[arg]
del kwargs[arg]
elif arg in self.defaults:
result = self.defaults[arg]
else:
return None
if self.varargs:
result[len(self.args)] = args
elif args:
return None
if self.varkw:
result[-1] = kwargs
elif kwargs:
return None
return tuple(result)

class coroutine(object):
"""Convert a function to be a simple coroutine.

A simple coroutine is a generator bound to act as much as possible like a
normal function. Callers call the function as usual while the coroutine
produces new return values and receives new arguments with `yield'.
"""
def __init__(self, function):
self.function = function
self.gname = ''.join(['__', function.__name__, '_generator'])
self.packer = ArgPacker(function)
coroutine = self
def method(self, *args, **kwargs):
return coroutine.generate(self, self, *args, **kwargs)
self.method = method
functools.update_wrapper(self, function)
functools.update_wrapper(method, function)

def __get__(self, obj, objtype=None):
return types.MethodType(self.method, obj, objtype)

def __call__(self, *args, **kwargs):
return self.generate(self, *args, **kwargs)

def generate(self, obj, *args, **kwargs):
try:
generator = getattr(obj, self.gname)
except AttributeError:
generator = self.function(*args, **kwargs)
setattr(obj, self.gname, generator)
retval = generator.next()
else:
packed = self.packer.pack(*args, **kwargs)
if packed is None:
self.function(*args, **kwargs) # Should raise TypeError
raise RuntimeError("ArgPacker reported spurious error")
retval = generator.send(packed)
return retval

 

[Bjoern Schliessmann]

I'd seen the consumer decorator, and it certainly is cleaner than
just using a generator. I don't like how it hides the parameter
signature in the middle of the consumer function though, and it
also doesn't provide for argument default values.

Mh, that may be. I didn't use it much yet.

Thanks in any case for the replies! Since I've apparently decided
my ArgPacker is worth it, the complete code for my coroutine
decorator follows.

Thanks for reporting back, and happy coding.

Regards,


Björn