My progress through the transducers in Python blog post series http://sixty-north.com/blog/series/understanding-transducers-through-python

transducers.py

square = lambda x: x*x
big = lambda x: x > 50
odd = lambda x: x % 2

# Stage 0
tuple(filter(big, map(square, range(10))))

# Stage 1
def my_map(fn, seq):
  return reduce(lambda result, x: result + (fn(x),), seq, ())

def my_filter(pred, seq):
  return reduce(lambda result, x: result + (x,) if pred(x) else result, seq, ())

my_filter(big, my_map(square, range(10)))

# Stage 2
def mapper(reducer, fn):
  return lambda result, x: reducer(result, fn(x))

def filterer(reducer, pred):
  def f(result, x):
    return reducer(result, x) if pred(x) else result
  return f

def collector(result, x):
  return result + (x,)

reduce(mapper(filterer(collector, big), square), range(10), ())

# Stage 3
def mapping(fn):
  return lambda reducer: lambda result, x: reducer(result, fn(x))

def filtering(pred):
  def filter_transducer(reducer):
    def filterer(result, x):
      return reducer(result, x) if pred(x) else result
    return filterer
  return filter_transducer

reduce(mapping(square)(filtering(big)(collector)), range(10), ())

# Stage 4
def compose(f, *fs):
  functions = (f,) + fs
  def composition(*args, **kwargs):
    initial = functions[0](*args, **kwargs)
    return reduce(lambda result, x: x(result), functions[1:], initial)
  return composition

reduce(compose(filtering(big), mapping(square))(collector), range(10), ())

# Stage 5
UNSET = object()

def transduce(transducer, reducer, items, initial=UNSET):
  reducer = transducer(reducer)
  if initial is UNSET:
    initial = reducer()
  return reducer(reduce(reducer, items, initial))

def collector(*args):
  if len(args) == 0:
    return ()
  elif len(args) == 1:
    result = args[0]
    return result
  else:
    result, inputs = args[0], args[1:]
    return result + inputs

def mapping(fn):
  def map_transducer(reducer):
    def mapper(*args):
      if len(args) < 2:
        return reducer(*args)
      else:
        result, input = args
        return reducer(result, fn(input))
    return mapper
  return map_transducer

def filtering(fn):
  def filter_transducer(reducer):
    def filterer(*args):
      if len(args) < 2:
        return reducer(*args)
      else:
        result, input = args
        if fn(input):
          return reducer(result, input)
        else:
          return result
    return filterer
  return filter_transducer

transduce(compose(filtering(big), mapping(square)), collector, range(10))

# Stage 6
def enumerating(start=0):
  def enumerating_transducer(reducer):
    state = {"i": start}
    def enumerator(*args):
      if len(args) < 2:
        return reducer(*args)
      else:
        result, input = args
        i = state["i"]
        state["i"] += 1
        return reducer(result, (i, input))
    return enumerator
  return enumerating_transducer

transduce(compose(enumerating(), mapping(square)), collector, range(10))

SKIP = object()

def transduce(transducer, reducer, items, initial=UNSET):
  reducer = transducer(reducer)
  accumulator = reducer() if initial is UNSET else intial
  for item in items:
    result = reducer(accumulator, item)
    if not result is SKIP:
      accumulator = result
  result = reducer(accumulator)
  return accumulator if result is SKIP else result

def collector(*args):
  if len(args) == 0:
    return ()
  elif len(args) == 1:
    result = args[0]
    return result
  else:
    result, inputs = args[0], args[1:]
    if result is SKIP:
      return SKIP
    else:
      return result + inputs

def grouping(group_size):
  def grouping_transducer(reducer):
    state = {"buffer": ()}
    def grouper(*args):
      if len(args) == 0:
        return reducer()
      elif len(args) == 1:
        result = args[0]
        if state["buffer"]:
          group = state["buffer"]
          state["buffer"] = ()
          return reducer(result, group)
        else:
          return reducer(result)
      else:
        result, inputs = args[0], args[1:]
        state["buffer"] += inputs
        if len(state["buffer"]) >= group_size:
          group = state["buffer"]
          state["buffer"] = ()
          return reducer(result, group)
        else:
          return SKIP
    return grouper
  return grouping_transducer

transduce(grouping(3), collector, range(8))

transduce(compose(enumerating(), grouping(3), filtering(big), mapping(square)),
          collector, range(15))

# Stage 7
class Reduced:
  def __init__(self, value):
    self.value = value

def transduce(transducer, reducer, items, initial=UNSET):
  reducer = transducer(reducer)
  accumulator = reducer() if initial is UNSET else intial
  for item in items:
    result = reducer(accumulator, item)
    if not result is SKIP:
      accumulator = result
    if isinstance(accumulator, Reduced):
      accumulator = accumulator.value
      break
  result = reducer(accumulator)
  return accumulator if result is SKIP else result

def firsting(predicate=UNSET):
  def firsting_transducer(reducer):
    def firster(*args):
      if len(args) < 2:
        return reducer(*args)
      else:
        result, input = args
        if predicate is UNSET or predicate(input):
          return Reduced(input)
        else:
          return SKIP
    return firster
  return firsting_transducer

transduce(compose(firsting(lambda i: i > 20), mapping(square), filtering(odd)), collector, range(10))

# Stage 8
def repeating(n=1):
  def repeating_transducer(reducer):
    def repeater(*args):
      if len(args) < 2:
        return reducer(*args)
      else:
        result, input = args
        for _ in range(n):
          new_result = reducer(result, input)
          if not new_result is SKIP:
            result = new_result
        return result
    return repeater
  return repeating_transducer

transduce(compose(grouping(3), repeating(4)), collector, range(20))

# Stage 9
def returner(*args):
  if len(args) == 1:
    raise(StopIteration)
  elif len(args) == 2:
    result, input = args
    if result is SKIP:
      return SKIP
    else:
      return input

def lazy_transduce(transducer, reducer, items, initial=UNSET):
  reducer = transducer(reducer)
  accumulator = reducer() if initial is UNSET else intial
  for item in items:
    result = reducer(accumulator, item)
    if not result is SKIP:
      accumulator = result
      if isinstance(accumulator, Reduced):
        accumulator = accumulator.value
        yield accumulator
        break
      yield accumulator
  accumulator = reducer(accumulator)
  if not accumulator is SKIP:
    yield accumulator

def logging_range(n):
  for i in range(n):
    print("i =", i)
    yield i

lazy_transduce(mapping(square), returner, logging_range(10))
[x for x in lazy_transduce(mapping(square), returner, logging_range(10))]
[g for g in lazy_transduce(grouping(3), returner, logging_range(8))]
next(lazy_transduce(firsting(lambda x: x > 5), returner, logging_range(10)))

# Stage 10
def push_transduce(transducer, reducer, initial=UNSET):
  reducer = transducer(reducer)
  accumulator = reducer() if initial is UNSET else intial
  while True:
    try:
      item = yield
    except GeneratorExit:
      break
    result = reducer(accumulator, item)
    if not result is SKIP:
      accumulator = result
      if isinstance(accumulator, Reduced):
        accumulator = accumulator.value
        yield accumulator
        break
      yield accumulator
  accumulator = reducer(accumulator)
  if not accumulator is SKIP:
    # FIXME - We can't yield anything here, as the generator has already closed!
    #         "RuntimeError: generator ignored GeneratorExit"
    yield accumulator