refactor: replaced 'os.path.join()' uses with 'pathlib.Path' / syntax

install/find_mpi.py

@@ -2,11 +2,12 @@
 
 import mpi4py
 from mpi4py import MPI
+from pathlib import Path
 
-path = mpi4py.__path__[0]
+path = Path(mpi4py.__path__[0])
 print("\nmpi4py path found is:", path)
 
-configfile = os.path.join(path, "mpi.cfg")
+configfile = path / "mpi.cfg"
 print("\nShowing config file: ", configfile, "\n")
 
 with open(configfile, "r") as confile_handle:

libensemble/executors/executor.py

@@ -175,11 +175,11 @@ def workdir_exists(self) -> bool | None:
 
     def file_exists_in_workdir(self, filename: str) -> bool:
         """Returns true if the named file exists in the task's workdir"""
-        return self.workdir and os.path.exists(os.path.join(self.workdir, filename))
+        return self.workdir and os.path.exists(Path(self.workdir) / filename)
 
     def read_file_in_workdir(self, filename: str) -> str:
         """Opens and reads the named file in the task's workdir"""
-        path = os.path.join(self.workdir, filename)
+        path = Path(self.workdir) / filename
         if not os.path.exists(path):
             raise ValueError(f"{filename} not found in working directory")
         with open(path) as f:

libensemble/manager.py

@@ -12,6 +12,7 @@
 import sys
 import time
 import traceback
+from pathlib import Path
 from typing import Any
 
 import numpy as np
