From 79b4ef43e2f309c8e84ca8da63770a9ff813cb5c Mon Sep 17 00:00:00 2001 From: Henri Casanova Date: Wed, 5 Aug 2026 14:34:52 -1000 Subject: [PATCH 1/3] Updated the schema version to 1.6 --- docs/source/introduction.rst | 2 +- tests/unit/wfinstances/test_instance_analyzer.py | 2 +- tests/unit/wfinstances/test_instances.py | 2 +- wfcommons/version.py | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/source/introduction.rst b/docs/source/introduction.rst index c9561185..1bdd9a84 100644 --- a/docs/source/introduction.rst +++ b/docs/source/introduction.rst @@ -68,6 +68,6 @@ simulation frameworks that support WfFormat can then use both types of instances interchangeably. WfFormat uses a JSON specification available in the `WfFormat Schema GitHub `_ repository. The current version of the WfCommons Python package uses the schema -version :code:`1.5`. The schema GitHub repository provides detailed explanation +version :code:`1.6`. The schema GitHub repository provides detailed explanation of WfFormat (including required fields), and also a validator script for verifying the compatibility of instances. diff --git a/tests/unit/wfinstances/test_instance_analyzer.py b/tests/unit/wfinstances/test_instance_analyzer.py index 2b941287..782a0edf 100644 --- a/tests/unit/wfinstances/test_instance_analyzer.py +++ b/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" diff --git a/tests/unit/wfinstances/test_instances.py b/tests/unit/wfinstances/test_instances.py index ebecc130..ed9cfc61 100644 --- a/tests/unit/wfinstances/test_instances.py +++ b/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" diff --git a/wfcommons/version.py b/wfcommons/version.py index 8d89615d..1a320879 100644 --- a/wfcommons/version.py +++ b/wfcommons/version.py @@ -9,4 +9,4 @@ # (at your option) any later version. __version__ = "1.5-dev" -__schema_version__ = "1.5" +__schema_version__ = "1.6" From 7e9be1d414f3702f0ed4d03b9e695d63e2e040b1 Mon Sep 17 00:00:00 2001 From: Henri Casanova Date: Wed, 5 Aug 2026 17:16:18 -1000 Subject: [PATCH 2/3] Completed move to WfFormat 1.6 - Bug fixes - Test fixes --- tests/unit/common/test_workflow.py | 104 ++++++++++++++++++++++++++++- wfcommons/common/workflow.py | 76 ++++++++++++++++++++- wfcommons/wfchef/utils.py | 68 +++++++++++-------- 3 files changed, 216 insertions(+), 32 deletions(-) diff --git a/tests/unit/common/test_workflow.py b/tests/unit/common/test_workflow.py index 81ae7291..bef2fdab 100644 --- a/tests/unit/common/test_workflow.py +++ b/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): diff --git a/wfcommons/common/workflow.py b/wfcommons/common/workflow.py index c0060c98..4e1e20a1 100644 --- a/wfcommons/common/workflow.py +++ b/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] + + + + + diff --git a/wfcommons/wfchef/utils.py b/wfcommons/wfchef/utils.py index 066f812a..0dc1ac67 100644 --- a/wfcommons/wfchef/utils.py +++ b/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]. """ From a8aa1f0b30d5f86af06e9eb3375cb98f64959a25 Mon Sep 17 00:00:00 2001 From: Henri Casanova Date: Wed, 5 Aug 2026 17:43:31 -1000 Subject: [PATCH 3/3] usual task id vs. name bug fix --- wfcommons/wfinstances/logs/makeflow.py | 2 +- wfcommons/wfinstances/logs/taskvine.py | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/wfcommons/wfinstances/logs/makeflow.py b/wfcommons/wfinstances/logs/makeflow.py index 8306fc0d..a5c9f188 100644 --- a/wfcommons/wfinstances/logs/makeflow.py +++ b/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) diff --git a/wfcommons/wfinstances/logs/taskvine.py b/wfcommons/wfinstances/logs/taskvine.py index 136872d9..bc960cb5 100644 --- a/wfcommons/wfinstances/logs/taskvine.py +++ b/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,