Source code for catalyst.utils.misc

from typing import Any, Iterable  # isort:skip

import collections
import copy
from datetime import datetime
from itertools import tee
from pathlib import Path
import shutil

import numpy as np


[docs]def pairwise(iterable: Iterable[Any]) -> Iterable[Any]: """ Iterate sequences by pairs Args: iterable: Any iterable sequence Returns: pairwise iterator Examples: >>> for i in pairwise([1, 2, 5, -3]): >>> print(i) (1, 2) (2, 5) (5, -3) """ a, b = tee(iterable) next(b, None) return zip(a, b)
[docs]def make_tuple(tuple_like): """ Creates a tuple if given ``tuple_like`` value isn't list or tuple Returns: tuple or list """ tuple_like = ( tuple_like if isinstance(tuple_like, (list, tuple)) else (tuple_like, tuple_like) ) return tuple_like
[docs]def merge_dicts(*dicts: dict) -> dict: """ Recursive dict merge. Instead of updating only top-level keys, ``merge_dicts`` recurses down into dicts nested to an arbitrary depth, updating keys. Args: *dicts: several dictionaries to merge Returns: dict: deep-merged dictionary """ assert len(dicts) > 1 dict_ = copy.deepcopy(dicts[0]) for merge_dict in dicts[1:]: merge_dict = merge_dict or {} for k, v in merge_dict.items(): if ( k in dict_ and isinstance(dict_[k], dict) and isinstance(merge_dict[k], collections.Mapping) ): dict_[k] = merge_dicts(dict_[k], merge_dict[k]) else: dict_[k] = merge_dict[k] return dict_
[docs]def append_dict(dict1, dict2): """ Appends dict2 with the same keys as dict1 to dict1 """ for key in dict1.keys(): dict1[key] = np.concatenate((dict1[key], dict2[key])) return dict1
[docs]def flatten_dict(d: dict, parent_key: str = "", sep: str = "/") -> dict: """ Flatten nested dicts Args: d (dict): the dict that will be flattened parent_key (str): prefix nested keys with string `parent_key` sep (str): delimiter between `parent_key` and `key` to use """ items = [] for k, v in d.items(): new_key = parent_key + sep + k if parent_key else k if isinstance(v, collections.MutableMapping): items.extend(flatten_dict(v, new_key, sep=sep).items()) else: items.append((new_key, v)) return collections.OrderedDict(items)
[docs]def maybe_recursive_call( object_or_dict, method: str, recursive_args=None, recursive_kwargs=None, **kwargs, ): """ Calls the ``method`` recursively for the object_or_dict Args: object_or_dict (Any): some object or a dictinary of objects method (str): method name to call recursive_args: list of arguments to pass to the ``method`` recursive_kwargs: list of key-arguments to pass to the ``method`` **kwargs: Arbitrary keyword arguments """ if isinstance(object_or_dict, dict): result = type(object_or_dict)() for k, v in object_or_dict.items(): r_args = \ None if recursive_args is None else recursive_args[k] r_kwargs = \ None if recursive_kwargs is None else recursive_kwargs[k] result[k] = maybe_recursive_call( v, method, recursive_args=r_args, recursive_kwargs=r_kwargs, **kwargs, ) return result r_args = recursive_args or [] if not isinstance(r_args, (list, tuple)): r_args = [r_args] r_kwargs = recursive_kwargs or {} return getattr(object_or_dict, method)(*r_args, **r_kwargs, **kwargs)
[docs]def is_exception(ex: Any) -> bool: """ Check if the argument is of Exception type """ result = (ex is not None) and isinstance(ex, BaseException) return result
[docs]def copy_directory(input_dir: Path, output_dir: Path) -> None: """ Recursively copies the input directory Args: input_dir (Path): input directory output_dir (Path): output directory """ output_dir.mkdir(exist_ok=True, parents=True) for path in input_dir.iterdir(): if path.is_dir(): path_name = path.name copy_directory(path, output_dir / path_name) else: shutil.copy2(path, output_dir)
[docs]def get_utcnow_time(format: str = None) -> str: """ Return string with current utc time in chosen format Args: format (str): format string. if None "%y%m%d.%H%M%S" will be used. Returns: str: formatted utc time string """ if format is None: format = "%y%m%d.%H%M%S" result = datetime.utcnow().strftime(format) return result
[docs]def format_metric(name: str, value: float) -> str: """ Format metric. Metric will be returned in the scientific format if 4 decimal chars are not enough (metric value lower than 1e-4) Args: name (str): metric name value (float): value of metric """ if value < 1e-4: return f"{name}={value:1.3e}" return f"{name}={value:.4f}"