intersticial to merge with scott

tests/test_components.py

@@ -46,7 +46,6 @@ class ComponentTestCase(TestCase):
         setup_logger(self)
 
     def tearDown(self):
-        #self.ctx.term()
         teardown_logger(self)
 
     def test_source(self):
@@ -74,6 +73,9 @@ class ComponentTestCase(TestCase):
         )
 
         launch_monitor(monitor)
+        iter_a = iter(comp_a)
+        ev = iter_a.next()
+        return
 
         for event in comp_a:
             log.info(event)
@@ -142,16 +144,8 @@ class ComponentTestCase(TestCase):
             DATASOURCE_UNFRAME
         )
 
-        names = [
-                    comp_a.get_id,
-                    comp_b.get_id,
-                    comp_c.get_id
-                ]
-
-        monitor.manage(set(names))
         launch_monitor(monitor)
-
-        sorted_out = date_sorted_sources([comp_a, comp_b, comp_c])
+        sorted_out = date_sorted_sources(comp_a, comp_b, comp_c)
 
         prev = None
         sort_count = 0

zipline/core/component.py

@@ -98,6 +98,11 @@ class Component(object):
         self.guid                   = uuid.uuid4()
         self.huid                   = humanhash.humanize(self.guid.hex)
 
+        # ------------
+        # Generator
+        # ------------
+        self.gen                    = None
+
 
     # ------------
     # Core Methods
@@ -188,7 +193,10 @@ class Component(object):
         return self.run_safe(self._run_in)
 
     def __iter__(self):
-        return self._launch()
+        if not self.gen:
+            self.gen = self._launch()
+
+        return self.gen
 
     # ----------------------------
     #  Cleanup & Modes of Failure
@@ -601,6 +609,9 @@ class Component(object):
         """
         return self.prefix + self.component_id
 
+    def get_hash(self):
+        return self.component_id
+
     def debug(self):
         """
         Debug information about the component.

zipline/gens/composites.py

@@ -11,25 +11,29 @@ from zipline.gens.transform import stateful_transform
 SourceBundle = namedtuple("SourceBundle", ['source', 'args', 'kwargs'])
 TransformBundle = namedtuple("TransformBundle", ['tnfm', 'args', 'kwargs'])
 
-def date_sorted_sources(*sources):
+def date_sorted_sources(bundles):
     """
     Takes an iterable of SortBundles, generating namestrings and initialized datasources
     for each before piping them into a date_sort.
     """
+    assert isinstance(bundles, (list, tuple))
+    for bundle in bundles:
+        assert isinstance(bundle, SourceBundle)
 
-    for source in sources:
-        assert iter(source), "Source %s not iterable" % source
-        assert source.__class__.__dict__.has_key('get_hash'), "No get_hash"
+    # Calculate namestring hashes to pass to date_sort.
+    names = [bundle.source.__name__ + hash_args(*bundle.args, **bundle.kwargs)
+             for bundle in bundles]
 
-    # Get name hashes to pass to date_sort.
-    names = [source.get_hash() for source in sources]
+    # Pass each source its arguments.
+    source_gens = [bundle.source(*bundle.args, **bundle.kwargs)
+                   for bundle in bundles]
 
     # Convert the list of generators into a flat stream by pulling
     # one element at a time from each.
-    stream_in = roundrobin(sources, names)
-    
+    stream_in = roundrobin(source_gens, names)
+
     # Guarantee the flat stream will be sorted by date, using source_id as
-    # tie-breaker, which is fully deterministic (given deterministic string 
+    # tie-breaker, which is fully deterministic (given deterministic string
     # representation for all args/kwargs)
     return date_sort(stream_in, names)
 
@@ -50,15 +54,15 @@ def merged_transforms(sorted_stream, bundles):
 
     # Create a copy of the stream for each transform.
     split = tee(sorted_stream, len(bundles))
-    # Package a stream copy with each bundle 
+    # Package a stream copy with each bundle
     tnfms_with_streams = zip(split, bundles)
 
     # Convert the copies into transform streams.
     tnfm_gens = [
         stateful_transform(
-            stream_copy, 
-            bundle.tnfm, 
-            *bundle.args, 
+            stream_copy,
+            bundle.tnfm,
+            *bundle.args,
             **bundle.kwargs
         )
         for stream_copy, bundle in tnfms_with_streams