from consecution import Node

class LogNode(Node):
    def process(self, item):
        # any logic you want for processing single item 
        print('{: >15} processing {}'.format(self.name, item))

        # send item downstream
        self.push(item)

from consecution import Node, Pipeline

# This is the same node class we defined above
class LogNode(Node):
    def process(self, item):
        print('{} processing {}'.format(self.name, item))
        self.push(item)

# Connect nodes with pipe symbols to create pipeline for consuming any iterable.
pipe = Pipeline(
    LogNode('extract') | LogNode('transform') | LogNode('load')
)

# Create a pipeline.png file in your working directory
pipe.plot()  

# Interactively display the pipeline visualization in an IPython notebook
# by simply making the final expression in a cell evaluate to a pipeline.
pipe

print(pipe)

pipe.consume(range(5))

from consecution import Node, Pipeline

class LogNode(Node):
    def process(self, item):
        print('{} processing {}'.format(self.name, item))
        self.push(item)

# pipe to a list of nodes to broadcast items
pipe = Pipeline(
    LogNode('extract') 
    | LogNode('transform') 
    | [LogNode('load_redis'), LogNode('load_postgres'), LogNode('load_mongo')]
)
pipe.plot()
pipe.consume(range(2))

from consecution import Node, Pipeline

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)
        
def parity(item):
    if item % 2 == 0:
        return 'transform_even'
    else:
        return 'transform_odd'

# pipe to a list containing a callable to achieve routing behaviour
pipe = Pipeline(
    LogNode('extract') 
    | [LogNode('transform_even'), LogNode('transform_odd'), parity] 
)
pipe.plot()
pipe.consume(range(4))

from consecution import Node, Pipeline

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)
        
def parity(item):
    if item % 2 == 0:
        return 'transform_even'
    else:
        return 'transform_odd'

# piping from a list back to a single node merges items into downstream node
pipe = Pipeline(
    LogNode('extract') 
    | [LogNode('transform_even'), LogNode('transform_odd'), parity] 
    | LogNode('load')
)
pipe.plot()
pipe.consume(range(4))

from consecution import Node, Pipeline

class LogNode(Node):
    def begin(self):
        self.sum = 0
        print('{}.begin()'.format(self.name))

    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.sum += item
        self.push(item)

    def end(self):
        print('sum = {:d} in {}.end()'.format(self.sum, self.name))

# Identical pipeline to merge example above, but with modified LogNode
pipe = Pipeline(
    LogNode('extract') 
    | [LogNode('transform_even'), LogNode('transform_odd'), parity] 
    | LogNode('load')
)
pipe.consume(range(4))

from consecution import Node, Pipeline, GlobalState

class LogNode(Node):
    def process(self, item):
        self.global_state.messages.append(
            '{: >15} processing {}'.format(self.name, item)
        )
        self.push(item)
        
# create a global state object with a messages attribute
global_state = GlobalState(messages=[])

# Assign the predefined global_state to the pipeline
pipe = Pipeline(
    LogNode('extract') | LogNode('transform') | LogNode('load'),
    global_state=global_state)
)
pipe.consume(range(3))

# print the content of the global state message list
for msg in pipe.global_state.messages:
    print msg

from consecution import Node, Pipeline
class Mapper(Node):
    def process(self, item):
        self.push(2 * item)

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)

pipe = Pipeline(
    LogNode('extractor') | Mapper('mapper') | LogNode('loader')
)

pipe.consume(range(3))

from consecution import Node, Pipeline
class Reducer(Node):
    def begin(self):
        self.result = 0
        
    def process(self, item):
        self.result += item
        
    def end(self):
        self.push(self.result)

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)

pipe = Pipeline(
    LogNode('extractor') | Reducer('reducer') | LogNode('loader')
)

pipe.consume(range(3))

from consecution import Node, Pipeline
class Filter(Node):
    def process(self, item):
        if item > 3:
            self.push(item)

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)

pipe = Pipeline(
    LogNode('extractor') | Filter('filter') | LogNode('loader')
)

pipe.consume(range(6))

class LogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)

class Batcher(GroupByNode):
    def key(self, item):
        return item // 4
    
    def process(self, batch):
        sum_val = sum(batch)
        self.push(sum_val)
        
pipe = Pipeline(
    Batcher('batcher') | LogNode('logger') 
)

pipe.consume(range(16))

# This function defines a pipeline that can use swappable processing nodes.
# We don't worry about how we are going to do logging or aggregating.
# We just focus on how the nodes are connected.
def pipeline_factory(log_node, agg_node):
    pipe = Pipeline(
        log_node('extractor') | agg_node('aggregator') | log_node('result_logger')
    )
    return pipe


# Now we define a node for left-justified logging
class LeftLogNode(Node):
    def process(self, item):
        print('{: <15} processing {}'.format(self.name, item))
        self.push(item)

