#22: using pathlib to handle files and directories

Author: rafaelfsilva

wfcommons/utils.py

@@ -28,7 +28,8 @@ def __repr__(self):
 
 
 def read_json(instance_filename: str) -> Dict[str, Any]:
-    """Read the JSON from the file path.
+    """
+    Read the JSON from the file path.
 
     :param instance_filename: The absolute path of the instance file.
     :type instance_filename: str
@@ -41,7 +42,8 @@ def read_json(instance_filename: str) -> Dict[str, Any]:
 
 
 def best_fit_distribution(data: List[float], logger: Optional[Logger] = None) -> Tuple:
-    """Fit a list of values to a distribution.
+    """
+    Fit a list of values to a distribution.
 
     :param data: List of values to be fitted to a distribution.
     :type data: List[float]
@@ -95,12 +97,13 @@ def best_fit_distribution(data: List[float], logger: Optional[Logger] = None) ->
             except Exception as e:
                 print(f"WARNING: distribution \"{dist_name}\" failed ({e})")
 
-    logger.debug('Best distribution fit: {}'.format(best_distribution))
+    logger.debug(f'Best distribution fit: {best_distribution}')
     return best_distribution, best_params
 
 
 def generate_rvs(distribution: Dict, min_value: float, max_value: float) -> float:
-    """Generate a random variable from a distribution.
+    """
+    Generate a random variable from a distribution.
 
     :param distribution: Distribution dictionary (name and parameters).
     :type distribution: Dict
@@ -122,7 +125,8 @@ def generate_rvs(distribution: Dict, min_value: float, max_value: float) -> floa
 
 
 def ncr(n: int, r: int) -> int:
-    """Calculate the number of combinations.
+    """
+    Calculate the number of combinations.
 
     :param n: The number of items.
     :type n: int

wfcommons/wfgen/abstract_recipe.py

@@ -14,7 +14,7 @@
 from abc import ABC, abstractmethod
 from os import path
 from logging import Logger
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional
 
 from wfcommons.common.file import File, FileLink
 from wfcommons.common.task import Task, TaskType
@@ -23,22 +23,23 @@
 
 
 class WorkflowRecipe(ABC):
-    """An abstract class of workflow recipes for creating synthetic workflow instances.
+    """
+    An abstract class of workflow recipes for creating synthetic workflow instances.
 
     :param name: The workflow recipe name.
     :type name: str
     :param data_footprint: The upper bound for the workflow total data footprint (in bytes).
-    :type data_footprint: int
+    :type data_footprint: Optional[int]
     :param num_tasks: The upper bound for the total number of tasks in the workflow.
-    :type num_tasks: int
+    :type num_tasks: Optional[int]
     :param runtime_factor: The factor of which tasks runtime will be increased/decreased.
-    :type runtime_factor: float
+    :type runtime_factor: Optional[float]
     :param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
-    :type input_file_size_factor: float
+    :type input_file_size_factor: Optional[float]
     :param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
-    :type output_file_size_factor: float
+    :type output_file_size_factor: Optional[float]
     :param logger: The logger where to log information/warning or errors (optional).
