Merge pull request #16 from BAMresearch/basic_pipeline_generator

Runner/pipeline draft based on graphlibs TopologicalSorter

Anja says everything is ok.

Author: toqduj

src/modacor/runner/__init__.py

@@ -0,0 +1,77 @@
+# src/modacor/runner/pipeline.py
+# # -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from graphlib import TopologicalSorter
+from pathlib import Path
+
+from attrs import define, field
+from attrs import validators as v
+
+from ..dataclasses.process_step import ProcessStep
+
+__all__ = ["Pipeline"]
+
+
+@define
+class Pipeline(TopologicalSorter):
+    """
+    Pipeline nodes are assumed to be of type ProcessStep
+    """
+
+    name: str = field(default="Unnamed Pipeline")
+    graph: dict[ProcessStep] = field(factory=dict)
+
+    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")
+
+    @classmethod
+    def from_dict(cls, graph_dict: dict, name=""):
+        return cls(name=name, graph=graph_dict)
+
+    def add_incoming_branch(self, branch: Self, branching_node):
+        """
+        Add a pipeline as a branch whose outcome shall be combined the existing pipeline.
+
+        This assumes that the branch to be added has a single exit point.
+
+        """
+        pipeline_to_add = Pipeline(graph=branch.graph)
+        pipeline_to_add_ordered = [*pipeline_to_add.static_order()]
+        # add the last node of the incoming as a predecessor to the connection point
+        self.graph[branching_node].update({pipeline_to_add_ordered[-1]})
+        # add the rest of the graph
+        self.graph = self.graph | branch.graph
+        # reinitialize the TopologicalSorter
+        super().__init__(graph=self.graph)
+
+    def add_outgoing_branch(self, branch: Self, branching_node):
+        """
+        Add a pipeline as a branch whose input is based on the existing pipeline.
+
+        This assumes that the branch to be added has a single entry point.
+
+        """
+        pipeline_to_add = Pipeline(graph=branch.graph)
+        pipeline_to_add_ordered = [*pipeline_to_add.static_order()]
+        # add the connection node as a predecessor to the first node of the outgoing branch
+        branch.graph[pipeline_to_add_ordered[0]].update({branching_node})
+        # add the rest of the graph
+        self.graph = self.graph | branch.graph
+        # reinitialize the TopologicalSorter
+        super().__init__(graph=self.graph)
+
+    def run(self, data: DataBundle, **kwargs):
+        """
+        run pipeline. to be extended for different schedulers
+        """
+        self.prepare()
+        while self.is_active():
+            for node in self.get_ready():
+                node.execute(data, **kwargs)
+                self.done(node)

src/modacor/runner/pipeline.py

@@ -0,0 +1,77 @@
+import pytest
+from pathlib import Path
+
+from ..runner.pipeline import Pipeline
+from ..dataclasses.process_step import ProcessStep
+from ..dataclasses.process_step_describer import ProcessStepDescriber
+
+
+@pytest.fixture
+def linear_pipeline():
+    return {3: {2, 1}, 2: {1}}
+
+
+class DummyIoSources:
+    pass
+
+
+class DummyProcessStepDescriber:
+    pass
+
+
+class DummyProcessStep:
+    pass
+
+
+def test_linear_pipeline(linear_pipeline):
+    "tests the sequence is expected for a linear graph"
+    pipeline = Pipeline(graph=linear_pipeline)
+    pipeline.prepare()
+    sequence = []
+    while pipeline.is_active():
+        for node in pipeline.get_ready():
+            sequence.append(node)
+            pipeline.done(node)
+    assert sequence == [1, 2, 3]
+
+
+def test_node_addition(linear_pipeline):
+    pipeline = Pipeline.from_dict(linear_pipeline)
+    ps = DummyProcessStep()
+    pipeline.add(ps, *[1, 2, 3])
+    pipeline.prepare()
+    sequence = []
+    while pipeline.is_active():
+        for node in pipeline.get_ready():
+            sequence.append(node)
+            pipeline.done(node)
+    assert sequence == [1, 2, 3, ps]
+
+
+def test_branch_addition(linear_pipeline, pipeline_to_add={5: {6}}, at_node=2):
+    """
+    add a pipeline as a branch on an existing pipeline, using the inherited add method
+
+    """
+    pipeline_1 = Pipeline(graph=linear_pipeline)
+    pipeline_2 = Pipeline(graph=pipeline_to_add)
+    pipeline_1.add(at_node, *pipeline_2.static_order())
+    assert [*pipeline_1.static_order()] == [1, 6, 5, 2, 3]
+
+
+def test_branch_addition_method(linear_pipeline, branch_graph={5: {6}}, branching_node=2):
+    pipeline = Pipeline(graph=linear_pipeline)
+    branch = Pipeline(graph=branch_graph)
+    pipeline.add_incoming_branch(branch, branching_node=2)
+    assert [*pipeline.static_order()] == [1, 6, 5, 2, 3]
+    assert pipeline.graph == {3: {2, 1}, 2: {1, 5}, 5: {6}}
+
+
+def test_diverging_branch_addition(
+    linear_pipeline, branch_graph={5: {6}, 6: set()}, branching_node=2
+):
+    pipeline = Pipeline(graph=linear_pipeline)
+    branch = Pipeline(graph=branch_graph)
+    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}}