renaming generator to wfgen

Author: rafaelfsilva

.github/workflows/build.yml

@@ -23,3 +23,7 @@ jobs:
       - name: Check package install
         run: |
           pip install .
+      - name: Build documentation
+        run: |
+          cd docs
+          make html

docs/source/generating_workflows.rst

@@ -110,16 +110,9 @@ workflow instances for every size defined in the array :code:`num_tasks`: ::
       for i, workflow in enumerate(workflows):
         workflow.write_json(f'blast-workflow-{task}-{i}.json')
 
-
-
-
-
-Examples
---------
-
 The following example generates 10 *Epigenomics* synthetic workflow instances
-based on the number of tasks entered by the user (1000), builds the synthetic workflow instances, and writes the
-synthetic instances to JSON files. ::
+based on the number of tasks entered by the user (1000), builds the synthetic
+workflow instances, and writes the synthetic instances to JSON files. ::
 
     from wfcommons.wfchef.recipes import EpigenomicsRecipe
     from wfcommons.generator import WorkflowGenerator
@@ -128,9 +121,9 @@ synthetic instances to JSON files. ::
     for i, workflow in enumerate(generator.build_workflows(10)):
         workflow.write_json(f'epigenomics-workflow-{i}.json')
 
-The example below generates a *Cycles* (agroecosystem) synthetic workflow instance based on the number
-of tasks entered by the user (250), builds the synthetic workflow instance, and writes the synthetic
-instance to a JSON file. ::
+The example below generates a *Cycles* (agroecosystem) synthetic workflow instance
+based on the number of tasks entered by the user (250), builds the synthetic workflow
+instance, and writes the synthetic instance to a JSON file. ::
 
     from wfcommons.wfchef.recipes import CyclesRecipe
     from wfcommons.generator import WorkflowGenerator
@@ -139,5 +132,3 @@ instance to a JSON file. ::
     workflow = generator.build_workflow()
     workflow.write_json(f'cycles-workflow.json')
 
-..
-    maybe we should pout examples only on generator, because we need it

wfcommons/__init__.py

@@ -16,7 +16,7 @@
 
 import logging
 
-from .generator import WorkflowGenerator
+from .wfgen import WorkflowGenerator
 from .trace import Trace, TraceAnalyzer, TraceElement
 
 logging.getLogger('wfcommons').addHandler(logging.NullHandler())

wfcommons/generator/workflow/__init__.py

@@ -21,12 +21,12 @@
 import traceback
 
 from typing import Dict, Optional, Union
-from stringcase import snakecase, capitalcase
+from stringcase import capitalcase
 
 from .duplicate import duplicate, NoMicrostructuresError
 from .find_microstructures import save_microstructures
 from .utils import create_graph
-from ..generator.workflow.abstract_recipe import WorkflowRecipe
+from ..wfgen.abstract_recipe import WorkflowRecipe
 from ..trace.trace import Trace
 from ..trace.trace_analyzer import TraceAnalyzer
 
@@ -176,7 +176,6 @@ def ls_recipe():
     Inspired by UNIX `ls` command, it lists the recipes already installed into the system and 
     how to import it to use.
     """
-    import inspect
     rows = []
     for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
         try:

wfcommons/wfchef/chef.py

@@ -8,113 +8,78 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-from typing import Dict, Optional, Set
-
-from wfcommons.common.workflow import Workflow
+import json
+import pathlib
 
-from wfcommons.generator.workflow.abstract_recipe import WorkflowRecipe
-from wfcommons.wfchef.duplicate import duplicate
+from logging import Logger
+from typing import Dict, Optional, Set
 
-import pathlib 
-import pickle
-import networkx as nx
-import numpy as np
-import pandas as pd
-import json
+from wfcommons.wfgen.abstract_recipe import WorkflowRecipe
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
 class BlastRecipe(WorkflowRecipe):
     """A Blast workflow recipe class for creating synthetic workflow traces.
 
-    :param num_pairs: The number of pair of signals to estimate earthquake STFs.
-    :type num_pairs: int
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
     :type data_footprint: int
     :param num_tasks: The upper bound for the total number of tasks in the workflow.
     :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
