#19: making WfCommons compatible with WfFormat v1.2

Author: rafaelfsilva

wfcommons/common/workflow.py

@@ -17,7 +17,7 @@
 from ..common.task import Task
 from ..version import __version__
 
-from ..wfchef.utils import annotate, create_graph
+from ..wfchef.utils import create_graph
 import tempfile
 
 
@@ -54,7 +54,7 @@ def __init__(self,
         """Create an object of a workflow representation."""
         self.description = description if description else 'Instance generated with WfCommons - https://wfcommons.org'
         self.created_at = str(datetime.utcnow().isoformat())
-        self.schema_version = '1.0'
+        self.schema_version = '1.2'
         self.wms_name = 'WfCommons' if not wms_name else wms_name
         self.wms_version = str(__version__) if not wms_version else wms_version
         self.wms_url = 'https://docs.wfcommons.org/en/v{}/'.format(__version__) if not wms_url else wms_url

wfcommons/wfinstances/logs/makeflow.py

@@ -113,14 +113,13 @@ def _parse_workflow_file(self):
                     inputs = line.split(':')[1].split()
 
                     for file in itertools.chain(outputs, inputs):
-                        if not file in self.files_map:
+                        if file not in self.files_map:
                             self.files_map[file] = {'task_name': None, 'children': [], 'file': []}
 
                 elif len(line.strip()) > 0:
                     # task execution command
                     prefix = line.replace('./', '').replace('perl', '').strip().split()[1 if 'LOCAL' in line else 0]
                     task_name = "{}_ID{:07d}".format(prefix, task_id_counter)
-                    task_id_counter += 1
 
                     # create list of task files
                     list_files = []
@@ -130,14 +129,18 @@ def _parse_workflow_file(self):
                     # create task
                     args = ' '.join(line.replace('LOCAL', '').replace('perl', '').strip().split())
                     task = Task(name=task_name,
+                                task_id="ID{:07d}".format(task_id_counter),
+                                category=prefix,
                                 task_type=TaskType.COMPUTE,
                                 runtime=0,
+                                program=prefix,
                                 args=args.split(),
                                 cores=1,
                                 files=list_files,
                                 logger=self.logger)
                     self.workflow.add_node(task_name, task=task)
                     self.args_map[args] = task
+                    task_id_counter += 1
 
         # adding edges
         for file in self.files_map:
@@ -219,7 +222,7 @@ def _parse_resource_monitor_logs(self):
                 # task
                 task = self.args_map[data['command'].replace('perl', '').strip()]
                 task.runtime = float(data['wall_time'][0])
-                task.cores = int(data['cores'][0])
+                task.cores = float(data['cores'][0])
                 task.memory = int(data['memory'][0]) * 1000  # MB to KB
                 task.bytes_read = int(data['bytes_read'][0] * 1000)  # MB to KB
                 task.bytes_written = int(data['bytes_written'][0] * 1000)  # MB to KB

wfcommons/wfinstances/logs/pegasus.py

@@ -156,6 +156,8 @@ def _parse_workflow(self):
                     task_name,
                     task=Task(
                         name=task_name,
+                        task_id=j['id'],
+                        category=j['name'],
                         task_type=TaskType.COMPUTE,
                         runtime=0,
                         args=j['arguments'],
@@ -212,6 +214,8 @@ def _parse_dax(self):
                     task_name,
                     task=Task(
                         name=task_name,
+                        task_id=str(j.get('id')),
+                        category=str(j.get('name')),
                         task_type=TaskType.COMPUTE,
                         runtime=0,
                         args=[],
@@ -406,6 +410,8 @@ def _parse_job_output_latest(self, task, output_file):
         with open(tmp_file) as f:
             data = yaml.load(f, Loader=yaml.FullLoader)[0]
 
+            task.program = data['transformation']
+
             if data['transformation'].startswith('pegasus:') or task.name.lower().startswith('chmod_'):
                 task.type = TaskType.AUXILIARY
 
@@ -486,6 +492,7 @@ def _parse_job_output_legacy(self, task, output_file):
         try:
             e = xml.etree.ElementTree.parse(output_file).getroot()
             # main job information
+            task.program = e.get('transformation')
             if e.get('transformation').startswith('pegasus:') or task.name.lower().startswith('chmod_'):
                 task.type = TaskType.AUXILIARY