#22: using pathlib to handle files and directories

Author: rafaelfsilva

docs/source/analyzing_instances.rst

@@ -32,8 +32,10 @@ provides an *instance loader* tool for importing workflow execution instances
 for analysis. For instance, the code snippet below shows how an instance can
 be loaded using the :class:`~wfcommons.wfinstances.instance.Instance` class: ::
 
+    import pathlib
     from wfcommons import Instance
-    instance = Instance(input_instance='/path/to/instance/file.json')
+    input_instance = pathlib.Path('/path/to/instance/file.json')
+    instance = Instance(input_instance=input_instance)
 
 The :class:`~wfcommons.wfinstances.instance.Instance` class provides a number of
 methods for interacting with the workflow instance, including:
@@ -115,20 +117,19 @@ distribution fitting for task *prefixes* of the Seismology workflow
 plots (runtime, and input and output files) into the :code:`fits` folder using
 :code:`seismology` as a prefix for each generated plot: ::
 
+    import pathlib
     from wfcommons import Instance, InstanceAnalyzer
-    from os import listdir
-    from os.path import isfile, join
 
     # obtaining list of instance files in the folder
-    INSTANCES_PATH = "/path/to/some/instance/folder/"
-    instance_files = [f for f in listdir(INSTANCES_PATH) if isfile(join(INSTANCES_PATH, f))]
+    INSTANCES_PATH = pathlib.Path('/path/to/some/instance/folder/')
+    instance_files = [f for f in INSTANCES_PATH.glob('*') if INSTANCES_PATH.joinpath(f).is_file()]
 
     # creating the instance analyzer object
     analyzer = InstanceAnalyzer()
 
     # appending instance files to the instance analyzer
     for instance_file in instance_files:
-        instance = Instance(input_instance=INSTANCES_PATH + instance_file)
+        instance = Instance(input_instance=INSTANCES_PATH.joinpath(instance_file))
         analyzer.append_instance(instance)
 
     # list of workflow task name prefixes to be analyzed in each instance

docs/source/generating_workflows.rst

@@ -86,48 +86,52 @@ The following example generates a *Seismology* synthetic workflow instance
 os 300 tasks, builds a synthetic workflow instance, and writes the
 synthetic instance to a JSON file. ::
 
+    import pathlib
     from wfcommons.wfchef.recipes import SeismologyRecipe
     from wfcommons import WorkflowGenerator
 
-    generator = WorkflowGenerator(SeismologyRecipe.from_num_tasks(250)) 
+    generator = WorkflowGenerator(SeismologyRecipe.from_num_tasks(250))
     workflow = generator.build_workflow()
-    workflow.write_json(f'seismology-workflow.json')
+    workflow.write_json(pathlib.Path('seismology-workflow.json'))
 
 
 The example below generates a number of 10 *Blast* synthetic
 workflow instances for every size defined in the array :code:`num_tasks`: ::
 
+    import pathlib
     from wfcommons.wfchef.recipes import BlastRecipe
     from wfcommons import WorkflowGenerator
 
     num_tasks = [100, 250, 370, 800]
-    
+
     for task in num_tasks:
-      generator = WorkflowGenerator(BlastRecipe.from_num_tasks(task))
-      workflows = generator.build_workflows(10)
-      
-      for i, workflow in enumerate(workflows):
-        workflow.write_json(f'blast-workflow-{task}-{i}.json')
+        generator = WorkflowGenerator(BlastRecipe.from_num_tasks(task))
+        workflows = generator.build_workflows(10)
+
+        for i, workflow in enumerate(workflows):
+            workflow.write_json(pathlib.Path(f'blast-workflow-{task}-{i}.json'))
+
 
 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. ::
 
+    import pathlib
     from wfcommons.wfchef.recipes import EpigenomicsRecipe
     from wfcommons import WorkflowGenerator
 
     generator = WorkflowGenerator(EpigenomicsRecipe.from_num_tasks(1000))
     for i, workflow in enumerate(generator.build_workflows(10)):
