#22: using pathlib to handle files and directories

Author: rafaelfsilva

wfcommons/common/file.py

@@ -11,7 +11,7 @@
 import logging
 
 from logging import Logger
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from ..utils import NoValue
 
@@ -32,22 +32,22 @@ class File:
     :param link: Type of file link.
     :type link: FileLink
     :param logger: The logger where to log information/warning or errors.
-    :type logger: Logger
+    :type logger: Optional[Logger]
     """
 
     def __init__(self, name: str, size: int, link: FileLink, logger: Optional[Logger] = None) -> None:
         """A file used by tasks."""
-        self.logger: Logger = logging.getLogger(__name__) if logger is None else logger
+        self.logger: Logger = logger if logger else logging.getLogger(__name__)
 
         self.name: str = name
         self.size: int = size
         self.link: FileLink = link
 
-    def as_dict(self) -> Dict:
+    def as_dict(self) -> Dict[str, Union[str, int, FileLink]]:
         """A JSON representation of the file.
 
         :return: A JSON object representation of the file.
-        :rtype: Dict
+        :rtype: Dict[str, Union[str, int, FileLink]]
         """
         return {
             'link': self.link.value,

wfcommons/common/machine.py

@@ -78,15 +78,13 @@ def __init__(self,
         self.cpu_flops: int = cpu['count'] * cpu['speed'] * 10 ^ 6 if 'speed' in cpu else 0
         self.cpu_vendor: str = cpu['vendor'] if 'vendor' in cpu else None
 
-        self.logger.debug("created machine: {0} with {1} cores and {2} FLOPS.".format(
-            self.name, self.cpu_cores, self.cpu_flops)
-        )
+        self.logger.debug(f"created machine: {self.name} with {self.cpu_cores} cores and {self.cpu_flops} FLOPS.")
 
-    def as_dict(self) -> Dict:
+    def as_dict(self) -> Dict[str, Union[int, str]]:
         """A JSON representation of the machine.
 
         :return: A JSON object representation of the machine.