# And one for right-justified logging
class RightLogNode(Node):
    def process(self, item):
        print('{: >15} processing {}'.format(self.name, item))
        self.push(item)

# We can aggregate by summing
class SumNode(Node):
    def begin(self):
        self.result = 0
        
    def process(self, item):
        self.result += item
        
    def end(self):
        self.push(self.result)

# Or we can aggregate by multiplying
class ProdNode(Node):
    def begin(self):
        self.result = 1
        
    def process(self, item):
        self.result *= item
        
    def end(self):
        self.push(self.result)


# Now we plug in nodes to create a pipeline that left-prints sums
sum_pipeline = pipeline_factory(log_node=LeftLogNode, agg_node=SumNode)

# And a different pipeline that right prints products
prod_pipeline = pipeline_factory(log_node=RightLogNode, agg_node=ProdNode)

print 'aggregate with sum, left justified\n' + '-'*40
sum_pipeline.consume(range(1, 5))

print '\naggregate with product, right justified\n' + '-'*40
prod_pipeline.consume(range(1, 5))

from __future__ import print_function
from collections import namedtuple
from pprint import pprint
import csv
from consecution import Node, Pipeline, GlobalState

# Named tuples are nice immutable containers 
# for passing data between nodes
Person = namedtuple('Person', 'gender age spent')

# Create a pipeline that aggregates by gender and age
# In creating the pipeline we focus on connectivity and don't
# worry about defining node behavior.
def pipe_factory(Extractor, Agg, gender_router, age_router):
    # Consecution provides a generic GlobalState class.  Any object can be used
    # as the global_state in a pipeline, but the GlobalState object provides a
    # nice abstraction where attributes can be accessed either by dot notation
    # (e.g. global_state.my_attribute) or by dictionary notation (e.g.
    # global_state['my_attribute'].  Furthermore, GlobalState objects can be
    # instantiated with initialized attributes using key-word arguments as shown
    # here.
    global_state = GlobalState(segment_totals={})

    # Notice, we haven't even defined the behavior of these nodes yet.  They
    # will be defined later and are, for now, just passed into the factory
    # function as arguments while we focus on getting the topology right.
    pipe = Pipeline(
        Extractor('make_person') |
        [
            gender_router,
            (Agg('male') | [age_router, Agg('male_child'), Agg('male_adult')]),
            (Agg('female') | [age_router, Agg('female_child'), Agg('female_adult')]),
        ],
        global_state=global_state
    )

    # Nodes can be created outside of a pipeline definition
    adult = Agg('adult')
    child = Agg('child')
    total = Agg('total')

    # Sometimes the topology you want to create cannot easily be expressed
    # using the pipeline abstraction for wiring nodes together.  You can
    # drop down to a lower level of abstraction by explicitly wiring nodes 
    # together using the .add_downstream() method.
    adult.add_downstream(total)
    child.add_downstream(total)

    # Once a pipeline has been created, you can access individual nodes
    # with dictionary-like indexing on the pipeline.
    pipe['male_child'].add_downstream(child)
    pipe['female_child'].add_downstream(child)
    pipe['male_adult'].add_downstream(adult)
    pipe['female_adult'].add_downstream(adult)

    return pipe

# Now that we have the topology of our pipeline defined, we can think about the
# logic that needs to go into each node.  We start by defining a node that takes
# a row from a csv file and tranforms it into a namedtuple.
class MakePerson(Node):
    def process(self, item):
        item['age'] = int(item['age'])
        item['spent'] = float(item['spent'])
        self.push(Person(**item))

# We now define a node to perform our aggregations.  Mutable global state comes
# with a lot of baggage and should be used with care.  This node illustrates
# how to use global state to put all aggregations in a central location that
# remains accessible when the pipeline finishes processing.
class Sum(Node):
    def begin(self):
        # initialize the node-local sum to zero
        self.total = 0

    def process(self, item):
        # increment the node-local total and push the item down stream
        self.total += item.spent
        self.push(item)

    def end(self):
        # when pipeline is done, update global state with sum
        self.global_state.segment_totals[self.name] = round(self.total, 2)


# This function routes tuples based on their associated gender
def by_gender(item):
    return '{}'.format(item.gender)

# This function routes tuples based on whether the purchaser was an adult or
# child
def by_age(item):
    if item.age >= 18:
        return '{}_adult'.format(item.gender)
    else:
        return '{}_child'.format(item.gender)

# Here we plug our node definitions into our topology to create a fully-defined
# pipeline.
pipe = pipe_factory(MakePerson, Sum, by_gender, by_age)

# We can now visualize pipeline.
pipe.plot()

# Now we feed our pipeline with rows from the csv file
with open('sample_data.csv') as f:
    pipe.consume(csv.DictReader(f))

# The global_state is also available as an attribute on the pipeline allowing
# us to access it when the pipeline is finished.  This is a good way to "return"
# an object from a pipeline.  Here we simply print the result.
print()
pprint(pipe.global_state.segment_totals)