@@ -338,7 +339,7 @@ def _save_every_k(self, fname: str, count: int, k: int, complete: bool) -> None:
     def _save_every_k_sims(self, complete: bool) -> None:
         """Saves history every kth sim step"""
         self._save_every_k(
-            os.path.join(self.libE_specs["workflow_dir_path"], "{}_{}after_sim_{}.npy"),
+            str(Path(self.libE_specs["workflow_dir_path"]) / "{}_{}after_sim_{}.npy"),
             self.hist.sim_ended_count,
             self.libE_specs["save_every_k_sims"],
             complete,
@@ -347,7 +348,7 @@ def _save_every_k_sims(self, complete: bool) -> None:
     def _save_every_k_gens(self, complete: bool) -> None:
         """Saves history every kth gen step"""
         self._save_every_k(
-            os.path.join(self.libE_specs["workflow_dir_path"], "{}_{}after_gen_{}.npy"),
+            str(Path(self.libE_specs["workflow_dir_path"]) / "{}_{}after_gen_{}.npy"),
             self.hist.index,
             self.libE_specs["save_every_k_gens"],
             complete,

libensemble/resources/resources.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import socket
+from pathlib import Path
 
 from libensemble.resources import node_resources
 from libensemble.resources.env_resources import EnvResources
@@ -295,8 +296,8 @@ def get_global_nodelist(node_file=Resources.DEFAULT_NODEFILE, rundir=None, env_r
 
         In dedicated mode, any node with a libE worker is removed from the list.
         """
-        top_level_dir = rundir or os.getcwd()
-        node_filepath = os.path.join(top_level_dir, node_file)
+        top_level_dir = Path(rundir) if rundir else Path.cwd()
+        node_filepath = top_level_dir / node_file
         global_nodelist = []
         if os.path.isfile(node_filepath):
             with open(node_filepath, "r") as f:

libensemble/tests/run_tests.py

@@ -35,16 +35,16 @@
 # Test Directories - all relative to project root dir
 CODE_DIR = "libensemble"
 LIBE_SRC_DIR = CODE_DIR
-TESTING_DIR = os.path.join(CODE_DIR, "tests")
+TESTING_DIR = Path(CODE_DIR) / "tests"
 UNIT_TEST_SUBDIRS = [
     "unit_tests",
     "unit_tests_mpi_import",
     "unit_tests_nompi",
     "unit_tests_logger",
 ]
-UNIT_TEST_DIRS = [os.path.join(TESTING_DIR, subdir) for subdir in UNIT_TEST_SUBDIRS]
-REG_TEST_SUBDIR = os.path.join(TESTING_DIR, "regression_tests")
-FUNC_TEST_SUBDIR = os.path.join(TESTING_DIR, "functionality_tests")
+UNIT_TEST_DIRS = [TESTING_DIR / subdir for subdir in UNIT_TEST_SUBDIRS]
+REG_TEST_SUBDIR = TESTING_DIR / "regression_tests"
+FUNC_TEST_SUBDIR = TESTING_DIR / "functionality_tests"
 
 # Coverage merge and report dir
 COV_MERGE_DIR = TESTING_DIR
@@ -132,12 +132,13 @@ def cleanup(root_dir):
     ]
     dirs_to_clean = UNIT_TEST_DIRS + [REG_TEST_SUBDIR, FUNC_TEST_SUBDIR]
     for dir_path in dirs_to_clean:
-        full_path = os.path.join(root_dir, dir_path)
-        if "libensemble/tests/" not in full_path.replace("\\", "/"):
-            cprint(f"Safety check failed for {full_path}. Check directory", style="red")
+        full_path = Path(root_dir) / dir_path
+        full_path_str = str(full_path)
+        if "libensemble/tests/" not in full_path_str.replace("\\", "/"):
+            cprint(f"Safety check failed for {full_path_str}. Check directory", style="red")
             sys.exit(2)
         for pattern in patterns:
-            for file_path in glob.glob(os.path.join(full_path, pattern)):
+            for file_path in glob.glob(str(full_path / pattern)):
                 try:
                     if os.path.isfile(file_path) or os.path.islink(file_path):
                         os.remove(file_path)
@@ -199,8 +200,8 @@ def print_test_failed(test_num, test_script_name, comm, nprocs, duration):
 def merge_coverage_reports(root_dir):
     """Merge coverage data from multiple tests and generate a report."""
     print_heading("Generating coverage reports")
-    tests_dir = os.path.join(root_dir, "libensemble", "tests")
-    cov_files = glob.glob(os.path.join(tests_dir, "**", ".cov_*"), recursive=True)
+    tests_dir = Path(root_dir) / "libensemble" / "tests"
+    cov_files = glob.glob(str(tests_dir / "**" / ".cov_*"), recursive=True)
 
     if cov_files:
         try:
@@ -262,11 +263,11 @@ def is_open_mpi():
 def build_forces(root_dir):
     """Build forces.x using mpicc."""
     cprint("Building forces.x before running regression tests...", style="yellow", newline=True)
-    forces_app_dir = os.path.join(root_dir, "libensemble/tests/scaling_tests/forces/forces_app")
+    forces_app_dir = Path(root_dir) / "libensemble/tests/scaling_tests/forces/forces_app"
     subprocess.run(["mpicc", "-O3", "-o", "forces.x", "forces.c", "-lm"], cwd=forces_app_dir, check=True)
-    destination_dir = os.path.join(root_dir, "libensemble/tests/forces_app")
+    destination_dir = Path(root_dir) / "libensemble/tests/forces_app"
     os.makedirs(destination_dir, exist_ok=True)
-    shutil.copy(os.path.join(forces_app_dir, "forces.x"), destination_dir)
+    shutil.copy(forces_app_dir / "forces.x", destination_dir)
 
 
 def skip_test(directives, args, current_os):
@@ -335,7 +336,7 @@ def run_unit_tests(root_dir, python_exec, args):
     print_heading(f"Running unit tests (with pytest)")
     for dir_path in UNIT_TEST_DIRS:
         cprint(f"Entering unit test dir: {dir_path}", style="yellow", newline=True)
-        full_path = os.path.join(root_dir, dir_path)
+        full_path = Path(root_dir) / dir_path
         cov_rep = cov_report_type + ":cov_unit"
         cmd = python_exec + ["-m", "pytest", "--color=yes", "--timeout=120", "--cov", "--cov-report", cov_rep]
         if args.e:
@@ -366,8 +367,8 @@ def run_regression_tests(root_dir, python_exec, args, current_os):
     reg_test_list = REG_TEST_LIST
     reg_test_files = []
     for dir_path in test_dirs:
-        full_path = os.path.join(root_dir, dir_path)
-        reg_test_files.extend(glob.glob(os.path.join(full_path, reg_test_list)))
+        full_path = Path(root_dir) / dir_path
+        reg_test_files.extend(glob.glob(str(full_path / reg_test_list)))
 
     reg_test_files = sorted(reg_test_files)
     reg_pass = 0

libensemble/tests/scaling_tests/forces/forces_adv/forces_simf.py

@@ -1,5 +1,6 @@
 import os
 import time
+from pathlib import Path
 
 import numpy as np
 
@@ -101,7 +102,7 @@ def run_forces(H, persis_info, sim_specs, libE_info):
 
     # Stat file to check for bad runs
     statfile = "forces.stat"
-    filepath = os.path.join(task.workdir, statfile)
+    filepath = Path(task.workdir) / statfile
     line = None
 
     poll_interval = 0.1  # secs

libensemble/tests/scaling_tests/forces/forces_adv/forces_support.py

@@ -1,4 +1,5 @@
 import os
+from pathlib import Path
 
 outfiles = ["err.txt", "forces.stat", "out.txt"]
 
@@ -18,6 +19,7 @@ def test_libe_stats(status):
 
 
 def test_ensemble_dir(libE_specs, dir, nworkers, sim_max):
+    dir = Path(dir)
     if not os.path.isdir(dir):
         print(f"Specified ensemble directory {dir} not found.")
         return
@@ -36,11 +38,11 @@ def test_ensemble_dir(libE_specs, dir, nworkers, sim_max):
 
         worker_dirs = [i for i in os.listdir(dir) if i.startswith("worker")]
         for worker_dir in worker_dirs:
-            sim_dirs = [i for i in os.listdir(os.path.join(dir, worker_dir)) if i.startswith("sim")]
+            sim_dirs = [i for i in os.listdir(dir / worker_dir) if i.startswith("sim")]
             num_sim_dirs += len(sim_dirs)
 
             for sim_dir in sim_dirs:
-                files_found.append(all([i in os.listdir(os.path.join(dir, worker_dir, sim_dir)) for i in outfiles]))
+                files_found.append(all([i in os.listdir(dir / worker_dir / sim_dir) for i in outfiles]))
 
         assert (
             num_sim_dirs == sim_max
@@ -62,7 +64,7 @@ def test_ensemble_dir(libE_specs, dir, nworkers, sim_max):
 
         files_found = []
         for sim_dir in sim_dirs:
-            files_found.append(all([i in os.listdir(os.path.join(dir, sim_dir)) for i in outfiles]))
+            files_found.append(all([i in os.listdir(dir / sim_dir) for i in outfiles]))
 
         assert all(
             files_found

libensemble/tests/scaling_tests/forces/forces_adv/run_libe_forces.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 import os
 import sys
+from pathlib import Path
 
 import numpy as np
 from forces_simf import run_forces  # Sim func from current dir
@@ -28,7 +29,7 @@
 
 nworkers, is_manager, libE_specs, _ = parse_args()
 
-sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+sim_app = Path.cwd() / "../forces_app/forces.x"
 
 if not os.path.isfile(sim_app):
     sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_adv/run_libe_forces_from_yaml.py

@@ -4,13 +4,14 @@
 
 import numpy as np
 
+from pathlib import Path
 from libensemble.ensemble import Ensemble
 from libensemble.executors.mpi_executor import MPIExecutor
 from libensemble.tools import add_unique_random_streams
 
 ####################
 
-sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+sim_app = Path.cwd() / "../forces_app/forces.x"
 
 if not os.path.isfile(sim_app):
     sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_gpu/run_libe_forces.py

@@ -17,6 +17,7 @@
 
 import os
 import sys
+from pathlib import Path
 
 import numpy as np
 from forces_simf import run_forces  # Sim func from current dir
@@ -30,7 +31,7 @@
 if __name__ == "__main__":
     # Initialize MPI Executor
     exctr = MPIExecutor()
-    sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+    sim_app = Path.cwd() / "../forces_app/forces.x"
 
     if not os.path.isfile(sim_app):
         sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_gpu_var_resources/run_libe_forces.py

@@ -20,6 +20,7 @@
 
 import os
 import sys
+from pathlib import Path
 
 import numpy as np
 from forces_simf import run_forces  # Sim func from current dir
@@ -33,7 +34,7 @@
 if __name__ == "__main__":
     # Initialize MPI Executor
     exctr = MPIExecutor()
-    sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+    sim_app = Path.cwd() / "../forces_app/forces.x"
 
     if not os.path.isfile(sim_app):
         sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_multi_app/run_libe_forces.py

@@ -24,6 +24,7 @@
 
 import os
 import sys
+from pathlib import Path
 
 import numpy as np
 from forces_simf import run_forces  # Sim func from current dir
@@ -39,8 +40,8 @@
     exctr = MPIExecutor()
 
     # Register simulation executable with executor
-    cpu_app = os.path.join(os.getcwd(), "../forces_app/forces_cpu.x")
-    gpu_app = os.path.join(os.getcwd(), "../forces_app/forces_gpu.x")
+    cpu_app = Path.cwd() / "../forces_app/forces_cpu.x"
+    gpu_app = Path.cwd() / "../forces_app/forces_gpu.x"
 
     if not os.path.isfile(cpu_app):
         sys.exit(f"{cpu_app} not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_simple/run_libe_forces.py

@@ -3,6 +3,7 @@
 import sys
 
 import numpy as np
+from pathlib import Path
 from forces_simf import run_forces  # Sim func from current dir
 
 from libensemble import Ensemble
@@ -16,7 +17,7 @@
     exctr = MPIExecutor()
 
     # Register simulation executable with executor
-    sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+    sim_app = Path.cwd() / "../forces_app/forces.x"
 
     if not os.path.isfile(sim_app):
         sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/forces_simple_with_input_file/run_libe_forces.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 import os
 import sys
+from pathlib import Path
 
 import numpy as np
 from forces_simf import run_forces  # Sim func from current dir
@@ -16,7 +17,7 @@
     exctr = MPIExecutor()
 
     # Register simulation executable with executor
-    sim_app = os.path.join(os.getcwd(), "../forces_app/forces.x")
+    sim_app = Path.cwd() / "../forces_app/forces.x"
 
     if not os.path.isfile(sim_app):
         sys.exit("forces.x not found - please build first in ../forces_app dir")

libensemble/tests/scaling_tests/forces/globus_compute_forces/forces_simf.py

@@ -2,6 +2,7 @@ def run_forces_globus_compute(H, persis_info, sim_specs, libE_info):
     import os
     import secrets
     import time
+    from pathlib import Path
 
     import numpy as np
 
@@ -44,7 +45,7 @@ def read_last_line(filepath):
     exctr = MPIExecutor()
     exctr.register_app(full_path=sim_app, app_name="forces")
 
-    calc_dir = os.path.join(sim_specs["user"]["remote_ensemble_dir"], secrets.token_hex(nbytes=4))
+    calc_dir = Path(sim_specs["user"]["remote_ensemble_dir"]) / Path(secrets.token_hex(nbytes=4))
     os.makedirs(calc_dir, exist_ok=True)
     os.chdir(calc_dir)
 
@@ -90,7 +91,7 @@ def read_last_line(filepath):
 
     # Stat file to check for bad runs
     statfile = "forces.stat"
-    filepath = os.path.join(task.workdir, statfile)
+    filepath = Path(task.workdir) / statfile
     line = None
 
     poll_interval = 1  # secs

libensemble/tools/parse_args.py

@@ -1,6 +1,7 @@
 import argparse
 import os
 import sys
+from pathlib import Path
 
 # ==================== Command-line argument parsing ===========================
 
@@ -115,9 +116,9 @@ def _tcp_parse_args(args):
 def _ssh_parse_args(args):
     """Parses arguments for SSH with reverse tunnel."""
     nworkers = len(args.workers)
-    worker_pwd = args.worker_pwd or os.getcwd()
+    worker_pwd = Path(args.worker_pwd) if args.worker_pwd else Path.cwd()
     script_dir, script_name = os.path.split(sys.argv[0])
-    worker_script_name = os.path.join(worker_pwd, script_name)
+    worker_script_name = worker_pwd / script_name
     ssh = ["ssh", "-R", "{tunnel_port}:localhost:{manager_port}", "{worker_ip}"]
     cmd = [
         args.worker_python,