Skip to content

eris

Top-level module, including resource and optional dependency management.

DependencyWarning

Bases: ErisWarning

Custom warning class for missing optional dependencies.

Source code in eris/__init__.py 
119
120
121
class DependencyWarning(ErisWarning):
    """Custom warning class for missing optional dependencies."""
    pass

ErisWarning

Bases: Warning

A warning class for this package, making it easy to silence all our warning messages should you wish to. Consult the python.warnings module documentation for more details.

Examples:

>>> import warnings
>>> from eris import ErisWarning
... warnings.simplefilter('ignore', ErisWarning)
Source code in eris/__init__.py 
105
106
107
108
109
110
111
112
113
114
115
116
class ErisWarning(Warning):
    """
    A warning class for this package, making it easy to silence all our warning messages should you wish to.
    Consult the `python.warnings` module documentation for more details.

    Examples:
        >>> import warnings
        >>> from eris import ErisWarning
        ... warnings.simplefilter('ignore', ErisWarning)
    """

    pass

Resources

Holds global resources for this package which are generated on demand.

Attributes:

Name Type Description
package str

Name of the package
optional_packages set[str]

Set of optional packages to check for
Source code in eris/__init__.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class Resources:
    """
    Holds global resources for this package which are generated on demand.

    Attributes:
        package: Name of the package
        optional_packages: Set of optional packages to check for
    """
    def __init__(self, *optional_packages: str):
        """
        Parameters:
            optional_packages: Optional packages to check for, e.g. 'numpy', 'pandas'
        """
        self.package: str = Path(__file__).parent.name
        self.optional_packages: set[str] = set(filter(self._check_module, optional_packages))
        self._data: 'Traversible' = None  # Generated on demand
        self._metadata: 'PackageMetadata' = None  # Generated on demand
        self._available_cpus: int = None  # Generated on demand
        self._rng: 'Random' = None  # Generated on demand
        self._pool: 'Executor' = None  # Generated on demand

    @property
    def data(self) -> 'Traversable':
        """Path to the package data"""
        if self._data is None:
            from importlib.resources import files
            self._data = files(self.package) / 'data'
        return self._data

    @property
    def metadata(self) -> 'PackageMetadata':
        """Package metadata"""
        if self._metadata is None:
            from importlib.metadata import metadata
            self._metadata = metadata(self.package)
        return self._metadata

    @property
    def rng(self) -> 'Random':
        """A random number generator instance, can be reused"""
        if self._rng is None:
            from random import Random
            self._rng = Random()
        return self._rng

    @property
    def available_cpus(self) -> int:
        """Number of available CPUs"""
        if self._available_cpus is None:
            try:
                from os import process_cpu_count as cpu_count
            except ImportError:
                from os import cpu_count
            self._available_cpus = cpu_count()
        return self._available_cpus

    @property
    def pool(self) -> 'Executor':
        """A concurrent.futures.Executor instance"""
        if self._pool is None:
            from concurrent.futures import ThreadPoolExecutor
            self._pool = ThreadPoolExecutor(min(32, self.available_cpus + 4))
        return self._pool

    @staticmethod
    def _check_module(module_name: str) -> bool:
        """Checks if a module can be imported.

        Args:
            module_name (str): The name of the module to check.

        Returns:
            bool: True if the module can be imported, False otherwise.
        """
        try:
            import_module(module_name)
            return True
        except ImportError:
            return False

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._pool is not None:
            self._pool.shutdown(wait=False, cancel_futures=True)

    def __del__(self):
        if self._pool is not None:
            self._pool.shutdown(wait=False, cancel_futures=True)

available_cpus property

Number of available CPUs

data property

Path to the package data

metadata property

Package metadata

pool property

A concurrent.futures.Executor instance

rng property

A random number generator instance, can be reused

__init__(*optional_packages)

Parameters:

Name Type Description Default
optional_packages str

Optional packages to check for, e.g. 'numpy', 'pandas'

 ()
Source code in eris/__init__.py 
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, *optional_packages: str):
    """
    Parameters:
        optional_packages: Optional packages to check for, e.g. 'numpy', 'pandas'
    """
    self.package: str = Path(__file__).parent.name
    self.optional_packages: set[str] = set(filter(self._check_module, optional_packages))
    self._data: 'Traversible' = None  # Generated on demand
    self._metadata: 'PackageMetadata' = None  # Generated on demand
    self._available_cpus: int = None  # Generated on demand
    self._rng: 'Random' = None  # Generated on demand
    self._pool: 'Executor' = None  # Generated on demand

require(*packages)

A decorator to check for required optional packages before executing a function.

Parameters:

Name Type Description Default
*packages str

Variable number of package names (strings) that are required.

 ()

Returns:

Type Description
Callable

A decorator that wraps the function. If any required packages are missing,
Callable

it issues a DependencyWarning and returns None. Otherwise, it executes
Callable

the original function.

Examples:

>>> from eris import require
... @require('numpy')
... def some_numpy_func():
... ...
Source code in eris/__init__.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def require(*packages: str) -> Callable:
    """
    A decorator to check for required optional packages before executing a function.

    Args:
        *packages: Variable number of package names (strings) that are required.

    Returns:
        A decorator that wraps the function. If any required packages are missing,
        it issues a DependencyWarning and returns None. Otherwise, it executes
        the original function.

    Examples:
        >>> from eris import require
        ... @require('numpy')
        ... def some_numpy_func():
        ... ...
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            if missing_deps := [dep for dep in packages if dep not in RESOURCES.optional_packages]:
                warn(
                    f"Function '{func.__name__}' requires the following missing dependencies: "
                    f"{', '.join(missing_deps)}. Skipping execution.",
                    DependencyWarning
                )
                return None
            return func(*args, **kwargs)
        return wrapper
    return decorator