migrating to WfFormat v1.3

rafaelfsilva · rafaelfsilva · commit 7ea8488498fd · 2022-01-30T16:47:44.000-08:00
diff --git a/wfcommons/common/task.py b/wfcommons/common/task.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2020-2021 The WfCommons Team.
+# Copyright (c) 2020-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -36,9 +36,9 @@ class Task:
     :type runtime: float
     :param cores: Number of cores required by the task.
     :type cores: float
-    :param task_id: Job unique ID (e.g., ID0000001).
+    :param task_id: Task unique ID (e.g., ID0000001).
     :type task_id: Optional[str]
-    :param category: Job category (can be used, for example, to define jobs that use the same program).
+    :param category: Task category (can be used, for example, to define tasks that use the same program).
     :type category: Optional[str]
     :param machine: Machine on which is the task has been executed.
     :type machine: Optional[Machine]
diff --git a/wfcommons/common/workflow.py b/wfcommons/common/workflow.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2020-2021 The WfCommons Team.
+# Copyright (c) 2020-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -54,9 +54,9 @@ def __init__(self,
                  ) -> None:
         """Create an object of a workflow representation."""
         self.description: Optional[
-            str] = description if description else 'Instance generated with WfCommons - https://wfcommons.org'
+            str] = description if description else "Instance generated with WfCommons - https://wfcommons.org"
         self.created_at: str = str(datetime.utcnow().isoformat())
-        self.schema_version: str = "1.2"
+        self.schema_version: str = "1.3"
         self.wms_name: Optional[str] = "WfCommons" if not wms_name else wms_name
         self.wms_version: Optional[str] = str(__version__) if not wms_version else wms_version
         self.wms_url: Optional[str] = f"https://docs.wfcommons.org/en/v{__version__}/" if not wms_url else wms_url
