Source code for colour.utilities.common

"""
Common Utilities
================

Defines the common utilities objects that don't fall in any specific category.

References
----------
-   :cite:`Kienzle2011a` : Kienzle, P., Patel, N., & Krycka, J. (2011).
    refl1d.numpyerrors - Refl1D v0.6.19 documentation. Retrieved January 30,
    2015, from
    http://www.reflectometry.org/danse/docs/refl1d/_modules/refl1d/\
numpyerrors.html
"""

from __future__ import annotations

import inspect
import functools
import numpy as np
import re
import subprocess  # nosec
import types
import warnings
from contextlib import contextmanager
from copy import copy
from pprint import pformat

from colour.constants import INTEGER_THRESHOLD
from colour.hints import (
    Any,
    Boolean,
    Callable,
    Dict,
    Generator,
    Integer,
    Iterable,
    Literal,
    Mapping,
    Optional,
    RegexFlag,
    Sequence,
    TypeVar,
    Union,
)
from colour.utilities import CaseInsensitiveMapping, Lookup

__author__ = "Colour Developers"
__copyright__ = "Copyright 2013 Colour Developers"
__license__ = "New BSD License - https://opensource.org/licenses/BSD-3-Clause"
__maintainer__ = "Colour Developers"
__email__ = "colour-developers@colour-science.org"
__status__ = "Production"

__all__ = [
    "CacheRegistry",
    "CACHE_REGISTRY",
    "handle_numpy_errors",
    "ignore_numpy_errors",
    "raise_numpy_errors",
    "print_numpy_errors",
    "warn_numpy_errors",
    "ignore_python_warnings",
    "attest",
    "batch",
    "disable_multiprocessing",
    "multiprocessing_pool",
    "is_ctlrender_installed",
    "is_matplotlib_installed",
    "is_networkx_installed",
    "is_opencolorio_installed",
    "is_openimageio_installed",
    "is_pandas_installed",
    "is_sklearn_installed",
    "is_tqdm_installed",
    "is_trimesh_installed",
    "required",
    "is_iterable",
    "is_string",
    "is_numeric",
    "is_integer",
    "is_sibling",
    "filter_kwargs",
    "filter_mapping",
    "first_item",
    "copy_definition",
    "validate_method",
    "optional",
]


