Merge branch 'main' into pipeline_with_processsteps

Author: toqduj

requirements.txt

@@ -10,3 +10,4 @@ stamina
 tox
 scipy
 uncertainties
+pyyaml

src/modacor/runner/pipeline.py

@@ -4,6 +4,7 @@
 
 from graphlib import TopologicalSorter
 from pathlib import Path
+import yaml
 
 from attrs import define, field
 from attrs import validators as v
@@ -26,9 +27,25 @@ def __attrs_post_init__(self):
         super().__init__(graph=self.graph)
 
     @classmethod
-    def from_json(cls, path_to_json: Path):
-        # functionality postponed
-        return cls(name="dummy")
+    def from_yaml(cls, path_to_yaml: Path):
+        """
+        Instantiate a Pipeline from a yaml configuration file.
+        """
+
+        yaml_obj = yaml.safe_load(path_to_yaml)
+        process_step_instances = {}
+        id_graph = {}
+        for module_name, module_data in yaml_obj["steps"].items():
+            # we need to instantiate ProcessSteps here, but
+            # need to implement a ProcessStep registry
+            step_id = module_data.get("step_id")
+            process_step_instances[step_id] = ProcessStep(io_sources=None)
+            id_graph[step_id] = module_data.get("requires_steps")
+        # translate step_id graph into ProcessStep graph
+        graph = {}
+        for k, v in id_graph.items():
+            graph[process_step_instances[k]] = {process_step_instances[i] for i in v}
+        return cls(name=yaml_obj["name"], graph=graph)
 
     @classmethod
     def from_dict(cls, graph_dict: dict, name=""):

src/modacor/tests/test_pipeline.py

@@ -1,5 +1,6 @@
 import pytest
 from pathlib import Path
+import yaml
 
 from ..runner.pipeline import Pipeline
 from ..dataclasses.process_step import ProcessStep
@@ -23,6 +24,47 @@ class DummyProcessStep:
     pass
 
 
+@pytest.fixture
+def yaml_one_step():
+    return """
+    name: one_step
+    steps:
+      multiply by xy:
+        module: PoissonUncertainties
+        step_id: 3
+        requires_steps: []
+        configuration:
+          multiplier: 3
+          signal: sample::signal
+        io_sources:
+        - sample
+    """
+
+
+@pytest.fixture
+def yaml_linear_pipeline():
+    return """
+    name: simple_pipeline
+    steps:
+      poisson:
+        module: PoissonUncertainties
+        step_id: 1
+      another poisson:
+        module: PoissonUncertainties
+        step_id: 2
+        requires_steps: [1]
+      multiply by xy:
+        module: PoissonUncertainties
+        step_id: 3
+        requires_steps: [2]
+        configuration:
+          multiplier: 3
+          signal: sample::signal
+        io_sources:
+        - sample
+    """
+
+
 def test_linear_pipeline(linear_pipeline):
     "tests the sequence is expected for a linear graph"
     pipeline = Pipeline(graph=linear_pipeline)
@@ -75,3 +117,16 @@ def test_diverging_branch_addition(
     pipeline.add_outgoing_branch(branch, branching_node)
     assert [*pipeline.static_order()] == [1, 2, 3, 6, 5]
     assert pipeline.graph == {3: {2, 1}, 2: {1}, 5: {6}, 6: {2}}
+
+
+def test_yaml_format(yaml_linear_pipeline):
+    yaml_obj = yaml.safe_load(yaml_linear_pipeline)
+    assert "steps" in yaml_obj
+    assert "poisson" in yaml_obj["steps"]
+    assert type(yaml_obj["steps"]["multiply by xy"]["configuration"]) == dict
+
+
+def test_pipeline_from_yaml(yaml_one_step):
+    pipeline = Pipeline.from_yaml(yaml_one_step)
+    assert pipeline.name == "one_step"
+    assert type(pipeline) == Pipeline