@@ -76,23 +76,23 @@ def write_json(self, json_file_path: Optional[pathlib.Path] = None) -> None:
         workflow_tasks = []
 
         workflow_json = {
-            'name': self.name,
-            'description': self.description,
-            'createdAt': self.created_at,
-            'schemaVersion': self.schema_version,
-            'author': {
-                'name': str(getpass.getuser()),
-                'email': 'support@wfcommons.org'
+            "name": self.name,
+            "description": self.description,
+            "createdAt": self.created_at,
+            "schemaVersion": self.schema_version,
+            "author": {
+                "name": str(getpass.getuser()),
+                "email": "support@wfcommons.org"
             },
-            'wms': {
-                'name': self.wms_name,
-                'version': self.wms_version,
-                'url': self.wms_url
+            "wms": {
+                "name": self.wms_name,
+                "version": self.wms_version,
+                "url": self.wms_url
             },
-            'workflow': {
-                'executedAt': self.executed_at,
-                'makespan': self.makespan,
-                'jobs': workflow_tasks
+            "workflow": {
+                "executedAt": self.executed_at,
+                "makespan": self.makespan,
+                "tasks": workflow_tasks
             }
         }
 
@@ -101,18 +101,18 @@ def write_json(self, json_file_path: Optional[pathlib.Path] = None) -> None:
         for edge in self.edges:
             for task_name in edge:
                 if task_name not in tasks_dependencies:
-                    tasks_dependencies[task_name] = {'parents': [], 'children': []}
-            tasks_dependencies[edge[0]]['children'].append(edge[1])
-            tasks_dependencies[edge[1]]['parents'].append(edge[0])
+                    tasks_dependencies[task_name] = {"parents": [], "children": []}
+            tasks_dependencies[edge[0]]["children"].append(edge[1])
+            tasks_dependencies[edge[1]]["parents"].append(edge[0])
 
         # add tasks to the workflow json object
         for node in self.nodes:
-            task: Task = self.nodes[node]['task']
+            task: Task = self.nodes[node]["task"]
             task_obj = task.as_dict()
 
             # manage task dependencies
-            task_obj['parents'] = tasks_dependencies[task.name]['parents']
-            task_obj['children'] = tasks_dependencies[task.name]['children']
+            task_obj["parents"] = tasks_dependencies[task.name]["parents"]
+            task_obj["children"] = tasks_dependencies[task.name]["children"]
 
             workflow_tasks.append(task_obj)
 
@@ -122,12 +122,12 @@ def write_json(self, json_file_path: Optional[pathlib.Path] = None) -> None:
                 workflow_machines.append(task.machine.as_dict())
 
         if workflow_machines:
-            workflow_json['workflow']['machines'] = workflow_machines
+            workflow_json["workflow"]["machines"] = workflow_machines
 
         # write to file
         if not json_file_path:
-            json_file_path = pathlib.Path(f'{self.name.lower()}.json')
-        with open(json_file_path, 'w') as outfile:
+            json_file_path = pathlib.Path(f"{self.name.lower()}.json")
+        with open(json_file_path, "w") as outfile:
             outfile.write(json.dumps(workflow_json, indent=4))
 
     def write_dot(self, dot_file_path: Optional[pathlib.Path] = None) -> None:
diff --git a/wfcommons/version.py b/wfcommons/version.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2020-2021 The WfCommons Team.
+# Copyright (c) 2020-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
diff --git a/wfcommons/wfchef/utils.py b/wfcommons/wfchef/utils.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2021 The WfCommons Team.
+# Copyright (c) 2021-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -56,23 +56,22 @@ def create_graph(path: pathlib.Path) -> nx.DiGraph:
 
         id_count = 0
 
-        for job in content['workflow']['jobs']:
+        for task in content["workflow"]["tasks"]:
 
             # specific for epigenomics -- have to think about how to do it in general
-            if "genome-dax" in content['name']:
-                _type, *_ = job['name'].split('_')
-                graph.add_node(job['name'], label=_type, type=_type, id=str(id_count))
+            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 = job['name'].split('_ID')
+                    _type, _id = task["name"].split("_ID")
                 except ValueError:
-                    _type, _id = job['name'].split('_0')
-                graph.add_node(job['name'], label=_type, type=_type, id=_id)
+                    _type, _id = task["name"].split("_0")
+                graph.add_node(task["name"], label=_type, type=_type, id=_id)
 
-            # for job in content['workflow']['jobs']:
-            for parent in job['parents']:
-                graph.add_edge(parent, job['name'])
+            for parent in task["parents"]:
+                graph.add_edge(parent, task["name"])
 
         for node in graph.nodes:
 
diff --git a/wfcommons/wfinstances/instance.py b/wfcommons/wfinstances/instance.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2020-2021 The WfCommons Team.
+# Copyright (c) 2020-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -103,7 +103,7 @@ def __init__(self, input_instance: pathlib.Path,
 
         # Tasks
         self.workflow: Workflow = Workflow(name=self.name, makespan=self.makespan)
-        for task in self.instance['workflow']['jobs']:
+        for task in self.instance["workflow"]["tasks"]:
             # Required arguments are defined in the JSON scheme
             # Here name, type and runtime are required
             # By default the value is set to None if we do not find the value
@@ -147,9 +147,8 @@ def __init__(self, input_instance: pathlib.Path,
                     logger=self.logger
                 )
             )
-
         # TODO: handle the case of the output files of the leaves tasks (not taken into account yet)
-        for task in self.instance['workflow']['jobs']:
+        for task in self.instance["workflow"]["tasks"]:
             for parent in task['parents']:
                 self.workflow.add_edge(parent, task['name'], weight=0)
 
diff --git a/wfcommons/wfinstances/schema.py b/wfcommons/wfinstances/schema.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
-# Copyright (c) 2020-2021 The WfCommons Team.
+# Copyright (c) 2020-2022 The WfCommons Team.
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -113,14 +113,14 @@ def _semantic_validation(self, data: Dict[str, Any]):
             self.logger.debug('Skipping machines processing.')
 
         tasks_ids = []
-        for j in data['workflow']['jobs']:
+        for j in data["workflow"]["tasks"]:
             tasks_ids.append(j['name'])
             if 'machine' in j and j['machine'] not in machine_ids:
                 self.logger.error(f"Machine \"{j['machine']}\" is not declared in the list of machines.")
                 has_error = True
 
         # since tasks may be declared out of order, their dependencies are only verified here
-        for j in data['workflow']['jobs']:
+        for j in data["workflow"]["tasks"]:
             for p in j['parents']:
                 if p not in tasks_ids:
                     self.logger.error(f"Parent task \"{p['parentId']}\" is not declared in the list of workflow tasks.")

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`#!/usr/bin/env python`
`2`	`2`	`# -- coding: utf-8 --`
`3`	`3`	`#`
`4`		`-# Copyright (c) 2020-2021 The WfCommons Team.`
	`4`	`+# Copyright (c) 2020-2022 The WfCommons Team.`
`5`	`5`	`#`
`6`	`6`	`# This program is free software: you can redistribute it and/or modify`
`7`	`7`	`# it under the terms of the GNU General Public License as published by`