+    :param logger: The logger where to log information/warning or errors (optional).
+    :type logger: Logger
     """
 
     def __init__(self,
                  data_footprint: Optional[int] = 0,
                  num_tasks: Optional[int] = 3,
                  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,
+                 logger: Optional[Logger] = None,
                  **kwargs) -> None:
-        super().__init__("Blast", data_footprint, num_tasks)
-        self.exclude_graphs = exclude_graphs
-
-    def generate_nx_graph(self) -> nx.DiGraph:
-        summary_path = this_dir.joinpath("microstructures", "summary.json")
-        summary = json.loads(summary_path.read_text())
-
-        metric_path = this_dir.joinpath("microstructures", "metric", "err.csv")
-        df = pd.read_csv(str(metric_path), index_col=0)
-        df = df.drop(self.exclude_graphs, axis=0, errors="ignore")
-        df = df.drop(self.exclude_graphs, axis=1, errors="ignore")
-        for col in df.columns:
-            df.loc[col, col] = np.nan
-
-        reference_orders = [summary["base_graphs"][col]["order"] for col in df.columns]
-        idx = np.argmin([abs(self.num_tasks - ref_num_tasks) for ref_num_tasks in reference_orders])
-        reference = df.columns[idx]
-
-        base = df.index[df[reference].argmin()]
-        graph = duplicate(this_dir.joinpath("microstructures"), base, self.num_tasks)
-        return graph
+        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()) -> 'BlastRecipe':
+    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)
-
-    def _load_base_graph(self) -> nx.DiGraph:
-        return pickle.loads(this_dir.joinpath("base_graph.pickle").read_bytes())
-
-    def _load_microstructures(self) -> Dict:
-        return json.loads(this_dir.joinpath("microstructures.json").read_text())
-
-    def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
-        """Generate a synthetic workflow trace of a Blast workflow.
-
-        :param workflow_name: The workflow name
-        :type workflow_name: int
-
-        :return: A synthetic workflow trace object.
-        :rtype: Workflow
-        """
-        workflow = Workflow(name=self.name + "-synthetic-trace" if not workflow_name else workflow_name, makespan=None)
-        graph = self.generate_nx_graph()
-
-        task_names = {}
-        for node in graph.nodes:
-            if node in ["SRC", "DST"]:
-                continue
-            node_type = graph.nodes[node]["type"]
-            task_name = self._generate_task_name(node_type)
-            task = self._generate_task(node_type, task_name)
-            workflow.add_node(task_name, task=task)
-
-            task_names[node] = task_name
-
-        for (src, dst) in graph.edges:
-            if src in ["SRC", "DST"] or dst in ["SRC", "DST"]:
-                continue
-            workflow.add_edge(task_names[src], task_names[dst])        
-        
-        workflow.nxgraph = graph
-        self.workflows.append(workflow)
-        return workflow
+        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:
         """

wfcommons/wfchef/recipes/blast/recipe.py

@@ -8,113 +8,78 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-from typing import Dict, Optional, Set
-
-from wfcommons.common.workflow import Workflow
+import json
+import pathlib
 
-from wfcommons.generator.workflow.abstract_recipe import WorkflowRecipe
-from wfcommons.wfchef.duplicate import duplicate
+from logging import Logger
+from typing import Dict, Optional, Set
 
-import pathlib 
-import pickle
-import networkx as nx
-import numpy as np
-import pandas as pd
-import json
+from wfcommons.wfgen.abstract_recipe import WorkflowRecipe
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
 class BwaRecipe(WorkflowRecipe):
     """A Bwa workflow recipe class for creating synthetic workflow traces.
 
-    :param num_pairs: The number of pair of signals to estimate earthquake STFs.
-    :type num_pairs: int
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
     :type data_footprint: int
     :param num_tasks: The upper bound for the total number of tasks in the workflow.
     :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
+    :param logger: The logger where to log information/warning or errors (optional).
+    :type logger: Logger
     """
 
     def __init__(self,
                  data_footprint: Optional[int] = 0,
                  num_tasks: Optional[int] = 3,
                  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,
+                 logger: Optional[Logger] = None,
                  **kwargs) -> None:
-        super().__init__("Bwa", data_footprint, num_tasks)
-        self.exclude_graphs = exclude_graphs
-
-    def generate_nx_graph(self) -> nx.DiGraph:
-        summary_path = this_dir.joinpath("microstructures", "summary.json")
-        summary = json.loads(summary_path.read_text())
-
-        metric_path = this_dir.joinpath("microstructures", "metric", "err.csv")
-        df = pd.read_csv(str(metric_path), index_col=0)
-        df = df.drop(self.exclude_graphs, axis=0, errors="ignore")
-        df = df.drop(self.exclude_graphs, axis=1, errors="ignore")
-        for col in df.columns:
-            df.loc[col, col] = np.nan
-
-        reference_orders = [summary["base_graphs"][col]["order"] for col in df.columns]
-        idx = np.argmin([abs(self.num_tasks - ref_num_tasks) for ref_num_tasks in reference_orders])
-        reference = df.columns[idx]
-
-        base = df.index[df[reference].argmin()]
-        graph = duplicate(this_dir.joinpath("microstructures"), base, self.num_tasks)
-        return graph
+        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()) -> 'BwaRecipe':
+    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)
-
-    def _load_base_graph(self) -> nx.DiGraph:
-        return pickle.loads(this_dir.joinpath("base_graph.pickle").read_bytes())
-
-    def _load_microstructures(self) -> Dict:
-        return json.loads(this_dir.joinpath("microstructures.json").read_text())
-
-    def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
-        """Generate a synthetic workflow trace of a Bwa workflow.
-
-        :param workflow_name: The workflow name
-        :type workflow_name: int
-
-        :return: A synthetic workflow trace object.
-        :rtype: Workflow
-        """
-        workflow = Workflow(name=self.name + "-synthetic-trace" if not workflow_name else workflow_name, makespan=None)
-        graph = self.generate_nx_graph()
-
-        task_names = {}
-        for node in graph.nodes:
-            if node in ["SRC", "DST"]:
-                continue
-            node_type = graph.nodes[node]["type"]
-            task_name = self._generate_task_name(node_type)
-            task = self._generate_task(node_type, task_name)
-            workflow.add_node(task_name, task=task)
-
-            task_names[node] = task_name
-
-        for (src, dst) in graph.edges:
-            if src in ["SRC", "DST"] or dst in ["SRC", "DST"]:
-                continue
-            workflow.add_edge(task_names[src], task_names[dst])        
-        
-        workflow.nxgraph = graph
-        self.workflows.append(workflow)
-        return workflow
+        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:
         """