Changes some files to be in compliance with wfformat v5

Author: tainagdcoleman

wfcommons/common/task.py

@@ -69,7 +69,7 @@ class Task:
 
     def __init__(self,
                  name: str,
-                 task_type: TaskType,
+                #  task_type: TaskType,
                  runtime: float,
                  cores: float = 1.0,
                  task_id: Optional[str] = None,
@@ -93,7 +93,7 @@ def __init__(self,
         self.logger: Logger = logging.getLogger(
             __name__) if logger is None else logger
         self.name: str = name
-        self.type: TaskType = task_type
+        # self.type: TaskType = task_type
         self.runtime: float = runtime
         self.cores: Optional[float] = cores
         self.task_id: Optional[str] = task_id
@@ -112,7 +112,7 @@ def __init__(self,
         self.launch_dir: Optional[str] = launch_dir
         self.start_time: Optional[str] = str(datetime.now().astimezone().isoformat()) if not start_time else start_time
         self.logger.debug(
-            f"created {self.type} task {self.name}: runtime => {self.runtime} seconds.")
+            f"created {self.task_id} task {self.name}: runtime => {self.runtime} seconds.")
 
     def as_dict(self) -> Dict:
         """A JSON representation of the task.

wfcommons/wfchef/chef.py

@@ -193,19 +193,50 @@ def ls_recipe():
     print(get_recipes())
 
 
-def uninstall_recipe(module_name: str):
+def uninstall_recipe(wf_name: str,
+                     savedir: pathlib.Path = this_dir.joinpath("recipes")):
     """
     Uninstalls a recipe installed in the system.
     """
 
-    for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
-        if entry_point.module_name == module_name:
-            print(f"Uninstalling package: wfchef.recipe.{module_name}")
-            proc = subprocess.Popen(["pip", "uninstall", f"wfchef.recipe.{module_name}"])
-            proc.wait()
-            return
-
-    print(f"Could not find recipe with module name {module_name} installed")
+    dst = pathlib.Path(savedir, f"{savedir.stem}_recipes", wf_name).resolve()
+    try:
+        # Removing package from setup.py
+        with this_dir.joinpath(dst.parent.parent.joinpath("setup.py")).open("r") as fp:
+            setup_str = fp.read()
+
+        # Find and remove the specific line that added the package
+        for line in setup_str.split("\n"):
+            if wf_name in setup_str:
+                setup_str = setup_str.replace(line, "")
+
+        with this_dir.joinpath(dst.parent.parent.joinpath("setup.py")).open("w") as fp:
+            fp.write(setup_str)
+
+        # Removing the import line from __init__.py
+        init_file = dst.parent.joinpath("__init__.py")
+        if init_file.exists():
+            with init_file.open("r") as fp:
+                init_str = fp.read()
+
+            # Remove the line that imports the package
+
+            for line in init_str.split("\n"):
+                if wf_name in line:
+                    init_str = init_str.replace(line, "")
+                    break
+
+            with init_file.open("w") as fp:
+                fp.write(init_str)
+            # for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
+            #     if entry_point.module_name == module_name:
+            #         print(f"Uninstalling package: {module_name}")
+            #         proc = subprocess.Popen(["pip", "uninstall", module_name])
+            #         proc.wait()
+            #         return
+    except Exception as e:
+        traceback.print_exc()
+    # print(f"Could not find recipe with module name {module_name} installed")
 
 
 def create_recipe(path_to_instances: Union[str, pathlib.Path],
@@ -300,7 +331,9 @@ def get_parser() -> argparse.ArgumentParser:
 
     uninstall_parser = subparsers.add_parser("uninstall")
     uninstall_parser.set_defaults(action=uninstall_recipe)
-    uninstall_parser.add_argument("module_name", help="name of recipe module to uninstall")
+    # uninstall_parser.add_argument("module_name", help="name of recipe module to uninstall")
+    uninstall_parser.add_argument("-n", "--name", help="name of the workflow to uninstall")
+    uninstall_parser.add_argument("-o", "--out", help="directory where the recipe is located")
 
     create_parser = subparsers.add_parser("create")
     create_parser.set_defaults(action=create_recipe)
@@ -352,7 +385,7 @@ def main():
     if args.action == ls_recipe:
         ls_recipe()
     elif args.action == uninstall_recipe:
-        uninstall_recipe(args.module_name)
+        uninstall_recipe(args.name, pathlib.Path(args.out))
     elif args.action == create_recipe:
         create_recipe(args.path, args.out, args.name, cutoff=args.cutoff, verbose=True)
 

wfcommons/wfchef/utils.py

@@ -56,7 +56,7 @@ def create_graph(path: pathlib.Path) -> nx.DiGraph:
 
         id_count = 0
 
-        for task in content["workflow"]["tasks"]:
+        for task in content["workflow"]["specification"]["tasks"]:
 
             # specific for epigenomics -- have to think about how to do it in general
             if "genome-dax" in content["name"]:

wfcommons/wfinstances/instance.py

@@ -76,7 +76,7 @@ def __init__(self, input_instance: pathlib.Path,
 
         # WMS properties
         self.wms: Dict[str, str] = {
-            u: v for u, v in self.instance['wms'].items()
+            u: v for u, v in self.instance['runtimeSystem'].items()
         }
 
         # Author properties
@@ -87,8 +87,8 @@ def __init__(self, input_instance: pathlib.Path,
         # Workflow properties
         # Global properties
         self.executed_at: datetime = dateutil.parser.parse(
-            self.instance['workflow']['executedAt'])
-        self.makespan: int = self.instance['workflow']['makespanInSeconds']
+            self.instance['workflow']['execution']['executedAt'])
+        self.makespan: int = self.instance['workflow']['execution']['makespanInSeconds']
 
         # Machines
         if 'machines' in self.instance['workflow'].keys():
@@ -109,7 +109,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"]["tasks"]:
+        for task in self.instance["workflow"]['specification']["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
@@ -135,7 +135,7 @@ def __init__(self, input_instance: pathlib.Path,
                     name=task['name'],
                     task_id=task.get('id', None),
                     category=task.get('category', None),
-                    task_type=TaskType(task['type']),
+                    # task_type=TaskType(task['type']),
                     runtime=task['runtimeInSeconds'] if 'runtimeInSeconds' in task else 0,
                     machine=machine,
                     program=command.get('program', None) if command else None,
@@ -153,7 +153,7 @@ def __init__(self, input_instance: pathlib.Path,
                 )
             )
         # TODO: handle the case of the output files of the leaves tasks (not taken into account yet)
-        for task in self.instance["workflow"]["tasks"]:
+        for task in self.instance["workflow"]["specification"]["tasks"]:
             for parent in task['parents']:
                 self.workflow.add_dependency(parent, task['name'])
 

wfcommons/wfinstances/schema.py

@@ -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"]["tasks"]:
+        for j in data["workflow"]["specification"]["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"]["tasks"]:
+        for j in data["workflow"]["specification"]["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.")