Merge pull request #4 from FAST-HEP/BK_add_future_instantiations

Add option for future instantiation of stages

example/fflow_eg_mods.py

@@ -5,7 +5,7 @@
 
 
 class Generator():
-    def __init__(self, mean, variance, quantity):
+    def __init__(self, name, out_dir, mean, variance, quantity):
         self.mean = mean
         self.variance = variance
         self.quantity = quantity
@@ -15,7 +15,7 @@ def __call__(self, data):
 
 
 class Summarize():
-    def __init__(self, methods, replace_values=False):
+    def __init__(self, name, out_dir, methods, replace_values=False):
         self.methods = {m: getattr(np, m) for m in methods}
         self.replace_values = replace_values
 

fast_flow/__init__.py

@@ -1,3 +1,3 @@
 import logging
 logging.basicConfig()
-__version__ = "0.2.1"
+__version__ = "0.3.0"

fast_flow/v1/__init__.py

@@ -1,13 +1,21 @@
 from __future__ import absolute_import
-from .dict_config import read_sequence_dict
+from .dict_config import read_sequence_dict, compile_sequence_dict
 from .yaml_config import config_dict_from_yaml
 
 
-__all__ = ["read_sequence_yaml", "read_sequence_dict"]
+__all__ = ["read_sequence_yaml", "compile_sequence_yaml",
+           "read_sequence_dict", "compile_sequence_dict"]
 
 
 def read_sequence_yaml(cfg_filename, output_dir=None, backend=None):
     cfg = config_dict_from_yaml(cfg_filename,
                                 output_dir=output_dir,
                                 backend=backend)
     return read_sequence_dict(**cfg)
+
+
+def compile_sequence_yaml(cfg_filename, output_dir=None, backend=None):
+    cfg = config_dict_from_yaml(cfg_filename,
+                                output_dir=output_dir,
+                                backend=backend)
+    return compile_sequence_dict(**cfg)

fast_flow/v1/dict_config.py

@@ -21,18 +21,38 @@ class BadStageList(BadConfig):
 
 
 def read_sequence_dict(stages, general={}, **stage_descriptions):
+    return read_sequence_dict_internal(stages, general,
+                                       stage_descriptions,
+                                       return_future=False)
+
+
+def compile_sequence_dict(stages, general={}, **stage_descriptions):
+    sequence = read_sequence_dict_internal(stages, general,
+                                           stage_descriptions,
+                                           return_future=True)
+
+    def build():
+        return [s() for s in sequence]
+
+    return build
+
+
+def read_sequence_dict_internal(stages, general={},
+                                stage_descriptions={},
+                                return_future=False):
     output_dir = general.get("output_dir", os.getcwd())
     default_module = general.get("backend", None)
     if default_module:
         default_module = importlib.import_module(default_module)
     stages = _create_stages(stages, output_dir, stage_descriptions,
                             this_dir=general.get("this_dir", None),
-                            default_module=default_module)
+                            default_module=default_module,
+                            return_future=return_future)
     return stages
 
 
 def _create_stages(stages, output_dir, stage_descriptions,
-                   this_dir=None, default_module=None):
+                   this_dir=None, default_module=None, return_future=False):
     if not isinstance(stages, list):
         msg = "Bad stage list: Should be a list"
         logger.error(msg + ", but instead got a '{}'".format(type(stages)))
@@ -41,42 +61,57 @@ def _create_stages(stages, output_dir, stage_descriptions,
     for i, stage_cfg in enumerate(stages):
         name, stage_type = infer_stage_name_class(i, stage_cfg)
         if name == "IMPORT":
-            out_stages += import_yaml(stage_type, output_dir, this_dir)
+            out_stages += import_yaml(stage_type, output_dir, this_dir,
+                                      return_future=return_future)
             continue
 
         out_stages += instantiate_stage(name, stage_type, output_dir,
                                         stage_descriptions=stage_descriptions,
-                                        default_module=default_module)
+                                        default_module=default_module,
+                                        return_future=return_future,
+                                        )
     return out_stages
 
 
-def import_yaml(filepath, output_dir, this_dir):
+def import_yaml(filepath, output_dir, this_dir, return_future=False):
     filepath = filepath.format(this_dir=this_dir)
     cfg = config_dict_from_yaml(filepath, output_dir=output_dir)
-    return read_sequence_dict(**cfg)
+    stages = cfg.pop("stages")
+    general = cfg.pop("general", {})
+    return read_sequence_dict_internal(stages, general,
+                                       stage_descriptions=cfg,
+                                       return_future=return_future)
 
 
-def instantiate_stage(name, stage_type, output_dir, stage_descriptions, default_module=None):
+def instantiate_stage(name, stage_type, output_dir, stage_descriptions,
+                      default_module=None, return_future=False):
     stage_class = get_stage_class(stage_type, default_module, raise_exception=False)
     if not stage_class:
         raise BadStagesDescription("Unknown type for stage '{}': {}".format(name, stage_type))
