Merge pull request #167 from wfcommons/wfformat_1.6

Updated the code to be WfFormat-1.6-compliant

Author: henricasanova

docs/source/introduction.rst

@@ -80,6 +80,6 @@ machines on which the workflow was executed.
 WfFormat uses a JSON schema available in the
 `WfFormat Schema GitHub <https://github.com/wfcommons/WfFormat>`_ repository.
 The current version of the WfCommons Python package uses schema version
-:code:`1.5`. The schema repository provides a detailed explanation of WfFormat
+:code:`1.6`. The schema repository provides a detailed explanation of WfFormat
 (including required fields) and a validator script for verifying the
 compatibility of instances.

tests/unit/common/test_workflow.py

@@ -12,6 +12,9 @@
 import pytest
 import requests
 import json
+import math
+from numbers import Real
+from typing import Any
 
 from datetime import datetime
 from wfcommons.common import Task, Workflow
@@ -26,6 +29,97 @@
 ]
 
 
+def assert_json_close(
+    actual: Any,
+    expected: Any,
+    *,
+    rel_tol: float = 1e-9,
+    abs_tol: float = 1e-12,
+    path: str = "$",
+) -> None:
+    """Assert that two JSON-like objects are equal, with close numerics."""
+
+    actual_is_number = (
+        isinstance(actual, Real)
+        and not isinstance(actual, bool)
+    )
+    expected_is_number = (
+        isinstance(expected, Real)
+        and not isinstance(expected, bool)
+    )
+
+    if actual_is_number or expected_is_number:
+        assert actual_is_number and expected_is_number, (
+            f"Numeric type mismatch at {path}: "
+            f"actual={actual!r}, expected={expected!r}"
+        )
+
+        assert math.isclose(
+            actual,
+            expected,
+            rel_tol=rel_tol,
+            abs_tol=abs_tol,
+        ), (
+            f"Numbers differ at {path}: "
+            f"actual={actual!r}, expected={expected!r}, "
+            f"rel_tol={rel_tol}, abs_tol={abs_tol}"
+        )
+        return
+
+    if isinstance(actual, dict) or isinstance(expected, dict):
+        assert isinstance(actual, dict) and isinstance(expected, dict), (
+            f"Type mismatch at {path}: "
+            f"actual={type(actual).__name__}, "
+            f"expected={type(expected).__name__}"
+        )
+
+        assert actual.keys() == expected.keys(), (
+            f"Dictionary keys differ at {path}: "
+            f"actual-only={actual.keys() - expected.keys()}, "
+            f"expected-only={expected.keys() - actual.keys()}"
+        )
+
+        for key in actual:
+            assert_json_close(
+                actual[key],
+                expected[key],
+                rel_tol=rel_tol,
+                abs_tol=abs_tol,
+                path=f"{path}.{key}",
+            )
+        return
+
+    if isinstance(actual, list) or isinstance(expected, list):
+        assert isinstance(actual, list) and isinstance(expected, list), (
+            f"Type mismatch at {path}: "
+            f"actual={type(actual).__name__}, "
+            f"expected={type(expected).__name__}"
+        )
+
+        assert len(actual) == len(expected), (
+            f"List lengths differ at {path}: "
+            f"actual={len(actual)}, expected={len(expected)}"
+        )
+
+        for index, (actual_item, expected_item) in enumerate(
+            zip(actual, expected)
+        ):
+            assert_json_close(
+                actual_item,
+                expected_item,
+                rel_tol=rel_tol,
+                abs_tol=abs_tol,
+                path=f"{path}[{index}]",
+            )
+        return
+
+    assert actual == expected, (
+        f"Values differ at {path}: "
+        f"actual={actual!r}, expected={expected!r}"
+    )
+
+
+
 class TestWorkflow:
 
     @pytest.fixture
