#19: making WfCommons compatible with WfFormat v1.2

Author: rafaelfsilva

wfcommons/common/task.py

@@ -35,9 +35,15 @@ class Task:
     :param runtime: Task runtime in seconds.
     :type runtime: float
     :param cores: Number of cores required by the task.
-    :type cores: int
+    :type cores: float
+    :param task_id: Job unique ID (e.g., ID0000001).
+    :type task_id: str
+    :param category: Job category (can be used, for example, to define jobs that use the same program).
+    :type category: str
     :param machine: Machine on which is the task has been executed.
     :type machine: Machine
+    :param program: Program name.
+    :type program: str
     :param args: List of task arguments.
     :type args: List[str]
     :param avg_cpu: Average CPU utilization in %.
@@ -64,8 +70,11 @@ def __init__(self,
                  name: str,
                  task_type: TaskType,
                  runtime: float,
-                 cores: int,
+                 cores: float = 1.0,
+                 task_id: Optional[str] = None,
+                 category: Optional[str] = None,
                  machine: Optional[Machine] = None,
+                 program: Optional[str] = None,
                  args: List[str] = [],
                  avg_cpu: Optional[float] = None,
                  bytes_read: Optional[int] = None,
@@ -82,7 +91,10 @@ def __init__(self,
         self.name: str = name
         self.type: TaskType = task_type
         self.runtime: float = runtime
-        self.cores: Optional[int] = cores
+        self.cores: Optional[float] = cores
+        self.task_id: Optional[str] = task_id
+        self.category: Optional[str] = category
+        self.program: Optional[str] = program
         self.args: List[str] = args
         self.avg_cpu: Optional[float] = avg_cpu
         self.bytes_read: Optional[int] = bytes_read
@@ -112,6 +124,7 @@ def as_dict(self) -> Dict:
             'name': self.name,
             'type': self.type.value,
             'runtime': self.runtime,
+            'command': {},
             'parents': [],
             'children': [],
             'files': task_files,
@@ -132,8 +145,10 @@ def as_dict(self) -> Dict:
             task_obj['avgPower'] = self.avg_power
         if self.priority:
             task_obj['priority'] = self.priority
+        if self.program:
+            task_obj['command']['program'] = self.program
         if self.args:
-            task_obj['arguments'] = self.args
+            task_obj['command']['arguments'] = self.args
         if self.machine:
             task_obj['machine'] = self.machine.name
 

wfcommons/common/workflow.py

@@ -20,6 +20,7 @@
 from ..wfchef.utils import annotate, create_graph
 import tempfile
 
+
 class Workflow(nx.DiGraph):
     """
     Representation of a workflow. The workflow representation is an extension of the
@@ -137,4 +138,4 @@ def write_dot(self, dot_filename: str = None) -> None:
     def to_nx_digraph(self) -> nx.DiGraph:
         with tempfile.NamedTemporaryFile() as temp:
             self.write_json(temp.name)
-            return create_graph(temp.name)
+            return create_graph(temp.name)

wfcommons/wfinstances/instance.py

@@ -117,14 +117,20 @@ def __init__(self, input_instance: str, schema_file: Optional[str] = None, logge
             machine = task.get('machine', None)
             machine = None if machine is None else self.machines[machine]
 
+            # Fetch the command associated to this task
+            command = task.get('command', None)
+
             self.workflow.add_node(
                 task['name'],
                 task=Task(
                     name=task['name'],
+                    task_id=task.get('id', None),
+                    category=task.get('category', None),
                     task_type=TaskType(task['type']),
                     runtime=task['runtime'],
                     machine=machine,
-                    args=task.get('arguments', None),
+                    program=command.get('program', None) if command else None,
+                    args=command.get('arguments', None) if command else None,
                     cores=task.get('cores', None),
                     avg_cpu=task.get('avgCPU', None),
                     bytes_read=task.get('bytesRead', None),

wfcommons/wfinstances/logs/makeflow.py

@@ -119,7 +119,7 @@ def _parse_workflow_file(self):
                 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{:06d}".format(prefix, task_id_counter)
+                    task_name = "{}_ID{:07d}".format(prefix, task_id_counter)
                     task_id_counter += 1
 
                     # create list of task files

wfcommons/wfinstances/schema.py

@@ -70,7 +70,7 @@ def _load_schema(self, schema_file: Optional[str] = None):
             return json.loads(open(schema_path).read())
 
         # fetching latest schema file from GitHub repository
-        url = 'https://raw.githubusercontent.com/wfcommons/workflow-schema/master/wfcommons-schema.json'
+        url = 'https://raw.githubusercontent.com/wfcommons/wfformat/master/wfcommons-schema.json'
         response = requests.get(url)
         schema = json.loads(response.content)
         with open(schema_path, 'w') as outfile: