diff --git a/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile1.txt b/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile1.txt new file mode 100644 index 000000000..e69de29bb diff --git a/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile2.py b/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile2.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile3.bat b/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opd/regFile3.bat new file mode 100644 index 000000000..e69de29bb diff --git a/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opf b/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opf index 76092fcfc..afd1dba46 100644 Binary files a/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opf and b/src/ansys/optislang/core/examples/00_run_script/files/calculator_with_params.opf differ diff --git a/src/ansys/optislang/core/examples/00_run_script/files/omdb_files.opf b/src/ansys/optislang/core/examples/00_run_script/files/omdb_files.opf new file mode 100644 index 000000000..856f5ed5d Binary files /dev/null and b/src/ansys/optislang/core/examples/00_run_script/files/omdb_files.opf differ diff --git a/src/ansys/optislang/core/examples/examples.py b/src/ansys/optislang/core/examples/examples.py index 70dd02832..877b1133a 100644 --- a/src/ansys/optislang/core/examples/examples.py +++ b/src/ansys/optislang/core/examples/examples.py @@ -64,6 +64,7 @@ "nodes_connection": (module_path / "00_run_script" / "files" / "connect_nodes.opf",), "nested_systems": (module_path / "00_run_script" / "files" / "nested_systems.opf",), "ten_bar_truss": (module_path / "00_run_script" / "files" / "ten_bar_truss.opf",), + "omdb_files": (module_path / "00_run_script" / "files" / "omdb_files.opf",), } # dictionary of scripts to be run @@ -141,4 +142,5 @@ "nodes_connection": None, "nested_systems": None, "ten_bar_truss": None, + "omdb_files": None, } diff --git a/src/ansys/optislang/core/io.py b/src/ansys/optislang/core/io.py new file mode 100644 index 000000000..7294b7ae1 --- /dev/null +++ b/src/ansys/optislang/core/io.py @@ -0,0 +1,298 @@ +"""Module for input/output functionality.""" +from __future__ import annotations + +from enum import Enum +from pathlib import Path +import time +from typing import Union + +from ansys.optislang.core.utils import enum_from_str + + +class File: + """Provides for operating on files.""" + + def __init__(self, path: Union[Path, str]) -> None: + """Create a ``File`` instance. + + Parameters + ---------- + path : Union[pathlib.Path, str] + Path to the file. + id: str, optional + File ID. + comment: str, optional + Description of the file. + """ + self.__path = Path(path) + + def __eq__(self, other: File) -> bool: + r"""Compare properties of two instances of the ``File`` class. + + Parameters + ---------- + other: File + File for comparison. + + Returns + ------- + bool + ``True`` if all properties match, ``False`` otherwise. + """ + if type(self) == type(other): + checks = {} + checks["exists"] = self.exists == other.exists + checks["filename"] = self.filename == other.filename + checks["last_modified_seconds"] = ( + self.last_modified_seconds == other.last_modified_seconds + ) + checks["last_modified_str"] = self.last_modified_str == other.last_modified_str + checks["path"] = self.path == other.path + checks["size"] = self.size == other.size + return False not in checks.values() + else: + return False + + @property + def exists(self) -> bool: + """Check whether file exists. + + Returns + ------- + bool + Information, whether file exists. + """ + return self.path.exists() + + @property + def filename(self) -> str: + """Get file name. + + Returns + ------- + str + File name + """ + return self.path.name + + @property + def last_modified_seconds(self) -> Union[float, None]: + """Last modified time as a timestamp. + + Returns + ------- + Union[float, None] + Last modified time in seconds since the Epoch, `None` if file doesn't exist. + """ + if self.exists: + return self.path.stat().st_mtime + else: + return None + + @property + def last_modified_str(self) -> Union[str, None]: + """Last modified time as a datetime. + + Returns + ------- + Union[str, None] + Last modified time as string, `None` if file doesn't exist. + """ + if self.exists: + return time.ctime(self.last_modified_seconds) + else: + return None + + @property + def path(self) -> Path: + """Path to the file. + + Returns + ------- + Path + Path the the file + """ + return self.__path + + @property + def size(self) -> Union[int, None]: + """File size in bytes. + + Returns + ------- + Union[int, None] + File size in bytes, `None` in file doesn't exist. + """ + if self.exists: + return self.path.stat().st_size + else: + return None + + +class RegisteredFile(File): + """Provides for operating on registered files.""" + + def __init__( + self, + path: Union[Path, str], + id: str, + comment: str, + tag: str, + usage: Union[RegisteredFileUsage, str], + ) -> None: + """Create a ``RegisteredFile`` instance. + + Parameters + ---------- + path : Union[pathlib.Path, str] + Path to the file. + id: str + File ID. + comment: str + Description of the file. + tag: str + File uid. + usage: Union[RegisteredFileUsage, str] + Usage of registered file. + """ + super().__init__(path=path) + self.__id = id + self.__comment = comment + self.__tag = tag + if not isinstance(usage, RegisteredFileUsage): + usage = RegisteredFileUsage.from_str(usage) + self.__usage = usage + + def __eq__(self, other: File) -> bool: + r"""Compare properties of two instances of the ``RegisteredFile`` class. + + Parameters + ---------- + other: RegisteredFile + File for comparison. + + Returns + ------- + bool + ``True`` if all properties match, ``False`` otherwise. + """ + if type(self) == type(other): + checks = {} + checks["exists"] = self.exists == other.exists + checks["filename"] = self.filename == other.filename + checks["last_modified_seconds"] = ( + self.last_modified_seconds == other.last_modified_seconds + ) + checks["last_modified_str"] = self.last_modified_str == other.last_modified_str + checks["path"] = self.path == other.path + checks["size"] = self.size == other.size + checks["comment"] = self.comment == other.comment + checks["id"] = self.id == other.id + checks["tag"] = self.tag == other.tag + checks["usage"] = self.usage == other.usage + return False not in checks.values() + else: + return False + + @property + def comment(self) -> str: + """File description. + + Returns + ------- + str + File description. + """ + return self.__comment + + @property + def id(self) -> str: + """File id. + + Returns + ------- + str + File id. + """ + return self.__id + + @property + def tag(self) -> str: + """File uid. + + Returns + ------- + str + File uid. + """ + return self.__tag + + @property + def usage(self) -> RegisteredFileUsage: + """Usage of registered file. + + Returns + ------- + RegisteredFileUsage + Usage of registered file. + """ + return self.__usage + + +class FileOutputFormat(Enum): + """Provides options for storing files.""" + + JSON = 0 + CSV = 1 + + def to_str(self) -> str: + """Convert file type to suffix. + + Returns + ------- + str + File suffix. + """ + return "." + self.name.lower() + + +class RegisteredFileUsage(Enum): + """Provides options for registered files usage.""" + + UNDETERMINED = 0 + INTERMEDIATE_RESULT = 1 + INPUT_FILE = 2 + OUTPUT_FILE = 3 + + @staticmethod + def from_str(string: str) -> RegisteredFileUsage: + """Convert string to an instance of the ``RegisteredFileUsage`` class. + + Parameters + ---------- + string: str + String to be converted. + + Returns + ------- + RegisteredFileUsage + Instance of the ``RegisteredFileUsage`` class. + + Raises + ------ + TypeError + Raised when an invalid type of ``string`` is given. + ValueError + Raised when an invalid value of ``string`` is given. + """ + return enum_from_str(string=string, enum_class=__class__, replace=(" ", "_")) + + def to_str(self) -> str: + """Convert usage type to string. + + Returns + ------- + str + Usage type. + """ + return self.name[0] + self.name[1:].lower().replace("_", " ") diff --git a/src/ansys/optislang/core/nodes.py b/src/ansys/optislang/core/nodes.py index 0b7bd49c0..f532706fa 100644 --- a/src/ansys/optislang/core/nodes.py +++ b/src/ansys/optislang/core/nodes.py @@ -1,10 +1,16 @@ """Contains classes for a node, system, parametric system, and root system.""" from __future__ import annotations +from collections import OrderedDict import copy +import csv from enum import Enum +from io import StringIO +import json +from pathlib import Path from typing import TYPE_CHECKING, Dict, Iterable, List, Tuple, Union +from ansys.optislang.core.io import File, FileOutputFormat, RegisteredFile, RegisteredFileUsage from ansys.optislang.core.project_parametric import ( ConstraintCriterion, CriteriaManager, @@ -89,25 +95,6 @@ def uid(self) -> str: """ return self.__uid - def _get_info(self) -> dict: - """Get the raw server output with the node info. - - Returns - ------- - dict - Dictionary with the node info. - - Raises - ------ - OslCommunicationError - Raised when an error occurs while communicating with the server. - OslCommandError - Raised when a command or query fails. - TimeoutError - Raised when the timeout float value expires. - """ - return self._osl_server.get_actor_info(self.uid) - def get_name(self) -> str: """Get the name of the node. @@ -203,13 +190,13 @@ def get_properties(self) -> dict: """ return self._osl_server.get_actor_properties(self.uid) - def get_status(self) -> str: - """Get the status of the node. + def get_registered_files(self) -> Tuple[RegisteredFile]: + """Get node's registered files. Returns ------- - str - Status of the node. + Tuple[RegisteredFile] + Tuple of registered files. Raises ------ @@ -220,16 +207,80 @@ def get_status(self) -> str: TimeoutError Raised when the timeout float value expires. """ - actor_info = self._osl_server.get_actor_info(uid=self.__uid) - return actor_info["status"] + registered_files_uids = self._get_info()["registered_files"] + if not registered_files_uids: + return () + project_registered_files = self._osl_server.get_basic_project_info()["projects"][0][ + "registered_files" + ] + return tuple( + [ + RegisteredFile( + path=Path(file["local_location"]["split_path"]["head"]) + / file["local_location"]["split_path"]["tail"], + id=file["ident"], + comment=file["comment"], + tag=file["tag"], + usage=file["usage"], + ) + for file in project_registered_files + if file["tag"] in registered_files_uids + ] + ) - def get_type(self) -> str: - """Get the type of the node. + def get_result_files(self) -> Tuple[RegisteredFile]: + """Get node's result files. + + Returns + ------- + Tuple[RegisteredFile] + Tuple of result files. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + return tuple( + filter( + lambda file: file.usage == RegisteredFileUsage.OUTPUT_FILE, + self.get_registered_files(), + ) + ) + + def get_states_ids(self) -> Tuple[str]: + """Get available actor states ids. + + Returns + ------- + Tuple[str] + Actor states ids. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with server. + OslCommandError + Raised when the command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + states = self._osl_server.get_actor_states(self.uid) + if not states.get("states", None): + return tuple([]) + return tuple([state["hid"] for state in states["states"]]) + + def get_status(self) -> str: + """Get the status of the node. Returns ------- str - Type of the node. + Status of the node. Raises ------ @@ -241,15 +292,15 @@ def get_type(self) -> str: Raised when the timeout float value expires. """ actor_info = self._osl_server.get_actor_info(uid=self.__uid) - return actor_info["type"] + return actor_info["status"] - def _get_parent_uid(self) -> str: - """Get the unique ID of the parent node. + def get_type(self) -> str: + """Get the type of the node. - Return - ------ + Returns + ------- str - Unique ID of the parent node. + Type of the node. Raises ------ @@ -260,14 +311,8 @@ def _get_parent_uid(self) -> str: TimeoutError Raised when the timeout float value expires. """ - project_tree = self._osl_server.get_full_project_tree_with_properties() - root_system_uid = project_tree["projects"][0]["system"]["uid"] - parent_tree = project_tree["projects"][0]["system"] - return Node._find_parent_node_uid( - tree=parent_tree, - parent_uid=root_system_uid, - node_uid=self.uid, - ) + actor_info = self._osl_server.get_actor_info(uid=self.__uid) + return actor_info["type"] def _create_nodes_from_properties_dicts( self, properties_dicts_list: List[dict] @@ -310,6 +355,71 @@ def _create_nodes_from_properties_dicts( return tuple(nodes_list) + def _get_info(self) -> dict: + """Get the raw server output with the node info. + + Returns + ------- + dict + Dictionary with the node info. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + return self._osl_server.get_actor_info(self.uid) + + def _get_parent_uid(self) -> str: + """Get the unique ID of the parent node. + + Return + ------ + str + Unique ID of the parent node. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + project_tree = self._osl_server.get_full_project_tree_with_properties() + root_system_uid = project_tree["projects"][0]["system"]["uid"] + parent_tree = project_tree["projects"][0]["system"] + return Node._find_parent_node_uid( + tree=parent_tree, + parent_uid=root_system_uid, + node_uid=self.uid, + ) + + def _get_status_info(self) -> Tuple[dict]: + """Get node's status info for each state. + + Returns + ------- + Tuple[dict] + Tuple with status info dictionary for each state. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + hids = self.get_states_ids() + return tuple([self._osl_server.get_actor_status_info(self.uid, hid) for hid in hids]) + def _is_parametric_system(self, uid: str) -> bool: """Check if the system is parametric. @@ -732,6 +842,187 @@ def response_manager(self) -> ResponseManager: """ return self.__response_manager + def get_omdb_files(self) -> Tuple[File]: + """Get paths to omdb files. + + Returns + ------- + Tuple[File] + Tuple with File objects containing path. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + statuses_info = self._get_status_info() + wdirs = [Path(status_info["working dir"]) for status_info in statuses_info] + omdb_files = [] + for wdir in wdirs: + omdb_files.extend([File(path) for path in wdir.glob("*.omdb")]) + return tuple(omdb_files) + + def save_designs_as( + self, + hid: str, + file_name: str, + format: FileOutputFormat = FileOutputFormat.JSON, + dir: Union[Path, str] = None, + ) -> File: + """Save designs for a given state. + + Parameters + ---------- + hid : str + Actor's state. + file_name : str + Name of the file. + format : FileOutputFormat, optional + Format of the file, by default ``FileOutputFormat.JSON``. + dir : Union[Path, str], optional + Directory, where file should be saved, by default ``None``. + Project's working directory is used by default. + + Returns + ------- + File + Object representing saved file. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + TypeError + Raised when incorrect type of ``dir`` is passed. + ValueError + Raised when unsupported value of ``format`` or non-existing ``hid`` + is passed. + """ + if dir is not None and isinstance(dir, str): + dir = Path(dir) + elif dir is None: + dir = self._osl_server.get_working_dir() + + if not isinstance(dir, Path): + raise TypeError(f"Unsupported type of dir: `{type(dir)}`.") + + designs = self._get_designs_dicts() + if not designs.get(hid): + raise ValueError(f"Design for given hid: `{hid}` not available.") + + if format == FileOutputFormat.JSON: + output_file = json.dumps(designs[hid]) + newline = None + elif format == FileOutputFormat.CSV: + output_file = self.__class__.__convert_design_dict_to_csv(designs[hid]) + newline = "" + else: + raise ValueError(f"Output type `{format}` is not supported.") + + output_file_path = dir / (file_name + format.to_str()) + with open(output_file_path, "w", newline=newline) as f: + f.write(output_file) + return File(output_file_path) + + def _get_designs_dicts(self) -> OrderedDict: + """Get parametric system's designs. + + Returns + ------- + OrderedDict + Ordered dictionary of designs, key is hid and value + is list of corresponding designs. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + statuses_info = self._get_status_info() + if not statuses_info: + return {} + designs = {} + # TODO: sort by hid? -> delete / use OrderedDict + for status_info in statuses_info: + designs[status_info["hid"]] = status_info["designs"] + for idx, design in enumerate(designs[status_info["hid"]]["values"]): + self.__class__.__append_status_info_to_design( + design, status_info["design_status"][idx] + ) + for i in range(2, len(statuses_info[0]["designs"]["values"][0]["hid"].split("."))): + designs = self.__class__.__sort_dict_by_key_hid( + designs, max(1, len(statuses_info[0]["designs"]["values"][0]["hid"].split(".")) - i) + ) + for hid, design in designs.items(): + # sort by design number, stripped from hid prefix + # (0.10, 0.9 ..., 0.1) -> (0.1, ..., 0.9, 0.10) + design["values"] = self.__class__.__sort_list_of_dicts_by_hid( + design["values"], len(hid.split(".")) + ) + return designs + + @staticmethod + def __append_status_info_to_design(design: dict, status_info: dict) -> None: + if design["hid"] != status_info["id"]: + raise ValueError(f'{design["hid"]} != {status_info["id"]}') + to_append = { + key: status_info[key] for key in ("feasible", "status", "pareto_design", "directory") + } + design.update(to_append) + + @staticmethod + def __convert_design_dict_to_csv(designs: dict) -> str: + csv_buffer = StringIO() + try: + csv_writer = csv.writer(csv_buffer) + header = ["Design"] + header.append("Feasible") + header.append("Status") + header.append("Pareto") + header.extend(designs["constraint_names"]) + header.extend(designs["limit_state_names"]) + header.extend(designs["objective_names"]) + header.extend(designs["parameter_names"]) + header.extend(designs["response_names"]) + csv_writer.writerow(header) + for design in designs["values"]: + line = [design["hid"]] + line.append(design["feasible"]) + line.append(design["status"]) + line.append(design["pareto_design"]) + line.extend(design["constraint_values"]) + line.extend(design["limit_state_values"]) + line.extend(design["objective_values"]) + line.extend(design["parameter_values"]) + line.extend(design["response_values"]) + csv_writer.writerow(line) + return csv_buffer.getvalue() + finally: + if csv_buffer is not None: + csv_buffer.close() + + @staticmethod + def __sort_list_of_dicts_by_hid(unsorted_list: List[dict], sort_by_position: int) -> List[dict]: + sort_key = lambda x: int(x["hid"].split(".")[sort_by_position]) + return sorted(unsorted_list, key=sort_key) + + @staticmethod + def __sort_dict_by_key_hid(unsorted_dict: dict, sort_by_position: int) -> dict: + sort_key = lambda item: int(item[0].split(".")[sort_by_position]) + return OrderedDict(sorted(unsorted_dict.items(), key=sort_key)) + class RootSystem(ParametricSystem): """Provides for creating and operating on a project system.""" diff --git a/src/ansys/optislang/core/optislang.py b/src/ansys/optislang/core/optislang.py index 7d9d012f1..bf091edfa 100644 --- a/src/ansys/optislang/core/optislang.py +++ b/src/ansys/optislang/core/optislang.py @@ -11,7 +11,7 @@ from ansys.optislang.core.tcp_osl_server import TcpOslServer if TYPE_CHECKING: - from ansys.optislang.core.logging import OslLogger + from ansys.optislang.core.logging import OslCustomAdapter from ansys.optislang.core.osl_server import OslServer @@ -324,7 +324,7 @@ def name(self) -> str: return self.__name @property - def log(self) -> OslLogger: + def log(self) -> OslCustomAdapter: """Instance logger.""" return self.__logger diff --git a/src/ansys/optislang/core/project.py b/src/ansys/optislang/core/project.py index 74f19cf27..fa00fa222 100644 --- a/src/ansys/optislang/core/project.py +++ b/src/ansys/optislang/core/project.py @@ -2,8 +2,9 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Tuple +from ansys.optislang.core.io import RegisteredFile, RegisteredFileUsage from ansys.optislang.core.nodes import RootSystem if TYPE_CHECKING: @@ -152,6 +153,61 @@ def get_reference_design(self) -> Design: """ return self.root_system.get_reference_design() + def get_registered_files(self) -> Tuple[RegisteredFile]: + """Get all registered files in the current project. + + Returns + ------- + Tuple[RegisteredFile] + Tuple with registered files. + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + project_registered_files_dicts = self.__osl_server.get_basic_project_info()["projects"][0][ + "registered_files" + ] + return tuple( + [ + RegisteredFile( + path=Path(file["local_location"]["split_path"]["head"]), + id=file["ident"], + comment=file["comment"], + tag=file["tag"], + usage=file["usage"], + ) + for file in project_registered_files_dicts + ] + ) + + def get_result_files(self) -> Tuple[RegisteredFile]: + """Get result files. + + Returns + ------- + Tuple[RegisteredFile] + Tuple with result files + + Raises + ------ + OslCommunicationError + Raised when an error occurs while communicating with the server. + OslCommandError + Raised when a command or query fails. + TimeoutError + Raised when the timeout float value expires. + """ + registered_files = self.get_registered_files() + return tuple( + [file for file in registered_files if file.usage == RegisteredFileUsage.OUTPUT_FILE] + ) + def get_status(self) -> str: """Get the status of the optiSLang project. diff --git a/tests/test_io.py b/tests/test_io.py new file mode 100644 index 000000000..defa05ebd --- /dev/null +++ b/tests/test_io.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from enum import Enum +from pathlib import Path + +import pytest + +from ansys.optislang.core.io import File, FileOutputFormat, RegisteredFile, RegisteredFileUsage + +CURRENT_FILE = __file__ +NON_EXISTING_FILE = Path().cwd() / "non_existing.py" + + +# region TEST ENUMERATION METHODS: +def enumeration_test_method(enumeration_class: Enum, enumeration_name: str): + """Test instance creation, method `from_str` and spelling.""" + mixed_name = "" + for index, char in enumerate(enumeration_name): + if index % 2 == 1: + mixed_name += char.lower() + else: + mixed_name += char + try: + enumeration_from_str = enumeration_class.from_str(string=mixed_name) + except: + assert False + assert isinstance(enumeration_from_str, enumeration_class) + assert isinstance(enumeration_from_str.name, str) + assert enumeration_from_str.name == enumeration_name + + +@pytest.mark.parametrize( + "file_output_format, name", + [ + (FileOutputFormat, "CSV"), + (FileOutputFormat, "JSON"), + ], +) +def test_file_output_format(file_output_format: FileOutputFormat, name: str): + """Test `FileOutputFormat`.""" + enumeration_test_method(enumeration_class=file_output_format, enumeration_name=name) + + +@pytest.mark.parametrize( + "registered_file_usage, name", + [ + (RegisteredFileUsage, "INPUT_FILE"), + (RegisteredFileUsage, "INTERMEDIATE_RESULT"), + (RegisteredFileUsage, "OUTPUT_FILE"), + (RegisteredFileUsage, "UNDETERMINED"), + ], +) +def test_file_output_format(registered_file_usage: RegisteredFileUsage, name: str): + """Test `RegisteredFileUsage`.""" + enumeration_test_method(enumeration_class=registered_file_usage, enumeration_name=name) + + +# endregion + + +# region TEST FILE CLASSES +def test_file(): + """Test ``File`` class.""" + existing_file = File(CURRENT_FILE) + non_existing_file = File(NON_EXISTING_FILE) + + assert isinstance(existing_file.exists, bool) + assert isinstance(non_existing_file.exists, bool) + assert existing_file.exists + assert not non_existing_file.exists + + assert isinstance(existing_file.filename, str) + assert existing_file.filename == "test_io.py" + assert non_existing_file.filename == "non_existing.py" + + assert isinstance(existing_file.last_modified_seconds, float) + assert non_existing_file.last_modified_seconds == None + assert isinstance(existing_file.last_modified_str, str) + assert non_existing_file.last_modified_str == None + assert isinstance(existing_file.path, Path) + assert isinstance(non_existing_file.path, Path) + assert isinstance(existing_file.size, int) + assert non_existing_file.size == None + + assert existing_file == existing_file + assert not existing_file == non_existing_file + + +def test_registered_file(): + """Test ``RegisteredFile`` class.""" + registered_file = RegisteredFile( + CURRENT_FILE, "python_module", "comment", "xxxx-test-uid", RegisteredFileUsage.UNDETERMINED + ) + non_existing_registered_file = RegisteredFile( + "xxx", "python_module", "comment", "xxxx-test-uid", RegisteredFileUsage.UNDETERMINED + ) + + assert isinstance(registered_file.comment, str) + assert isinstance(registered_file.id, str) + assert isinstance(registered_file.tag, str) + assert isinstance(registered_file.usage, RegisteredFileUsage) + assert not registered_file == non_existing_registered_file + + +# endregion diff --git a/tests/test_nodes.py b/tests/test_nodes.py index 285c866e0..587b9b13d 100644 --- a/tests/test_nodes.py +++ b/tests/test_nodes.py @@ -1,15 +1,18 @@ from contextlib import nullcontext as does_not_raise +from pathlib import Path import pytest from ansys.optislang.core import Optislang, examples +from ansys.optislang.core.io import File, FileOutputFormat, RegisteredFile from ansys.optislang.core.nodes import Node, ParametricSystem, RootSystem, System from ansys.optislang.core.project_parametric import ParameterManager pytestmark = pytest.mark.local_osl -single_node = examples.get_files("calculator_with_params")[1][0] +calculator_w_parameters = examples.get_files("calculator_with_params")[1][0] nested_systems = examples.get_files("nested_systems")[1][0] connect_nodes = examples.get_files("nodes_connection")[1][0] +omdb_files = examples.get_files("omdb_files")[1][0] @pytest.fixture() @@ -29,7 +32,7 @@ def optislang(scope="function", autouse=False) -> Optislang: # TEST NODE def test_node_initialization(optislang: Optislang): """Test `Node` initialization.""" - optislang.open(file_path=single_node) + optislang.open(file_path=calculator_w_parameters) project = optislang.project root_system = project.root_system node = root_system.get_nodes()[0] @@ -39,7 +42,7 @@ def test_node_initialization(optislang: Optislang): def test_node_properties(optislang: Optislang): """Test properties of the instance of `Node` class.""" - optislang.open(file_path=single_node) + optislang.open(file_path=calculator_w_parameters) project = optislang.project root_system = project.root_system node = root_system.get_nodes()[0] @@ -50,7 +53,7 @@ def test_node_properties(optislang: Optislang): def test_node_queries(optislang: Optislang): """Test get methods of the instance of `Node` class.""" - optislang.open(file_path=single_node) + optislang.open(file_path=calculator_w_parameters) project = optislang.project root_system = project.root_system node = root_system.find_nodes_by_name("Calculator")[0] @@ -72,6 +75,18 @@ def test_node_queries(optislang: Optislang): properties = node.get_properties() assert isinstance(properties, dict) + reg_files = node.get_registered_files() + assert len(reg_files) == 1 + assert isinstance(reg_files[0], RegisteredFile) + + res_files = node.get_result_files() + assert len(res_files) == 1 + assert isinstance(res_files[0], RegisteredFile) + + states_ids = node.get_states_ids() + assert len(states_ids) == 1 + assert isinstance(states_ids[0], str) + status = node.get_status() assert isinstance(status, str) @@ -198,3 +213,56 @@ def test_get_parameter_manager(optislang: Optislang): assert isinstance(parameter_manager, ParameterManager) optislang.dispose() assert dnr is None + + +def test_get_omdb_files(tmp_path: Path): + """Test `get_omdb_files()` method.""" + optislang = Optislang(project_path=omdb_files) + optislang.set_timeout(30) + optislang.reset() + optislang.start() + project = optislang.project + root_system = project.root_system + + sensitivity: ParametricSystem = root_system.find_nodes_by_name("Sensitivity")[0] + s_omdb_file = sensitivity.get_omdb_files() + assert len(s_omdb_file) > 0 + assert isinstance(s_omdb_file[0], File) + + most_inner_sensitivity: ParametricSystem = root_system.find_nodes_by_name( + "MostInnerSensitivity", search_depth=3 + )[0] + mis_omdb_file = most_inner_sensitivity.get_omdb_files() + assert len(mis_omdb_file) > 0 + assert isinstance(mis_omdb_file[0], File) + optislang.dispose() + + +def test_save_designs_as(tmp_path: Path): + """Test `save_designs_as` method.""" + optislang = Optislang(project_path=omdb_files) + optislang.set_timeout(30) + optislang.reset() + optislang.start() + project = optislang.project + root_system = project.root_system + + sensitivity: ParametricSystem = root_system.find_nodes_by_name("Sensitivity")[0] + s_hids = sensitivity.get_states_ids() + s_file = sensitivity.save_designs_as(s_hids[0], "FirstDesign", FileOutputFormat.CSV, tmp_path) + assert isinstance(s_file, File) + assert s_file.exists + assert s_file.path.suffix == ".csv" + + most_inner_sensitivity: ParametricSystem = root_system.find_nodes_by_name( + "MostInnerSensitivity", 3 + )[0] + mis_hids = most_inner_sensitivity.get_states_ids() + mis_file = most_inner_sensitivity.save_designs_as(mis_hids[0], "InnerDesign") + assert isinstance(mis_file, File) + assert mis_file.exists + assert mis_file.path.suffix == ".json" + mis_file.path.unlink() + assert not mis_file.exists + + optislang.dispose() diff --git a/tests/test_project.py b/tests/test_project.py index 799cf3823..b848694fc 100644 --- a/tests/test_project.py +++ b/tests/test_project.py @@ -4,11 +4,13 @@ import pytest -from ansys.optislang.core import Optislang +from ansys.optislang.core import Optislang, examples +from ansys.optislang.core.io import RegisteredFile from ansys.optislang.core.nodes import RootSystem from ansys.optislang.core.project_parametric import Design, ParameterManager pytestmark = pytest.mark.local_osl +calculator_w_parameters = examples.get_files("calculator_with_params")[1][0] @pytest.fixture() @@ -25,48 +27,33 @@ def optislang(scope="function", autouse=True) -> Optislang: return osl -def test_get_description(optislang: Optislang): - """Test `get_description`.""" +def test_project_queries(optislang: Optislang): + """Test project queries.""" project = optislang.project + description = project.get_description() assert isinstance(description, str) - with does_not_raise() as dnr: - optislang.dispose() - time.sleep(3) - assert dnr is None - -def test_get_location(optislang: Optislang): - """Test ``get_location``.""" - project = optislang.project location = project.get_location() assert isinstance(location, Path) - with does_not_raise() as dnr: - optislang.dispose() - time.sleep(3) - assert dnr is None - -def test_get_name(optislang: Optislang): - """Test ``get_name``.""" - project = optislang.project name = project.get_name() assert isinstance(name, str) - with does_not_raise() as dnr: - optislang.dispose() - time.sleep(3) - assert dnr is None - -def test_get_status(optislang: Optislang): - """Test ``get_status``.""" - project = optislang.project status = project.get_status() assert isinstance(status, str) - with does_not_raise() as dnr: - optislang.dispose() - time.sleep(3) - assert dnr is None + + optislang.open(file_path=calculator_w_parameters) + + reg_files = project.get_registered_files() + assert len(reg_files) == 3 + assert isinstance(reg_files[0], RegisteredFile) + + res_files = project.get_result_files() + assert len(res_files) == 1 + assert isinstance(res_files[0], RegisteredFile) + + optislang.dispose() def test_project_properties(optislang: Optislang):