FEAT(PLOT,.TC): +Overwrites+Failed+Canceled

Author: ankostis

graphtik/plot.py

@@ -166,6 +166,9 @@ def build_pydot(
 
     steps_thickness = 3
     fill_color = "wheat"
+    failed_color = "LightCoral"
+    cancel_color = "Grey"
+    overwrite_color = "SkyBlue"
     steps_color = "#009999"
     new_clusters = {}
 
@@ -214,7 +217,11 @@ def get_node_name(a):
             # LABEL change with solution.
             if solution and nx_node in solution:
                 kw["style"] = "filled"
-                kw["fillcolor"] = fill_color
+                kw["fillcolor"] = (
+                    overwrite_color
+                    if nx_node in getattr(solution, "overwrites", ())
+                    else fill_color
+                )
                 ## NOTE: SVG tooltips not working without URL:
                 #  https://gitlab.com/graphviz/graphviz/issues/1425
                 kw["tooltip"] = str(solution.get(nx_node))
@@ -225,9 +232,15 @@ def get_node_name(a):
             if steps and nx_node in steps:
                 kw["penwdth"] = steps_thickness
             shape = "egg" if isinstance(nx_node, NetworkOperation) else "oval"
-            if nx_node in getattr(solution, "executed", ()):
+            if nx_node in getattr(solution, "failures", ()):
+                kw["style"] = "filled"
+                kw["fillcolor"] = failed_color
+            elif nx_node in getattr(solution, "executed", ()):
                 kw["style"] = "filled"
                 kw["fillcolor"] = fill_color
+            elif nx_node in getattr(solution, "canceled", ()):
+                kw["style"] = "filled"
+                kw["fillcolor"] = cancel_color
             try:
                 kw["URL"] = f"file://{inspect.getfile(nx_node.fn)}"
             except Exception as ex:

test/test_graphtik.py

@@ -504,9 +504,15 @@ def test_pruning_multiouts_not_override_intermediates1(exemethod):
     assert solution == exp
     assert solution.overwrites == {"overidden": [5, 1]}
 
+    # Check plotting Overwrites.
+    assert "SkyBlue" in str(solution.plot())
+
     solution = pipeline.compute(inp1, "asked")
     assert solution.overwrites == {}
 
+    # Check not plotting Overwrites.
+    assert "SkyBlue" not in str(solution.plot())
+
 
 @pytest.mark.xfail(
     sys.version_info < (3, 6),
@@ -1103,6 +1109,11 @@ def test_execution_endurance(exemethod):
     sol = graph.compute(inp, outputs=["a+2b", "cc"])
     assert sol == {"a+2b": 5}
 
+    # Check plotting Fail & Cancel.
+    #
+    dot = str(sol.plot())
+    assert "LightCoral" in dot # failed
+    assert "Grey" in dot # Canceled
 
 def test_multithreading_plan_execution():
     # From Huygn's test-code given in yahoo/graphkit#31