Merge pull request #20 from wfcommons/feature/wfchef

Feature/wfchef

Author: rafaelfsilva

setup.py

@@ -60,7 +60,8 @@
     python_requires='>=3.6',
     entry_points={
         'console_scripts': [
-            'wfchef=wfcommons.wfchef.chef:main',
+            'wfchef=wfcommons.wfchef.chef:main'
+
         ],
         'workflow_recipes': [
             'epigenomics_recipe = wfcommons.wfchef.recipes:EpigenomicsRecipe',

wfcommons-schema.json

@@ -0,0 +1 @@
+{"$schema": "http://json-schema.org/schema#", "type": "object", "title": "WfFormat", "properties": {"name": {"description": "Workflow instance name", "type": "string", "minLength": 1}, "description": {"description": "General description of the instance", "type": "string", "minLength": 1}, "createdAt": {"description": "Instance creation timestamp", "type": "string", "minLength": 1, "format": "date-time"}, "schemaVersion": {"description": "Version of the schema used", "type": "string", "enum": ["1.0", "1.1"]}, "wms": {"type": "object", "properties": {"name": {"description": "WMS name", "type": "string", "minLength": 1}, "version": {"description": "WMS version", "type": "string", "minLength": 1}, "url": {"description": "WMS url", "type": "string", "minLength": 1, "format": "uri"}}, "required": ["name", "version"]}, "author": {"type": "object", "properties": {"name": {"description": "Author name", "type": "string", "minLength": 1}, "email": {"description": "Author email", "type": "string", "minLength": 1, "format": "email"}, "institution": {"description": "Author institution", "type": "string", "minLength": 1}, "country": {"description": "Author country", "type": "string", "minLength": 1}}, "required": ["name", "email"]}, "workflow": {"type": "object", "properties": {"makespan": {"description": "Workflow makespan in seconds", "type": "number"}, "executedAt": {"description": "Workflow start execution timestamp", "type": "string", "minLength": 1}, "machines": {"description": "Sets of machines used for workflow jobs", "type": "array", "minItems": 1, "items": {"description": "Description of the machine (node) that ran the job", "type": "object", "properties": {"system": {"description": "Machine system", "type": "string", "enum": ["linux", "macos", "windows"]}, "architecture": {"description": "Machine architecture", "type": "string", "minLength": 1}, "nodeName": {"description": "Machine node name", "type": "string", "format": "hostname", "minLength": 1}, "release": {"description": "Machine release", "type": "string", "minLength": 1}, "memory": {"description": "Total machine's RAM memory in KB", "type": "integer", "minimum": 1}, "cpu": {"description": "Machine's CPU information", "type": "object", "properties": {"count": {"description": "Number of CPU cores", "type": "integer", "minimum": 1}, "speed": {"description": "CPU speed in MHz", "type": "integer", "minimum": 1}, "vendor": {"description": "CPU vendor", "type": "string", "minLength": 1}}}}, "required": ["nodeName"]}}, "jobs": {"description": "Sets of workflow jobs", "type": "array", "minItems": 1, "items": {"type": "object", "properties": {"name": {"description": "Job name", "type": "string", "minLength": 1}, "type": {"description": "Job type (whether it is a compute or an auxiliary job)", "type": "string", "enum": ["compute", "transfer", "auxiliary"]}, "arguments": {"description": "Sets of job arguments", "type": "array", "items": {"description": "A job argument", "type": "string", "minLength": 1}}, "parents": {"description": "Sets of parent jobs", "type": "array", "items": {"type": "string", "description": "Id of the parent job", "minLength": 1, "pattern": "^[0-9a-zA-Z-_]*$"}}, "files": {"description": "Sets of input/output data", "type": "array", "items": {"type": "object", "properties": {"name": {"description": "A human-readable name for the file", "type": "string", "minLength": 1}, "size": {"description": "File size in KB", "type": "integer", "minimum": 0}, "link": {"description": "Whether it is an input or output data", "type": "string", "enum": ["input", "output"]}}, "required": ["name", "size", "link"]}}, "runtime": {"description": "Job runtime in seconds", "type": "number"}, "cores": {"description": "Number of cores required by the job", "type": "integer", "minimum": 1}, "avgCPU": {"description": "Average CPU utilization in %", "type": "number"}, "bytesRead": {"description": "Total bytes read in KB", "type": "number"}, "bytesWritten": {"description": "Total bytes written in KB", "type": "number"}, "memory": {"description": "Memory (resident set) size of the process in KB", "type": "number"}, "energy": {"description": "Total energy consumption in kWh", "type": "number"}, "avgPower": {"description": "Average power consumption in W", "type": "number"}, "priority": {"description": "Job priority", "type": "number"}, "machine": {"description": "Machine name used", "type": "string", "minLength": 1}}, "required": ["name", "type", "runtime"]}}}, "required": ["makespan", "executedAt", "jobs"]}}, "required": ["name", "schemaVersion", "workflow"]}

wfcommons/wfchef/chef.py

@@ -8,9 +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
+import traceback
+from typing import Dict, Optional, Union
+import argparse
 import networkx as nx
 import numpy as np
 import pandas as pd
@@ -19,16 +23,15 @@
 import pkg_resources
 import subprocess
 import traceback
-
 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")
@@ -170,12 +173,13 @@ def analyzer_summary(path_to_instances: pathlib.Path) -> Dict:
 
     return stats_dict
 
+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:
+            return entry_point.load()
 
-def ls_recipe():
-    """
-    Inspired by UNIX `ls` command, it lists the recipes already installed into the system and 
-    how to import it to use.
-    """
+def get_recipes() -> pd.DataFrame:
     rows = []
     for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
         try:
@@ -185,8 +189,14 @@ def ls_recipe():
         except Exception as e:
             traceback.print_exc()
             print(f"Could not load {entry_point.module_name}")
-    df = pd.DataFrame(rows, columns=["name", "module", "import command"])
-    print(df.to_string(index=None))
+    return pd.DataFrame(rows, columns=["name", "module", "import command"])
+
+def ls_recipe():
+    """
+    Inspired by UNIX `ls` command, it lists the recipes already installed into the system and 
+    how to import it to use.
+    """
+    print(get_recipes())
 
 
 def uninstall_recipe(module_name: str):
@@ -209,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.
@@ -218,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
@@ -242,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
+        )