[docs]class CacheRegistry: """ A registry for mapping-based caches. Attributes ---------- - :attr:`~colour.utilities.CacheRegistry.registry` Methods ------- - :meth:`~colour.SpectralShape.__init__` - :meth:`~colour.SpectralShape.__str__` - :meth:`~colour.SpectralShape.register_cache` - :meth:`~colour.SpectralShape.unregister_cache` - :meth:`~colour.SpectralShape.clear_cache` - :meth:`~colour.SpectralShape.clear_all_caches` Examples -------- >>> cache_registry = CacheRegistry() >>> cache_a = cache_registry.register_cache('Cache A') >>> cache_a['Foo'] = 'Bar' >>> cache_b = cache_registry.register_cache('Cache B') >>> cache_b['John'] = 'Doe' >>> cache_b['Luke'] = 'Skywalker' >>> print(cache_registry) {'Cache A': '1 item(s)', 'Cache B': '2 item(s)'} >>> cache_registry.clear_cache('Cache A') >>> print(cache_registry) {'Cache A': '0 item(s)', 'Cache B': '2 item(s)'} >>> cache_registry.unregister_cache('Cache B') >>> print(cache_registry) {'Cache A': '0 item(s)'} >>> print(cache_b) {} """
[docs] def __init__(self): self._registry = {}
@property def registry(self) -> Dict: """ Getter property for the cache registry. Returns ------- :class:`dict` Cache registry. """ return self._registry
[docs] def __str__(self) -> str: """ Return a formatted string representation of the cache registry. Returns ------- :class:`str` Formatted string representation. """ return pformat( { name: f"{len(self._registry[name])} item(s)" for name in sorted(self._registry) } )
[docs] def register_cache(self, name: str) -> Dict: """ Register a new cache with given name in the registry. Parameters ---------- name Cache name for the registry. Returns ------- :class:`dict` Registered cache. Examples -------- >>> cache_registry = CacheRegistry() >>> cache_a = cache_registry.register_cache('Cache A') >>> cache_a['Foo'] = 'Bar' >>> cache_b = cache_registry.register_cache('Cache B') >>> cache_b['John'] = 'Doe' >>> cache_b['Luke'] = 'Skywalker' >>> print(cache_registry) {'Cache A': '1 item(s)', 'Cache B': '2 item(s)'} """ self._registry[name] = {} return self._registry[name]
[docs] def unregister_cache(self, name: str): """ Unregister cache with given name in the registry. Parameters ---------- name Cache name in the registry. Notes ----- - The cache is cleared before being unregistered. Examples -------- >>> cache_registry = CacheRegistry() >>> cache_a = cache_registry.register_cache('Cache A') >>> cache_a['Foo'] = 'Bar' >>> cache_b = cache_registry.register_cache('Cache B') >>> cache_b['John'] = 'Doe' >>> cache_b['Luke'] = 'Skywalker' >>> print(cache_registry) {'Cache A': '1 item(s)', 'Cache B': '2 item(s)'} >>> cache_registry.unregister_cache('Cache B') >>> print(cache_registry) {'Cache A': '1 item(s)'} >>> print(cache_b) {} """ self.clear_cache(name) del self._registry[name]
[docs] def clear_cache(self, name: str): """ Clear the cache with given name. Parameters ---------- name Cache name in the registry. Examples -------- >>> cache_registry = CacheRegistry() >>> cache_a = cache_registry.register_cache('Cache A') >>> cache_a['Foo'] = 'Bar' >>> print(cache_registry) {'Cache A': '1 item(s)'} >>> cache_registry.clear_cache('Cache A') >>> print(cache_registry) {'Cache A': '0 item(s)'} """ self._registry[name].clear()
[docs] def clear_all_caches(self): """ Clear all the caches in the registry. Examples -------- >>> cache_registry = CacheRegistry() >>> cache_a = cache_registry.register_cache('Cache A') >>> cache_a['Foo'] = 'Bar' >>> cache_b = cache_registry.register_cache('Cache B') >>> cache_b['John'] = 'Doe' >>> cache_b['Luke'] = 'Skywalker' >>> print(cache_registry) {'Cache A': '1 item(s)', 'Cache B': '2 item(s)'} >>> cache_registry.clear_all_caches() >>> print(cache_registry) {'Cache A': '0 item(s)', 'Cache B': '0 item(s)'} """ for key in self._registry: self.clear_cache(key)
CACHE_REGISTRY: CacheRegistry = CacheRegistry() """ *Colour* cache registry referencing all the caches used for repetitive or long processes. CACHE_REGISTRY """
[docs]def handle_numpy_errors(**kwargs: Any) -> Callable: """ Decorate a function to handle *Numpy* errors. Other Parameters ---------------- kwargs Keywords arguments. Returns ------- Callable References ---------- :cite:`Kienzle2011a` Examples -------- >>> import numpy >>> @handle_numpy_errors(all='ignore') ... def f(): ... 1 / numpy.zeros(3) >>> f() """ context = np.errstate(**kwargs) def wrapper(function: Callable) -> Callable: """Wrap given function wrapper.""" @functools.wraps(function) def wrapped(*args: Any, **kwargs: Any) -> Any: """Wrap given function.""" with context: return function(*args, **kwargs) return wrapped return wrapper
ignore_numpy_errors = handle_numpy_errors(all="ignore") raise_numpy_errors = handle_numpy_errors(all="raise") print_numpy_errors = handle_numpy_errors(all="print") warn_numpy_errors = handle_numpy_errors(all="warn")
[docs]def ignore_python_warnings(function: Callable) -> Callable: """ Decorate a function to ignore *Python* warnings. Parameters ---------- function Function to decorate. Returns ------- Callable Examples -------- >>> @ignore_python_warnings ... def f(): ... warnings.warn('This is an ignored warning!') >>> f() """ @functools.wraps(function) def wrapper(*args: Any, **kwargs: Any) -> Any: """Wrap given function.""" with warnings.catch_warnings(): warnings.simplefilter("ignore") return function(*args, **kwargs) return wrapper
[docs]def attest(condition: Boolean, message: str = ""): """ Provide the `assert` statement functionality without being disabled by optimised Python execution. Parameters ---------- condition Condition to attest/assert. message Message to display when the assertion fails. """ if not condition: raise AssertionError(message)
[docs]def batch(sequence: Sequence, k: Union[Integer, Literal[3]] = 3) -> Generator: """ Return a batch generator from given sequence. Parameters ---------- sequence Sequence to create batches from. k Batch size. Yields ------ Generator Batch generator. Examples -------- >>> batch(tuple(range(10)), 3) # doctest: +ELLIPSIS <generator object batch at 0x...> """ for i in range(0, len(sequence), k): yield sequence[i : i + k]
_MULTIPROCESSING_ENABLED: Boolean = True """*Colour* multiprocessing state."""
[docs]class disable_multiprocessing: """ Define a context manager and decorator to temporarily disabling *Colour* multiprocessing state. """ def __enter__(self) -> disable_multiprocessing: """ Disable *Colour* multiprocessing state upon entering the context manager. """ global _MULTIPROCESSING_ENABLED _MULTIPROCESSING_ENABLED = False return self def __exit__(self, *args: Any): """ Enable *Colour* multiprocessing state upon exiting the context manager. """ global _MULTIPROCESSING_ENABLED _MULTIPROCESSING_ENABLED = True def __call__(self, function: Callable) -> Callable: """Call the wrapped definition.""" @functools.wraps(function) def wrapper(*args: Any, **kwargs: Any) -> Any: """Wrap given function.""" with self: return function(*args, **kwargs) return wrapper
def _initializer(kwargs: Any): """ Initialize a multiprocessing pool. It is used to ensure that processes on *Windows* inherit correctly from the current domain-range scale. Parameters ---------- kwargs Initialisation arguments. """ # NOTE: No coverage information is available as this code is executed in # sub-processes. import colour.utilities.array # pragma: no cover colour.utilities.array._DOMAIN_RANGE_SCALE = kwargs.get( "scale", "reference" ) # pragma: no cover import colour.algebra.common # pragma: no cover colour.algebra.common._SDIV_MODE = kwargs.get( "sdiv_mode", "Ignore Zero Conversion" ) # pragma: no cover colour.algebra.common._SPOW_ENABLED = kwargs.get( "spow_enabled", True ) # pragma: no cover
[docs]@contextmanager def multiprocessing_pool(*args: Any, **kwargs: Any) -> Generator: """ Define a context manager providing a multiprocessing pool. Other Parameters ---------------- args Arguments. kwargs Keywords arguments. Yields ------ Generator Multiprocessing pool. Examples -------- >>> from functools import partial >>> def _add(a, b): ... return a + b >>> with multiprocessing_pool() as pool: ... pool.map(partial(_add, b=2), range(10)) ... # doctest: +SKIP [2, 3, 4, 5, 6, 7, 8, 9, 10, 11] """ from colour.algebra import get_sdiv_mode, is_spow_enabled from colour.utilities import get_domain_range_scale class _DummyPool: """ A dummy multiprocessing pool that does not perform multiprocessing. Other Parameters ---------------- args Arguments. kwargs Keywords arguments. """ def __init__(self, *args: Any, **kwargs: Any): pass def map(self, func, iterable, chunksize=None): """Apply given function to each element of given iterable.""" return [func(a) for a in iterable] def terminate(self): """Terminate the process.""" pass kwargs["initializer"] = _initializer kwargs["initargs"] = ( { "scale": get_domain_range_scale(), "sdiv_mode": get_sdiv_mode(), "spow_enabled": is_spow_enabled(), }, ) pool_factory: Callable if _MULTIPROCESSING_ENABLED: import multiprocessing pool_factory = multiprocessing.Pool else: pool_factory = _DummyPool pool = pool_factory(*args, **kwargs) try: yield pool finally: pool.terminate()
[docs]def is_ctlrender_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *ctlrender* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *ctlrender* is unavailable. Returns ------- :class:`bool` Whether *ctlrender* is installed. Raises ------ :class:`ImportError` If *ctlrender* is not installed. """ try: # pragma: no cover stdout = subprocess.run( ["ctlrender", "-help"], capture_output=True ).stdout.decode( "utf-8" ) # nosec if "transforms an image using one or more CTL scripts" not in stdout: raise FileNotFoundError() return True except FileNotFoundError as error: # pragma: no cover if raise_exception: raise FileNotFoundError( '"ctlrender" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_matplotlib_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *Matplotlib* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *Matplotlib* is unavailable. Returns ------- :class:`bool` Whether *Matplotlib* is installed. Raises ------ :class:`ImportError` If *Matplotlib* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import matplotlib # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"Matplotlib" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_networkx_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *NetworkX* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *NetworkX* is unavailable. Returns ------- :class:`bool` Whether *NetworkX* is installed. Raises ------ :class:`ImportError` If *NetworkX* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import networkx # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"NetworkX" related API features, e.g. the automatic colour ' f'conversion graph, are not available: "{error}".\nPlease refer ' "to the installation guide for more information: " "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_opencolorio_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *OpenColorIO* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *OpenColorIO* is unavailable. Returns ------- :class:`bool` Whether *OpenColorIO* is installed. Raises ------ :class:`ImportError` If *OpenColorIO* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import PyOpenColorIO # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"OpenColorIO" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_openimageio_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *OpenImageIO* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *OpenImageIO* is unavailable. Returns ------- :class:`bool` Whether *OpenImageIO* is installed. Raises ------ :class:`ImportError` If *OpenImageIO* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import OpenImageIO # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"OpenImageIO" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_pandas_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *Pandas* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *Pandas* is unavailable. Returns ------- :class:`bool` Whether *Pandas* is installed. Raises ------ :class:`ImportError` If *Pandas* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import pandas # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( f'"Pandas" related API features are not available: "{error}".\n' "See the installation guide for more information: " "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_sklearn_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *Scikit-Learn* (sklearn) is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *Scikit-Learn* (sklearn) is unavailable. Returns ------- :class:`bool` Whether *Scikit- isLearn* (sklearn) installed. Raises ------ :class:`ImportError` If *Scikit-Learn* (sklearn) is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import sklearn # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"Scikit-Learn" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_tqdm_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *tqdm* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *tqdm* is unavailable. Returns ------- :class:`bool` Whether *tqdm* is installed. Raises ------ :class:`ImportError` If *tqdm* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import tqdm # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( f'"tqdm" related API features are not available: "{error}".\n' "See the installation guide for more information: " "https://www.colour-science.org/installation-guide/" ) return False
[docs]def is_trimesh_installed(raise_exception: Boolean = False) -> Boolean: """ Return whether *Trimesh* is installed and available. Parameters ---------- raise_exception Whether to raise an exception if *Trimesh* is unavailable. Returns ------- :class:`bool` Whether *Trimesh* is installed. Raises ------ :class:`ImportError` If *Trimesh* is not installed. """ try: # pragma: no cover # pylint: disable=W0612 import trimesh # noqa return True except ImportError as error: # pragma: no cover if raise_exception: raise ImportError( '"Trimesh" related API features are not available: ' f'"{error}".\nSee the installation guide for more information: ' "https://www.colour-science.org/installation-guide/" ) return False
_REQUIREMENTS_TO_CALLABLE: CaseInsensitiveMapping = CaseInsensitiveMapping( { "ctlrender": is_ctlrender_installed, "Matplotlib": is_matplotlib_installed, "NetworkX": is_networkx_installed, "OpenColorIO": is_opencolorio_installed, "OpenImageIO": is_openimageio_installed, "Pandas": is_pandas_installed, "Scikit-Learn": is_sklearn_installed, "tqdm": is_tqdm_installed, "trimesh": is_trimesh_installed, } ) """ Mapping of requirements to their respective callables. """
[docs]def required( *requirements: Literal[ "ctlrender", "Matplotlib", "NetworkX", "OpenColorIO", "OpenImageIO", "Pandas", "Scikit-Learn", "tqdm", "trimesh", ] ) -> Callable: """ Decorate a function to check whether various ancillary package requirements are satisfied. Other Parameters ---------------- requirements Requirements to check whether they are satisfied. Returns ------- Callable """ def wrapper(function: Callable) -> Callable: """Wrap given function wrapper.""" @functools.wraps(function) def wrapped(*args: Any, **kwargs: Any) -> Any: """Wrap given function.""" for requirement in requirements: _REQUIREMENTS_TO_CALLABLE[requirement](raise_exception=True) return function(*args, **kwargs) return wrapped return wrapper
[docs]def is_iterable(a: Any) -> Boolean: """ Return whether given variable :math:`a` is iterable. Parameters ---------- a Variable :math:`a` to check the iterability. Returns ------- :class:`bool` Whether variable :math:`a` is iterable. Examples -------- >>> is_iterable([1, 2, 3]) True >>> is_iterable(1) False """ return is_string(a) or (True if getattr(a, "__iter__", False) else False)
[docs]def is_string(a: Any) -> Boolean: """ Return whether given variable :math:`a` is a :class:`str`-like variable. Parameters ---------- a Variable :math:`a` to test. Returns ------- :class:`bool` Whether variable :math:`a` is a :class:`str`-like variable. Examples -------- >>> is_string("I'm a string!") True >>> is_string(["I'm a string!"]) False """ return True if isinstance(a, str) else False
[docs]def is_numeric(a: Any) -> Boolean: """ Return whether given variable :math:`a` is a :class:`Number`-like variable. Parameters ---------- a Variable :math:`a` to test. Returns ------- :class:`bool` Whether variable :math:`a` is a :class:`Number`-like variable. Examples -------- >>> is_numeric(1) True >>> is_numeric((1,)) False """ return isinstance( a, ( int, float, complex, np.integer, np.floating, ), )
[docs]def is_integer(a: Any) -> Boolean: """ Return whether given variable :math:`a` is an :class:`numpy.integer`-like variable under given threshold. Parameters ---------- a Variable :math:`a` to test. Returns ------- :class:`bool` Whether variable :math:`a` is an :class:`numpy.integer`-like variable. Notes ----- - The determination threshold is defined by the :attr:`colour.algebra.common.INTEGER_THRESHOLD` attribute. Examples -------- >>> is_integer(1) True >>> is_integer(1.01) False """ return abs(a - np.around(a)) <= INTEGER_THRESHOLD
[docs]def is_sibling(element: Any, mapping: Mapping) -> Boolean: """ Return whether given element type is present in given mapping types. Parameters ---------- element Element to check whether its type is present in the mapping types. mapping Mapping types. Returns ------- :class:`bool` Whether given element type is present in given mapping types. """ return isinstance( element, tuple({type(element) for element in mapping.values()}) )
[docs]def filter_kwargs(function: Callable, **kwargs: Any) -> Dict: """ Filter keyword arguments incompatible with the given function signature. Parameters ---------- function Callable to filter the incompatible keyword arguments. Other Parameters ---------------- kwargs Keywords arguments. Returns ------- dict Filtered keyword arguments. Examples -------- >>> def fn_a(a): ... return a >>> def fn_b(a, b=0): ... return a, b >>> def fn_c(a, b=0, c=0): ... return a, b, c >>> fn_a(1, **filter_kwargs(fn_a, b=2, c=3)) 1 >>> fn_b(1, **filter_kwargs(fn_b, b=2, c=3)) (1, 2) >>> fn_c(1, **filter_kwargs(fn_c, b=2, c=3)) (1, 2, 3) """ kwargs = copy(kwargs) try: args = list(inspect.signature(function).parameters.keys()) except ValueError: # pragma: no cover return {} for key in set(kwargs.keys()) - set(args): kwargs.pop(key) return kwargs
[docs]def filter_mapping( mapping: Mapping, filterers: Union[str, Sequence[str]], anchors: Boolean = True, flags: Union[Integer, RegexFlag] = re.IGNORECASE, ) -> Dict: """ Filter given mapping with given filterers. Parameters ---------- mapping Mapping to filter. filterers Filterer pattern for given mapping elements or a list of filterers. anchors Whether to use Regex line anchors, i.e. *^* and *$* are added, surrounding the filterer pattern. flags Regex flags. Returns ------- dict Filtered mapping elements. Notes ----- - To honour the filterers ordering, the return value is an :class:`dict` class instance. Examples -------- >>> class Element: ... pass >>> mapping = { ... 'Element A': Element(), ... 'Element B': Element(), ... 'Element C': Element(), ... 'Not Element C': Element(), ... } >>> filter_mapping(mapping, '\\w+\\s+A') # doctest: +ELLIPSIS {'Element A': <colour.utilities.common.Element object at 0x...>} >>> sorted(filter_mapping(mapping, 'Element.*')) ['Element A', 'Element B', 'Element C'] """ def filter_mapping_with_filter( mapping: Mapping, filterer: str, anchors: Boolean = True, flags: Union[Integer, RegexFlag] = re.IGNORECASE, ) -> Dict: """ Filter given mapping with given filterer. Parameters ---------- mapping Mapping to filter. filterer Filterer pattern for given mapping elements. anchors Whether to use Regex line anchors, i.e. *^* and *$* are added, surrounding the filterer pattern. flags Regex flags. Returns ------- dict Filtered mapping elements. """ if anchors: filterer = f"^{filterer}$" filterer = filterer.replace("^^", "^").replace("$$", "$") elements = [ mapping[element] for element in mapping if re.match(filterer, element, flags) ] lookup = Lookup(mapping) return { lookup.first_key_from_value(element): element for element in elements } filterers = [str(filterers)] if is_string(filterers) else filterers filtered_mapping = {} for filterer in filterers: filtered_mapping.update( filter_mapping_with_filter(mapping, filterer, anchors, flags) ) return filtered_mapping
[docs]def first_item(a: Iterable) -> Any: """ Return the first item of given iterable. Parameters ---------- a Iterable to get the first item from. Returns ------- :class:`object` Raises ------ :class:`StopIteration` If the iterable is empty. Examples -------- >>> a = range(10) >>> first_item(a) 0 """ return next(iter(a))
[docs]def copy_definition( definition: Callable, name: Optional[str] = None ) -> Callable: """ Copy a definition using the same code, globals, defaults, closure, and name. Parameters ---------- definition Definition to be copied. name Optional definition copy name. Returns ------- Callable Definition copy. """ copy = types.FunctionType( definition.__code__, definition.__globals__, # type: ignore[attr-defined] str(name or definition.__name__), definition.__defaults__, # type: ignore[attr-defined] definition.__closure__, # type: ignore[attr-defined] ) copy.__dict__.update(definition.__dict__) return copy
[docs]def validate_method( method: str, valid_methods: Union[Sequence, Mapping], message: str = '"{0}" method is invalid, it must be one of {1}!', ) -> str: """ Validate whether given method exists in the given valid methods and returns the method lower cased. Parameters ---------- method Method to validate. valid_methods Valid methods. message Message for the exception. Returns ------- :class:`str` Method lower cased. Raises ------ :class:`ValueError` If the method does not exist. Examples -------- >>> validate_method('Valid', ['Valid', 'Yes', 'Ok']) 'valid' """ valid_methods = [str(valid_method) for valid_method in valid_methods] method_lower = method.lower() if method_lower not in [ valid_method.lower() for valid_method in valid_methods ]: raise ValueError(message.format(method, valid_methods)) return method_lower
T = TypeVar("T")
[docs]def optional(value: Optional[T], default: T) -> T: """ Handle optional argument value by providing a default value. Parameters ---------- value Optional argument value. default Default argument value if ``value`` is *None*. Returns ------- T Argument value. Examples -------- >>> optional('Foo', 'Bar') 'Foo' >>> optional(None, 'Bar') 'Bar' """ if value is None: return default else: return value