# Path: featureflow/extractor.py
# class Node(object):
#     def __init__(self, needs=None):
#         super(Node, self).__init__()
#         if not inspect.isgeneratorfunction(self._process):
#             raise InvalidProcessMethod(self.__class__)
# 
#         self._cache = None
#         self._listeners = []
# 
#         self._needs = dictify(needs)
#         self._dependecy_names = dict(
#             (id(v), k) for k, v in list(self._needs.items()))
# 
#         for n in self.dependencies:
#             n.add_listener(self)
# 
#         self._finalized_dependencies = set()
#         self._enqueued_dependencies = set()
# 
#     @property
#     def dependencies(self):
#         if isinstance(self._needs, dict):
#             return list(self._needs.values())
#         return self._needs
# 
#     def _dependency_name(self, pusher):
#         return self._dependecy_names[id(pusher)]
# 
#     def __repr__(self):
#         return self.__class__.__name__
# 
#     def __str__(self):
#         return self.__repr__()
# 
#     def __enter__(self):
#         return self
# 
#     def __exit__(self, t, value, traceback):
#         pass
# 
#     @property
#     def version(self):
#         return self.__class__.__name__
# 
#     @property
#     def needs(self):
#         return self._needs
# 
#     @property
#     def dependency_count(self):
#         return len(self._needs)
# 
#     @property
#     def is_root(self):
#         return not self._needs
# 
#     @property
#     def is_leaf(self):
#         return not self._listeners
# 
#     def add_listener(self, listener):
#         self._listeners.append(listener)
# 
#     def find_listener(self, predicate):
#         for l in self._listeners:
#             return l if predicate(l) else l.find_listener(predicate)
#         return None
# 
#     def disconnect(self):
#         for e in self.dependencies:
#             try:
#                 e._listeners.remove(self)
#             except ValueError:
#                 pass
# 
#     def _enqueue(self, data, pusher):
#         self._cache = data
# 
#     def _dequeue(self):
#         if self._cache is None:
#             raise NotEnoughData()
# 
#         v, self._cache = self._cache, None
#         return v
# 
#     def _process(self, data):
#         yield data
# 
#     def _first_chunk(self, data):
#         return data
# 
#     def _last_chunk(self):
#         return iter(())
# 
#     def _finalize(self, pusher):
#         pass
# 
#     @property
#     def _finalized(self):
#         """
#         Return true if all dependencies have informed this node that they'll
#         be sending no more data (by calling _finalize()), and that they have
#         sent at least one batch of data (by calling enqueue())
#         """
#         return \
#             len(self._finalized_dependencies) >= self.dependency_count \
#             and len(self._enqueued_dependencies) >= self.dependency_count
# 
#     def _push(self, data, queue=None):
#         queue.appendleft((
#             id(self),
#             self.process.__name__,
#             {'pusher': self, 'data': data, 'queue': queue}))
# 
#     def _finish(self, pusher=None, queue=None):
#         self._finalize(pusher)
# 
#         if pusher in self.dependencies:
#             self._finalized_dependencies.add(id(pusher))
# 
#         if pusher:
#             return
#         queue.appendleft((
#             id(self),
#             self._finish.__name__,
#             {'pusher': self, 'queue': queue}))
# 
#     def process(self, data=None, pusher=None, queue=None):
#         if data is not None:
#             self._enqueued_dependencies.add(id(pusher))
#             self._enqueue(data, pusher)
# 
#         try:
#             inp = self._dequeue()
#             inp = self._first_chunk(inp)
#             self._first_chunk = lambda x: x
#             for d in self._process(inp):
#                 yield self._push(d, queue=queue)
#         except NotEnoughData:
#             yield None
# 
#         if self.is_root or self._finalized:
#             for chunk in self._last_chunk():
#                 yield self._push(chunk, queue=queue)
#             self._finish(pusher=None, queue=queue)
#             self._push(None, queue=queue)
#             yield None
# 
# class Aggregator(object):
#     """
#     A mixin for Node-derived classes that addresses the case when the processing
#     node cannot do its computation until all input has been received
#     """
# 
#     def __init__(self, needs=None):
#         super(Aggregator, self).__init__(needs=needs)
# 
#     def _dequeue(self):
#         if not self._finalized:
#             raise NotEnoughData()
#         return super(Aggregator, self)._dequeue()


import json
import bz2
from .extractor import Node, Aggregator
from dill import dumps, HIGHEST_PROTOCOL


class IdentityEncoder(Node):
    content_type = 'application/octet-stream'

    def __init__(self, needs=None):
        super(IdentityEncoder, self).__init__(needs=needs)

    def _seed(self):
        return b''

    def _enqueue(self, data, pusher):
        self._cache = data if data else self._seed()


class TextEncoder(IdentityEncoder):
    content_type = 'text/plain'

    def __init__(self, needs=None):
        super(TextEncoder, self).__init__(needs=needs)

    def _seed(self):
        return ''

    def _process(self, data):
        yield data.encode()



