Updates wfchef abstract recipe interfaces

Author: tainagdcoleman

wfcommons/wfchef/chef.py

@@ -8,11 +8,13 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
+import pathlib
 import argparse
+from email.mime import base
 import json
-import math
-from modulefinder import Module
-from importlib_metadata import entry_points
+import traceback
+from typing import Dict, Optional, Union
+import argparse
 import networkx as nx
 import numpy as np
 import pandas as pd
@@ -21,16 +23,15 @@
 import pkg_resources
 import subprocess
 import traceback
-
-from typing import Dict, List, Optional, Union
+from typing import Dict, Optional, Union
 from stringcase import capitalcase
-
 from .duplicate import duplicate, NoMicrostructuresError
 from .find_microstructures import save_microstructures
 from .utils import create_graph
 from ..wfgen.abstract_recipe import WorkflowRecipe
 from ..wfinstances.instance import Instance
 from ..wfinstances.instance_analyzer import InstanceAnalyzer
+import math
 
 this_dir = pathlib.Path(__file__).resolve().parent
 skeleton_path = this_dir.joinpath("skeletons")
@@ -172,8 +173,7 @@ def analyzer_summary(path_to_instances: pathlib.Path) -> Dict:
 
     return stats_dict
 
-def get_recipe(recipe: str) -> Module:
-    modules = []
+def get_recipe(recipe: str) -> "Module":
     for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
         att  = entry_point.attrs[0]
         if att == recipe:
@@ -219,7 +219,7 @@ def create_recipe(path_to_instances: Union[str, pathlib.Path],
                   wf_name: str,
                   cutoff: int = 4000,
                   verbose: bool = False,
-                  runs: int = 1) -> WorkflowRecipe:
+                  runs: int = 1) -> "WorkflowRecipe":
     """
     Creates a recipe for a workflow application by automatically replacing custom information 
     from the recipe skeleton.
@@ -228,7 +228,7 @@ def create_recipe(path_to_instances: Union[str, pathlib.Path],
     :type path_to_instances: str or pathlib.Path 
     :param savedir: path to save the recipe.
     :type savedir: pathlib.Path
-    :param wf_name: name of the workflow apllication.
+    :param wf_name: name of the workflow application.
     :type wf_name: str 
     :param cutoff: when set, only consider instances of smaller or equal sizes.
     :type cutoff: int
@@ -252,8 +252,8 @@ def create_recipe(path_to_instances: Union[str, pathlib.Path],
     err_savepath = microstructures_path.joinpath("metric", "err.csv")
     err_savepath.parent.mkdir(exist_ok=True, parents=True)
     df = find_err(microstructures_path, runs=runs)
-
     err_savepath.write_text(df.to_csv())
+    
     # Recipe 
     with skeleton_path.joinpath("recipe.py").open() as fp:
         skeleton_str = fp.read()

wfcommons/wfchef/duplicate.py

@@ -12,15 +12,10 @@
 import json
 import pickle
 import networkx as nx
-from typing import Set, Optional, List, Union, Dict
+from typing import Set, List, Union, Dict
 from uuid import uuid4
-
 import numpy as np
-from .utils import draw
 import random
-import argparse
-import pandas as pd
-from functools import partial
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
@@ -73,23 +68,24 @@ def duplicate(path: pathlib.Path,
     :param path: path to the summary JSON file.
     :type path: pathlib.Path.
     :param base: name (for samples available in WfCommons) or path to the specific 
-                graph to be used as base (if not set WfChef chooses the best fitting one). 
+                 graph to be used as base (if not set WfChef chooses the best fitting one). 
     :type base: str or pathlib.Path.
     :param num_nodes: total amount of nodes desired in the synthetic instance.
     :type num_nodes: int.
 
-
     :return: graph with the desired number of tasks.
     :rtype: networkX DiGraph.
     """
     summary = json.loads(path.joinpath("summary.json").read_text())
+   
     if base:
         base_path = pathlib.Path(base)
         if not base_path.is_absolute():
             base_path = path.joinpath(base_path)
     else:
         base_path = path.joinpath(min(summary["base_graphs"].keys(), key=lambda k: summary["base_graphs"][k]["order"]))