@@ -53,7 +147,11 @@ def test_workflow_creation(self, workflow: Workflow) -> None:
             "workflow": {
                 "specification": {
                     "tasks": [],
-                    "files": []
+                    "files": [],
+                    "metrics": {
+                        "numberOfTasks": 0,
+                        "numberOfFiles": 0
+                    }
                 },
                 "execution": {
                     "makespanInSeconds": 100.0,
@@ -139,7 +237,9 @@ def test_workflow_json_generation(self):
         written_json["workflow"]["specification"]["files"] = sorted(written_json["workflow"]["specification"]["files"], key=lambda x: x['id'])
 
         # Compare the two jsons!
-        assert(original_json == written_json)
+        # assert(original_json == written_json)
+        # assert(original_json == pytest.approx(written_json))
+        assert_json_close(original_json,written_json)
 
     @pytest.mark.unit
     def test_workflow_dot_file(self):

tests/unit/wfinstances/test_instance_analyzer.py

@@ -23,7 +23,7 @@ def instance(self) -> pathlib.Path:
             "name": "workflow_test",
             "description": "Instance generate for WfCommons Test",
             "createdAt": "2020-12-30T02:19:01.238077",
-            "schemaVersion": "1.5",
+            "schemaVersion": "1.6",
             "author": {
                 "name": "wfcommons",
                 "email": "support@wfcommons.org"

tests/unit/wfinstances/test_instances.py

@@ -23,7 +23,7 @@ def instance(self) -> pathlib.Path:
             "name": "workflow_test",
             "description": "Instance generate for WfCommons Test",
             "createdAt": "2020-12-30T02:19:01.238077",
-            "schemaVersion": "1.5",
+            "schemaVersion": "1.6",
             "author": {
                 "name": "wfcommons",
                 "email": "support@wfcommons.org"

wfcommons/common/workflow.py

@@ -19,7 +19,7 @@
 from ..common.task import Task, TaskType
 from ..version import __version__, __schema_version__
 
-from ..wfchef.utils import create_graph
+from ..wfchef.utils import create_graph, create_graph_from_json_object
 import tempfile
 
 
@@ -210,6 +210,75 @@ def generate_json(self) -> None:
 
         self.workflow_json = workflow_json
 
+        # Augment the workflow_json with the metrics
+        specification_metrics, execution_metrics = self.compute_metrics(workflow_json)
+        if specification_metrics != {}:
+            self.workflow_json["workflow"]["specification"]["metrics"] = specification_metrics
+        if execution_metrics != {}:
+            self.workflow_json["workflow"]["execution"]["metrics"] = execution_metrics
+
+
+    def compute_metrics(self, workflow_json: json) -> json:
+        """
+        Compute workflow specification and execution metrics.
+        :param workflow_json: Workflow instance as a JSON object.
+        :type workflow_json: json
+
+        :return: specification and execution metrics.
+        :rtype: json
+        """
+
+        # Specification metrics
+        specification_metrics : json = {}
+        specification_metrics["numberOfTasks"] = len(workflow_json["workflow"]["specification"]["tasks"])
+        if "files" in workflow_json["workflow"]["specification"]:
+            number_of_files = len(workflow_json["workflow"]["specification"]["files"])
+            specification_metrics["numberOfFiles"] = number_of_files
+            if number_of_files > 0:
+                specification_metrics["sumOfFileSizesInBytes"] = sum(
+                    file.get("sizeInBytes", 0)
+                    for file in workflow_json["workflow"]["specification"]["files"]
+                )
+        graph = create_graph_from_json_object(workflow_json)
+        levels = [
+            tuple(generation)
+            for generation in nx.topological_generations(graph)
+        ][1:-1] # Remove the fictitious SRC and DST levels
+
+        if len(levels) > 0:
+            widths = [len(level) for level in levels]  # Remove the fictitious SRC and DST levels
+
+            specification_metrics["numberOfLevels"] = len(widths)
+            specification_metrics["minimumWidth"] = min(widths)
+            specification_metrics["maximumWidth"] = max(widths)
+
+        # Execution metrics
+        execution_metrics : json = {}
+        if "execution" in workflow_json["workflow"]:
+            total_runtime_in_seconds = sum(
+                task.get("runtimeInSeconds", 0)
+                for task in workflow_json["workflow"]["execution"]["tasks"]
+            )
+            if total_runtime_in_seconds > 0:
+                execution_metrics["sumTaskRuntimesInSeconds"] = total_runtime_in_seconds
+
+            total_read_bytes = sum(
+                task.get("readBytes", 0)
+                for task in workflow_json["workflow"]["execution"]["tasks"]
+            )
+            if total_runtime_in_seconds > 0:
+                execution_metrics["totalNumBytesRead"] = total_read_bytes
+
+            total_written_bytes = sum(
+                task.get("writtenBytes", 0)
+                for task in workflow_json["workflow"]["execution"]["tasks"]
+            )
+            if total_written_bytes > 0:
+                execution_metrics["totalNumBytesWritten"] = total_written_bytes
+
+        return specification_metrics, execution_metrics
+
+
     def write_dot(self, dot_file_path: Optional[pathlib.Path] = None) -> None:
         """
         Write a dot file of the workflow instance.
@@ -264,3 +333,8 @@ def roots(self) -> List[str]:
 
     def leaves(self) -> List[str]:
         return [n for n,d in self.out_degree() if d==0]
+
+
+
+
+

wfcommons/version.py

@@ -9,4 +9,4 @@
 # (at your option) any later version.
 
 __version__ = "1.6-dev"
-__schema_version__ = "1.5"
+__schema_version__ = "1.6"

wfcommons/wfchef/utils.py

@@ -47,42 +47,54 @@ def create_graph(path: pathlib.Path) -> nx.DiGraph:
     path = pathlib.Path(path)
     with path.open() as fp:
         content = json.load(fp)
+        return create_graph_from_json_object(content)
 
-        graph = nx.DiGraph()
 
-        # Add src/dst nodes
-        graph.add_node("SRC", label="SRC", type="SRC", id="SRC")
-        graph.add_node("DST", label="DST", type="DST", id="DST")
+def create_graph_from_json_object(workflow_instance: dict) -> nx.DiGraph:
+    """
+    Creates a networkX DiGraph from a JSON file in the WfFormat.
 
-        id_count = 0
+    :param workflow_instance: A workflow instance as a JSON object.
+    :type workflow_instance: json
 
-        for task in content["workflow"]["specification"]["tasks"]:
+    :return: graph.
+    :rtype: networkX DiGraph.
+    """
+    graph = nx.DiGraph()
 
-            # specific for epigenomics -- have to think about how to do it in general
-            if "genome-dax" in content["name"]:
-                _type, *_ = task["name"].split("_")
-                graph.add_node(task["name"], label=_type, type=_type, id=str(id_count))
-                id_count += 1
-            else:
-                try:
-                    _type, _id = task["name"].split("_ID")
-                except ValueError:
-                    _type, _id = task["name"].split("_0")
-                graph.add_node(task["name"], label=_type, type=_type, id=_id)
+    # Add src/dst nodes
+    graph.add_node("SRC", label="SRC", type="SRC", id="SRC")
+    graph.add_node("DST", label="DST", type="DST", id="DST")
 
-            for parent in task["parents"]:
-                graph.add_edge(parent, task["name"])
+    id_count = 0
 
-        for node in graph.nodes:
+    for task in workflow_instance["workflow"]["specification"]["tasks"]:
 
-            if node in ["SRC", "DST"]:
-                continue
-            if graph.in_degree(node) <= 0:
-                graph.add_edge("SRC", node)
-            if graph.out_degree(node) <= 0:
-                graph.add_edge(node, "DST")
+        # specific for epigenomics -- have to think about how to do it in general
+        if "genome-dax" in workflow_instance["name"]:
+            _type, *_ = task["name"].split("_")
+            graph.add_node(task["name"], label=_type, type=_type, id=str(id_count))
+            id_count += 1
+        else:
+            try:
+                _type, _id = task["id"].split("_ID")
+            except ValueError:
+                _type, _id = task["id"].split("_0")
+            graph.add_node(task["name"], label=_type, type=_type, id=_id)
 
-        return graph
+        for parent in task["parents"]:
+            graph.add_edge(parent, task["name"])
+
+    for node in graph.nodes:
+
+        if node in ["SRC", "DST"]:
+            continue
+        if graph.in_degree(node) <= 0:
+            graph.add_edge("SRC", node)
+        if graph.out_degree(node) <= 0:
+            graph.add_edge(node, "DST")
+
+    return graph
 
 
 def annotate(g: nx.DiGraph) -> None:
@@ -176,8 +188,6 @@ def draw(g: nx.DiGraph,
     :param subgraph: nodes that were added by replication and will be colored green.
     :type subgraph: Set[str].
 
-    
-
     :return: the figure and the axis used.
     :rtype:  Tuple[plt.Figure, plt.Axes].
     """

wfcommons/wfinstances/logs/makeflow.py

@@ -130,7 +130,7 @@ def _parse_workflow_file(self) -> None:
                 elif '\t' in line:
                     # task execution command (likely olf here)
                     prefix = line.replace('./', '').strip().split()[1 if 'LOCAL' in line else 0]
-                    task_name = "ID{:07d}".format(task_id_counter)
+                    task_name = "{}_ID{:07d}".format(prefix, task_id_counter)
 
                     # create list of input and output files
                     output_files = self._create_files(outputs, "output", task_name)

wfcommons/wfinstances/logs/taskvine.py

@@ -276,7 +276,8 @@ def _construct_workflow(self) -> None:
         # Create all tasks
         task_map = {}
         for task_id in self.known_task_ids:
-            task_name = "Task_%d" % task_id
+            task_name = "Task_ID{:07d}".format(task_id)
+            print(f"DEBUGHERE: {task_name} ")
             task = Task(name=task_name,
                         task_id=task_name,
                         task_type=TaskType.COMPUTE,