-    :type logger: Logger
+    :type logger: Optional[Logger]
     """
 
     def __init__(self, name: str,
@@ -57,23 +58,24 @@ def __init__(self, name: str,
         if output_file_size_factor <= 0.0:
             raise ValueError("The output file size factor should be a number higher than 0.0.")
 
-        self.logger = logging.getLogger(__name__) if logger is None else logger
+        self.logger: Optional[Logger] = logging.getLogger(__name__) if logger is None else logger
         self.name = name
-        self.data_footprint = data_footprint
-        self.num_tasks = num_tasks
-        self.runtime_factor = runtime_factor
-        self.input_file_size_factor = input_file_size_factor
-        self.output_file_size_factor = output_file_size_factor
+        self.data_footprint: Optional[int] = data_footprint
+        self.num_tasks: Optional[int] = num_tasks
+        self.runtime_factor: Optional[float] = runtime_factor
+        self.input_file_size_factor: Optional[float] = input_file_size_factor
+        self.output_file_size_factor: Optional[float] = output_file_size_factor
         self.tasks_files: Dict[str, List[File]] = {}
         self.tasks_files_names: Dict[str, List[str]] = {}
-        self.task_id_counter = 1
+        self.task_id_counter: int = 1
         self.tasks_map = {}
         self.tasks_children = {}
         self.tasks_parents = {}
 
     @abstractmethod
     def _workflow_recipe(self) -> Dict[str, Any]:
-        """Recipe for generating synthetic instances for a workflow. Recipes can be
+        """
+        Recipe for generating synthetic instances for a 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.
@@ -96,11 +98,11 @@ def from_num_tasks(cls,
         :param num_tasks: The upper bound for the total number of tasks in the workflow.
         :type num_tasks: int
         :param runtime_factor: The factor of which tasks runtime will be increased/decreased.
-        :type runtime_factor: float
+        :type runtime_factor: Optional[float]
         :param input_file_size_factor: The factor of which tasks input files size will be increased/decreased.
-        :type input_file_size_factor: float
+        :type input_file_size_factor: Optional[float]
         :param output_file_size_factor: The factor of which tasks output files size will be increased/decreased.
-        :type output_file_size_factor: float
+        :type output_file_size_factor: Optional[float]
 
         :return: A workflow recipe object that will generate synthetic workflows up to
                  the total number of tasks provided.
@@ -110,18 +112,20 @@ def from_num_tasks(cls,
 
     @abstractmethod
     def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
-        """Generate a synthetic workflow instance.
+        """
+        Generate a synthetic workflow instance.
 
         :param workflow_name: The workflow name
-        :type workflow_name: int
+        :type workflow_name: Optional[str]
 
         :return: A synthetic workflow instance object.
         :rtype: Workflow
         """
         raise NotImplementedError
 
     def _generate_task(self, task_name: str, task_id: str) -> Task:
-        """Generate a synthetic task.
+        """
+        Generate a synthetic task.
 
         :param task_name: task name.
         :type task_name: str
@@ -165,7 +169,8 @@ def _generate_task(self, task_name: str, task_id: str) -> Task:
         return task
 
     def _generate_task_name(self, prefix: str) -> str:
-        """Generate a task name from a prefix appended with an ID.
+        """
+        Generate a task name from a prefix appended with an ID.
 
         :param prefix: task prefix.
         :type prefix: str
@@ -178,7 +183,8 @@ def _generate_task_name(self, prefix: str) -> str:
         return task_name
 
     def _generate_task_files(self, task: Task) -> List[File]:
-        """Generate input and output files for a task.
+        """
+        Generate input and output files for a task.
 
         :param task: task object.
         :type task: Task
@@ -211,7 +217,8 @@ def _generate_task_files(self, task: Task) -> List[File]:
         return output_files_list
 
     def _generate_files(self, task_id: str, recipe: Dict[str, Any], link: FileLink) -> List[File]:
-        """Generate files for a specific task ID.
+        """
+        Generate files for a specific task ID.
 
         :param task_id: task ID.
         :type task_id: str
@@ -242,7 +249,8 @@ def _generate_files(self, task_id: str, recipe: Dict[str, Any], link: FileLink)
         return files_list
 
     def _generate_file(self, extension: str, recipe: Dict[str, Any], link: FileLink) -> File:
-        """Generate a file according to a file recipe.
+        """
+        Generate a file according to a file recipe.
 
         :param extension:
         :type extension: str
@@ -263,7 +271,8 @@ def _generate_file(self, extension: str, recipe: Dict[str, Any], link: FileLink)
                     size=size)
 
     def _get_files_by_task_and_link(self, task_id: str, link: FileLink) -> List[File]:
