feat: UI

* chore: bump packages

* fix(type): mypy issue

* feat: visualisation mock of UI

* feat: working graph visualisation with parallel nodes

* feat: getting the data from the backend

* feat: modal added to show metadata

* feat: zoom on workflow

* feat: Getting graph data from the actual graph and nodes

* feat: Getting the source and function from the UI

* feat: Uploading the source file via browser

* feat: Failure paths are clearly shown

* feat: parallel graph rendering works

* feat: stub image added along with alias nodes

* feat: Richer python task modal

* feat: map node data

* feat: bounding boxes around branches

* feat: bounding boxes around branches

* feat: cli entry point to see the graph

* feat: dark mode

* feat: dark mode

Author: vijayvammi

examples/02-sequential/conditional.py

@@ -1,4 +1,4 @@
-from runnable import Conditional, Pipeline, PythonTask
+from runnable import Conditional, Pipeline, PythonTask, Stub
 
 
 def when_heads_function():
@@ -50,7 +50,9 @@ def main():
         function=toss_function,
         returns=["toss"],
     )
-    pipeline = Pipeline(steps=[toss_task, conditional])
+
+    continue_to = Stub(name="continue to")
+    pipeline = Pipeline(steps=[toss_task, conditional, continue_to])
 
     pipeline.execute()
 

examples/06-parallel/parallel.py

@@ -45,14 +45,15 @@ def traversal():
 def main():
     parallel_step = Parallel(
         name="parallel_step",
-        terminate_with_success=True,
         branches={
             "branch1": traversal(),
             "branch2": traversal(),
         },
     )
 
-    pipeline = Pipeline(steps=[parallel_step])
+    continue_to = Stub(name="continue to")
+
+    pipeline = Pipeline(steps=[parallel_step, continue_to])
 
     pipeline.execute()
     return pipeline

examples/06-parallel/parallel.yaml

@@ -39,7 +39,10 @@ dag:
   steps:
     parallel_step:
       type: parallel
-      next: success
+      next: continue_to
       branches:
         branch1: *branch
         branch2: *branch
+    continue_to:
+      type: stub
+      next: success

examples/07-map/map.py

@@ -9,7 +9,7 @@
     process_chunk,
     read_processed_chunk,
 )
-from runnable import Map, NotebookTask, Pipeline, PythonTask, ShellTask
+from runnable import Map, NotebookTask, Pipeline, PythonTask, ShellTask, Stub
 
 
 def iterable_branch(execute: bool = True):
@@ -101,9 +101,10 @@ def main():
     collect = PythonTask(
         name="collect",
         function=assert_default_reducer,
-        terminate_with_success=True,
     )
 
+    continue_to = Stub(name="continue to")
+
     pipeline = Pipeline(steps=[map_state, collect])
 
     pipeline.execute(parameters_file="examples/common/initial_parameters.yaml")

extensions/nodes/conditional.py

@@ -7,7 +7,7 @@
 from runnable import console, defaults
 from runnable.datastore import Parameter
 from runnable.graph import Graph, create_graph
-from runnable.nodes import CompositeNode, MapVariableType
+from runnable.nodes import CompositeNode, MapVariableType, NodeInD3
 
 logger = logging.getLogger(defaults.LOGGER_NAME)
 
@@ -241,3 +241,21 @@ def fan_in(self, map_variable: MapVariableType = None):
             step_log.status = defaults.FAIL
 
         self._context.run_log_store.add_step_log(step_log, self._context.run_id)
+
+    def to_d3_node(self) -> NodeInD3:
+        def get_display_string() -> str:
+            display = f"match {self.parameter}:\n"
+            for case in self.branches.keys():
+                display += f'    case "{case}":\n        ...\n'
+            if self.default:
+                display += "    case _:\n        ...\n"
+            return display
+
+        return NodeInD3(
+            id=self.internal_name,
+            label="conditional",
+            metadata={
+                "conditioned on": self.parameter,
+                "display": get_display_string(),
+            },
+        )

extensions/nodes/fail.py

@@ -6,7 +6,7 @@
 from runnable import datastore, defaults
 from runnable.datastore import StepLog
 from runnable.defaults import MapVariableType
-from runnable.nodes import TerminalNode
+from runnable.nodes import NodeInD3, TerminalNode
 
 
 class FailNode(TerminalNode):
@@ -70,3 +70,9 @@ def execute(
         step_log.attempts.append(attempt_log)
 
         return step_log
+
+    def to_d3_node(self) -> NodeInD3:
+        return NodeInD3(
+            id=self.internal_name,
+            label="fail",
+        )

extensions/nodes/map.py

@@ -18,7 +18,7 @@
 )
 from runnable.defaults import MapVariableType
 from runnable.graph import Graph, create_graph
-from runnable.nodes import CompositeNode
+from runnable.nodes import CompositeNode, NodeInD3
 
 logger = logging.getLogger(defaults.LOGGER_NAME)
 
@@ -348,3 +348,18 @@ def update_param(
         self._context.run_log_store.set_parameters(
             parameters=params, run_id=self._context.run_id
         )
+
+    def to_d3_node(self) -> NodeInD3:
+        return NodeInD3(
+            id=self.internal_name,
+            label="map",
+            metadata={
+                "node_type": "map",
+                "iterate_on": self.iterate_on,  # Parameter name containing the iterable
+                "iterate_as": self.iterate_as,  # Name used for each iteration
+                "map_branch_id": self.internal_name
+                + "."
+                + defaults.MAP_PLACEHOLDER,  # The branch identifier pattern
+                "is_composite": True,  # Flag indicating this is a composite node
+            },
+        )

extensions/nodes/parallel.py

@@ -6,7 +6,7 @@
 from runnable import defaults
 from runnable.defaults import MapVariableType
 from runnable.graph import Graph, create_graph
-from runnable.nodes import CompositeNode
+from runnable.nodes import CompositeNode, NodeInD3
 
 
 class ParallelNode(CompositeNode):
@@ -157,3 +157,9 @@ def fan_in(self, map_variable: MapVariableType = None):
             step_log.status = defaults.FAIL
 
         self._context.run_log_store.add_step_log(step_log, self._context.run_id)
+
+    def to_d3_node(self) -> NodeInD3:
+        return NodeInD3(
+            id=self.internal_name,
+            label="parallel",
+        )

extensions/nodes/stub.py

@@ -7,7 +7,7 @@
 from runnable import datastore, defaults
 from runnable.datastore import StepLog
 from runnable.defaults import MapVariableType
-from runnable.nodes import ExecutableNode
+from runnable.nodes import ExecutableNode, NodeInD3
 
 logger = logging.getLogger(defaults.LOGGER_NAME)
 
@@ -87,3 +87,9 @@ def execute(
         step_log.attempts.append(attempt_log)
 
         return step_log
+
+    def to_d3_node(self) -> NodeInD3:
+        return NodeInD3(
+            id=self.internal_name,
+            label="stub",
+        )

extensions/nodes/success.py

@@ -6,7 +6,7 @@
 from runnable import datastore, defaults
 from runnable.datastore import StepLog
 from runnable.defaults import MapVariableType
-from runnable.nodes import TerminalNode
+from runnable.nodes import NodeInD3, TerminalNode
 
 
 class SuccessNode(TerminalNode):
@@ -70,3 +70,9 @@ def execute(
         step_log.attempts.append(attempt_log)
 
         return step_log
+
+    def to_d3_node(self) -> NodeInD3:
+        return NodeInD3(
+            id=self.internal_name,
+            label="success",
+        )