-        :rtype: Dict
+        :rtype: Dict[str, Union[int, str]]
         """
         machine = {"nodeName": self.name}
         if self.system:

wfcommons/common/task.py

@@ -37,33 +37,33 @@ class Task:
     :param cores: Number of cores required by the task.
     :type cores: float
     :param task_id: Job unique ID (e.g., ID0000001).
-    :type task_id: str
+    :type task_id: Optional[str]
     :param category: Job category (can be used, for example, to define jobs that use the same program).
-    :type category: str
+    :type category: Optional[str]
     :param machine: Machine on which is the task has been executed.
-    :type machine: Machine
+    :type machine: Optional[Machine]
     :param program: Program name.
-    :type program: str
+    :type program: Optional[str]
     :param args: List of task arguments.
-    :type args: List[str]
+    :type args: Optional[List[str]]
     :param avg_cpu: Average CPU utilization in %.
-    :type avg_cpu: float
+    :type avg_cpu: Optional[float]
     :param bytes_read: Total bytes read in KB.
-    :type bytes_read: int
+    :type bytes_read: Optional[int]
     :param bytes_written: Total bytes written in KB.
-    :type bytes_written: int
+    :type bytes_written: Optional[int]
     :param memory: Memory (resident set) size of the process in KB.
-    :type memory: int
+    :type memory: Optional[int]
     :param energy: Total energy consumption in kWh.
-    :type energy: int
+    :type energy: Optional[int]
     :param avg_power: Average power consumption in W.
-    :type avg_power: float
+    :type avg_power: Optional[float]
     :param priority: Task priority.
-    :type priority: int
+    :type priority: Optional[int]
     :param files: List of input/output files used by the task.
-    :type files: List[File]
+    :type files: Optional[List[File]]
     :param logger: The logger where to log information/warning or errors.
-    :type logger: Logger
+    :type logger: Optional[Logger]
     """
 
     def __init__(self,
@@ -75,15 +75,15 @@ def __init__(self,
                  category: Optional[str] = None,
                  machine: Optional[Machine] = None,
                  program: Optional[str] = None,
-                 args: List[str] = [],
+                 args: Optional[List[str]] = None,
                  avg_cpu: Optional[float] = None,
                  bytes_read: Optional[int] = None,
                  bytes_written: Optional[int] = None,
                  memory: Optional[int] = None,
                  energy: Optional[int] = None,
                  avg_power: Optional[float] = None,
                  priority: Optional[int] = None,
-                 files: List[File] = [],
+                 files: Optional[List[File]] = None,
                  logger: Optional[Logger] = None
                  ) -> None:
         """A task in a workflow."""
@@ -95,20 +95,18 @@ def __init__(self,
         self.task_id: Optional[str] = task_id
         self.category: Optional[str] = category
         self.program: Optional[str] = program
-        self.args: List[str] = args
+        self.args: List[str] = args if args else []
         self.avg_cpu: Optional[float] = avg_cpu
         self.bytes_read: Optional[int] = bytes_read
         self.bytes_written: Optional[int] = bytes_written
         self.memory: Optional[int] = memory
         self.energy: Optional[int] = energy
         self.avg_power: Optional[float] = avg_power
-        self.files: List[File] = files
+        self.files: List[File] = files if files else []
         self.machine: Machine = machine
-        self.priority = priority
+        self.priority: Optional[int] = priority
 
-        self.logger.debug("created {0} task {1}: runtime => {2} seconds.".format(
-            self.type, self.name, self.runtime)
-        )
+        self.logger.debug(f"created {self.type} task {self.name}: runtime => {self.runtime} seconds.")
 
     def as_dict(self) -> Dict:
         """A JSON representation of the task.

wfcommons/common/workflow.py

@@ -11,6 +11,7 @@
 import getpass
 import json
 import networkx as nx
+import pathlib
 
 from datetime import datetime
 from typing import Optional
@@ -29,17 +30,17 @@ class Workflow(nx.DiGraph):
     :param name: Workflow name.
     :type name: str
     :param description: Workflow instance description.
-    :type description: str
+    :type description: Optional[str]
     :param wms_name: WMS name.
-    :type wms_name: str
+    :type wms_name: Optional[str]
     :param wms_version: WMS version.
-    :type wms_version: str
+    :type wms_version: Optional[str]
     :param wms_url: URL for the WMS website.
-    :type wms_url: str
+    :type wms_url: Optional[str]
     :param executed_at: Workflow start timestamp in the ISO 8601 format.
-    :type executed_at: str
+    :type executed_at: Optional[str]
     :param makespan: Workflow makespan in seconds.
-    :type makespan: int
+    :type makespan: Optional[int]
     """
 
     def __init__(self,
@@ -52,21 +53,23 @@ def __init__(self,
                  makespan: Optional[int] = 0.0
                  ) -> None:
         """Create an object of a workflow representation."""
-        self.description = description if description else 'Instance generated with WfCommons - https://wfcommons.org'
-        self.created_at = str(datetime.utcnow().isoformat())
-        self.schema_version = "1.2"
-        self.wms_name = "WfCommons" if not wms_name else wms_name
-        self.wms_version = str(__version__) if not wms_version else wms_version
-        self.wms_url = f"https://docs.wfcommons.org/en/v{__version__}/" if not wms_url else wms_url
-        self.executed_at = datetime.now().astimezone().strftime("%Y%m%dT%H%M%S%z") if not executed_at else executed_at
-        self.makespan = makespan
+        self.description: Optional[
+            str] = description if description else 'Instance generated with WfCommons - https://wfcommons.org'
+        self.created_at: str = str(datetime.utcnow().isoformat())
+        self.schema_version: str = "1.2"
+        self.wms_name: Optional[str] = "WfCommons" if not wms_name else wms_name
+        self.wms_version: Optional[str] = str(__version__) if not wms_version else wms_version
+        self.wms_url: Optional[str] = f"https://docs.wfcommons.org/en/v{__version__}/" if not wms_url else wms_url
+        self.executed_at: Optional[str] = datetime.now().astimezone().strftime(
+            "%Y%m%dT%H%M%S%z") if not executed_at else executed_at
+        self.makespan: Optional[int] = makespan
         super().__init__(name=name, makespan=self.makespan, executedat=self.executed_at)
 
-    def write_json(self, json_filename: Optional[str] = None) -> None:
+    def write_json(self, json_file_path: Optional[pathlib.Path] = None) -> None:
         """Write a JSON file of the workflow instance.
 
-        :param json_filename: JSON output file name.
-        :type json_filename: str
+        :param json_file_path: JSON output file name.
+        :type json_file_path: Optional[pathlib.Path]
         """
         workflow_machines = []
         machines_list = []
@@ -122,22 +125,22 @@ def write_json(self, json_filename: Optional[str] = None) -> None:
             workflow_json['workflow']['machines'] = workflow_machines
 
         # write to file
-        if not json_filename:
-            json_filename = f'{self.name.lower()}.json'
-        with open(json_filename, 'w') as outfile:
+        if not json_file_path:
+            json_file_path = pathlib.Path(f'{self.name.lower()}.json')
+        with open(json_file_path, 'w') as outfile:
             outfile.write(json.dumps(workflow_json, indent=4))
 
-    def write_dot(self, dot_filename: str = None) -> None:
+    def write_dot(self, dot_file_path: Optional[pathlib.Path] = None) -> None:
         """Write a dot file of the workflow instance.
 
-        :param dot_filename: DOT output file name.
-        :type dot_filename: str
+        :param dot_file_path: DOT output file name.
+        :type dot_file_path: Optional[pathlib.Path]
         """
-        if not dot_filename:
-            dot_filename = f"{self.name.lower()}.dot"
-        nx.nx_agraph.write_dot(self, dot_filename)
+        if not dot_file_path:
+            dot_file_path = pathlib.Path(f"{self.name.lower()}.dot")
+        nx.nx_agraph.write_dot(self, dot_file_path)
 
     def to_nx_digraph(self) -> nx.DiGraph:
         with tempfile.NamedTemporaryFile() as temp:
-            self.write_json(temp.name)
-            return create_graph(temp.name)
+            self.write_json(pathlib.Path(temp.name))
+            return create_graph(pathlib.Path(temp.name))

wfcommons/wfchef/utils.py

@@ -8,14 +8,16 @@
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 
-import networkx as nx
+import json
 import pathlib
+
+import matplotlib.patches as mpatches
 import matplotlib.pyplot as plt
+import networkx as nx
+
+from hashlib import sha256
 from matplotlib import cm
-import matplotlib.patches as mpatches
 from typing import Iterable, Union, Set, Optional, Tuple, Hashable
-import json
-from hashlib import sha256
 
 this_dir = pathlib.Path(__file__).resolve().parent
 
@@ -32,12 +34,12 @@ def combine_hashes(*hashes: str) -> str:
     return string_hash(sorted(hashes))
 
 
-def create_graph(path: Union[str, pathlib.Path]) -> nx.DiGraph:
+def create_graph(path: pathlib.Path) -> nx.DiGraph:
     """
     Creates a networkX DiGraph from a JSON file in the WfFormat.
 
     :param path: name (for samples available in WfCommons) or the path to graphs JSON.
-    :type path: str or pathlib.Path.
+    :type path: pathlib.Path
     
     :return: graph.
     :rtype: networkX DiGraph.