diff --git a/src/ansys/optislang/core/examples/downloads.py b/src/ansys/optislang/core/examples/downloads.py index d0d309b15..ae07ad258 100644 --- a/src/ansys/optislang/core/examples/downloads.py +++ b/src/ansys/optislang/core/examples/downloads.py @@ -1,6 +1,8 @@ """Download files from repository.""" import os +from pathlib import Path +from typing import Tuple import ansys.optislang.core.examples as examples @@ -8,31 +10,43 @@ # EXAMPLE_REPO = "https://github.com/pyansys/pyoptislang/tree/main/examples/files" -def _download_file(scriptname: str) -> str: - """Check if file exists, otherwise download it. Return path to file then.""" +def _download_files(scriptname: str) -> Tuple[Path, ...]: + """Check if files exists, otherwise download them. Return path to files then.""" # check if file was downloaded for file_path in examples.example_files[scriptname]: - if not os.path.isfile(file_path): + if not file_path.is_file(): NotImplementedError("Automatic download not implemented.") return examples.example_files[scriptname] -def _download_script(scriptname: str) -> str: +def _download_script(scriptname: str) -> Path: """Check if file exists, otherwise download it. Return path to file then.""" # check if script was downloaded, download it if not local_path = examples.example_scripts[scriptname] - if not os.path.isfile(local_path): + if not local_path.is_file(): NotImplementedError("Automatic download not implemented.") return local_path -def get_files(scriptname: str) -> str: - """Insert name of the example script (without ``.py``) and get local_path.""" +def get_files(scriptname: str) -> Tuple[Path, Tuple[Path, ...]]: + """Get tuple of files necessary for running example. + + Parameters + ---------- + scriptname: str + Name of the example script (without ``.py``). + + Returns + ------- + Tuple[Path, Tuple[Path, ...]] + Tuple[0]: path to script + Tuple[1]: tuple of paths to files necessary for running script + """ if examples.example_files[scriptname] is not None: - file_path = _download_file(scriptname) + file_path = _download_files(scriptname) else: file_path = None diff --git a/src/ansys/optislang/core/examples/examples.py b/src/ansys/optislang/core/examples/examples.py index bc5380c27..0748b7a72 100644 --- a/src/ansys/optislang/core/examples/examples.py +++ b/src/ansys/optislang/core/examples/examples.py @@ -2,138 +2,133 @@ import inspect import os +from pathlib import Path -module_path = os.path.dirname(inspect.getfile(inspect.currentframe())) +module_path = Path(inspect.getfile(inspect.currentframe())).parent # dictionary of files, that must be available to run scripts example_files = { "ansys_workbench_portscan": None, "arsm_ten_bar_truss": ( - os.path.join(module_path, "00_run_script", "ten_bar_truss", "files", "ten_bar_truss.s"), - os.path.join(module_path, "00_run_script", "ten_bar_truss", "files", "ten_bar_truss.out"), + module_path / "00_run_script" / "ten_bar_truss" / "files" / "ten_bar_truss.s", + module_path / "00_run_script" / "ten_bar_truss" / "files" / "ten_bar_truss.out", ), "ansys_workbench_ten_bar_truss": ( - os.path.join( - module_path, "00_run_script", "ten_bar_truss", "files", "ten_bar_truss_apdl.wbpz" - ) + module_path / "00_run_script" / "ten_bar_truss" / "files" / "ten_bar_truss_apdl.wbpz", ), "ten_bar_modify_parameters": None, "ten_bar_truss_lc2": ( - os.path.join(module_path, "00_run_script", "ten_bar_truss", "files", "ten_bar_truss2.s"), - os.path.join(module_path, "00_run_script", "ten_bar_truss", "files", "ten_bar_truss2.out"), + module_path / "00_run_script" / "ten_bar_truss" / "files" / "ten_bar_truss2.s", + module_path / "00_run_script" / "ten_bar_truss" / "files" / "ten_bar_truss2.out", ), "oscillatorcalibration_system_ascii": ( - os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py" - ), - os.path.join(module_path, "00_run_script", "oscillator", "files", "oscillator.s"), - os.path.join(module_path, "00_run_script", "oscillator", "files", "oscillator.bat"), - os.path.join(module_path, "00_run_script", "oscillator", "files", "oscillator.sh"), - os.path.join(module_path, "00_run_script", "oscillator", "files", "oscillator_signal.txt"), - os.path.join( - module_path, "00_run_script", "oscillator", "files", "oscillator_reference.txt" - ), + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator.s", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator.bat", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator.sh", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator_signal.txt", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator_reference.txt", ), "oscillatorcalibration_system_python": ( - os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py" - ), - os.path.join( - module_path, "00_run_script", "oscillator", "files", "oscillator_reference.txt" - ), + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", + module_path / "00_run_script" / "oscillator" / "files" / "oscillator_reference.txt", ), "oscillator_optimization_ea": ( - os.path.join(module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py") + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", ), "oscillator_optimization_on_mop": ( - os.path.join(module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py") + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", ), "oscillator_robustness_arsm": ( - os.path.join(module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py") + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", ), "oscillator_sensitivity_mop": ( - os.path.join(module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py") + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", ), "oscillator_system_python": ( - os.path.join(module_path, "00_run_script", "oscillator", "scripts", "_create_oscillator.py") + module_path / "00_run_script" / "oscillator" / "scripts" / "_create_oscillator.py", ), "create_all_possible_nodes": None, "etk_abaqus": ( - os.path.join(module_path, "00_run_script", "etk_abaqus", "files", "oscillator.inp"), - os.path.join(module_path, "00_run_script", "etk_abaqus", "files", "oscillator.odb"), + module_path / "00_run_script" / "etk_abaqus" / "files" / "oscillator.inp", + module_path / "00_run_script" / "etk_abaqus" / "files" / "oscillator.odb", ), "python_help": None, "python_node": None, "optimizer_settings": None, "sensitivity_settings": None, - "simple_calculator": os.path.join(module_path, "00_run_script", "files", "calculator.opf"), + "simple_calculator": (module_path / "00_run_script" / "files" / "calculator.opf",), } # dictionary of scripts to be run example_scripts = { - "ansys_workbench_portscan": os.path.join( - module_path, "00_run_script", "scripts", "ansys_workbench_portscan.py" - ), - "arsm_ten_bar_truss": os.path.join( - module_path, "00_run_script", "ten_bar_truss", "scripts", "arsm_ten_bar_truss.py" - ), - "ansys_workbench_ten_bar_truss": os.path.join( - module_path, "00_run_script", "ten_bar_truss", "scripts", "ansys_workbench_ten_bar_truss.py" - ), - "ten_bar_modify_parameters": os.path.join( - module_path, "00_run_script", "ten_bar_truss", "scripts", "ten_bar_modify_parameters.py" - ), - "ten_bar_truss_lc2": os.path.join( - module_path, "00_run_script", "ten_bar_truss", "scripts", "ten_bar_truss_lc2.py" - ), - "oscillatorcalibration_system_ascii": os.path.join( - module_path, - "00_run_script", - "oscillator", - "scripts", - "oscillatorcalibration_system_ascii.py", - ), - "oscillatorcalibration_system_python": os.path.join( - module_path, - "00_run_script", - "oscillator", - "scripts", - "oscillatorcalibration_system_python.py", - ), - "oscillator_optimization_ea": os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "oscillator_optimization_ea.py" - ), - "oscillator_optimization_on_mop": os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "oscillator_optimization_on_mop.py" - ), - "oscillator_robustness_arsm": os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "oscillator_robustness_arsm.py" - ), - "oscillator_sensitivity_mop": os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "oscillator_sensitivity_mop.py" - ), - "oscillator_system_python": os.path.join( - module_path, "00_run_script", "oscillator", "scripts", "oscillator_system_python.py" - ), - "create_all_possible_nodes": os.path.join( - module_path, "00_run_script", "scripts", "create_all_possible_nodes.py" - ), - "etk_abaqus": os.path.join( - module_path, "00_run_script", "etk_abaqus", "scripts", "etk_abaqus.py" - ), - "python_help": os.path.join( - module_path, "00_run_script", "python", "scripts", "python_help.py" - ), - "python_node": os.path.join( - module_path, "00_run_script", "python", "scripts", "python_node.py" - ), - "optimizer_settings": os.path.join( - module_path, "00_run_script", "scripts", "optimizer_settings.py" - ), - "sensitivity_settings": os.path.join( - module_path, "00_run_script", "scripts", "sensitivity_settings.py" - ), - "simple_calculator": os.path.join( - module_path, "00_run_script", "scripts", "simple_calculator.py" - ), + "ansys_workbench_portscan": module_path + / "00_run_script" + / "scripts" + / "ansys_workbench_portscan.py", + "arsm_ten_bar_truss": module_path + / "00_run_script" + / "ten_bar_truss" + / "scripts" + / "arsm_ten_bar_truss.py", + "ansys_workbench_ten_bar_truss": module_path + / "00_run_script" + / "ten_bar_truss" + / "scripts" + / "ansys_workbench_ten_bar_truss.py", + "ten_bar_modify_parameters": module_path + / "00_run_script" + / "ten_bar_truss" + / "scripts" + / "ten_bar_modify_parameters.py", + "ten_bar_truss_lc2": module_path + / "00_run_script" + / "ten_bar_truss" + / "scripts" + / "ten_bar_truss_lc2.py", + "oscillatorcalibration_system_ascii": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillatorcalibration_system_ascii.py", + "oscillatorcalibration_system_python": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillatorcalibration_system_python.py", + "oscillator_optimization_ea": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillator_optimization_ea.py", + "oscillator_optimization_on_mop": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillator_optimization_on_mop.py", + "oscillator_robustness_arsm": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillator_robustness_arsm.py", + "oscillator_sensitivity_mop": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillator_sensitivity_mop.py", + "oscillator_system_python": module_path + / "00_run_script" + / "oscillator" + / "scripts" + / "oscillator_system_python.py", + "create_all_possible_nodes": module_path + / "00_run_script" + / "scripts" + / "create_all_possible_nodes.py", + "etk_abaqus": module_path / "00_run_script" / "etk_abaqus" / "scripts" / "etk_abaqus.py", + "python_help": module_path / "00_run_script" / "python" / "scripts" / "python_help.py", + "python_node": module_path / "00_run_script" / "python" / "scripts" / "python_node.py", + "optimizer_settings": module_path / "00_run_script" / "scripts" / "optimizer_settings.py", + "sensitivity_settings": module_path / "00_run_script" / "scripts" / "sensitivity_settings.py", + "simple_calculator": module_path / "00_run_script" / "scripts" / "simple_calculator.py", } diff --git a/src/ansys/optislang/core/optislang.py b/src/ansys/optislang/core/optislang.py index 4808ae1ce..377c3f474 100644 --- a/src/ansys/optislang/core/optislang.py +++ b/src/ansys/optislang/core/optislang.py @@ -1,4 +1,5 @@ """Contains Optislang class which provides python API for optiSLang application.""" +from pathlib import Path from typing import Sequence, Tuple, Union from importlib_metadata import version @@ -25,10 +26,10 @@ class Optislang: Defaults to ``None``. port : int, optional A numeric port number of running optiSLang server. Defaults to ``None``. - executable : str, optional + executable : Union[str, Path], optional Path to the optiSLang executable file which supposed to be executed on localhost. It is ignored when the host and port parameters are specified. Defaults to ``None``. - project_path : str, optional + project_path : Union[str, Path], optional Path to the optiSLang project file which is supposed to be used by new local optiSLang server. It is ignored when the host and port parameters are specified. - If the project file exists, it is opened. @@ -77,8 +78,8 @@ def __init__( self, host: str = None, port: int = None, - executable: str = None, - project_path: str = None, + executable: Union[str, Path] = None, + project_path: Union[str, Path] = None, no_save: bool = False, ini_timeout: Union[int, float] = 20, name: str = None, @@ -89,8 +90,8 @@ def __init__( """Initialize a new instance of the ``Optislang`` class.""" self.__host = host self.__port = port - self.__executable = executable - self.__project_path = project_path + self.__executable = Path(executable) if executable is not None else None + self.__project_path = Path(project_path) if project_path is not None else None self.__no_save = no_save self.__ini_timeout = ini_timeout self.__name = name @@ -216,12 +217,12 @@ def get_project_description(self) -> str: """ return self.__osl_server.get_project_description() - def get_project_location(self) -> str: + def get_project_location(self) -> Path: """Get path to the optiSLang project file. Returns ------- - str + Path Path to the optiSLang project file. If no project is loaded in the optiSLang, returns ``None``. @@ -299,12 +300,12 @@ def get_timeout(self) -> Union[float, None]: """ return self.__osl_server.get_timeout() - def get_working_dir(self) -> str: + def get_working_dir(self) -> Path: """Get path to the optiSLang project working directory. Returns ------- - str + Path Path to the optiSLang project working directory. If no project is loaded in the optiSLang, returns ``None``. @@ -365,14 +366,14 @@ def run_python_script( def run_python_file( self, - file_path: str, + file_path: Union[str, Path], args: Union[Sequence[object], None] = None, ) -> Tuple[str, str]: """Read python script from the file, load it in a project context and execute it. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the Python script file which content is supposed to be executed on the server. args : Sequence[object], None, optional Sequence of arguments used in Python script. Defaults to ``None``. @@ -395,12 +396,12 @@ def run_python_file( """ return self.__osl_server.run_python_file(file_path, args) - def save_copy(self, file_path: str) -> None: + def save_copy(self, file_path: Union[str, Path]) -> None: """Save the current project as a copy to a location. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where to save the project copy. Raises diff --git a/src/ansys/optislang/core/osl_process.py b/src/ansys/optislang/core/osl_process.py index e8e34e5ea..4740e0bae 100644 --- a/src/ansys/optislang/core/osl_process.py +++ b/src/ansys/optislang/core/osl_process.py @@ -2,6 +2,7 @@ from enum import Enum import logging import os +from pathlib import Path import subprocess import tempfile from threading import Thread @@ -43,9 +44,9 @@ class OslServerProcess: Parameters ---------- - executable : str + executable : Union[str, Path] Path to the optiSLang executable file. - project_path : str, optional + project_path : Union[str, Path], optional Path to the optiSLang project file. - If the project file exists, it is opened. - If the project file does not exist, a new project is created on the specified path. @@ -61,7 +62,7 @@ class OslServerProcess: no_save : bool, optional Determines whether not to save the specified project after all other actions are completed. Defaults to ``False``. - server_info : str, optional + server_info : Union[str, Path], optional Path to the server information file. If an absolute path is not supplied, it is considered to be relative to the project working directory. If ``None``, no server information file will be written. Defaults to ``None``. @@ -115,13 +116,13 @@ class OslServerProcess: def __init__( self, - executable: str = None, - project_path: str = None, + executable: Union[str, Path] = None, + project_path: Union[str, Path] = None, batch: bool = True, port_range: Tuple[int, int] = None, password: str = None, no_save: bool = False, - server_info: str = None, + server_info: Union[str, Path] = None, log_server_events: bool = False, listener: Tuple[str, int] = None, listener_id: str = None, @@ -147,7 +148,7 @@ def __init__( if executable is not None: if not os.path.isfile(executable): raise FileNotFoundError(f"optiSLang executable cannot be found: {executable}") - self.__executable = executable + self.__executable = Path(executable) else: osl_exec = utils.get_osl_exec() if osl_exec is not None: @@ -162,19 +163,35 @@ def __init__( if project_path == None: self.__tempdir = tempfile.TemporaryDirectory() - self.__project_path = os.path.join( - self.__tempdir.name, self.__class__.DEFAULT_PROJECT_FILE + project_path = Path(self.__tempdir.name) / self.__class__.DEFAULT_PROJECT_FILE + + if isinstance(project_path, str): + project_path = Path(project_path) + + if not isinstance(project_path, Path): + raise TypeError( + f"Invalid type of project_path: {type(project_path)}," + "Union[str, Path] is supported." ) - else: - if not project_path.endswith(".opf"): - raise ValueError("Invalid optiSLang project file.") - self.__project_path = project_path + + if not project_path.suffix == ".opf": + raise ValueError("Invalid optiSLang project file.") + + self.__project_path = project_path + + if isinstance(server_info, str): + server_info = Path(server_info) + elif not (isinstance(server_info, Path) or server_info is None): + raise TypeError( + f"Invalid type of server_info: {type(server_info)}," + "Union[str, Path] is supported." + ) + self.__server_info = server_info self.__batch = batch self.__port_range = port_range self.__password = password self.__no_save = no_save - self.__server_info = server_info self.__log_server_events = log_server_events self.__listener = listener self.__listener_id = listener_id @@ -186,23 +203,23 @@ def __init__( self.__additional_args = kwargs @property - def executable(self) -> str: + def executable(self) -> Path: """Path to the optiSLang executable file. Returns ------- - str + Path Path to the optiSLang executable file. """ return self.__executable @property - def project_path(self) -> str: + def project_path(self) -> Path: """Path to the optiSLang project file. Returns ------- - str + Path Path to the optiSLang project file. """ return self.__project_path @@ -254,12 +271,12 @@ def no_save(self) -> bool: return self.__no_save @property - def server_info(self) -> str: + def server_info(self) -> Union[Path, None]: """Path to the server information file. Returns ------- - str + Union[Path, None] Path to the server information file, if defined; ``None`` otherwise. """ return self.__server_info @@ -410,23 +427,23 @@ def _get_process_args(self) -> List[str]: List of command line arguments. """ args = [] - args.append(self.__executable) + args.append(str(self.__executable)) if self.__batch: args.append("-b") # Start batch mode - if not os.path.isfile(self.__project_path): + if not self.__project_path.is_file(): # Creates a new project in the provided path. if IRON_PYTHON: - args.append(f'--new="{self.__project_path}"') + args.append(f'--new="{str(self.__project_path)}"') else: - args.append(f"--new={self.__project_path}") + args.append(f"--new={str(self.__project_path)}") else: # Opens existing project if IRON_PYTHON: - args.append(f'"{self.__project_path}"') + args.append(f'"{str(self.__project_path)}"') else: - args.append(self.__project_path) + args.append(str(self.__project_path)) if self.__batch: args.append("--no-run") # Does not run the specified projects. @@ -449,9 +466,9 @@ def _get_process_args(self) -> List[str]: # Writes the server information file using the file path specified. If an absolute path # is not supplied, it is considered to be relative to the project working directory. if IRON_PYTHON: - args.append(f'--write-server-info="{self.__server_info}"') + args.append(f'--write-server-info="{str(self.__server_info)}"') else: - args.append(f"--write-server-info={self.__server_info}") + args.append(f"--write-server-info={str(self.__server_info)}") if self.__no_save: # Do not save the specified projects after all other actions have been completed. diff --git a/src/ansys/optislang/core/osl_server.py b/src/ansys/optislang/core/osl_server.py index 4d0aa5842..45dae5213 100644 --- a/src/ansys/optislang/core/osl_server.py +++ b/src/ansys/optislang/core/osl_server.py @@ -1,6 +1,7 @@ """Contains abstract optiSLang server class.""" from abc import ABC, abstractmethod +from pathlib import Path from typing import Sequence, Tuple, Union @@ -85,12 +86,12 @@ def get_project_description(self) -> str: pass @abstractmethod - def get_project_location(self) -> str: + def get_project_location(self) -> Path: """Get path to the optiSLang project file. Returns ------- - str + Path Path to the optiSLang project file. If no project is loaded in the optiSLang, returns ``None``. @@ -172,12 +173,12 @@ def get_timeout(self) -> Union[float, None]: pass @abstractmethod - def get_working_dir(self) -> str: + def get_working_dir(self) -> Path: """Get path to the optiSLang project working directory. Returns ------- - str + Path Path to the optiSLang project working directory. If no project is loaded in the optiSLang, returns ``None``. @@ -210,7 +211,7 @@ def new(self) -> None: @abstractmethod def open( self, - file_path: str, + file_path: Union[str, Path], force: bool, restore: bool, reset: bool, @@ -219,7 +220,7 @@ def open( Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the optiSLang project file to open. force : bool # TODO: description of this parameter is missing in ANSYS help @@ -288,14 +289,14 @@ def run_python_script( @abstractmethod def run_python_file( self, - file_path: str, + file_path: Union[str, Path], args: Union[Sequence[object], None] = None, ) -> Tuple[str, str]: """Read python script from the file, load it in a project context and execute it. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the Python script file which content is supposed to be executed on the server. args : Sequence[object], None, optional Sequence of arguments used in Python script. Defaults to ``None``. @@ -336,7 +337,7 @@ def save(self) -> None: @abstractmethod def save_as( self, - file_path: str, + file_path: Union[str, Path], force: bool, restore: bool, reset: bool, @@ -345,7 +346,7 @@ def save_as( Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where to save the project file. force : bool # TODO: description of this parameter is missing in ANSYS help @@ -366,12 +367,12 @@ def save_as( pass @abstractmethod - def save_copy(self, file_path: str) -> None: + def save_copy(self, file_path: Union[str, Path]) -> None: """Save the current project as a copy to a location. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where to save the project copy. Raises diff --git a/src/ansys/optislang/core/tcp_osl_server.py b/src/ansys/optislang/core/tcp_osl_server.py index 33d56a5a2..64fab361d 100644 --- a/src/ansys/optislang/core/tcp_osl_server.py +++ b/src/ansys/optislang/core/tcp_osl_server.py @@ -3,6 +3,7 @@ import json import logging import os +from pathlib import Path from queue import Queue import re import select @@ -227,12 +228,12 @@ def send_msg(self, msg: str, timeout: Union[float, None] = 5) -> None: self.__socket.settimeout(timeout) self.__socket.sendall(header + data) - def send_file(self, file_path: str, timeout: Union[float, None] = 5) -> None: + def send_file(self, file_path: Union[str, Path], timeout: Union[float, None] = 5) -> None: """Send content of the file to the server. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the file whose content is to be sent to the server. timeout : Union[float, None], optional Timeout in seconds to send the buffer of the read part of the file. If a non-zero value @@ -321,12 +322,12 @@ def receive_msg(self, timeout: Union[float, None] = 5) -> str: return force_text(data) - def receive_file(self, file_path: str, timeout: Union[float, None] = 5) -> None: + def receive_file(self, file_path: Union[str, Path], timeout: Union[float, None] = 5) -> None: """Receive file from the server. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where the received file is to be saved. timeout : Union[float, None], optional Timeout in seconds to receive a buffer of the file part. The function will raise @@ -472,14 +473,16 @@ def _receive_bytes(self, count: int, timeout: Union[float, None]) -> bytes: received_len += len(chunk) return received - def _fetch_file(self, file_len: int, file_path: str, timeout: Union[float, None]) -> None: + def _fetch_file( + self, file_len: int, file_path: Union[str, Path], timeout: Union[float, None] + ) -> None: """Write received bytes from the server to the file. Parameters ---------- file_len : int Number of bytes to be written. - file_path : str + file_path : Union[str, Path] Path to the file to which the received data is to be written. timeout : Union[float, None], optional Timeout in seconds to receive bytes from the server and write them to the file. @@ -812,10 +815,10 @@ class TcpOslServer(OslServer): Defaults to ``None``. port : int, optional A numeric port number of running optiSLang server. Defaults to ``None``. - executable : str, optional + executable : Union[str, Path], optional Path to the optiSLang executable file which supposed to be executed on localhost. It is ignored when the host and port parameters are specified. Defaults to ``None``. - project_path : str, optional + project_path : Union[str, Path], optional Path to the optiSLang project file which is supposed to be used by new local optiSLang server. It is ignored when the host and port parameters are specified. - If the project file exists, it is opened. @@ -872,8 +875,8 @@ def __init__( self, host: str = None, port: int = None, - executable: str = None, - project_path: str = None, + executable: Union[str, Path] = None, + project_path: Union[str, Path] = None, no_save: bool = False, ini_timeout: float = 20, password: str = None, @@ -884,14 +887,14 @@ def __init__( self.__host = host self.__port = port self.__timeout = None - self.__executable = executable if logger is None: self._logger = logging.getLogger(__name__) else: self._logger = logger - self.__project_path = project_path + self.__executable = Path(executable) if executable is not None else None + self.__project_path = Path(project_path) if project_path is not None else None self.__no_save = no_save self.__password = password self.__osl_process = None @@ -1070,12 +1073,12 @@ def get_project_description(self) -> str: return None return project_info["projects"][0]["settings"]["short_description"] - def get_project_location(self) -> str: + def get_project_location(self) -> Path: """Get path to the optiSLang project file. Returns ------- - str + Path Path to the optiSLang project file. If no project is loaded in the optiSLang, returns ``None``. @@ -1091,7 +1094,7 @@ def get_project_location(self) -> str: project_info = self._get_basic_project_info() if len(project_info["projects"]) == 0: return None - return project_info["projects"][0]["location"] + return Path(project_info["projects"][0]["location"]) def get_project_name(self) -> str: """Get name of the optiSLang project. @@ -1158,12 +1161,12 @@ def get_timeout(self) -> Union[float, None]: """ return self.__timeout - def get_working_dir(self) -> str: + def get_working_dir(self) -> Path: """Get path to the optiSLang project working directory. Returns ------- - str + Path Path to the optiSLang project working directory. If no project is loaded in the optiSLang, returns ``None``. @@ -1179,7 +1182,7 @@ def get_working_dir(self) -> str: project_info = self._get_basic_project_info() if len(project_info["projects"]) == 0: return None - return project_info["projects"][0]["working_dir"] + return Path(project_info["projects"][0]["working_dir"]) def new(self) -> None: """Create a new project. @@ -1202,7 +1205,7 @@ def new(self) -> None: def open( self, - file_path: str, + file_path: Union[str, Path], force: bool, restore: bool, reset: bool, @@ -1211,7 +1214,7 @@ def open( Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the optiSLang project file to open. force : bool # TODO: description of this parameter is missing in ANSYS help @@ -1286,14 +1289,14 @@ def run_python_script( def run_python_file( self, - file_path: str, + file_path: Union[str, Path], args: Union[Sequence[object], None] = None, ) -> Tuple[str, str]: """Read python script from the file, load it in a project context and execute it. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path to the Python script file which content is supposed to be executed on the server. args : Sequence[object], None, optional Sequence of arguments used in Python script. Defaults to ``None``. @@ -1334,7 +1337,7 @@ def save(self) -> None: def save_as( self, - file_path: str, + file_path: Union[str, Path], force: bool, restore: bool, reset: bool, @@ -1343,7 +1346,7 @@ def save_as( Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where to save the project file. force : bool # TODO: description of this parameter is missing in ANSYS help @@ -1359,12 +1362,12 @@ def save_as( """ raise NotImplementedError("Currently, command is not supported in batch mode.") - def save_copy(self, file_path: str) -> None: + def save_copy(self, file_path: Union[str, Path]) -> None: """Save the current project as a copy to a location. Parameters ---------- - file_path : str + file_path : Union[str, Path] Path where to save the project copy. Raises @@ -1376,7 +1379,7 @@ def save_copy(self, file_path: str) -> None: TimeoutError Raised when the timeout float value expires. """ - self._send_command(commands.save_copy(file_path, self.__password)) + self._send_command(commands.save_copy(str(file_path), self.__password)) def set_timeout(self, timeout: Union[float, None] = None) -> None: """Set timeout value for execution of commands. diff --git a/src/ansys/optislang/core/utils.py b/src/ansys/optislang/core/utils.py index 442b6746a..3fb1dc815 100644 --- a/src/ansys/optislang/core/utils.py +++ b/src/ansys/optislang/core/utils.py @@ -1,14 +1,14 @@ """Utitilies module.""" import collections -from glob import glob import os +from pathlib import Path import re from typing import Dict, Iterable, OrderedDict, Tuple, Union from ansys.optislang.core import FIRST_SUPPORTED_VERSION -def get_osl_exec(osl_version: Union[int, str, None] = None) -> Union[Tuple[int, str], None]: +def get_osl_exec(osl_version: Union[int, str, None] = None) -> Union[Tuple[int, Path], None]: """Return path to the optiSLang executable file. Parameters @@ -19,7 +19,7 @@ def get_osl_exec(osl_version: Union[int, str, None] = None) -> Union[Tuple[int, Returns ------- - Tuple[int, str], None + Tuple[int, Path], None optiSLang version and path to the corresponding executable, if exists. If both ANSYS and standalone installations are present, ANSYS installation is returned. If no executable is found for specified version, returns ``None``. @@ -42,12 +42,12 @@ def get_osl_exec(osl_version: Union[int, str, None] = None) -> Union[Tuple[int, return (osl_version, osl_execs[osl_version][0]) -def find_all_osl_exec() -> OrderedDict[int, Tuple[str, ...]]: +def find_all_osl_exec() -> OrderedDict[int, Tuple[Path, ...]]: """Find available optiSLang executables. Returns ------- - OrderedDict[int, Tuple[str, ...]] + OrderedDict[int, Tuple[Path, ...]] Ordered dictionary of found optiSLang executables. The dictionary key corresponds to the optiSLang version and value to the tuple of optiSLang executable paths (ansys installation is first in the tuple, standalone installations then). The dictionary is @@ -69,12 +69,12 @@ def find_all_osl_exec() -> OrderedDict[int, Tuple[str, ...]]: raise NotImplementedError(f"Unsupported OS {os.name}.") -def _find_all_osl_exec_in_windows() -> OrderedDict[int, Tuple[str, ...]]: +def _find_all_osl_exec_in_windows() -> OrderedDict[int, Tuple[Path, ...]]: """Find all optiSLang executables on Windows operating system. Returns ------- - OrderedDict[int, Tuple[str, ...]] + OrderedDict[int, Tuple[Path, ...]] Ordered dictionary of found optiSLang executables. The dictionary key corresponds to the optiSLang version and value to the tuple of optiSLang executable paths. The dictionary is sorted by the version number in descending order. @@ -90,12 +90,12 @@ def _find_all_osl_exec_in_windows() -> OrderedDict[int, Tuple[str, ...]]: return _sort_osl_execs(all_osl_execs) -def _find_all_osl_exec_in_posix() -> OrderedDict[int, Tuple[str, ...]]: +def _find_all_osl_exec_in_posix() -> OrderedDict[int, Tuple[Path, ...]]: """Find all optiSLang executables on POSIX compliant operating systems. Returns ------- - OrderedDict[int, Tuple[str, ...]] + OrderedDict[int, Tuple[Path, ...]] Ordered dictionary of found optiSLang executables. The dictionary key corresponds to the optiSLang version and value to the tuple of optiSLang executable paths. The dictionary is sorted by the version number in descending order. @@ -107,7 +107,7 @@ def _find_all_osl_exec_in_posix() -> OrderedDict[int, Tuple[str, ...]]: return _sort_osl_execs(all_osl_execs) -def _find_ansys_osl_execs_in_windows_envars() -> Dict[int, str]: +def _find_ansys_osl_execs_in_windows_envars() -> Dict[int, Path]: """Find optiSLang executables based on environmental variables on Windows operating system. Ansys AWP_ROOT environment variable is used to determine the root directory of the ANSYS @@ -115,61 +115,61 @@ def _find_ansys_osl_execs_in_windows_envars() -> Dict[int, str]: Returns ------- - Dict[int, str] + Dict[int, Path] Dictionary of found optiSLang executables. Dictionary key is an optiSLang version and dictionary value is a path to the corresponding optiSLang executable. """ osl_execs = {} awp_root_envars = _get_environ_vars(pattern=f"^AWP_ROOT.*") for awp_root_key, awp_root_value in awp_root_envars.items(): - osl_exec_path = os.path.join(awp_root_value, "optiSLang", "optislang.com") - if os.path.isfile(osl_exec_path): + osl_exec_path = Path(awp_root_value) / "optiSLang" / "optislang.com" + if osl_exec_path.is_file(): ansys_version = _try_cast_str_to_int(awp_root_key[-3:]) if ansys_version >= FIRST_SUPPORTED_VERSION: osl_execs[ansys_version] = osl_exec_path return osl_execs -def _find_ansys_osl_execs_in_windows_program_files() -> Dict[int, str]: +def _find_ansys_osl_execs_in_windows_program_files() -> Dict[int, Path]: """Find optiSLang executables in Program Files directory on Windows operating system. Search is performed in standard installation directory of ANSYS products. Returns ------- - Dict[int, str] + Dict[int, Path] Dictionary of found optiSLang executables. Dictionary key is an optiSLang version and dictionary value is a path to the corresponding optiSLang executable. """ osl_execs = {} program_files_path = _get_program_files_path() - for ansys_version_dir in glob(os.path.join(program_files_path, "ANSYS Inc", "v*")): - ansys_version = _try_cast_str_to_int(ansys_version_dir[-3:]) + for ansys_version_dir in (program_files_path / "ANSYS Inc").glob("v*/"): + ansys_version = _try_cast_str_to_int(ansys_version_dir.name[1:]) if ansys_version >= FIRST_SUPPORTED_VERSION: - osl_exec_path = os.path.join(ansys_version_dir, "optiSLang", "optislang.com") - if os.path.isfile(osl_exec_path): + osl_exec_path = ansys_version_dir / "optiSLang" / "optislang.com" + if osl_exec_path.is_file(): osl_execs[ansys_version] = osl_exec_path return osl_execs -def _find_standalone_osl_execs_in_windows() -> Dict[int, str]: +def _find_standalone_osl_execs_in_windows() -> Dict[int, Path]: """Find executables of standalone optiSLang installations on Windows operating system. Returns ------- - Dict[int, str] + Dict[int, Path] Dictionary of found optiSLang executables. Dictionary key is an optiSLang version and dictionary value is a path to the corresponding optiSLang executable. """ osl_execs = {} - root_install_dir = os.path.join(_get_program_files_path(), "Dynardo", "Ansys optiSLang") - if os.path.isdir(root_install_dir): - for osl_version_dir in glob(os.path.join(root_install_dir, "*")): - osl_exec_path = os.path.join(root_install_dir, osl_version_dir, "optislang.com") + root_install_dir = _get_program_files_path() / "Dynardo" / "Ansys optiSLang" + if root_install_dir.is_dir(): + for osl_version_dir in root_install_dir.glob("*/"): + osl_exec_path = osl_version_dir / "optislang.com" if os.path.isfile(osl_exec_path): # convert version fmt "yyyy Rx" to "yyx" # TODO: add '$' at the end of regex pattern to disable dev version - match = re.findall("[2][0][1-9][0-9] R[1-9]", osl_version_dir) + match = re.findall("[2][0][1-9][0-9] R[1-9]", str(osl_version_dir)) if match: ansys_version = _try_cast_str_to_int(match[0][2:4] + match[0][6]) if ansys_version >= FIRST_SUPPORTED_VERSION: @@ -177,50 +177,50 @@ def _find_standalone_osl_execs_in_windows() -> Dict[int, str]: return osl_execs -def _find_ansys_osl_execs_in_posix() -> Dict[int, str]: +def _find_ansys_osl_execs_in_posix() -> Dict[int, Path]: """Find optiSLang executables in default ANSYS paths on POSIX compliant operating systems. Search is performed in standard installation paths of ANSYS products. Returns ------- - Dict[int, str] + Dict[int, Path] Dictionary of found optiSLang executables. Dictionary key is an optiSLang version and dictionary value is a path to the corresponding optiSLang executable. """ osl_execs = {} base_path = None - for ansys_dir in ["/usr/ansys_inc", "/ansys_inc"]: - if os.path.isdir(ansys_dir): + for ansys_dir in [Path("/usr/ansys_inc"), Path("/ansys_inc")]: + if ansys_dir.is_dir(): base_path = ansys_dir if base_path is not None: - for ansys_version_dir in glob(os.path.join(base_path, "v*")): - ansys_version = _try_cast_str_to_int(ansys_version_dir[-3:]) + for ansys_version_dir in base_path.glob("v*/"): + ansys_version = _try_cast_str_to_int(ansys_version_dir.name[1:]) if ansys_version >= FIRST_SUPPORTED_VERSION: - osl_exec_path = os.path.join(ansys_version_dir, "optiSLang", "optislang") - if os.path.isfile(osl_exec_path): + osl_exec_path = ansys_version_dir / "optiSLang" / "optislang" + if osl_exec_path.is_file(): osl_execs[ansys_version] = osl_exec_path return osl_execs -def _find_standalone_osl_execs_in_posix() -> Dict[int, str]: +def _find_standalone_osl_execs_in_posix() -> Dict[int, Path]: """Find executables of standalone optiSLang installations on POSIX compliant operating systems. Returns ------- - Dict[int, str] + Dict[int, Path] Dictionary of found optiSLang executables. Dictionary key is an optiSLang version and dictionary value is a path to the corresponding optiSLang executable. """ osl_execs = {} - root_install_dir = "/opt/dynardo" - if os.path.isdir(root_install_dir): - for osl_version_dir in glob(os.path.join(root_install_dir, "*")): - osl_exec_path = os.path.join(root_install_dir, osl_version_dir, "optislang") - if os.path.isfile(osl_exec_path): + root_install_dir = Path("/opt/dynardo") + if root_install_dir.is_dir(): + for osl_version_dir in root_install_dir.glob("*/"): + osl_exec_path = osl_version_dir / "optislang" + if osl_exec_path.is_file(): # convert version fmt "yyyy Rx" to "yyx" # TODO: add '$' at the end of regex pattern to disable dev version - match = re.findall("[2][0][1-9][0-9]R[1-9]", osl_version_dir) + match = re.findall("[2][0][1-9][0-9]R[1-9]", str(osl_version_dir)) if match: ansys_version = _try_cast_str_to_int(match[0][2:4] + match[0][5]) if ansys_version >= FIRST_SUPPORTED_VERSION: @@ -228,18 +228,20 @@ def _find_standalone_osl_execs_in_posix() -> Dict[int, str]: return osl_execs -def _merge_osl_exec_dicts(osl_execs_dicts: Iterable[Dict[int, str]]) -> Dict[int, Tuple[str, ...]]: +def _merge_osl_exec_dicts( + osl_execs_dicts: Iterable[Dict[int, Path]] +) -> Dict[int, Tuple[Path, ...]]: """Merge dictionaries of optiSLang executables into one dictionary. Parameters ---------- - osl_execs_dicts : Iterable[Dict[int, str]] + osl_execs_dicts : Iterable[Dict[int, Path]] Iterable of dictionaries of optiSLang executables. In each dictionary, key is an optiSLang version and value is a corresponding path to the optiSLang executable. Returns ------- - Dict[int, Tuple[str, ...]] + Dict[int, Tuple[Path, ...]] Merged dictionary in which the key is an optiSLang version and value is a tuple of paths to the corresponding optiSLang executables. """ @@ -277,7 +279,7 @@ def _sort_osl_execs(osl_execs: Dict[int, Tuple[str, ...]]) -> OrderedDict[int, T return collections.OrderedDict(sorted(osl_execs.items(), reverse=True)) -def _get_program_files_path() -> str: +def _get_program_files_path() -> Path: """Get the "Program Files" directory path on Windows operating system. Returns @@ -285,7 +287,7 @@ def _get_program_files_path() -> str: str Path to the "Program Files" directory. """ - return os.environ.get("ProgramFiles", "C:\\Program Files") + return Path(os.environ.get("ProgramFiles", "C:\\Program Files")) def _get_environ_vars(pattern: str = ".*") -> Dict: diff --git a/tests/test_find_executable.py b/tests/test_find_executable.py index b8feb59fc..6a354c17c 100644 --- a/tests/test_find_executable.py +++ b/tests/test_find_executable.py @@ -1,4 +1,5 @@ import os +from pathlib import Path from typing import OrderedDict import pytest @@ -25,7 +26,7 @@ def test_get_osl_executable(): auto_exe_path = utils.get_osl_exec() assert isinstance(auto_exe_path, tuple) assert isinstance(auto_exe_path[0], int) - assert isinstance(auto_exe_path[1], str) + assert isinstance(auto_exe_path[1], Path) exe_path = utils.get_osl_exec(osl_version="999") assert exe_path is None @@ -36,5 +37,5 @@ def test_find_all_osl_exec(): version = next(iter(dictionary.keys())) assert isinstance(dictionary, OrderedDict) assert isinstance(version, int) - assert isinstance(dictionary[version][0], str) - assert isinstance(dictionary[version], (str, tuple)) + assert isinstance(dictionary[version][0], Path) + assert isinstance(dictionary[version], (Path, tuple)) diff --git a/tests/test_optislang.py b/tests/test_optislang.py index ab87a82ca..f2b32638b 100644 --- a/tests/test_optislang.py +++ b/tests/test_optislang.py @@ -1,5 +1,6 @@ from contextlib import nullcontext as does_not_raise import os +from pathlib import Path import pytest @@ -17,7 +18,9 @@ def optislang(scope="function", autouse=True) -> Optislang: Optislang: Connects to the optiSLang application and provides an API to control it. """ - return Optislang() + osl = Optislang() + osl.set_timeout(20) + return osl def test_get_osl_version_string(optislang): @@ -53,7 +56,7 @@ def test_get_project_description(optislang): def test_get_project_location(optislang): "Test ``get_project_location``." location = optislang.get_project_location() - assert isinstance(location, str) + assert isinstance(location, Path) with does_not_raise() as dnr: optislang.shutdown() assert dnr is None @@ -80,7 +83,7 @@ def test_get_project_status(optislang): def test_get_working_dir(optislang): "Test ``get_working_dir``." working_dir = optislang.get_working_dir() - assert isinstance(working_dir, str) + assert isinstance(working_dir, Path) with does_not_raise() as dnr: optislang.shutdown() assert dnr is None @@ -172,8 +175,6 @@ def test_stop(optislang): ) optislang.start(wait_for_finished=False) optislang.stop() - optislang.start() - optislang.stop() assert dnr is None with does_not_raise() as dnr: optislang.shutdown() @@ -199,8 +200,6 @@ def test_stop_gently(optislang): ) optislang.start(wait_for_finished=False) optislang.stop_gently() - optislang.start() - optislang.stop() assert dnr is None with does_not_raise() as dnr: optislang.shutdown() diff --git a/tests/test_tcp_osl_server.py b/tests/test_tcp_osl_server.py index 5c8f30f0f..2ef6235e0 100644 --- a/tests/test_tcp_osl_server.py +++ b/tests/test_tcp_osl_server.py @@ -1,5 +1,6 @@ from contextlib import nullcontext as does_not_raise import os +from pathlib import Path import socket import time @@ -36,8 +37,7 @@ def tcp_client() -> tos.TcpClient: TcpOslServer: Class which provides access to optiSLang server using plain TCP/IP communication protocol. """ - client = tos.TcpClient() - return client + return tos.TcpClient() @pytest.fixture(scope="function", autouse=False) @@ -81,9 +81,15 @@ def test_send_msg(osl_server_process, tcp_client): assert dnr is None -def test_send_file(osl_server_process, tcp_client, tmp_path): - "Test ``send_file`" - file_path = os.path.join(tmp_path, "testfile.txt") +@pytest.mark.parametrize("path_type", [str, Path]) +def test_send_file(osl_server_process, tcp_client, tmp_path, path_type): + "Test ``send_file``" + file_path = tmp_path / "testfile.txt" + if path_type == str: + file_path = str(file_path) + elif path_type != Path: + assert False + with open(file_path, "w") as testfile: testfile.write(_msg) with does_not_raise() as dnr: @@ -104,22 +110,31 @@ def test_receive_msg(osl_server_process, tcp_client): assert isinstance(msg, str) -def test_receive_file(osl_server_process, tcp_client, tmp_path): +@pytest.mark.parametrize("path_type", [str, Path]) +def test_receive_file(osl_server_process, tcp_client, tmp_path, path_type): "Test ``receive_file`" - file_path = os.path.join(tmp_path, "testfile.txt") + file_path = tmp_path / "testfile.txt" + received_path = tmp_path / "received.txt" + if path_type == str: + file_path = str(file_path) + received_path = str(received_path) + elif path_type != Path: + assert False + with open(file_path, "w") as testfile: testfile.write(_msg) tcp_client.connect(host=_host, port=_port) tcp_client.send_file(file_path) with does_not_raise() as dnr: - tcp_client.receive_file(os.path.join(tmp_path, "received.txt")) + tcp_client.receive_file(received_path) + assert os.path.isfile(received_path) tcp_client.disconnect() osl_server_process.terminate() assert dnr is None # TcpOslServer -def test_get_server_info(osl_server_process, tcp_osl_server): +def test_get_server_info(tcp_osl_server): """Test ``_get_server_info``.""" server_info = tcp_osl_server._get_server_info() tcp_osl_server.shutdown() @@ -165,7 +180,7 @@ def test_get_project_location(osl_server_process, tcp_osl_server): """Test ``get_project_location``.""" project_location = tcp_osl_server.get_project_location() tcp_osl_server.shutdown() - assert isinstance(project_location, str) + assert isinstance(project_location, Path) assert bool(project_location) @@ -189,7 +204,7 @@ def test_get_working_dir(osl_server_process, tcp_osl_server): """Test ``get_working_dir``.""" working_dir = tcp_osl_server.get_working_dir() tcp_osl_server.shutdown() - assert isinstance(working_dir, str) + assert isinstance(working_dir, Path) assert bool(working_dir) @@ -217,7 +232,8 @@ def test_reset(osl_server_process, tcp_osl_server): assert dnr is None -def test_run_python_file(osl_server_process, tcp_osl_server, tmp_path): +@pytest.mark.parametrize("path_type", [str, Path]) +def test_run_python_file(tcp_osl_server, tmp_path, path_type): """Test ``run_python_file``.""" cmd = """ a = 5 @@ -225,7 +241,12 @@ def test_run_python_file(osl_server_process, tcp_osl_server, tmp_path): result = a + b print(result) """ - cmd_path = os.path.join(tmp_path, "commands.txt") + cmd_path = tmp_path / "commands.txt" + if path_type == str: + cmd_path = str(cmd_path) + elif path_type != Path: + assert False + with open(cmd_path, "w") as f: f.write(cmd) run_file = tcp_osl_server.run_python_file(file_path=cmd_path)