Transducers implemention with map, filter, and concat.

transducers.py

try:
    from collections.abc import Iterable  # noqa
except ImportError:
    from collections import Iterable  # noqa


def reducer(init, rfn):
    def ident(x):
        return x

    return init, ident, rfn


def tx_map(fn):
    def transducer(r):
        init, ident, reduce = r

        def map_reduce(acc, x):
            return reduce(acc, fn(x))

        return init, ident, map_reduce

    return transducer


def tx_filter(pred):
    def transducer(r):
        init, ident, reduce = r

        def filter_reduce(acc, x):
            if pred(x):
                return reduce(acc, x)
            return acc

        return init, ident, filter_reduce

    return transducer


def tx_concat():
    def transducer(r):
        init, ident, reduce = r

        def concat_init():
            return init()

        def concat_reduce(acc, xs):
            if isinstance(xs, (str, bytes)) or not isinstance(xs, Iterable):
                return reduce(acc, xs)

            it = iter(xs)
            try:
                x = next(it)
                return concat_reduce(reduce(acc, x), it)
            except StopIteration:
                return acc

        return reducer(concat_init, concat_reduce)

    return transducer


def tx_comp(*xforms):
    def composite(r):
        for xform in xforms[::-1]:
            r = xform(r)
        return r

    return composite


def transduce(xform, r, xs):
    init, ident, reduce = xform(r)
    if callable(init):
        acc = init()
    else:
        acc = init
    for x in xs:
        acc = reduce(acc, x)
    return ident(acc)


sum_reducer = reducer(lambda: 0, lambda acc, x: acc + x)

coll = [[1], 2, [], [3, 4], 5]
xform = tx_comp(
    tx_concat(),
    tx_filter(lambda x: x % 2 == 0),
    tx_map(lambda x: x**2),
)

result = transduce(xform, sum_reducer, coll)
print(result)  # prints 20 = (2 ** 2 + 4 ** 2)

push_reducer = reducer(lambda: [], lambda acc, x: [*acc, x])
result2 = transduce(xform, push_reducer, coll)
print(result2)  # prints [4, 16]


print(transduce(tx_concat(), push_reducer, [0, [1, 2, [3, 4]]]))


def trace(x):
    print(x)
    return x

def my_stream():
    import time
    count = 0
    while True:
        time.sleep(1)
        count+=1
        yield count



noop = reducer(lambda: None, lambda acc, x: None)
transduce(tx_comp(xform, tx_map(trace)), noop, my_stream())