From c2b4689668a019cfc45bddd47f582ba744102123 Mon Sep 17 00:00:00 2001 From: fawce Date: Thu, 2 Aug 2012 18:02:43 -0400 Subject: [PATCH] intersticial to merge with scott --- tests/test_components.py | 14 ++++---------- zipline/core/component.py | 13 ++++++++++++- zipline/gens/composites.py | 30 +++++++++++++++++------------- 3 files changed, 33 insertions(+), 24 deletions(-) diff --git a/tests/test_components.py b/tests/test_components.py index 3626e54c..45f632fd 100644 --- a/tests/test_components.py +++ b/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 diff --git a/zipline/core/component.py b/zipline/core/component.py index 37d22d16..4cb4c0d0 100644 --- a/zipline/core/component.py +++ b/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. diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index af2fdc8c..b4ecde8d 100644 --- a/zipline/gens/composites.py +++ b/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