Skip to content

Utils

utils

cache

cache = Cache() module-attribute

Cache

path: Path property
__init__(path: Union[str, Path, None] = None) -> None
cleanup()

collections

EventTarget

__init__(*args, **kwargs)
add_event_listener(type: str, listener: callable)
remove_event_listener(type: str, listener: callable)
dispatch_event(event: str)

MutableCollection

Bases: EventTarget

__init__(*args, **kwargs)
set_parent(parent: MutableCollection) -> MutableCollection
changed()
coerce(value: any)

MutableDict

Bases: MutableCollection, MutableMapping[K, V]

__init__(*args, **kwargs)
__getitem__(key: K) -> V
__setitem__(key: K, value: V)
__delitem__(key: K)
__iter__() -> Generator[K, None, None]
__len__() -> int
__repr__() -> str

MutableList

Bases: MutableCollection, MutableSequence[V]

__init__(*args, **kwargs)
__getitem__(key: int) -> V
__setitem__(key: int, value: V)
insert(index: int, value: V)
__delitem__(key: int)
__len__() -> int
__repr__() -> str
sort(*args, **kwargs) -> None

MutableSet

Bases: MutableCollection, MutableSet[V]

__init__(*args, **kwargs)
add(item: V)
update(*s: Iterable[V]) -> None
discard(item: V)
__contains__(item: V) -> bool
__iter__() -> Generator[V, None, None]
__len__() -> int
__repr__() -> str

flatten(data: Dict[str, Any], *, separator: str = '.', parent_key: Optional[str] = None) -> Dict[Union[str, Tuple[str]], Any]

Flatten a nested dictionary.

Parameters:

Name Type Description Default
data Dict[str, Any]

The nested dictionary to flatten.

 required
separator str

The separator, by default "."

 '.'
parent_key Optional[str]

The parent key, by default None

 None

Returns:

Type Description
Dict[str | tuple[str], Any]

The flattened dictionary.

config

T = TypeVar('T', bound=dataclass) module-attribute

ConfigWrapper

wrapped = wrapped instance-attribute
signature = inspect.signature(wrapped) instance-attribute
filter_keys = filter_keys instance-attribute
name = name instance-attribute
config = config instance-attribute
__init__(wrapped, config: Config, name: Optional[str] = None)
get_params(*args, **kwargs)
__call__(*args, **kwargs)

Config

Bases: MutableMapping

omconf = config if isinstance(config, OmegaConf) else OmegaConf.structured(config) if is_dataclass(config) else OmegaConf.create(config) instance-attribute
__new__(config: Union[Type[T], dict[str, Any]]) -> Union[T, Self] 
__new__(config: Type[T]) -> T

__new__(config: dict[str, Any]) -> Self
__init__(config: Any) -> None
__getitem__(name: str) -> Any
__setitem__(name: str, value: Any) -> None
__delitem__(name: str) -> None
__iter__() -> Any
__len__() -> int
__getattr__(name: str) -> Any
wraps(wrapped: Union[type, callable, str, None] = None, **kwargs: Any) -> Union[ConfigWrapper, functools.partial] 
wraps(wrapped: Union[type, callable], **kwargs) -> ConfigWrapper

wraps(wrapped: Union[str, None], **kwargs) -> functools.partial
load(path: str) -> Self classmethod
from_dotlist(dotlist: str) -> Self classmethod
from_cli(args: list[str]) -> Self classmethod

configmethod(func: T) -> T

escape

ESCAPED_CHARS = '\\\\`*_{}[]()#+\\-.!' module-attribute

ESCAPED_CHARS_RE = re.compile(f'([{re.escape(ESCAPED_CHARS)}])') module-attribute

UNESCAPED_CHARS_RE = re.compile(f'\\([{re.escape(ESCAPED_CHARS)}])') module-attribute

escape(text: str) -> str

Escape text.

unescape(text: str) -> str

Unescape text.

func

P = ParamSpec('P') module-attribute

T = TypeVar('T') module-attribute

bind(__func: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> Union[functools.partial, Callable[..., T]]

Bind arguments to a function and return a new function.

This function is similar to functools.partial but it allows to bind arguments by name and by position (converting positional arguments to keyword arguments when possible).

Parameters:

Name Type Description Default
__func Callable

The function to bind arguments to.

 required
*args args

Positional arguments to bind to the function.

 ()
**kwargs kwargs

Keyword arguments to bind to the function.

 {}

Returns:

Type Description
partial or Callable

A new function with the arguments bound.

hashing

Fast cryptographic hash of Python objects, with a special case for fast hashing of numpy arrays.

Pickler = pickle._Pickler module-attribute

T = TypeVar('T') module-attribute

P = ParamSpec('P') module-attribute

Hashable

Bases: Protocol

A class that can be hashed.

Hasher

Bases: Pickler

A subclass of pickler, to do cryptographic hashing, rather than pickling.

stream = io.BytesIO() instance-attribute
dispatch = Pickler.dispatch.copy() class-attribute instance-attribute
__init__()
hash(obj, return_digest=True)
save(obj)
memoize(obj)
save_global(obj, name=None, pack=struct.pack)
save_set(set_items)

NumpyHasher

Bases: Hasher

Special case the hasher for when numpy is loaded.

coerce_mmap = coerce_mmap instance-attribute
np = np instance-attribute
__init__(coerce_mmap=False)
save(obj)

Subclass the save method, to hash ndarray subclass, rather than pickling them. Off course, this is a total abuse of the Pickler class.

Wrapped

Bases: Generic[T]

cls = cls instance-attribute
__init__(cls: Callable[P, T])
__call__(*args: P.args, **kwargs: P.kwargs) -> T
__getattr__(name: str) -> Any

hash(*obj, coerce_mmap=False)

Quick calculation of a hash to identify uniquely Python objects containing numpy arrays.

Parameters:

Name Type Description Default
obj

The objects to hash

 ()
coerce_mmap

Make no difference between np.memmap and np.ndarray

 False

init_based_hash(cls: Type[T]) -> Wrapped[T]

objects

import_class(cls: str)

create_object(__cls: str, __partial: bool = False, *args, **kwargs)

rsync

RSyncError

Bases: Exception

rsync(src: Path, dest: Path, exclude: Optional[list[str]] = None, ignore_errors: Optional[bool] = False, append_dir: Optional[bool] = False) -> Generator[str, None, None]