# Path: featureflow/extractor.py
# class Node(object):
#     def __init__(self, needs=None):
#         super(Node, self).__init__()
#         if not inspect.isgeneratorfunction(self._process):
#             raise InvalidProcessMethod(self.__class__)
# 
#         self._cache = None
#         self._listeners = []
# 
#         self._needs = dictify(needs)
#         self._dependecy_names = dict(
#             (id(v), k) for k, v in list(self._needs.items()))
# 
#         for n in self.dependencies:
#             n.add_listener(self)
# 
#         self._finalized_dependencies = set()
#         self._enqueued_dependencies = set()
# 
#     @property
#     def dependencies(self):
#         if isinstance(self._needs, dict):
#             return list(self._needs.values())
#         return self._needs
# 
#     def _dependency_name(self, pusher):
#         return self._dependecy_names[id(pusher)]
# 
#     def __repr__(self):
#         return self.__class__.__name__
# 
#     def __str__(self):
#         return self.__repr__()
# 
#     def __enter__(self):
#         return self
# 
#     def __exit__(self, t, value, traceback):
#         pass
# 
#     @property
#     def version(self):
#         return self.__class__.__name__
# 
#     @property
#     def needs(self):
#         return self._needs
# 
#     @property
#     def dependency_count(self):
#         return len(self._needs)
# 
#     @property
#     def is_root(self):
#         return not self._needs
# 
#     @property
#     def is_leaf(self):
#         return not self._listeners
# 
#     def add_listener(self, listener):
#         self._listeners.append(listener)
# 
#     def find_listener(self, predicate):
#         for l in self._listeners:
#             return l if predicate(l) else l.find_listener(predicate)
#         return None
# 
#     def disconnect(self):
#         for e in self.dependencies:
#             try:
#                 e._listeners.remove(self)
#             except ValueError:
#                 pass
# 
#     def _enqueue(self, data, pusher):
#         self._cache = data
# 
#     def _dequeue(self):
#         if self._cache is None:
#             raise NotEnoughData()
# 
#         v, self._cache = self._cache, None
#         return v
# 
#     def _process(self, data):
#         yield data
# 
#     def _first_chunk(self, data):
#         return data
# 
#     def _last_chunk(self):
#         return iter(())
# 
#     def _finalize(self, pusher):
#         pass
# 
#     @property
#     def _finalized(self):
#         """
#         Return true if all dependencies have informed this node that they'll
#         be sending no more data (by calling _finalize()), and that they have
#         sent at least one batch of data (by calling enqueue())
#         """
#         return \
#             len(self._finalized_dependencies) >= self.dependency_count \
#             and len(self._enqueued_dependencies) >= self.dependency_count
# 
#     def _push(self, data, queue=None):
#         queue.appendleft((
#             id(self),
#             self.process.__name__,
#             {'pusher': self, 'data': data, 'queue': queue}))
# 
#     def _finish(self, pusher=None, queue=None):
#         self._finalize(pusher)
# 
#         if pusher in self.dependencies:
#             self._finalized_dependencies.add(id(pusher))
# 
#         if pusher:
#             return
#         queue.appendleft((
#             id(self),
#             self._finish.__name__,
#             {'pusher': self, 'queue': queue}))
# 
#     def process(self, data=None, pusher=None, queue=None):
#         if data is not None:
#             self._enqueued_dependencies.add(id(pusher))
#             self._enqueue(data, pusher)
# 
#         try:
#             inp = self._dequeue()
#             inp = self._first_chunk(inp)
#             self._first_chunk = lambda x: x
#             for d in self._process(inp):
#                 yield self._push(d, queue=queue)
#         except NotEnoughData:
#             yield None
# 
#         if self.is_root or self._finalized:
#             for chunk in self._last_chunk():
#                 yield self._push(chunk, queue=queue)
#             self._finish(pusher=None, queue=queue)
#             self._push(None, queue=queue)
#             yield None
# 
# class Aggregator(object):
#     """
#     A mixin for Node-derived classes that addresses the case when the processing
#     node cannot do its computation until all input has been received
#     """
# 
#     def __init__(self, needs=None):
#         super(Aggregator, self).__init__(needs=needs)
# 
#     def _dequeue(self):
#         if not self._finalized:
#             raise NotEnoughData()
#         return super(Aggregator, self)._dequeue()


# Path: featureflow/encoder.py
import json
import bz2
from .extractor import Node, Aggregator
from dill import dumps, HIGHEST_PROTOCOL



class IdentityEncoder(Node):
    content_type = 'application/octet-stream'

    def __init__(self, needs=None):
        super(IdentityEncoder, self).__init__(needs=needs)

    def _seed(self):
        return b''

    def _enqueue(self, data, pusher):
        self._cache = data if data else self._seed()


class TextEncoder(IdentityEncoder):
    content_type = 'text/plain'

    def __init__(self, needs=None):
        super(TextEncoder, self).__init__(needs=needs)

    def _seed(self):
        return ''

    def _process(self, data):
        yield data.encode()



class JSONEncoder(Node):