-
+    
+        
     graph = pickle.loads(base_path.joinpath("base_graph.pickle").read_bytes())
     if num_nodes < graph.order():
         raise ValueError(

wfcommons/wfchef/recipes/blast/recipe.py

@@ -8,18 +8,16 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import json
 import pathlib
-
 from logging import Logger
-from typing import Dict, Optional, Set
-
-from wfcommons.wfgen.abstract_recipe import WorkflowRecipe
+from typing import Optional, Set
+import pathlib 
+from wfcommons.wfchef.wfchef_abstract_recipe import BaseMethod, WfChefWorkflowRecipe
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
-class BlastRecipe(WorkflowRecipe):
+class BlastRecipe(WfChefWorkflowRecipe):
     """A Blast workflow recipe class for creating synthetic workflow instances.
 
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
@@ -46,47 +44,18 @@ def __init__(self,
                  input_file_size_factor: Optional[float] = 1.0,
                  output_file_size_factor: Optional[float] = 1.0,
                  logger: Optional[Logger] = None,
+                 base_method: BaseMethod = BaseMethod.ERROR_TABLE,
                  **kwargs) -> None:
-        super().__init__("Blast", data_footprint, num_tasks, exclude_graphs, runtime_factor, input_file_size_factor,
-                         output_file_size_factor, logger, this_dir)
-
-    @classmethod
-    def from_num_tasks(cls,
-                       num_tasks: int,
-                       exclude_graphs: Set[str] = set(),
-                       runtime_factor: Optional[float] = 1.0,
-                       input_file_size_factor: Optional[float] = 1.0,
-                       output_file_size_factor: Optional[float] = 1.0
-                       ) -> 'BlastRecipe':
-        """
-        Instantiate a Blast workflow recipe that will generate synthetic workflows up to
-        the total number of tasks provided.
-
-        :param num_tasks: The upper bound for the total number of tasks in the workflow (at least 3).
-        :type num_tasks: int
-        :param exclude_graphs:
-        :type exclude_graphs: Set
-        :param runtime_factor: The factor of which tasks runtime will be increased/decreased.
-        :type runtime_factor: float
-        :param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
-        :type input_file_size_factor: float
-        :param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
-        :type output_file_size_factor: float
-
-        :return: A Blast workflow recipe object that will generate synthetic workflows up
-                 to the total number of tasks provided.
-        :rtype: BlastRecipe
-        """
-        return BlastRecipe(num_tasks=num_tasks, exclude_graphs=exclude_graphs, runtime_factor=runtime_factor,
-                           input_file_size_factor=input_file_size_factor,
-                           output_file_size_factor=output_file_size_factor)
-
-    def _workflow_recipe(self) -> Dict:
-        """
-        Recipe for generating synthetic instances of the Blast workflow. Recipes can be
-        generated by using the :class:`~wfcommons.wfinstances.instance_analyzer.InstanceAnalyzer`.
-
-        :return: A recipe in the form of a dictionary in which keys are task prefixes.
-        :rtype: Dict[str, Any]
-        """
-        return json.loads(this_dir.joinpath("task_type_stats.json").read_text())
+        super().__init__(
+            name="Blast", 
+            data_footprint=data_footprint, 
+            num_tasks=num_tasks, 
+            exclude_graphs=exclude_graphs, 
+            runtime_factor=runtime_factor, 
+            input_file_size_factor=input_file_size_factor,
+            output_file_size_factor=output_file_size_factor, 
+            logger=logger, 
+            this_dir=this_dir, 
+            base_method=base_method,
+            **kwargs
+        )

wfcommons/wfchef/recipes/bwa/recipe.py

@@ -8,18 +8,16 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import json
 import pathlib
-
 from logging import Logger
-from typing import Dict, Optional, Set
-
-from wfcommons.wfgen.abstract_recipe import WorkflowRecipe
+from typing import Optional, Set
+import pathlib 
+from wfcommons.wfchef.wfchef_abstract_recipe import BaseMethod, WfChefWorkflowRecipe
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
-class BwaRecipe(WorkflowRecipe):
+class BwaRecipe(WfChefWorkflowRecipe):
     """A Bwa workflow recipe class for creating synthetic workflow instances.
 
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
@@ -46,47 +44,18 @@ def __init__(self,
                  input_file_size_factor: Optional[float] = 1.0,
                  output_file_size_factor: Optional[float] = 1.0,
                  logger: Optional[Logger] = None,
+                 base_method: BaseMethod = BaseMethod.ERROR_TABLE,
                  **kwargs) -> None:
-        super().__init__("Bwa", data_footprint, num_tasks, exclude_graphs, runtime_factor, input_file_size_factor,
-                         output_file_size_factor, logger, this_dir)
-
-    @classmethod
-    def from_num_tasks(cls,
-                       num_tasks: int,
-                       exclude_graphs: Set[str] = set(),
-                       runtime_factor: Optional[float] = 1.0,
-                       input_file_size_factor: Optional[float] = 1.0,
-                       output_file_size_factor: Optional[float] = 1.0
-                       ) -> 'BwaRecipe':
-        """
-        Instantiate a Bwa workflow recipe that will generate synthetic workflows up to
-        the total number of tasks provided.
-
-        :param num_tasks: The upper bound for the total number of tasks in the workflow (at least 3).
-        :type num_tasks: int
-        :param exclude_graphs:
-        :type exclude_graphs: Set
-        :param runtime_factor: The factor of which tasks runtime will be increased/decreased.
-        :type runtime_factor: float
-        :param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
-        :type input_file_size_factor: float
-        :param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
-        :type output_file_size_factor: float
-
-        :return: A Bwa workflow recipe object that will generate synthetic workflows up
-                 to the total number of tasks provided.
-        :rtype: BwaRecipe
-        """
-        return BwaRecipe(num_tasks=num_tasks, exclude_graphs=exclude_graphs, runtime_factor=runtime_factor,
-                         input_file_size_factor=input_file_size_factor,
-                         output_file_size_factor=output_file_size_factor)
-
-    def _workflow_recipe(self) -> Dict:
-        """
-        Recipe for generating synthetic instances of the Bwa workflow. Recipes can be
-        generated by using the :class:`~wfcommons.wfinstances.instance_analyzer.InstanceAnalyzer`.
-
-        :return: A recipe in the form of a dictionary in which keys are task prefixes.
-        :rtype: Dict[str, Any]
-        """
-        return json.loads(this_dir.joinpath("task_type_stats.json").read_text())
+        super().__init__(
+            name="Bwa", 
+            data_footprint=data_footprint, 
+            num_tasks=num_tasks, 
+            exclude_graphs=exclude_graphs, 
+            runtime_factor=runtime_factor, 
+            input_file_size_factor=input_file_size_factor,
+            output_file_size_factor=output_file_size_factor, 
+            logger=logger, 
+            this_dir=this_dir, 
+            base_method=base_method,
+            **kwargs
+        )

wfcommons/wfchef/recipes/cycles/recipe.py

@@ -8,18 +8,16 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import json
 import pathlib
-
 from logging import Logger
-from typing import Dict, Optional, Set
-
-from wfcommons.wfgen.abstract_recipe import WorkflowRecipe
+from typing import Optional, Set
+import pathlib 
+from wfcommons.wfchef.wfchef_abstract_recipe import BaseMethod, WfChefWorkflowRecipe
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
-class CyclesRecipe(WorkflowRecipe):
+class CyclesRecipe(WfChefWorkflowRecipe):
     """A Cycles workflow recipe class for creating synthetic workflow instances.
 
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
@@ -46,47 +44,18 @@ def __init__(self,
                  input_file_size_factor: Optional[float] = 1.0,
                  output_file_size_factor: Optional[float] = 1.0,
                  logger: Optional[Logger] = None,
+                 base_method: BaseMethod = BaseMethod.ERROR_TABLE,
                  **kwargs) -> None:
-        super().__init__("Cycles", data_footprint, num_tasks, exclude_graphs, runtime_factor, input_file_size_factor,
-                         output_file_size_factor, logger, this_dir)
-
-    @classmethod
-    def from_num_tasks(cls,
-                       num_tasks: int,
-                       exclude_graphs: Set[str] = set(),
-                       runtime_factor: Optional[float] = 1.0,
-                       input_file_size_factor: Optional[float] = 1.0,
-                       output_file_size_factor: Optional[float] = 1.0
-                       ) -> 'CyclesRecipe':
-        """
-        Instantiate a Cycles workflow recipe that will generate synthetic workflows up to
-        the total number of tasks provided.
-
-        :param num_tasks: The upper bound for the total number of tasks in the workflow (at least 3).
-        :type num_tasks: int
-        :param exclude_graphs:
-        :type exclude_graphs: Set
-        :param runtime_factor: The factor of which tasks runtime will be increased/decreased.
-        :type runtime_factor: float
-        :param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
-        :type input_file_size_factor: float
-        :param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
-        :type output_file_size_factor: float
-
-        :return: A Cycles workflow recipe object that will generate synthetic workflows up
-                 to the total number of tasks provided.
-        :rtype: CyclesRecipe
-        """
-        return CyclesRecipe(num_tasks=num_tasks, exclude_graphs=exclude_graphs, runtime_factor=runtime_factor,
-                            input_file_size_factor=input_file_size_factor,
-                            output_file_size_factor=output_file_size_factor)
-
-    def _workflow_recipe(self) -> Dict:
-        """
-        Recipe for generating synthetic instances of the Cycles workflow. Recipes can be
-        generated by using the :class:`~wfcommons.wfinstances.instance_analyzer.InstanceAnalyzer`.
-
-        :return: A recipe in the form of a dictionary in which keys are task prefixes.
-        :rtype: Dict[str, Any]
-        """
-        return json.loads(this_dir.joinpath("task_type_stats.json").read_text())
+        super().__init__(
+            name="Cycles", 
+            data_footprint=data_footprint, 
+            num_tasks=num_tasks, 
+            exclude_graphs=exclude_graphs, 
+            runtime_factor=runtime_factor, 
+            input_file_size_factor=input_file_size_factor,
+            output_file_size_factor=output_file_size_factor, 
+            logger=logger, 
+            this_dir=this_dir, 
+            base_method=base_method,
+            **kwargs
+        )