-        workflow.write_json(f'epigenomics-workflow-{i}.json')
+        workflow.write_json(pathlib.Path(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. ::
 
+    import pathlib
     from wfcommons.wfchef.recipes import CyclesRecipe
     from wfcommons import WorkflowGenerator
 
     generator = WorkflowGenerator(CyclesRecipe.from_num_tasks(250))
     workflow = generator.build_workflow()
-    workflow.write_json(f'cycles-workflow.json')
-
+    workflow.write_json(pathlib.Path('cycles-workflow.json'))

docs/source/parsing_logs.rst

@@ -32,17 +32,21 @@ executions. The following example shows the analysis of Makeflow execution logs,
 stored in a local folder (:code:`execution_dir`), for a workflow execution using the
 :class:`~wfcommons.wfinstances.logs.makeflow.MakeflowLogsParser` class: ::
 
+    import pathlib
     from wfcommons.wfinstances import MakeflowLogsParser
 
     # creating the parser for the Makeflow workflow
-    parser = MakeflowLogsParser(execution_dir='/path/to/makeflow/execution/dir/blast/chameleon-small-001/',
-                                resource_monitor_logs_dir='/path/to/makeflow/resource/monitor/logs/dir')
+    execution_dir = pathlib.Path('/path/to/makeflow/execution/dir/blast/chameleon-small-001/')
+    resource_monitor_logs_dir = pathlib.Path('/path/to/makeflow/resource/monitor/logs/dir')
+    parser = MakeflowLogsParser(execution_dir=execution_dir,
+                                resource_monitor_logs_dir=resource_monitor_logs_dir)
 
     # generating the workflow instance object
     workflow = parser.build_workflow('makeflow-workflow-test')
 
     # writing the workflow instance to a JSON file
-    workflow.write_json('makeflow-workflow.json')
+    workflow_path = pathlib.Path('./makeflow-workflow.json')
+    workflow.write_json(workflow_path)
 
 .. note::
     The :class:`~wfcommons.wfinstances.logs.makeflow.MakeflowLogsParser` class requires
@@ -63,16 +67,19 @@ based on the dataflow programming model. The following example shows the analysi
 Nextflow execution logs, stored in a local folder (:code:`execution_dir`), for a workflow
 execution using the :class:`~wfcommons.wfinstances.logs.nextflow.NextflowLogsParser` class: ::
 
+    import pathlib
     from wfcommons.wfinstances import NextflowLogsParser
 
     # creating the parser for the Nextflow workflow
-    parser = NextflowLogsParser(execution_dir='/path/to/nextflow/execution/dir/')
+    execution_dir = pathlib.Path('/path/to/nextflow/execution/dir/')
+    parser = NextflowLogsParser(execution_dir=execution_dir)
 
     # generating the workflow instance object
     workflow = parser.build_workflow('nextflow-workflow-test')
 
     # writing the workflow instance to a JSON file
-    workflow.write_json('nextflow-workflow.json')
+    workflow_path = pathlib.Path('./nextflow-workflow.json')
+    workflow.write_json(workflow_path)
 
 .. note::
     The :class:`~wfcommons.wfinstances.logs.nextflow.NextflowLogsParser` class assumes
@@ -95,16 +102,19 @@ the analysis of Pegasus execution logs, stored in a local folder (:code:`submit_
 workflow execution using the :class:`~wfcommons.wfinstances.logs.pegasus.PegasusLogsParser`
 class: ::
 
+    import pathlib
     from wfcommons.wfinstances import PegasusLogsParser
 
     # creating the parser for the Pegasus workflow
-    parser = PegasusLogsParser(submit_dir='/path/to/pegasus/submit/dir/seismology/chameleon-100p-001/')
+    submit_dir = pathlib.Path('/path/to/pegasus/submit/dir/seismology/chameleon-100p-001/')
+    parser = PegasusLogsParser(submit_dir=submit_dir)
 
     # generating the workflow instance object
     workflow = parser.build_workflow('pegasus-workflow-test')
 
     # writing the workflow instance to a JSON file
-    workflow.write_json('pegasus-workflow.json')
+    workflow_path = pathlib.Path('./pegasus-workflow.json')
+    workflow.write_json(workflow_path)
 
 .. warning::
     By default, the :class:`~wfcommons.wfinstances.logs.pegasus.PegasusLogsParser`

wfcommons/utils.py

@@ -11,6 +11,7 @@
 import json
 import logging
 import math
+import pathlib
 import scipy.stats
 import warnings
 import numpy as np
@@ -27,7 +28,7 @@ def __repr__(self):
         return '<%s.%s>' % (self.__class__.__name__, self.name)
 
 
-def read_json(instance_filename: str) -> Dict[str, Any]:
+def read_json(instance_filename: pathlib.Path) -> Dict[str, Any]:
     """
     Read the JSON from the file path.
 

wfcommons/wfchef/chef.py

@@ -8,13 +8,9 @@
 # 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 traceback
-from typing import Dict, Optional, Union
 import argparse
+import math
 import networkx as nx
 import numpy as np
 import pandas as pd
@@ -23,29 +19,29 @@
 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")
 
 
-def compare_rmse(synth_graph: nx.DiGraph, real_graph: nx.DiGraph):
+def compare_rmse(synth_graph: nx.DiGraph, real_graph: nx.DiGraph) -> float:
     """
-    Calculates the Root Mean Square Error of a synthetic instance created 
+    Calculate the Root Mean Square Error of a synthetic instance created
     based on the correspondent (in number of tasks) real-world sample.
 
     :param synth_graph: a synthetic instance created by WfCommons.
-    :type synth_graph: networkX DiGraph 
+    :type synth_graph: networkX.DiGraph
     :param real_graph: the correspondent (in number of tasks) real-world workflow instance.
-    :type real_graph: networkX DiGraph 
+    :type real_graph: networkX.DiGraph
 
     :return: The RMSE between the synthetic instance and the real instance.
     :rtype: float
@@ -72,10 +68,10 @@ def compare_rmse(synth_graph: nx.DiGraph, real_graph: nx.DiGraph):
     return mse / real_graph.order()
 
 
-def find_err(workflow: Union[str, pathlib.Path],
-             err_savepath: Optional[Union[str, pathlib.Path]] = None,
-             always_update: bool = False,
-             runs: int = 1) -> pd.DataFrame:
+def find_err(workflow: pathlib.Path,
+             err_savepath: Optional[pathlib.Path] = None,
+             always_update: Optional[bool] = False,
+             runs: Optional[int] = 1) -> pd.DataFrame:
     """
     Creates a dataframe with the Root Mean Square Error of the synthetic instances created 
     based on the correspondent, w.r.t. number of tasks, real-world samples available at 
@@ -84,13 +80,13 @@ def find_err(workflow: Union[str, pathlib.Path],
     repositories.
 
     :param workflow: name (for samples available in WfCommons) or path to the real workflow instances.
-    :type workflow: str or pathlib.Path 
+    :type workflow: pathlib.Path
     :param err_savepath: path to save the err (rmse) of all instances available into a csv.
-    :type real_graph: str or pathlib.Path
+    :type real_graph: Optional[pathlib.Path]
     :param always_update: flag to set if the err needs to be updated or not (True: if new instances are added, False: otherwise).
-    :type real_graph: bool 
+    :type real_graph: Optional[bool]
     :param runs: number of times to repeat the err calculation process (due to randomization).
-    :type runs: bool
+    :type runs: Optional[bool]
 
     :return: dataframe with RMSE of all available instances.
     :rtype: pd.DataFrame
@@ -100,7 +96,6 @@ def find_err(workflow: Union[str, pathlib.Path],
                            key=lambda name: summary["base_graphs"][name]["order"])
 
     if err_savepath:
-        err_savepath = pathlib.Path(err_savepath)
         err_savepath.parent.mkdir(exist_ok=True, parents=True)
 
     labels = [graph for graph in sorted_graphs]
@@ -147,17 +142,11 @@ def analyzer_summary(path_to_instances: pathlib.Path) -> Dict:
     and from Makeflow WMS GitHub <https://github.com/wfcommons/makeflow-instances>`_
     repositories.
 
-    :param workflow: name (for samples available in WfCommons) or path to the real workflow instances.
-    :type workflow: str or pathlib.Path 
-    :param err_savepath: path to save the err (rmse) of all instances available into a csv.
-    :type real_graph: str or pathlib.Path
-    :param always_update: flag to set if the err needs to be updated or not (True: if new instances are added, False: otherwise).
-    :type real_graph: bool 
-    :param runs: number of times to repeat the err calculation process (due to randomization).
-    :type runs:bool
+    :param path_to_instances:
+    :type path_to_instances: pathlib.Path
 
-    :return: dataframe with RMSE of all available instances.
-    :rtype: pd.DataFrame
+    :return:
+    :rtype: Dict
     """
     analyzer = InstanceAnalyzer()
     task_types = set()
@@ -173,12 +162,14 @@ 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]
+        att = entry_point.attrs[0]
         if att == recipe:
             return entry_point.load()
 
+
 def get_recipes() -> pd.DataFrame:
     rows = []
     for entry_point in pkg_resources.iter_entry_points('workflow_recipes'):
@@ -191,6 +182,7 @@ def get_recipes() -> pd.DataFrame:
             print(f"Could not load {entry_point.module_name}")
     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 
@@ -219,7 +211,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):
     """
     Creates a recipe for a workflow application by automatically replacing custom information 
     from the recipe skeleton.
@@ -253,7 +245,7 @@ def create_recipe(path_to_instances: Union[str, pathlib.Path],
     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

@@ -8,15 +8,16 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import pathlib
 import json
-import pickle
 import networkx as nx
-from typing import Set, List, Union, Dict
-from uuid import uuid4
 import numpy as np
+import pathlib
+import pickle
 import random
 
+from typing import Set, List, Union, Dict
+from uuid import uuid4
+
 this_dir = pathlib.Path(__file__).resolve().parent
 
 
@@ -77,15 +78,14 @@ def duplicate(path: pathlib.Path,
     :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(