Source code for

Handles configuration variables of the package

.. autosummary::

.. codeauthor:: David Zwicker <>

import collections
import contextlib
import importlib
import re
import sys
import warnings
from pathlib import Path
from typing import Any, Dict, List, Optional, Union

from .misc import module_available
from .parameters import Parameter

# define default parameter values
DEFAULT_CONFIG: List[Parameter] = [
        "Determines whether numba used the debug mode for compilation. If enabled, "
        "this emits extra information that might be useful for debugging.",
        "Determines whether the fastmath flag is set during compilation. This affects "
        "the precision of the mathematical calculations.",
        "Determines whether multiple threads are used in numba-compiled code.",
        "Minimal number of support points before multithreading is enabled in numba "

[docs]class Config(collections.UserDict): """class handling the package configuration""" def __init__(self, items: Optional[Dict[str, Any]] = None, mode: str = "update"): """ Args: items (dict, optional): Configuration values that should be added or overwritten to initialize the configuration. mode (str): Defines the mode in which the configuration is used. Possible values are * `insert`: any new configuration key can be inserted * `update`: only the values of pre-existing items can be updated * `locked`: no values can be changed Note that the items specified by `items` will always be inserted, independent of the `mode`. """ self.mode = "insert" # temporarily allow inserting items super().__init__({ p for p in DEFAULT_CONFIG}) if items: self.update(items) self.mode = mode def _translate_deprecated_key(self, key: str) -> str: """helper function that allows using deprecated config items""" # the depreciations have been introduced on 2022-09-04 and are scheduled to be # removed after 2023-03-04 if key == "numba.parallel": warnings.warn( "Option `numba.parallel` has been renamed to `numba.multithreading`", DeprecationWarning, ) return "numba.multithreading" elif key == "numba.parallel_threshold": warnings.warn( "Option `numba.parallel_threshold` has been renamed to " "`numba.multithreading_threshold`", DeprecationWarning, ) return "numba.multithreading_threshold" return key def __getitem__(self, key: str): """retrieve item `key`""" key = self._translate_deprecated_key(key) parameter =[key] if isinstance(parameter, Parameter): return parameter.convert() else: return parameter def __setitem__(self, key: str, value): """update item `key` with `value`""" key = self._translate_deprecated_key(key) if self.mode == "insert":[key] = value elif self.mode == "update": try: self[key] # test whether the key already exist (including magic keys) except KeyError: raise KeyError( f"{key} is not present and config is not in `insert` mode" )[key] = value elif self.mode == "locked": raise RuntimeError("Configuration is locked") else: raise ValueError(f"Unsupported configuration mode `{self.mode}`") def __delitem__(self, key: str): """removes item `key`""" key = self._translate_deprecated_key(key) if self.mode == "insert": del[key] else: raise RuntimeError("Configuration is not in `insert` mode")
[docs] def to_dict(self) -> Dict[str, Any]: """convert the configuration to a simple dictionary Returns: dict: A representation of the configuration in a normal :class:`dict`. """ return {k: v for k, v in self.items()}
def __repr__(self) -> str: """represent the configuration as a string""" return f"{self.__class__.__name__}({repr(self.to_dict())})" @contextlib.contextmanager def __call__(self, values: Optional[Dict[str, Any]] = None, **kwargs): """context manager temporarily changing the configuration Args: values (dict): New configuration parameters **kwargs: New configuration parameters """ data_initial = # save old configuration # set new configuration if values is not None: yield # return to caller # restore old configuration = data_initial
[docs]def get_package_versions( packages: List[str], *, na_str="not available" ) -> Dict[str, str]: """tries to load certain python packages and returns their version Args: packages (list): The names of all packages na_str (str): Text to return if package is not available Returns: dict: Dictionary with version for each package name """ versions: Dict[str, str] = {} for name in sorted(packages): try: module = importlib.import_module(name.replace("-", "_")) except ImportError: versions[name] = na_str else: versions[name] = module.__version__ return versions
[docs]def parse_version_str(ver_str: str) -> List[int]: """helper function converting a version string into a list of integers""" result = [] for token in ver_str.split(".")[:3]: try: result.append(int(token)) except ValueError: pass return result
[docs]def check_package_version(package_name: str, min_version: str): """checks whether a package has a sufficient version""" msg = f"`{package_name}` version {min_version} required for py-pde" try: # obtain version of the package version = importlib.import_module(package_name).__version__ except ImportError: warnings.warn(f"{msg} (but none installed)") else: # check whether it is installed and works if parse_version_str(version) < parse_version_str(min_version): warnings.warn(f"{msg} (installed: {version})")
[docs]def packages_from_requirements(requirements_file: Union[Path, str]) -> List[str]: """read package names from a requirements file Args: requirements_file (str or :class:`~pathlib.Path`): The file from which everything is read Returns: list of package names """ result = [] with open(requirements_file) as fp: for line in fp: line_s = line.strip() if line_s.startswith("#"): continue res ="[a-zA-Z0-9_\-]+", line_s) if res: result.append( return result
[docs]def environment() -> Dict[str, Any]: """obtain information about the compute environment Returns: dict: information about the python installation and packages """ import matplotlib as mpl from .. import __version__ as package_version from .. import config from . import mpi from .numba import numba_environment from .plotting import get_plotting_context PACKAGE_PATH = Path(__file__).resolve().parents[2] result: Dict[str, Any] = {} result["package version"] = package_version result["python version"] = sys.version result["platform"] = sys.platform # add the package configuration result["config"] = config.to_dict() # add details for mandatory packages packages_min = packages_from_requirements(PACKAGE_PATH / "requirements.txt") result["mandatory packages"] = get_package_versions(packages_min) result["matplotlib environment"] = { "backend": mpl.get_backend(), "plotting context": get_plotting_context().__class__.__name__, } # add details about optional packages packages = set( packages_from_requirements(PACKAGE_PATH / "tests" / "requirements_full.txt") ) packages |= set( packages_from_requirements(PACKAGE_PATH / "tests" / "requirements_mpi.txt") ) packages -= set(packages_min) result["optional packages"] = get_package_versions(sorted(packages)) if module_available("numba"): result["numba environment"] = numba_environment() # add information about MPI environment if mpi.size > 1: result["multiprocessing"] = {"initialized": True, "size": mpi.size} else: result["multiprocessing"] = {"initialized": False} return result