wmaee.core.utils

  1import os
  2import uuid
  3import shutil
  4import contextlib
  5import collections.abc
  6from typing import Dict, Type, Any, Optional
  7
  8
  9# this beautiful solution was taken from:
 10# https://stackoverflow.com/questions/2059482/temporarily-modify-the-current-processs-environment
 11@contextlib.contextmanager
 12def override_environ(*remove: str, **update):
 13    """
 14    Temporarily updates the ``os.environ`` dictionary in-place.
 15
 16    The ``os.environ`` dictionary is updated in-place so that the modification
 17    is sure to work in all situations.
 18
 19    :param remove: Environment variables to remove.
 20    :type remove: str
 21    :param update: Dictionary of environment variables and values to add/update.
 22    """
 23    env = os.environ
 24    update = update or {}
 25    remove = remove or []
 26
 27    # List of environment variables being updated or removed.
 28    stomped = (set(update.keys()) | set(remove)) & set(env.keys())
 29    # Environment variables and values to restore on exit.
 30    update_after = {k: env[k] for k in stomped}
 31    # Environment variables and values to remove on exit.
 32    remove_after = frozenset(k for k in update if k not in env)
 33
 34    try:
 35        env.update(update)
 36        [env.pop(k, None) for k in remove]
 37        yield
 38    finally:
 39        env.update(update_after)
 40        [env.pop(k) for k in remove_after]
 41
 42
 43# class working_directory(object):
 44#     """
 45#     A convenience class which syntactic sugar, allowing the user to change the directories.
 46#     Can also be nested.
 47#     """
 48
 49#     def __init__(self, name: Optional[str] = None, prefix: Optional[str] = None, delete: bool =False):
 50#         """
 51#         Constructs a working_directory object
 52#         :param name: name of the directory if None is given `os.getcwd()` will be used (default is `None`)
 53#         :type name: Optional[str]
 54#         :param prefix: a prefix where to locate the directory (default is `None`)
 55#         :type prefix: Optional[str]
 56#         :param delete: whether to delete the directory after a with clause (default is `False`)
 57#         :type delete: bool
 58#         """
 59#         self._name = str(uuid.uuid4()) if not name else name
 60#         self._delete = delete
 61#         self._curr_dir = os.getcwd()
 62#         self._active = False
 63#         if prefix is not None:
 64#             self._name = os.path.join(prefix, self._name)
 65
 66#     def __enter__(self):
 67#         if not os.path.exists(self._name):
 68#             os.mkdir(self._name)
 69#         os.chdir(self._name)
 70#         self._active = True
 71#         return self
 72
 73#     def __exit__(self, exc_type, exc_val, exc_tb):
 74#         os.chdir(self._curr_dir)
 75#         if self._delete:
 76#             shutil.rmtree(self._name)
 77#         self._active = False
 78
 79#     @property
 80#     def name(self):
 81#         return self._name
 82
 83#     @property
 84#     def active(self):
 85#         return self._active
 86
 87
 88def merge(*dicts: Dict[Any, Any], factory: Type = dict, **kwargs) -> Dict[Any, Any]:
 89    """
 90    Merge all specified dictionaries in {dicts} into a single one. Moreover, {kwargs} will be included. In case
 91    duplicate keys exist, the order of passing the dictionaries will determine which values will sustain for the keys.
 92    The last updated is carried out using {kwargs}
 93
 94    :param dicts: the dictionaries to merge
 95    :type dicts: Dict[Any, Any]
 96    :param factory: the constructor of the mapping type to create (default is `dict`)
 97    :type factory: Type
 98    :return: a merged dictionary of type {factory}
 99    :rtype: Dict[Any, Any]
100    """
101
102    merged = factory()
103    for dictionary in dicts:
104        merged.update(dictionary)
105    merged.update(kwargs)
106    return merged
107
108
109def _wrap_in_iterable(o, factory=tuple):
110    return factory((o,))
111
112
113def ensure_iterable(o, exclude=(str, bytes), factory=tuple):
114    """
115    wraps an object {o} into an iterable it is not and iterable. the type of the iterable is specified by {factory}
116    :param o: the object to wrap
117    :param exclude: type list of iterable objects which need wrapping (default is (str, bytes))
118    :type exclude: Iterable[type]
119    :param factory: the Iterable type in which {o} should be wrapped
120    :type factory: type
121    :rtype: Iterable
122    """
123    if isinstance(o, collections.abc.Iterable):
124        return o if not isinstance(o, exclude) else _wrap_in_iterable(o, factory=factory)
125    else:
126        return _wrap_in_iterable(o, factory=factory)