From 8cdf23f873d38084d1d805eeab56bcaf0f4d1661 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20Schm=C3=B6lder?= Date: Thu, 15 Aug 2024 12:23:46 +0200 Subject: [PATCH] Add docstrings / type annotations --- cadet/cadet.py | 342 +++++++++++++++++++-------- cadet/cadet_dll_parameterprovider.py | 124 ++++++---- 2 files changed, 321 insertions(+), 145 deletions(-) diff --git a/cadet/cadet.py b/cadet/cadet.py index f0a0197..c72cbad 100644 --- a/cadet/cadet.py +++ b/cadet/cadet.py @@ -1,122 +1,248 @@ +import os +from pathlib import Path import platform import shutil import subprocess +from typing import Optional import warnings -from pathlib import Path from cadet.h5 import H5 from cadet.runner import CadetRunnerBase, CadetFileRunner from cadet.cadet_dll import CadetDLLRunner -def is_dll(value): - suffix = Path(value).suffix +def is_dll(path: os.PathLike) -> bool: + """ + Determine if the given path points to a shared library. + + Parameters + ---------- + path : os.PathLike + Path to the file. + + Returns + ------- + bool + True if the file has a shared library extension (.so, .dll), False otherwise. + """ + suffix = Path(path).suffix return suffix in {'.so', '.dll'} class CadetMeta(type): """ - A meta class for the CADET interface. This allows calls to Cadet.cadet_path = "..." to set - the cadet_path for all subsequent Cadet() instances. + Meta class for the CADET interface. + + This meta class allows setting the `cadet_path` attribute for all instances of the + `Cadet` class. The `cadet_path` determines whether to use a DLL or file-based CADET + runner. """ - _cadet_runner_class = None - _is_file_class = None @property - def is_file(cls): + def is_file(cls) -> bool: + """ + Check if the current runner is file-based. + + Returns + ------- + bool + True if the current runner is file-based, False otherwise. + """ return bool(cls._is_file_class) @property - def cadet_path(cls): + def cadet_path(cls) -> Optional[Path]: + """ + Get the current CADET path. + + Returns + ------- + Optional[Path] + The current CADET path if set, otherwise None. + """ if cls._cadet_runner_class is not None: return cls._cadet_runner_class.cadet_path + return None @cadet_path.setter - def cadet_path(cls, value): - if cls._cadet_runner_class is not None and cls._cadet_runner_class.cadet_path != value: + def cadet_path(cls, cadet_path: os.PathLike) -> None: + """ + Set the CADET path and initialize the appropriate runner. + + Parameters + ---------- + cadet_path : os.PathLike + Path to the CADET executable or library. + + Notes + ----- + If the path is a DLL, a `CadetDLLRunner` runner is used. + Otherwise, a `CadetFileRunner` runner is used. + """ + cadet_path = Path(cadet_path) + if ( + cls._cadet_runner_class is not None + and + cls._cadet_runner_class.cadet_path != cadet_path + ): del cls._cadet_runner_class - if is_dll(value): - cls._cadet_runner_class = CadetDLLRunner(value) + if is_dll(cadet_path): + cls._cadet_runner_class = CadetDLLRunner(cadet_path) cls._is_file_class = False else: - cls._cadet_runner_class = CadetFileRunner(value) + cls._cadet_runner_class = CadetFileRunner(cadet_path) cls._is_file_class = True @cadet_path.deleter - def cadet_path(cls): + def cadet_path(cls) -> None: + """ + Delete the current CADET runner instance. + """ del cls._cadet_runner_class class Cadet(H5, metaclass=CadetMeta): - # cadet_path must be set in order for simulations to run - def __init__(self, *data): - super().__init__(*data) - self._cadet_runner = None - - self._is_file = None # Is CLI or DLL - # self.cadet_path # from Bill, declared in meta class -> path to CLI-file or DLL-file - - self.install_path = None # from Jo -> root of the CADET installation. + """ + CADET interface class. + + This class manages the CADET runner, whether it's based on a CLI executable or a DLL, + and provides methods for running simulations and loading results. + + Attributes + ---------- + install_path : Optional[Path] + The root directory of the CADET installation. + cadet_cli_path : Optional[Path] + Path to the 'cadet-cli' executable. + cadet_dll_path : Optional[Path] + Path to the 'cadet.dll' or equivalent shared library. + cadet_create_lwe_path : Optional[Path] + Path to the 'createLWE' executable. + return_information : Optional[dict] + Stores the information returned after a simulation run. + """ - self.cadet_cli_path = None - self.cadet_dll_path = None - self.cadet_create_lwe_path = None + def __init__(self, *data): + """ + Initialize a new instance of the Cadet class. - self.return_information = None + Parameters + ---------- + *data : tuple + Additional data to be passed to the base class. + """ + super().__init__(*data) + self._cadet_runner: Optional[CadetRunnerBase] = None + self._is_file: Optional[bool] = None + self.install_path: Optional[Path] = None + self.cadet_cli_path: Optional[Path] = None + self.cadet_dll_path: Optional[Path] = None + self.cadet_create_lwe_path: Optional[Path] = None + self.return_information: Optional[dict] = None @property - def is_file(self): + def is_file(self) -> Optional[bool]: + """ + Check if the current runner is file-based. + + Returns + ------- + Optional[bool] + True if the runner is file-based, False otherwise. None if undetermined. + """ if self._is_file is not None: return bool(self._is_file) if self._is_file_class is not None: return bool(self._is_file_class) + return None @property - def cadet_runner(self): + def cadet_runner(self) -> CadetRunnerBase: + """ + Get the current CADET runner instance. + + Returns + ------- + CadetRunnerBase + The current runner instance, either a DLL or file-based runner. + """ if self._cadet_runner is not None: return self._cadet_runner if hasattr(self, "_cadet_runner_class") and self._cadet_runner_class is not None: return self._cadet_runner_class + self.autodetect_cadet() + return self._cadet_runner @property - def cadet_path(self): + def cadet_path(self) -> Path: + """ + Get the path to the current CADET executable or library. + + Returns + ------- + Path + The path to the current CADET executable or library if set, otherwise None. + """ runner = self.cadet_runner if runner is not None: return runner.cadet_path + return None @cadet_path.setter - def cadet_path(self, value): - if self._cadet_runner is not None and self._cadet_runner.cadet_path != value: + def cadet_path(self, cadet_path: os.PathLike) -> None: + """ + Set the CADET path and initialize the appropriate runner. + + Parameters + ---------- + cadet_path : os.PathLike + Path to the CADET executable or library. + + Notes + ----- + If the path is a DLL, a `CadetDLLRunner` runner is used. + Otherwise, a `CadetFileRunner` runner is used. + """ + cadet_path = Path(cadet_path) + + if self._cadet_runner is not None and self._cadet_runner.cadet_path != cadet_path: del self._cadet_runner - if is_dll(value): - self._cadet_runner = CadetDLLRunner(value) + if is_dll(cadet_path): + self._cadet_runner = CadetDLLRunner(cadet_path) self._is_file = False else: - self._cadet_runner = CadetFileRunner(value) + self._cadet_runner = CadetFileRunner(cadet_path) self._is_file = True @cadet_path.deleter - def cadet_path(self): + def cadet_path(self) -> None: + """ + Delete the current CADET runner instance. + """ del self._cadet_runner - def autodetect_cadet(self): + def autodetect_cadet(self) -> Optional[Path]: """ - Autodetect installation CADET based on operating system and API usage. + Autodetect the CADET installation path. Returns ------- - cadet_root : Path - Installation path of the CADET program. + Optional[Path] + The root directory of the CADET installation. + + Raises + ------ + FileNotFoundError + If CADET cannot be found in the system path. """ executable = 'cadet-cli' if platform.system() == 'Windows': executable += '.exe' - # Searching for the executable in system path path = shutil.which(executable) if path is None: @@ -125,49 +251,33 @@ def autodetect_cadet(self): ) cli_path = Path(path) - - cadet_root = None - if cli_path is not None: - cadet_root = cli_path.parent.parent + cadet_root = cli_path.parent.parent if cli_path else None + if cadet_root: self.install_path = cadet_root return cadet_root @property - def install_path(self): - """str: Path to the installation of CADET. - - This can either be the root directory of the installation or the path to the - executable file 'cadet-cli'. If a file path is provided, the root directory will - be inferred. - - Raises - ------ - FileNotFoundError - If CADET cannot be found at the specified path. - - Warnings - -------- - If the specified install_path is not the root of the CADET installation, it will - be inferred from the file path. + def install_path(self) -> Optional[Path]: + """ + Path to the installation of CADET. - See Also - -------- - check_cadet + Returns + ------- + Optional[Path] + The root directory of the CADET installation or the path to 'cadet-cli'. """ return self._install_path @install_path.setter - def install_path(self, install_path): + def install_path(self, install_path: Optional[os.PathLike]) -> None: """ Set the installation path of CADET. Parameters ---------- - install_path : str or Path - Path to the root of the CADET installation. - It should either be the root directory of the installation or the path - to the executable file 'cadet-cli'. + install_path : Path | os.PathLike | None + Path to the root of the CADET installation or the executable file 'cadet-cli'. If a file path is provided, the root directory will be inferred. """ if install_path is None: @@ -203,7 +313,7 @@ def install_path(self, install_path): self.cadet_path = cadet_cli_path else: raise FileNotFoundError( - "CADET could not be found. Please check the path" + "CADET could not be found. Please check the path." ) cadet_create_lwe_path = cadet_root / 'bin' / lwe_executable @@ -217,46 +327,90 @@ def install_path(self, install_path): dll_path = cadet_root / 'lib' / 'lib_cadet.so' dll_debug_path = cadet_root / 'lib' / 'lib_cadet_d.so' - # Look for debug dll if dll is not found. if not dll_path.is_file() and dll_debug_path.is_file(): dll_path = dll_debug_path - # Look for debug dll if dll is not found. if dll_path.is_file(): self.cadet_dll_path = dll_path.as_posix() - def transform(self, x): + def transform(self, x: str) -> str: + """ + Transform the input string to uppercase. + + Parameters + ---------- + x : str + Input string. + + Returns + ------- + str + Transformed string in uppercase. + """ return str.upper(x) - def inverse_transform(self, x): - return str.lower(x) + def inverse_transform(self, x: str) -> str: + """ + Transform the input string to lowercase. - def load_results(self): - runner = self.cadet_runner - if runner is not None: - runner.load_results(self) + Parameters + ---------- + x : str + Input string. - def run(self, timeout=None, check=None): - data = self.cadet_runner.run(simulation=self.root.input, filename=self.filename, timeout=timeout, check=check) - # TODO: Why is this commented out? - # self.return_information = data - return data + Returns + ------- + str + Transformed string in lowercase. + """ + return str.lower(x) + + def run( + self, + timeout: Optional[int] = None, + ) -> None: + """ + Run the CADET simulation. - def run_load(self, timeout = None, check=None, clear=True): - data = self.cadet_runner.run( - simulation=self.root.input, - filename=self.filename, + Parameters + ---------- + timeout : Optional[int] + Maximum time allowed for the simulation to run, in seconds. + """ + self.cadet_runner.run( + self, timeout=timeout, - check=check ) - # TODO: Why is this commented out? - # self.return_information = data + + def run_load( + self, + timeout: Optional[int] = None, + clear: bool = True + ) -> None: + """ + Run the CADET simulation and load the results. + + Parameters + ---------- + timeout : Optional[int] + Maximum time allowed for the simulation to run, in seconds. + clear : bool + If True, clear previous results after loading new ones. + """ + self.run(timeout) self.load_results() + if clear: self.clear() - return data - def clear(self): + def load_results(self) -> None: + """Load the results of the last simulation run into the current instance.""" + runner = self.cadet_runner + if runner is not None: + runner.load_results(self) + + def clear(self) -> None: + """Clear the loaded results from the current instance.""" runner = self.cadet_runner if runner is not None: runner.clear() diff --git a/cadet/cadet_dll_parameterprovider.py b/cadet/cadet_dll_parameterprovider.py index 73e03d4..9b95153 100644 --- a/cadet/cadet_dll_parameterprovider.py +++ b/cadet/cadet_dll_parameterprovider.py @@ -1,47 +1,88 @@ - import ctypes import cadet.cadet_dll_utils as utils import addict +from typing import Any, Dict, Optional, Union -c_cadet_result = ctypes.c_int +c_cadet_result = ctypes.c_int array_double = ctypes.POINTER(ctypes.POINTER(ctypes.c_double)) - - point_int = ctypes.POINTER(ctypes.c_int) -def null(*args): + +def null(*args: Any) -> None: pass + if 0: log_print = print else: log_print = null + class NestedDictReader: + """ + Utility class to read and navigate through nested dictionaries. + """ - def __init__(self, data): + def __init__(self, data: Dict[str, Any]) -> None: self._root = data - self._cursor = [] - self._cursor.append(data) - self.buffer = None - - def push_scope(self, scope): + self._cursor = [data] + self.buffer: Optional[Any] = None + + def push_scope(self, scope: str) -> bool: + """ + Enter a nested scope within the dictionary. + + Parameters + ---------- + scope : str + The key representing the nested scope. + + Returns + ------- + bool + True if the scope exists and was entered, otherwise False. + """ if scope in self._cursor[-1]: - log_print('Entering scope {}'.format(scope)) + log_print(f'Entering scope {scope}') self._cursor.append(self._cursor[-1][scope]) return True - return False - def pop_scope(self): - self._cursor.pop() - log_print('Exiting scope') - - def current(self): + def pop_scope(self) -> None: + """ + Exit the current scope. + """ + if len(self._cursor) > 1: + self._cursor.pop() + log_print('Exiting scope') + + def current(self) -> Any: + """ + Get the current scope data. + + Returns + ------- + Any + The current data under the scope. + """ return self._cursor[-1] -def recursively_convert_dict(data): + +def recursively_convert_dict(data: Dict[str, Any]) -> addict.Dict: + """ + Recursively convert dictionary keys to uppercase while preserving nested structure. + + Parameters + ---------- + data : dict + The dictionary to convert. + + Returns + ------- + addict.Dict + A new dictionary with all keys converted to uppercase. + """ ans = addict.Dict() for key_original, item in data.items(): if isinstance(item, dict): @@ -53,20 +94,23 @@ def recursively_convert_dict(data): class PARAMETERPROVIDER(ctypes.Structure): - """Implemented the CADET Parameter Provider interface which allows - querying python for the necessary parameters for a CADET simulation - - _fields_ is a list of function names and signatures exposed by the - capi Parameter Provider interface - https://docs.python.org/3/library/ctypes.html - https://github.com/modsim/CADET/blob/master/src/libcadet/api/CAPIv1.cpp""" + """ + Implement the CADET Parameter Provider interface, allowing querying Python for parameters. - def __init__(self, simulation): - sim_input = recursively_convert_dict(simulation) + This class exposes various function pointers as fields in a ctypes structure + to be used with CADET's C-API. + Parameters + ---------- + simulation : Cadet + The simulation object containing the input data. + """ + + def __init__(self, simulation: "Cadet") -> None: + sim_input = recursively_convert_dict(simulation.root.input) self.userData = NestedDictReader(sim_input) - #figure out how to add this to the class + # Assign function pointers at instance level self.getDouble = self._fields_[1][1](utils.param_provider_get_double) self.getInt = self._fields_[2][1](utils.param_provider_get_int) self.getBool = self._fields_[3][1](utils.param_provider_get_bool) @@ -112,25 +156,3 @@ def __init__(self, simulation): ('pushScope', ctypes.CFUNCTYPE(c_cadet_result, ctypes.py_object, ctypes.c_char_p)), ('popScope', ctypes.CFUNCTYPE(c_cadet_result, ctypes.py_object)), ] - - #these don't work right now but work when places on an instance - #getDouble = _fields_[1][1](utils.param_provider_get_double) - #getInt = _fields_[2][1](utils.param_provider_get_int) - #getBool = _fields_[3][1](utils.param_provider_get_bool) - #getString = _fields_[4][1](utils.param_provider_get_string) - - #getDoubleArray = _fields_[5][1](utils.param_provider_get_double_array) - #getIntArray = _fields_[6][1](utils.param_provider_get_int_array) - #getBoolArray = ctypes.cast(None, _fields_[7][1]) - #getStringArray = ctypes.cast(None, _fields_[8][1]) - - #getDoubleArrayItem = _fields_[9][1](utils.param_provider_get_double_array_item) - #getIntArrayItem = _fields_[10][1](utils.param_provider_get_int_array_item) - #getBoolArrayItem = _fields_[11][1](utils.param_provider_get_bool_array_item) - #getStringArrayItem = _fields_[12][1](utils.param_provider_get_string_array_item) - - #exists = _fields_[13][1](utils.param_provider_exists) - #isArray = _fields_[14][1](utils.param_provider_is_array) - #numElements = _fields_[15][1](utils.param_provider_num_elements) - #pushScope = _fields_[16][1](utils.param_provider_push_scope) - #popScope = _fields_[17][1](utils.param_provider_pop_scope) \ No newline at end of file