-        """Get the list of files for a task ID and link type.
+        """
+        Get the list of files for a task ID and link type.
 
         :param task_id: task ID.
         :type task_id: str

wfcommons/wfgen/generator.py

@@ -25,7 +25,7 @@ class WorkflowGenerator:
     :param workflow_recipe: The workflow recipe to be used for this generator.
     :type workflow_recipe: WorkflowRecipe
     :param logger: The logger where to log information/warning or errors (optional).
-    :type logger: Logger
+    :type logger: Optional[Logger]
     """
 
     def __init__(self, workflow_recipe: WorkflowRecipe, logger: Optional[Logger] = None) -> None:
@@ -44,15 +44,14 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
         the generator.
 
         :param workflow_name: The workflow name.
-        :type workflow_name: str
+        :type workflow_name: Optional[str]
 
         :return: A synthetic workflow instance object.
         :rtype: Workflow
         """
         workflow: Workflow = self.workflow_recipe.build_workflow(workflow_name)
         self.workflows.append(workflow)
-        self.logger.info(
-            "Generated a synthetic workflow with {} tasks".format(len(workflow.nodes)))
+        self.logger.info(f"Generated a synthetic workflow with {len(workflow.nodes)} tasks")
         return workflow
 
     def build_workflows(self, num_workflows: int) -> List[Workflow]:

wfcommons/wfinstances/logs/makeflow.py

@@ -8,11 +8,10 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import glob
 import json
 import itertools
 import math
-import os
+import pathlib
 
 from datetime import datetime
 from logging import Logger
@@ -30,44 +29,43 @@ class MakeflowLogsParser(LogsParser):
     Parse Makeflow submit directory to generate workflow instance.
 
     :param execution_dir: Makeflow workflow execution directory (contains .mf and .makeflowlog files).
-    :type execution_dir: str
+    :type execution_dir: pathlib.Path
     :param resource_monitor_logs_dir: Resource Monitor log files directory.
-    :type resource_monitor_logs_dir: str
+    :type resource_monitor_logs_dir: pathlib.Path
     :param description: Workflow instance description.
-    :type description: str
+    :type description: Optional[str]
     :param logger: The logger where to log information/warning or errors (optional).
