Reflect Andrei's comments

Co-authored-by: Andrei Maslennikov <andreyma@nvidia.com>

Author: TaekyungHeo

src/cloudai/_core/parser.py

@@ -65,7 +65,7 @@ def parse(
         if not test_path.exists():
             raise FileNotFoundError(f"Test path '{test_path}' not found.")
 
-        system_parser = SystemParser(str(self.system_config_path))
+        system_parser = SystemParser(self.system_config_path)
         system = system_parser.parse()
         logging.debug("Parsed system config")
 

src/cloudai/_core/system_parser.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
+from pathlib import Path
 
 import toml
 
@@ -29,19 +29,19 @@ class SystemParser:
     Attributes
         _parsers (Dict[str, Type[BaseSystemParser]]): A mapping from system types to their corresponding parser
             classes.
-        file_path (str): The file path to the system configuration file.
+        file_path (Path): The file path to the system configuration file.
     """
 
     _parsers = {}
 
-    def __init__(self, file_path: str):
+    def __init__(self, file_path: Path):
         """
         Initialize a SystemParser instance.
 
         Args:
-            file_path (str): The file path to the system configuration file.
+            file_path (Path): The file path to the system configuration file.
         """
-        self.file_path: str = file_path
+        self.file_path: Path = file_path
 
     def parse(self) -> System:
         """
@@ -55,10 +55,10 @@ def parse(self) -> System:
         Returns
             System: The parsed system object.
         """
-        if not os.path.isfile(self.file_path):
+        if not self.file_path.is_file():
             raise FileNotFoundError(f"The file '{self.file_path}' does not exist.")
 
-        with open(self.file_path, "r") as file:
+        with self.file_path.open("r") as file:
             data = toml.load(file)
             scheduler = data.get("scheduler", "").lower()
             registry = Registry()

src/cloudai/installer/slurm_installer.py

@@ -17,6 +17,7 @@
 import contextlib
 import os
 import subprocess
+from pathlib import Path
 from typing import Dict, Iterable, cast
 
 import toml
@@ -35,9 +36,9 @@ class SlurmInstaller(BaseInstaller):
         CONFIG_FILE_NAME (str): The name of the configuration file.
         PREREQUISITES (List[str]): A list of required binaries for the installer.
         REQUIRED_SRUN_OPTIONS (List[str]): A list of required srun options to check.
-        install_path (str): Path where the benchmarks are to be installed. This is optional since uninstallation does
+        install_path (Path): Path where the benchmarks are to be installed. This is optional since uninstallation does
             not require it.
-        config_path (str): Path to the installation configuration file.
+        config_path (Path): Path to the installation configuration file.
     """
 
     CONFIG_FILE_NAME = ".cloudai.toml"
@@ -60,7 +61,7 @@ def __init__(self, system: System):
         super().__init__(system)
         slurm_system = cast(SlurmSystem, self.system)
         self.install_path = slurm_system.install_path
-        self.config_path = os.path.join(os.path.expanduser("~"), self.CONFIG_FILE_NAME)
+        self.config_path = Path.home() / self.CONFIG_FILE_NAME
 
     def _check_prerequisites(self) -> InstallStatusResult:
         """
@@ -108,16 +109,16 @@ def _check_srun_options(self) -> None:
 
     def _write_config(self) -> InstallStatusResult:
         """Write the installation configuration to a TOML file atomically."""
-        absolute_install_path = os.path.abspath(self.install_path)
-        config_data: Dict[str, str] = {"install_path": absolute_install_path}
+        absolute_install_path = self.install_path.resolve()
+        config_data: Dict[str, str] = {"install_path": str(absolute_install_path)}
 
         try:
-            with open(self.config_path, "w") as file:
+            with self.config_path.open("w") as file:
                 toml.dump(config_data, file)
             return InstallStatusResult(True)
         except Exception as e:
             with contextlib.suppress(OSError):
-                os.remove(self.config_path)
+                self.config_path.unlink()
             return InstallStatusResult(False, str(e))
 
     def _read_config(self) -> Dict[str, str]:
@@ -128,7 +129,7 @@ def _read_config(self) -> Dict[str, str]:
             Dict[str, str]: Configuration, including installation path.
         """
         try:
-            with open(self.config_path, "r") as file:
+            with self.config_path.open("r") as file:
                 return toml.load(file)
         except FileNotFoundError as e:
             raise FileNotFoundError(
@@ -138,8 +139,8 @@ def _read_config(self) -> Dict[str, str]:
 
     def _remove_config(self) -> None:
         """Remove the installation configuration file."""
-        if os.path.exists(self.config_path):
-            os.remove(self.config_path)
+        if self.config_path.exists():
+            self.config_path.unlink()
 
     def is_installed(self, test_templates: Iterable[TestTemplate]) -> InstallStatusResult:
         """
@@ -153,7 +154,7 @@ def is_installed(self, test_templates: Iterable[TestTemplate]) -> InstallStatusR
         Returns:
             InstallStatusResult: Result containing the installation status and error message if not installed.
         """
-        if not os.path.exists(self.config_path):
+        if not self.config_path.exists():
             return InstallStatusResult(
                 False,
                 f"Configuration file does not exist at {self.config_path}. "
@@ -189,11 +190,11 @@ def install(self, test_templates: Iterable[TestTemplate]) -> InstallStatusResult
             return prerequisites_result
 
         try:
-            os.makedirs(self.install_path, exist_ok=True)
+            self.install_path.mkdir(parents=True, exist_ok=True)
         except OSError as e:
             return InstallStatusResult(False, f"Failed to create installation directory at {self.install_path}: {e}")
 
-        if not os.access(self.install_path, os.W_OK):
+        if not self.install_path.is_dir() or not os.access(self.install_path, os.W_OK):
             return InstallStatusResult(False, f"The installation path {self.install_path} is not writable.")
 
         install_result = super().install(test_templates)

src/cloudai/schema/test_template/jax_toolbox/slurm_command_gen_strategy.py

@@ -173,21 +173,20 @@ def _parse_slurm_args(
             JaxToolboxSlurmInstallStrategy.DOCKER_IMAGE_FILENAME,
         ).docker_image_path
 
-        local_workspace_dir = os.path.abspath(cmd_args["output_path"])
-        # Use the dynamic key_prefix for accessing docker_workspace_dir
+        local_workspace_dir = Path(cmd_args["output_path"]).resolve()
         docker_workspace_dir = cmd_args[f"{key_prefix}.setup_flags.docker_workspace_dir"]
         container_mounts = f"{local_workspace_dir}:{docker_workspace_dir}"
 
         if "pgo_nsys_converter.profile_path" in cmd_args:
-            profile_path = cmd_args["pgo_nsys_converter.profile_path"]
+            profile_path = Path(cmd_args["pgo_nsys_converter.profile_path"]).resolve()
             container_mounts += f",{profile_path}:{profile_path}"
 
         base_args.update({"image_path": image_path, "container_mounts": container_mounts})
 
-        output_path = os.path.abspath(cmd_args["output_path"])
+        output_path = Path(cmd_args["output_path"]).resolve()
         output_suffix = "-%j.txt" if env_vars.get("UNIFIED_STDOUT_STDERR") == "1" else "-%j-%n-%t.txt"
-        base_args["output"] = os.path.join(output_path, f"output{output_suffix}")
-        base_args["error"] = os.path.join(output_path, f"error{output_suffix}")
+        base_args["output"] = str(output_path / f"output{output_suffix}")
+        base_args["error"] = str(output_path / f"error{output_suffix}")
 
         return base_args
 

src/cloudai/schema/test_template/jax_toolbox/slurm_install_strategy.py

@@ -14,7 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 
 from cloudai import InstallStatusResult
 from cloudai.systems.slurm.strategy import SlurmInstallStrategy
@@ -34,8 +33,8 @@ def is_installed(self) -> InstallStatusResult:
             return InstallStatusResult(success=True)
         else:
             if self.docker_image_cache_manager.cache_docker_images_locally:
-                expected_docker_image_path = os.path.join(
-                    self.docker_image_cache_manager.install_path, self.SUBDIR_PATH, self.DOCKER_IMAGE_FILENAME
+                expected_docker_image_path = (
+                    self.docker_image_cache_manager.install_path / self.SUBDIR_PATH / self.DOCKER_IMAGE_FILENAME
                 )
                 return InstallStatusResult(
                     success=False,

src/cloudai/schema/test_template/nccl_test/slurm_command_gen_strategy.py

@@ -14,7 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 from pathlib import Path
 from typing import Any, Dict, List
 
@@ -73,7 +72,7 @@ def _parse_slurm_args(
 
         container_mounts = ""
         if "NCCL_TOPO_FILE" in env_vars and "DOCKER_NCCL_TOPO_FILE" in env_vars:
-            nccl_graph_path = os.path.abspath(env_vars["NCCL_TOPO_FILE"])
+            nccl_graph_path = Path(env_vars["NCCL_TOPO_FILE"]).resolve()
             nccl_graph_file = env_vars["DOCKER_NCCL_TOPO_FILE"]
             container_mounts = f"{nccl_graph_path}:{nccl_graph_file}"
         elif "NCCL_TOPO_FILE" in env_vars:

src/cloudai/schema/test_template/nccl_test/slurm_install_strategy.py

@@ -14,7 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 
 from cloudai import InstallStatusResult
 from cloudai.systems.slurm.strategy import SlurmInstallStrategy
@@ -40,8 +39,8 @@ def is_installed(self) -> InstallStatusResult:
             return InstallStatusResult(success=True)
         else:
             if self.docker_image_cache_manager.cache_docker_images_locally:
-                expected_docker_image_path = os.path.join(
-                    self.docker_image_cache_manager.install_path, self.SUBDIR_PATH, self.DOCKER_IMAGE_FILENAME
+                expected_docker_image_path = (
+                    self.docker_image_cache_manager.install_path / self.SUBDIR_PATH / self.DOCKER_IMAGE_FILENAME
                 )
                 return InstallStatusResult(
                     success=False,
@@ -55,7 +54,7 @@ def is_installed(self) -> InstallStatusResult:
                 return InstallStatusResult(
                     success=False,
                     message=(
-                        f"Docker image for NCCL test is not accessible.\n" f"    - Error: {docker_image_result.message}"
+                        f"Docker image for NCCL test is not accessible.\n    - Error: {docker_image_result.message}"
                     ),
                 )
 

src/cloudai/schema/test_template/nemo_launcher/slurm_install_strategy.py

@@ -18,6 +18,7 @@
 import os
 import subprocess
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
 from typing import Any, Dict, List
 
 from cloudai import InstallStatusResult, System
@@ -89,15 +90,15 @@ def _validate_cmd_arg(self, cmd_args: Dict[str, Any], arg_name: str) -> str:
         return arg_value
 
     def is_installed(self) -> InstallStatusResult:
-        subdir_path = os.path.join(self.install_path, self.SUBDIR_PATH)
-        repo_path = os.path.join(subdir_path, self.REPOSITORY_NAME)
-        repo_installed = os.path.isdir(repo_path)
+        subdir_path = self.install_path / self.SUBDIR_PATH
+        repo_path = subdir_path / self.REPOSITORY_NAME
+        repo_installed = repo_path.is_dir()
 
         docker_image_installed = self.docker_image_cache_manager.check_docker_image_exists(
             self.docker_image_url, self.SUBDIR_PATH, self.DOCKER_IMAGE_FILENAME
         ).success
 
-        data_dir_path = self.default_cmd_args["data_dir"]
+        data_dir_path = Path(self.default_cmd_args["data_dir"])
         datasets_check_result = self._check_datasets_on_nodes(data_dir_path)
         if not datasets_check_result.success:
             return InstallStatusResult(
@@ -121,8 +122,8 @@ def is_installed(self) -> InstallStatusResult:
                     f"with commit hash {self.repository_commit_hash}"
                 )
             if not docker_image_installed:
-                docker_image_path = os.path.join(subdir_path, self.DOCKER_IMAGE_FILENAME)
-                missing_components.append(f"Docker image at {docker_image_path} " f"from URL {self.docker_image_url}")
+                docker_image_path = subdir_path / self.DOCKER_IMAGE_FILENAME
+                missing_components.append(f"Docker image at {docker_image_path} from URL {self.docker_image_url}")
             if not datasets_check_result.success:
                 missing_components.append(f"Datasets in {data_dir_path} on some nodes")
             return InstallStatusResult(
@@ -141,10 +142,10 @@ def install(self) -> InstallStatusResult:
         except PermissionError as e:
             return InstallStatusResult(success=False, message=str(e))
 
-        subdir_path = os.path.join(self.install_path, self.SUBDIR_PATH)
-        os.makedirs(subdir_path, exist_ok=True)
+        subdir_path = self.install_path / self.SUBDIR_PATH
+        subdir_path.mkdir(parents=True, exist_ok=True)
 
-        data_dir_path = self.default_cmd_args["data_dir"]
+        data_dir_path = Path(self.default_cmd_args["data_dir"])
         datasets_check_result = self._check_datasets_on_nodes(data_dir_path)
         if not datasets_check_result.success:
             return InstallStatusResult(
@@ -195,12 +196,12 @@ def _check_install_path_access(self):
             PermissionError: If the install path does not exist or if there is no permission to create directories and
                 files.
         """
-        if not os.path.exists(self.install_path):
+        if not self.install_path.exists():
             raise PermissionError(f"Install path {self.install_path} does not exist.")
-        if not os.access(self.install_path, os.W_OK):
+        if not self.install_path.is_dir() or not os.access(self.install_path, os.W_OK):
             raise PermissionError(f"No permission to write in install path {self.install_path}.")
 
-    def _check_datasets_on_nodes(self, data_dir_path: str) -> DatasetCheckResult:
+    def _check_datasets_on_nodes(self, data_dir_path: Path) -> DatasetCheckResult:
         """
         Verify the presence of specified dataset files and directories on all idle compute nodes.
 
@@ -210,7 +211,7 @@ def _check_datasets_on_nodes(self, data_dir_path: str) -> DatasetCheckResult:
         for systems with multiple nodes.
 
         Args:
-            data_dir_path (str): Path where dataset files and directories are stored.
+            data_dir_path (Path): Path where dataset files and directories are stored.
 
         Returns:
             DatasetCheckResult: Result object containing success status and nodes without datasets.
@@ -248,22 +249,20 @@ def _check_datasets_on_nodes(self, data_dir_path: str) -> DatasetCheckResult:
 
         return DatasetCheckResult(success=not nodes_without_datasets, nodes_without_datasets=nodes_without_datasets)
 
-    def _check_dataset_on_node(self, node: str, data_dir_path: str, dataset_items: List[str]) -> bool:
+    def _check_dataset_on_node(self, node: str, data_dir_path: Path, dataset_items: List[str]) -> bool:
         """
         Check if dataset files and directories exist on a single compute node.
 
         Args:
             node (str): The name of the compute node.
-            data_dir_path (str): Path to the data directory.
+            data_dir_path (Path): Path to the data directory.
             dataset_items (List[str]): List of dataset file and directory names to check.
 
         Returns:
             bool: True if all dataset files and directories exist on the node, False otherwise.
         """
         python_check_script = (
-            f"import os;print(all(os.path.isfile(os.path.join('{data_dir_path}', "
-            f"item)) or os.path.isdir(os.path.join('{data_dir_path}', item)) "
-            f"for item in {dataset_items}))"
+            f"import os;print(all(Path('{data_dir_path}') / item).exists() for item in {dataset_items})"
         )
         cmd = (
             f"srun --nodes=1 --nodelist={node} "
@@ -273,43 +272,43 @@ def _check_dataset_on_node(self, node: str, data_dir_path: str, dataset_items: L
         result = subprocess.run(cmd, shell=True, check=False, capture_output=True, text=True)
         return result.returncode == 0 and result.stdout.strip() == "True"
 
-    def _clone_repository(self, subdir_path: str) -> None:
+    def _clone_repository(self, subdir_path: Path) -> None:
         """
         Clones NeMo-Launcher repository into specified path if it does not already exist.
 
         Args:
-            subdir_path (str): Subdirectory path for installation.
+            subdir_path (Path): Subdirectory path for installation.
         """
-        repo_path = os.path.join(subdir_path, self.REPOSITORY_NAME)
+        repo_path = subdir_path / self.REPOSITORY_NAME
 
-        if os.path.exists(repo_path):
+        if repo_path.exists():
             logging.warning("Repository already exists at %s, clone skipped", repo_path)
         else:
             logging.debug("Cloning NeMo-Launcher repository into %s", repo_path)
-            clone_cmd = ["git", "clone", self.repository_url, repo_path]
+            clone_cmd = ["git", "clone", self.repository_url, str(repo_path)]
             result = subprocess.run(clone_cmd, capture_output=True, text=True)
             if result.returncode != 0:
                 raise RuntimeError(f"Failed to clone repository: {result.stderr}")
 
         logging.debug("Checking out specific commit %s in repository", self.repository_commit_hash)
         checkout_cmd = ["git", "checkout", self.repository_commit_hash]
-        result = subprocess.run(checkout_cmd, cwd=repo_path, capture_output=True, text=True)
+        result = subprocess.run(checkout_cmd, cwd=str(repo_path), capture_output=True, text=True)
         if result.returncode != 0:
             raise RuntimeError(f"Failed to checkout commit: {result.stderr}")
 
-    def _install_requirements(self, subdir_path: str) -> None:
+    def _install_requirements(self, subdir_path: Path) -> None:
         """
         Installs the required Python packages from the requirements.txt file in the cloned repository.
 
         Args:
-            subdir_path (str): Subdirectory path for installation.
+            subdir_path (Path): Subdirectory path for installation.
         """
-        repo_path = os.path.join(subdir_path, self.REPOSITORY_NAME)
-        requirements_file = os.path.join(repo_path, "requirements.txt")
+        repo_path = subdir_path / self.REPOSITORY_NAME
+        requirements_file = repo_path / "requirements.txt"
 
-        if os.path.isfile(requirements_file):
+        if requirements_file.is_file():
             logging.debug("Installing requirements from %s", requirements_file)
-            install_cmd = ["pip", "install", "-r", requirements_file]
+            install_cmd = ["pip", "install", "-r", str(requirements_file)]
             result = subprocess.run(install_cmd, capture_output=True, text=True)
             if result.returncode != 0:
                 raise RuntimeError(f"Failed to install requirements: {result.stderr}")