From 2bd976203b3b62896d8d94f4fde482e8f43056bb Mon Sep 17 00:00:00 2001 From: moe-ad Date: Fri, 6 Dec 2024 15:22:23 +0100 Subject: [PATCH] doc: modified two source files --- src/ansys/dpf/core/custom_operator.py | 59 ++++++++++++++------------- src/ansys/dpf/core/data_sources.py | 33 ++++++++------- 2 files changed, 49 insertions(+), 43 deletions(-) diff --git a/src/ansys/dpf/core/custom_operator.py b/src/ansys/dpf/core/custom_operator.py index f5e8941951..9ba5b9b49b 100644 --- a/src/ansys/dpf/core/custom_operator.py +++ b/src/ansys/dpf/core/custom_operator.py @@ -31,7 +31,7 @@ import abc import ctypes import os -import pathlib +from pathlib import Path import re import shutil import tempfile @@ -85,23 +85,23 @@ def update_virtual_environment_for_custom_operators( raise NotImplementedError( "Updating the dpf-site.zip of a DPF Server is only available when InProcess." ) - current_dpf_site_zip_path = os.path.join(server.ansys_path, "dpf", "python", "dpf-site.zip") + current_dpf_site_zip_path = Path(server.ansys_path) / "dpf" / "python" / "dpf-site.zip" # Get the path to where we store the original dpf-site.zip - original_dpf_site_zip_path = os.path.join( - server.ansys_path, "dpf", "python", "original", "dpf-site.zip" + original_dpf_site_zip_path = ( + Path(server.ansys_path) / "dpf" / "python" / "original" / "dpf-site.zip" ) # Restore the original dpf-site.zip if restore_original: - if os.path.exists(original_dpf_site_zip_path): + if original_dpf_site_zip_path.exists(): shutil.move(src=original_dpf_site_zip_path, dst=current_dpf_site_zip_path) - os.rmdir(os.path.dirname(original_dpf_site_zip_path)) + original_dpf_site_zip_path.parent.rmdir() else: warnings.warn("No original dpf-site.zip found. Current is most likely the original.") else: # Store original dpf-site.zip for this DPF Server if no original is stored - if not os.path.exists(os.path.dirname(original_dpf_site_zip_path)): - os.mkdir(os.path.dirname(original_dpf_site_zip_path)) - if not os.path.exists(original_dpf_site_zip_path): + if not original_dpf_site_zip_path.parent.exists(): + original_dpf_site_zip_path.parent.mkdir() + if not original_dpf_site_zip_path.exists(): shutil.move(src=current_dpf_site_zip_path, dst=original_dpf_site_zip_path) # Get the current paths to site_packages import site @@ -111,46 +111,47 @@ def update_virtual_environment_for_custom_operators( # Get the first one targeting an actual site-packages folder for path_to_site_packages in paths_to_current_site_packages: if path_to_site_packages[-13:] == "site-packages": - current_site_packages_path = pathlib.Path(path_to_site_packages) + current_site_packages_path = Path(path_to_site_packages) break if current_site_packages_path is None: warnings.warn("Could not find a currently loaded site-packages folder to update from.") return # If an ansys.dpf.core.path file exists, then the installation is editable - search_path = pathlib.Path(current_site_packages_path) + search_path = current_site_packages_path potential_editable = list(search_path.rglob("__editable__.ansys_dpf_core-*.pth")) if potential_editable: path_file = potential_editable[0] else: # Keep for older setuptools versions - path_file = os.path.join(current_site_packages_path, "ansys.dpf.core.pth") - if os.path.exists(path_file): + path_file = current_site_packages_path / "ansys.dpf.core.pth" + if path_file.exists(): # Treat editable installation of ansys-dpf-core - with open(path_file, "r") as f: + with path_file.open("r") as f: current_site_packages_path = f.readline().strip() with tempfile.TemporaryDirectory() as tmpdir: - os.mkdir(os.path.join(tmpdir, "ansys_dpf_core")) - ansys_dir = os.path.join(tmpdir, "ansys_dpf_core") - os.mkdir(os.path.join(ansys_dir, "ansys")) - os.mkdir(os.path.join(ansys_dir, "ansys", "dpf")) - os.mkdir(os.path.join(ansys_dir, "ansys", "grpc")) + tmpdir = Path(tmpdir) + ansys_dir = tmpdir / "ansys_dpf_core" + ansys_dir.mkdir() + ansys_dir.joinpath("ansys").mkdir() + ansys_dir.joinpath("ansys", "dpf").mkdir() + ansys_dir.joinpath("ansys", "grpc").mkdir() shutil.copytree( - src=os.path.join(current_site_packages_path, "ansys", "dpf", "core"), - dst=os.path.join(ansys_dir, "ansys", "dpf", "core"), + src=current_site_packages_path / "ansys" / "dpf" / "core", + dst=ansys_dir / "ansys" / "dpf" / "core", ignore=lambda directory, contents: ["__pycache__", "result_files"], ) shutil.copytree( - src=os.path.join(current_site_packages_path, "ansys", "dpf", "gate"), - dst=os.path.join(ansys_dir, "ansys", "dpf", "gate"), + src=current_site_packages_path / "ansys" / "dpf" / "gate", + dst=ansys_dir / "ansys" / "dpf" / "gate", ignore=lambda directory, contents: ["__pycache__"], ) shutil.copytree( - src=os.path.join(current_site_packages_path, "ansys", "grpc", "dpf"), - dst=os.path.join(ansys_dir, "ansys", "grpc", "dpf"), + src=current_site_packages_path / "ansys" / "grpc" / "dpf", + dst=ansys_dir / "ansys" / "grpc" / "dpf", ignore=lambda directory, contents: ["__pycache__"], ) # Find the .dist_info folder pattern = re.compile(r"^ansys_dpf_core\S*") - for p in pathlib.Path(current_site_packages_path).iterdir(): + for p in current_site_packages_path.iterdir(): if p.is_dir(): # print(p.stem) if re.search(pattern, p.stem): @@ -158,12 +159,12 @@ def update_virtual_environment_for_custom_operators( break shutil.copytree( src=dist_info_path, - dst=os.path.join(ansys_dir, dist_info_path.name), + dst=ansys_dir / dist_info_path.name, ) # Zip the files as dpf-site.zip - base_name = os.path.join(tmpdir, "ansys_dpf_core_zip") + base_name = tmpdir / "ansys_dpf_core_zip" base_dir = "." - root_dir = os.path.join(tmpdir, "ansys_dpf_core") # OK + root_dir = tmpdir / "ansys_dpf_core" # OK shutil.make_archive( base_name=base_name, root_dir=root_dir, base_dir=base_dir, format="zip" ) diff --git a/src/ansys/dpf/core/data_sources.py b/src/ansys/dpf/core/data_sources.py index d694f54b5a..0275f6f4f6 100644 --- a/src/ansys/dpf/core/data_sources.py +++ b/src/ansys/dpf/core/data_sources.py @@ -28,6 +28,7 @@ """ import os +from pathlib import Path import warnings import traceback from typing import Union @@ -142,7 +143,7 @@ def set_result_file_path(self, filepath, key=""): ['/tmp/file.rst'] """ - extension = os.path.splitext(filepath)[1] + extension = Path(filepath).suffix # Handle .res files from CFX if key == "" and extension == ".res": key = "cas" @@ -162,7 +163,7 @@ def set_result_file_path(self, filepath, key=""): def guess_result_key(filepath: str) -> str: """Guess result key for files without a file extension.""" result_keys = ["d3plot", "binout"] - base_name = os.path.basename(filepath) + base_name = Path(filepath).name # Handle files without extension for result_key in result_keys: if result_key in base_name: @@ -172,14 +173,13 @@ def guess_result_key(filepath: str) -> str: @staticmethod def guess_second_key(filepath: str) -> str: """For files with an h5 or cff extension, look for another extension.""" + + # These files usually end with .cas.h5 or .dat.h5 accepted = ["cas", "dat"] - without_ext = os.path.splitext(filepath)[0] - new_split = os.path.splitext(without_ext) + new_split = Path(filepath).suffixes new_key = "" - if len(new_split) > 1: - key = new_split[1][1:] - if key in accepted: - new_key = key + if new_split[0] in accepted: + new_key = new_split[0] return new_key def set_domain_result_file_path( @@ -241,9 +241,12 @@ def add_file_path(self, filepath, key="", is_domain: bool = False, domain_id=0): """ # The filename needs to be a fully qualified file name - if not os.path.dirname(filepath): + # if not os.path.dirname(filepath) + + filepath = Path(filepath) + if not filepath.parent.name: # append local path - filepath = os.path.join(os.getcwd(), os.path.basename(filepath)) + filepath = Path.cwd() / filepath.name if is_domain: if key == "": raise NotImplementedError("A key must be given when using is_domain=True.") @@ -280,9 +283,10 @@ def add_domain_file_path(self, filepath, key, domain_id): """ # The filename needs to be a fully qualified file name - if not os.path.dirname(filepath): + filepath = Path(filepath) + if not filepath.parent.name: # append local path - filepath = os.path.join(os.getcwd(), os.path.basename(filepath)) + filepath = Path.cwd() / filepath.name self._api.data_sources_add_domain_file_path_with_key_utf8( self, str(filepath), key, domain_id ) @@ -307,9 +311,10 @@ def add_file_path_for_specified_result(self, filepath, key="", result_key=""): The default is ``""``, in which case the key is found directly. """ # The filename needs to be a fully qualified file name - if not os.path.dirname(filepath): + filepath = Path(filepath) + if not filepath.parent.name: # append local path - filepath = os.path.join(os.getcwd(), os.path.basename(filepath)) + filepath = Path.cwd() / filepath.name self._api.data_sources_add_file_path_for_specified_result_utf8( self, str(filepath), key, result_key