-    :type logger: Logger
+    :type logger: Optional[Logger]
     """
 
     def __init__(self,
-                 execution_dir: str,
-                 resource_monitor_logs_dir: str,
+                 execution_dir: pathlib.Path,
+                 resource_monitor_logs_dir: pathlib.Path,
                  description: Optional[str] = None,
                  logger: Optional[Logger] = None) -> None:
         """Create an object of the makeflow log parser."""
         super().__init__('Makeflow', 'http://ccl.cse.nd.edu/software/makeflow/', description, logger)
 
         # Sanity check
-        if not os.path.isdir(execution_dir):
-            raise OSError('The provided path does not exist or is not a folder: {}'.format(execution_dir))
+        if not execution_dir.is_dir():
+            raise OSError(f'The provided path does not exist or is not a folder: {execution_dir}')
 
-        files = glob.glob('{}/*.mf'.format(execution_dir))
+        files: List[pathlib.Path] = list(execution_dir.glob('*.mf'))
         if len(files) == 0:
-            raise OSError('Unable to find .mf file in: {}'.format(execution_dir))
-        self.mf_file = files[0]
+            raise OSError(f'Unable to find .mf file in: {execution_dir}')
+        self.mf_file: pathlib.Path = files[0]
 
-        files = glob.glob('{}/*.makeflowlog'.format(execution_dir))
+        files = list(execution_dir.glob('*.makeflowlog'))
         if len(files) == 0:
-            raise OSError('Unable to find .makeflowlog file in: {}'.format(execution_dir))
-        self.mf_log_file = files[0]
+            raise OSError(f'Unable to find .makeflowlog file in: {execution_dir}')
+        self.mf_log_file: pathlib.Path = files[0]
 
-        if not os.path.isdir(resource_monitor_logs_dir):
-            raise OSError('The provided path does not exist or is not a folder: {}'.format(resource_monitor_logs_dir))
+        if not resource_monitor_logs_dir.is_dir():
+            raise OSError(f'The provided path does not exist or is not a folder: {resource_monitor_logs_dir}')
 
-        self.execution_dir = execution_dir
+        self.execution_dir: pathlib.Path = execution_dir
 
-        # self.mf_log_file = mf_log_file
-        self.resource_monitor_logs_dir = resource_monitor_logs_dir
+        self.resource_monitor_logs_dir: pathlib.Path = resource_monitor_logs_dir
         self.files_map = {}
         self.args_map = {}
 
@@ -76,7 +74,7 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
         Create workflow instance based on the workflow execution logs.
 
         :param workflow_name: The workflow name.
-        :type workflow_name: str
+        :type workflow_name: Optional[str]
 
         :return: A workflow instance object.
         :rtype: Workflow
@@ -100,7 +98,7 @@ def build_workflow(self, workflow_name: Optional[str] = None) -> Workflow:
 
         return self.workflow
 
-    def _parse_workflow_file(self):
+    def _parse_workflow_file(self) -> None:
         """Parse the makeflow workflow file and build the workflow structure."""
         task_id_counter = 1
 
@@ -148,8 +146,9 @@ def _parse_workflow_file(self):
                 if self.files_map[file]['task_name']:
                     self.workflow.add_edge(self.files_map[file]['task_name'], child)
 
-    def _create_files(self, files_list: List[str], link: FileLink, task_name: str):
-        """ Create a list of files objects.
+    def _create_files(self, files_list: List[str], link: FileLink, task_name: str) -> List[File]:
+        """
+        Create a list of files objects.
 
         :param files_list: list of file names.
         :rtype files_list: List[str]
@@ -168,10 +167,11 @@ def _create_files(self, files_list: List[str], link: FileLink, task_name: str):
                     self.files_map[file]['file'][0] if link == FileLink.INPUT else self.files_map[file]['file'][1])
             else:
                 size = 0
-                if os.path.isdir('{}/{}'.format(self.execution_dir, file)):
-                    size = sum(os.path.getsize(f) for f in os.listdir('.') if os.path.isfile(f))
-                elif os.path.isfile('{}/{}'.format(self.execution_dir, file)):
-                    size = int(math.ceil(os.stat('{}/{}'.format(self.execution_dir, file)).st_size / 1000))  # B to KB
+                file_path = self.execution_dir.joinpath(file)
+                if file_path.is_dir():
+                    size = sum(math.ceil(f.stat().st_size / 1000) for f in file_path.glob("*") if f.is_file())
+                elif file_path.is_file():
+                    size = int(math.ceil(file_path.stat().st_size / 1000))  # B to KB
 
                 file_obj_in = File(name=file,
                                    size=size,
@@ -206,7 +206,7 @@ def _parse_makeflow_log_file(self):
                 elif 'COMPLETED' in line:
                     self.workflow.makespan = float('%.2f' % ((int(line.split()[2]) - start_time) / 1000000))
 
-                elif line.startswith('# FILE') and not 'condorlog' in line:
+                elif line.startswith('# FILE') and 'condorlog' not in line:
                     file_name = line.split()[3]
                     if file_name in self.files_map:
                         size = int(math.ceil(int(line.split()[5]) / 1000))  # B to KB
@@ -215,7 +215,7 @@ def _parse_makeflow_log_file(self):
 
     def _parse_resource_monitor_logs(self):
         """Parse the log files produced by resource monitor"""
-        for file in glob.glob('{}/*.summary'.format(self.resource_monitor_logs_dir)):
+        for file in pathlib.Path.glob(f'{self.resource_monitor_logs_dir}/*.summary'):
             with open(file) as f:
                 data = json.load(f)