Ziplast operator

[andersea]

Just thought I'd share some ideas for streams that would be useful to me.

A ziplast stream. It would zip a number of async sequences, but yield a tuple for every item recieved on every stream with each element being the last element recieved on each stream. If no elements have yet been recieved on a stream, the value could be None on the corresponding tuple position. I am really often running into having to coordinate two sequences that yield values at different intervals, one every other second and another every ten seconds, for example.

Merge can kind of achieve this, if you are able tell each element apart, for example by tagging them somehow. If each item has different type, then it is doable, but if each item is a dict for example, then it becomes not so easy to tell where each item came from, unless you wrap it with some metadata.

Also a ratelimit stream. It yields a value at most every n seconds. The rest of the items are discarded.

[vxgmichel]

Just thought I'd share some ideas for streams that would be useful to me.

Great, thanks!

A ziplast stream

That would be the equivalent of ReactiveX combineLatest right?

Here's a possible implementation:

@operator(pipable=True)
def ziplatest(*sources):
    n = len(sources)

    # Add source index to the items
    new_sources = [
        stream.map(source, lambda x, i=i: {i: x})
        for i, source in enumerate(sources)]

    # Merge the sources
    merged = stream.merge.raw(*new_sources)

    # Accumulate the current state in a dict
    accumulated = stream.accumulate.raw(
        merged, lambda x, e: {**x, **e})

    # Convert the state dict to a tuple
    return stream.map.raw(
        accumulated, lambda x: tuple(map(x.get, range(n))))

And an example:

async def main():
    xs = stream.count(interval=2) | pipe.delay(2)
    ys = stream.count(interval=5) | pipe.delay(5)
    zs = xs | ziplatest.pipe(ys)
    ps = zs | pipe.print()
    await ps

With the corresponding output:

(0, None)
(1, None)
(1, 0)
(2, 0)
(3, 0)
(3, 1)
(4, 1)
(5, 1)
(6, 1)
(6, 2)
(7, 2)
[...]

I guess extra keywords arguments could also be provided, either to disable the producing of incomplete items, or to define a custom default value.

What do you think?

[andersea]

Awesome! That is exactly what I need.

[vxgmichel]

Great! I don't really have the time to write the docs and tests for this feature at the moment, so feel free to submit a PR if you're interested. Otherwise, I'll take care of it later.

[andersea]

Np, thanks for the effort. I really like this library and async generator functions is a pretty nice new python feature. It really cleans up a lot of my previous async context manager code in the project I am currently working at. You get a feeling that async/await is still a pretty young feature in python and there are definitely a few pitfalls to dodge here and there.

[vxgmichel]

Hi @andersea,

I just added the ziplatest operator (see PR #22), it will become available with version 0.3.1.

Thanks for the suggestion :)