-    result = _configure_stage(name, stage_class, output_dir, stage_descriptions)
+    result = _configure_stage(name, stage_class, output_dir,
+                              stage_descriptions, return_future=return_future)
     return [result]
 
 
-def _configure_stage(name, stage_class, out_dir, stage_descriptions):
+def _configure_stage(name, stage_class, out_dir,
+                     stage_descriptions, return_future=False):
     cfg = stage_descriptions.get(name, None)
     if cfg is None:
         raise BadStagesDescription("Missing description for stage '{}'".format(name))
-    if isinstance(cfg, dict):
-        cfg.setdefault("name", name)
-        cfg.setdefault("out_dir", out_dir)
-        stage = stage_class(**cfg)
-    elif isinstance(cfg, list):
-        stage = stage_class(*cfg)
-    else:
-        stage = stage_class(cfg, name=name)
-    return stage
+
+    def stage():
+        if isinstance(cfg, dict):
+            cfg.setdefault("name", name)
+            cfg.setdefault("out_dir", out_dir)
+            return stage_class(**cfg)
+        elif isinstance(cfg, list):
+            return stage_class(*cfg)
+        else:
+            return stage_class(cfg, name=name)
+
+    if return_future:
+        return stage
+    return stage()
 
 
 def infer_stage_name_class(index, stage_cfg):

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.1
+current_version = 0.3.0
 commit = True
 tag = False
 

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="fast-flow",
-    version="0.2.1",
+    version="0.3.0",
     author="Ben Krikler",
     author_email="[email protected]",
     description="YAML-based analysis flow description language",

tests/test_dict_config.py

@@ -114,3 +114,16 @@ def test_sequence_from_dict(a_stage_list, all_stage_configs, tmpdir):
     assert stages[0].an_int == 3
     assert stages[0].a_str == "hello world"
     assert len(stages[0].other_args) == 2
+
+
+def test_compile_sequence_from_dict(a_stage_list, all_stage_configs, tmpdir):
+    general = dict(backend="tests.fake_scribbler_to_test", output_dir=str(tmpdir))
+    compiled = dict_config.compile_sequence_dict(a_stage_list, general, **all_stage_configs)
+    stages = compiled()
+
+    assert len(stages) == 2
+    assert isinstance(stages[0], fakes.FakeScribblerArgs)
+    assert isinstance(stages[1], fakes.FakeScribbler)
+    assert stages[0].an_int == 3
+    assert stages[0].a_str == "hello world"
+    assert len(stages[0].other_args) == 2

tests/test_module.py

@@ -0,0 +1,48 @@
+from __future__ import absolute_import
+import pytest
+import fast_flow.v1 as fast_flow
+from . import fake_scribbler_to_test as fakes
+
+
+@pytest.fixture
+def config_1(tmpdir):
+    content = """
+    general:
+        backend: tests.fake_scribbler_to_test
+        output_dir: %(tmpdir)s
+
+    stages:
+        - my_first_stage: tests.fake_scribbler_to_test.FakeScribbler
+        - my_second_stage: FakeScribblerArgs
+
+    my_first_stage: {}
+    my_second_stage:
+        an_int: 3
+        a_str: hello world
+        some_other_arg: True
+        yet_more_arg: [0, 1, 2]
+    """ % dict(tmpdir=str(tmpdir))
+    out_file = tmpdir / "config.yml"
+    out_file.write(content)
+    return out_file
+
+
+def test_read_sequence_yaml(config_1):
+    stages = fast_flow.read_sequence_yaml(str(config_1))
+    assert len(stages) == 2
+    assert isinstance(stages[0], fakes.FakeScribbler)
+    assert isinstance(stages[1], fakes.FakeScribblerArgs)
+    assert stages[1].an_int == 3
+    assert stages[1].a_str == "hello world"
+    assert len(stages[1].other_args) == 2
+
+
+def test_compile_sequence_yaml(config_1):
+    stages = fast_flow.compile_sequence_yaml(str(config_1))
+    stages = stages()
+    assert len(stages) == 2
+    assert isinstance(stages[0], fakes.FakeScribbler)
+    assert isinstance(stages[1], fakes.FakeScribblerArgs)
+    assert stages[1].an_int == 3
+    assert stages[1].a_str == "hello world"
+    assert len(stages[1].other_args) == 2